Compare commits

..

1 Commits

Author SHA1 Message Date
bf4e23d503 added the basic router config 2025-01-08 21:20:38 -05:00
290 changed files with 1691 additions and 24220 deletions

View File

@@ -1,11 +1,9 @@
name: build_systems
name: build_jeeves
on:
workflow_dispatch:
pull_request:
push:
branches: [main]
schedule:
- cron: "0 22 * * *"
jobs:
build:
@@ -15,14 +13,11 @@ jobs:
matrix:
system:
- "bob"
- "brain"
- "jeeves"
- "leviathan"
- "muninn"
- "rhapsody-in-green"
continue-on-error: true
- "router"
steps:
- uses: actions/checkout@v4
- name: Build default package
run: "nixos-rebuild build --flake ./#${{ matrix.system }}"
- name: copy to nix-cache
run: nix copy --accept-flake-config --to unix:///host-nix/var/nix/daemon-socket/socket .#nixosConfigurations.${{ matrix.system }}.config.system.build.toplevel

View File

@@ -1,30 +0,0 @@
name: fix_eval_warnings
on:
workflow_run:
workflows: ["build_systems"]
types: [completed]
jobs:
check-warnings:
if: >-
github.event.workflow_run.conclusion != 'cancelled' &&
github.event.workflow_run.head_branch == 'main' &&
(github.event.workflow_run.event == 'push' || github.event.workflow_run.event == 'schedule')
runs-on: self-hosted
permissions:
contents: write
pull-requests: write
steps:
- uses: actions/checkout@v4
- name: Fix eval warnings
env:
GH_TOKEN: ${{ secrets.GH_TOKEN_FOR_UPDATES }}
run: >-
nix develop .#devShells.x86_64-linux.default -c
python -m python.eval_warnings.main
--run-id "${{ github.event.workflow_run.id }}"
--repo "${{ github.repository }}"
--ollama-url "${{ secrets.OLLAMA_URL }}"
--run-url "${{ github.event.workflow_run.html_url }}"

View File

@@ -1,29 +0,0 @@
name: merge_flake_lock_update
on:
workflow_dispatch:
schedule:
- cron: "0 2 * * 6"
jobs:
merge:
runs-on: ubuntu-latest
permissions:
contents: write
pull-requests: write
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: merge_flake_lock_update
run: |
pr_number=$(gh pr list --state open --author RichieCahill --label flake_lock_update --json number --jq '.[0].number')
echo "pr_number=$pr_number" >> $GITHUB_ENV
if [ -n "$pr_number" ]; then
gh pr merge "$pr_number" --rebase
else
echo "No open PR found with label flake_lock_update"
fi
env:
GITHUB_TOKEN: ${{ secrets.GH_TOKEN_FOR_UPDATES }}

View File

@@ -1,19 +0,0 @@
name: pytest
on:
push:
branches:
- main
pull_request:
branches:
- main
merge_group:
jobs:
pytest:
runs-on: self-hosted
steps:
- uses: actions/checkout@v4
- name: Run tests
run: nix develop .#devShells.x86_64-linux.default -c pytest tests

View File

@@ -1,15 +0,0 @@
name: treefmt
on:
workflow_dispatch:
pull_request:
push:
branches: [main]
jobs:
treefmt:
name: nix fmt
runs-on: self-hosted
steps:
- uses: actions/checkout@v4
- name: runs treefmt
run: "treefmt --ci"

View File

@@ -2,7 +2,7 @@ name: update-flake-lock
on:
workflow_dispatch:
schedule:
- cron: "0 0 * * 6"
- cron: "0 0 * * *"
jobs:
lockfile:
@@ -15,9 +15,7 @@ jobs:
- name: Update flake.lock
uses: DeterminateSystems/update-flake-lock@main
with:
token: ${{ secrets.GH_TOKEN_FOR_UPDATES }}
pr-title: "Update flake.lock"
pr-labels: |
dependencies
automated
flake_lock_update

7
.gitignore vendored
View File

@@ -162,10 +162,3 @@ cython_debug/
#.idea/
test.*
# syncthing
.stfolder
# Frontend build output
frontend/dist/
frontend/node_modules/

View File

@@ -1,14 +1,10 @@
# Generate AGE keys from SSH keys with:
# ssh-keygen -A
# nix-shell -p ssh-to-age --run 'cat /etc/ssh/ssh_host_ed25519_key.pub | ssh-to-age'
keys:
- &admin_richie age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c # cspell:disable-line
- &admin_richie age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
- &system_bob age1q47vup0tjhulkg7d6xwmdsgrw64h4ax3la3evzqpxyy4adsmk9fs56qz3y # cspell:disable-line
- &system_brain age1jhf7vm0005j60mjq63696frrmjhpy8kpc2d66mw044lqap5mjv4snmwvwm # cspell:disable-line
- &system_jeeves age13lmqgc3jvkyah5e3vcwmj4s5wsc2akctcga0lpc0x8v8du3fxprqp4ldkv # cspell:disable-line
- &system_leviathan age1l272y8udvg60z7edgje42fu49uwt4x2gxn5zvywssnv9h2krms8s094m4k # cspell:disable-line
- &system_rhapsody age1ufnewppysaq2wwcl4ugngjz8pfzc5a35yg7luq0qmuqvctajcycs5lf6k4 # cspell:disable-line
- &system_bob age1q47vup0tjhulkg7d6xwmdsgrw64h4ax3la3evzqpxyy4adsmk9fs56qz3y
- &system_jeeves age13lmqgc3jvkyah5e3vcwmj4s5wsc2akctcga0lpc0x8v8du3fxprqp4ldkv
- &system_router age1xzxryqq63x65yuza9lmmkud7crjjxpnkdew070yhx6xn7xe4tdws5twxsv
- &system_rhapsody age1ufnewppysaq2wwcl4ugngjz8pfzc5a35yg7luq0qmuqvctajcycs5lf6k4
creation_rules:
- path_regex: users/secrets\.yaml$
@@ -16,7 +12,6 @@ creation_rules:
- age:
- *admin_richie
- *system_bob
- *system_brain
- *system_jeeves
- *system_leviathan
- *system_router
- *system_rhapsody

93
.vscode/settings.json vendored
View File

@@ -2,15 +2,11 @@
"cSpell.words": [
"aboutwelcome",
"acltype",
"addopts",
"addstr",
"advplyr",
"ahci",
"aioesphomeapi",
"aiounifi",
"alsa",
"apiclient",
"apscheduler",
"archlinux",
"ashift",
"asrouter",
@@ -21,14 +17,11 @@
"auditd",
"autofetch",
"autologin",
"automations",
"autopull",
"autotrim",
"autoupdate",
"azuretools",
"bantime",
"bazarr",
"bgwriter",
"binhex",
"bitwarden",
"blkdiscard",
@@ -39,27 +32,20 @@
"captivedetect",
"cgroupdriver",
"charliermarsh",
"Checkpointing",
"cloudflared",
"codellama",
"codezombiech",
"compactmode",
"Compat",
"contentblocking",
"cookiebanners",
"createdb",
"createrole",
"crlite",
"cryptsetup",
"cuda",
"darkreader",
"datagrip",
"datareporting",
"davidanson",
"dconf",
"dearrow",
"debugpy",
"deepseek",
"dialout",
"diffie",
"direnv",
@@ -67,28 +53,23 @@
"dnodesize",
"dotfiles",
"drawio",
"duckdns",
"eamodio",
"ehci",
"emerg",
"endlessh",
"errorlens",
"esbenp",
"esphome",
"extest",
"fadvise",
"fastfetch",
"fastforwardteam",
"FASTFOX",
"ffmpegthumbnailer",
"filebot",
"filebrowser",
"fileroller",
"findbar",
"Fira",
"fmask",
"fontconfig",
"formfill",
"forwardfor",
"foxundermoon",
"FULLSCREEN",
"fwupd",
@@ -98,7 +79,6 @@
"getch",
"getmaxyx",
"ghdeploy",
"gitea",
"globalprivacycontrol",
"gparted",
"gtts",
@@ -114,32 +94,23 @@
"HPKP",
"hplip",
"htmlaboutaddons",
"httpchk",
"hurlenko",
"hwloc",
"ical",
"ignorelist",
"improv",
"INITDB",
"iocharset",
"ioit",
"iperf",
"isal",
"jellyfin",
"jnoortheen",
"jsbc",
"kagi",
"keyformat",
"keylocation",
"kuma",
"lazer",
"levelname",
"libglvnd",
"libmysqlclient",
"libsodium",
"libssh",
"libvirtd",
"llms",
"localtime",
"louislam",
"lsnew",
@@ -147,56 +118,43 @@
"lynis",
"mangohud",
"markdownlint",
"maxconn",
"maxpages",
"maxretry",
"maxtime",
"mechatroner",
"mediainfo",
"mixtral",
"mklabel",
"mkpart",
"modbus",
"modbuss",
"modesetting",
"mountpoint",
"mountpoints",
"mousewheel",
"mqtt",
"mtxr",
"mypy",
"muninn",
"ncdu",
"nemo",
"neofetch",
"nerdfonts",
"netdev",
"netdevs",
"Networkd",
"networkmanager",
"newtabpage",
"nixfmt",
"nixos",
"nixpkgs",
"nmap",
"noauto",
"nodev",
"noecho",
"nonsponsored",
"Noto",
"nprt",
"nvme",
"OCSP",
"oderwat",
"ollama",
"ondemand",
"oneshot",
"openwakeword",
"optimise",
"optoutstudies",
"overalljails",
"overscroll",
"overseerr",
"paho",
"partitionwise",
"pbmode",
"pciutils",
"pcscd",
@@ -204,17 +162,16 @@
"peerconnection",
"PESKYFOX",
"PGID",
"photoprism",
"pipewire",
"pkgs",
"plugdev",
"poppler",
"posixacl",
"powertop",
"primarycache",
"prismlauncher",
"privatebrowsing",
"PRIVOXY",
"protontricks",
"prowlarr",
"proxychains",
"prusa",
@@ -222,29 +179,21 @@
"PUID",
"pulseaudio",
"punycode",
"pychromecast",
"pydocstyle",
"pyfakefs",
"pylance",
"pylint",
"pymetno",
"pymodbus",
"pyopenweathermap",
"pyownet",
"pytest",
"qbit",
"qbittorrent",
"qbittorrentvpn",
"qbitvpn",
"quicksuggest",
"radarr",
"readahead",
"receiveencrypted",
"recordsize",
"Redistributable",
"referer",
"REFERERS",
"relatime",
"Rhosts",
"ripgrep",
"roboto",
"rokuecp",
"routable",
"rspace",
"rtkit",
@@ -253,7 +202,6 @@
"schemeless",
"scrollback",
"SECUREFOX",
"sessionmaker",
"sessionstore",
"shellcheck",
"signon",
@@ -265,10 +213,8 @@
"socialtracking",
"sonarr",
"sponsorblock",
"sqlalchemy",
"sqltools",
"ssdp",
"SSHOPTS",
"stdenv",
"subresource",
"substituters",
@@ -276,28 +222,19 @@
"sysstat",
"tabmanager",
"tamasfe",
"TCPIP",
"testdisk",
"tiktok",
"timonwong",
"titlebar",
"tmmworkshop",
"Tmpfs",
"topsites",
"topstories",
"treefmt",
"torrenting",
"twimg",
"typedmonarchmoney",
"typer",
"uaccess",
"ubiquiti",
"ublock",
"uiprotect",
"uitour",
"unifi",
"unrar",
"unsubmitted",
"uptimekuma",
"urlbar",
"urlclassifier",
"usbhid",
@@ -306,20 +243,15 @@
"usernamehw",
"userprefs",
"vfat",
"victron",
"virt",
"virtualisation",
"vpnpromourl",
"wakeonlan",
"webchannel",
"WEBRTC",
"WEBUI",
"wireplumber",
"wireshark",
"Workqueues",
"xattr",
"xcursorgen",
"xdist",
"xhci",
"yazi",
"yubikey",
@@ -331,10 +263,5 @@
"zoxide",
"zram",
"zstd"
],
"python-envs.defaultEnvManager": "ms-python.python:system",
"python-envs.pythonProjects": [],
"python.testing.pytestArgs": ["tests"],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true
]
}

View File

@@ -1,12 +0,0 @@
## Dev environment tips
- use treefmt to format all files
- make python code ruff compliant
- use pytest to test python code
- always use the minimum amount of complexity
- if judgment calls are easy to reverse make them. if not ask me first
- Match existing code style.
- Use builtin helpers getenv() over os.environ.get.
- Prefer single-purpose functions over “do everything” helpers.
- Avoid compatibility branches like PG_USER and POSTGRESQL_URL unless requested.
- Keep helpers only if reused or they simplify the code otherwise inline.

6
build.sh Normal file
View File

@@ -0,0 +1,6 @@
#!/bin/sh
nixos-rebuild build --flake /home/richie/projects/dotfiles#bob
nixos-rebuild build --flake /home/richie/projects/dotfiles#jeeves
nixos-rebuild build --flake /home/richie/projects/dotfiles#muninn
nixos-rebuild build --flake /home/richie/projects/dotfiles#rhapsody-in-green

View File

@@ -23,7 +23,7 @@
boot = {
tmp.useTmpfs = true;
kernelPackages = lib.mkDefault pkgs.linuxPackages_6_12;
zfs.package = lib.mkDefault pkgs.zfs_2_4;
zfs.package = lib.mkDefault pkgs.zfs;
};
hardware.enableRedistributableFirmware = true;
@@ -31,42 +31,21 @@
home-manager = {
useGlobalPkgs = true;
useUserPackages = true;
extraSpecialArgs = { inherit inputs outputs; };
backupFileExtension = "backup";
extraSpecialArgs = {inherit inputs outputs;};
};
nixpkgs = {
overlays = builtins.attrValues outputs.overlays;
config.allowUnfree = true;
};
services = {
# firmware update
fwupd.enable = true;
snapshot_manager = {
enable = lib.mkDefault true;
PYTHONPATH = "${inputs.self}/";
};
zfs = {
trim.enable = lib.mkDefault true;
autoScrub.enable = lib.mkDefault true;
config = {
allowUnfree = true;
};
};
powerManagement.powertop.enable = lib.mkDefault true;
services.fwupd.enable = true;
programs.zsh.enable = true;
security = {
auditd.enable = lib.mkDefault true;
sudo-rs = {
enable = true;
execWheelOnly = true;
};
sudo.enable = false;
};
security.auditd.enable = lib.mkDefault true;
users.mutableUsers = lib.mkDefault false;

View File

@@ -1,10 +1,4 @@
{ lib, pkgs, ... }:
let
libPath = pkgs.lib.makeLibraryPath [
pkgs.zlib
pkgs.stdenv.cc.cc.lib
];
in
{
programs.nix-ld = {
enable = lib.mkDefault true;
@@ -21,7 +15,6 @@ in
libxml2
openssl
stdenv.cc.cc
stdenv.cc.cc.lib
systemd
util-linux
xz
@@ -30,9 +23,4 @@ in
zstd
];
};
environment = {
sessionVariables.LD_LIBRARY_PATH = lib.mkDefault libPath;
variables.LD_LIBRARY_PATH = lib.mkDefault libPath;
};
}

View File

@@ -4,7 +4,8 @@
console.keyMap = lib.mkDefault "us";
i18n = {
defaultLocale = lib.mkDefault "en_US.UTF-8";
defaultLocale = lib.mkDefault "en_US.utf8";
supportedLocales = lib.mkDefault [ "en_US.UTF-8/UTF-8" ];
extraLocaleSettings = lib.mkDefault {
LC_ADDRESS = "en_US.UTF-8";
LC_IDENTIFICATION = "en_US.UTF-8";

View File

@@ -2,11 +2,9 @@
inputs,
lib,
...
}:
let
}: let
flakeInputs = lib.filterAttrs (_: lib.isType "flake") inputs;
in
{
in {
nix = {
settings = {
trusted-users = [
@@ -15,14 +13,17 @@ in
];
trusted-substituters = [
"https://cache.nixos.org"
"https://cache.tmmworkshop.com"
"https://nix-community.cachix.org"
];
substituters = [
"https://cache.nixos.org/?priority=2&want-mass-query=true"
"https://cache.tmmworkshop.com/?priority=2&want-mass-query=true"
"https://nix-community.cachix.org/?priority=10&want-mass-query=true"
];
trusted-public-keys = [
"cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY="
"cache.tmmworkshop.com:jHffkpgbmEdstQPoihJPYW9TQe6jnQbWR2LqkNGV3iA="
"nix-community.cachix.org-1:mB9FSh9qf2dCimDSUo8Zy7bkq5CX+/rkCWyvRCYg3Fs="
];
auto-optimise-store = lib.mkDefault true;
@@ -33,12 +34,10 @@ in
];
warn-dirty = false;
flake-registry = ""; # disable global flake registries
connect-timeout = 10;
fallback = true;
};
# Add each flake input as a registry and nix_path
registry = lib.mapAttrs (_: flake: { inherit flake; }) flakeInputs;
registry = lib.mapAttrs (_: flake: {inherit flake;}) flakeInputs;
nixPath = lib.mapAttrsToList (n: _: "${n}=flake:${n}") flakeInputs;
};
}

View File

@@ -2,6 +2,6 @@
{
environment.systemPackages = with pkgs; [
git
my_python
python313
];
}

View File

@@ -1,57 +1,37 @@
{
pkgs,
lib,
config,
...
}:
{ inputs, pkgs, lib, config, ... }:
let
cfg = config.services.snapshot_manager;
in
{
options = {
services.snapshot_manager = {
enable = lib.mkEnableOption "ZFS snapshot manager";
enable = lib.mkOption {
default = true;
example = true;
description = "Whether to enable k3s-net.";
type = lib.types.bool;
};
path = lib.mkOption {
type = lib.types.path;
description = "Path that needs to be updated via git pull";
default = ./snapshot_config.toml;
description = "Path to the snapshot_manager TOML config.";
};
PYTHONPATH = lib.mkOption {
type = lib.types.str;
description = ''
the PYTHONPATH to use for the snapshot_manager service.
'';
};
EnvironmentFile = lib.mkOption {
type = lib.types.nullOr (lib.types.coercedTo lib.types.path toString lib.types.str);
default = null;
description = ''
Single environment file for the service (e.g. /etc/snapshot-manager/env).
Use a leading "-" to ignore if missing (systemd feature).
'';
};
};
};
config = lib.mkIf cfg.enable {
systemd = {
services.snapshot_manager = {
services."snapshot_manager" = {
description = "ZFS Snapshot Manager";
requires = [ "zfs-import.target" ];
after = [ "zfs-import.target" ];
path = [ pkgs.zfs ];
environment = {
PYTHONPATH = cfg.PYTHONPATH;
};
serviceConfig = {
Type = "oneshot";
ExecStart = "${pkgs.my_python}/bin/python -m python.tools.snapshot_manager ${lib.escapeShellArg cfg.path}";
}
// lib.optionalAttrs (cfg.EnvironmentFile != null) {
EnvironmentFile = cfg.EnvironmentFile;
ExecStart = "${inputs.system_tools.packages.x86_64-linux.default}/bin/snapshot_manager --config-file='${cfg.path}'";
};
};
timers.snapshot_manager = {
timers."snapshot_manager" = {
wantedBy = [ "timers.target" ];
timerConfig = {
OnBootSec = "15m";
@@ -61,4 +41,4 @@ in
};
};
};
}
}

View File

@@ -37,8 +37,6 @@
TcpKeepAlive = "no";
X11Forwarding = lib.mkDefault false;
KexAlgorithms = [
"sntrup761x25519-sha512@openssh.com"
"mlkem768x25519-sha256"
"curve25519-sha256@libssh.org"
"diffie-hellman-group-exchange-sha256"
];
@@ -69,5 +67,5 @@
};
};
networking.firewall.allowedTCPPorts = [ 22 ];
networking.firewall.allowedTCPPorts = [ 22 ];
}

View File

@@ -1,18 +1,10 @@
{ pkgs, ... }:
{
boot = {
kernelPackages = pkgs.linuxPackages_6_18;
zfs.package = pkgs.zfs_2_4;
kernelPackages = pkgs.linuxPackages_6_12;
zfs.package = pkgs.zfs;
};
hardware.bluetooth = {
enable = true;
powerOnBoot = true;
};
# rtkit is optional but recommended for pipewire
security.rtkit.enable = true;
services = {
displayManager.sddm = {
enable = true;
@@ -29,8 +21,6 @@
};
};
pulseaudio.enable = false;
pipewire = {
enable = true;
alsa.enable = true;

View File

@@ -2,9 +2,6 @@
{
services.printing = {
enable = true;
drivers = with pkgs; [
gutenprint
hplip
];
drivers = with pkgs; [ gutenprint hplip ];
};
}

View File

@@ -10,9 +10,6 @@
authorizedKeys = config.users.users.richie.openssh.authorizedKeys.keys;
};
};
availableKernelModules = [
"igb"
"r8152"
];
availableKernelModules = [ "igb" ];
};
}

View File

@@ -1,9 +1,6 @@
{ pkgs, ... }:
{
environment.systemPackages = with pkgs; [
mangohud
steam-run
];
environment.systemPackages = with pkgs; [mangohud steam-run];
hardware.steam-hardware.enable = true;
programs = {
@@ -14,7 +11,7 @@
remotePlay.openFirewall = true;
localNetworkGameTransfers.openFirewall = true;
protontricks.enable = true;
extraCompatPackages = with pkgs; [ proton-ge-bin ];
extraCompatPackages = with pkgs; [proton-ge-bin];
extest.enable = true;
};
gamescope = {

View File

@@ -1,19 +1,19 @@
{ lib, ... }:
{
services.syncthing = {
enable = true;
user = "richie";
overrideDevices = true;
overrideFolders = lib.mkDefault true;
overrideFolders = true;
dataDir = "/home/richie/Syncthing";
configDir = "/home/richie/.config/syncthing";
settings.devices = {
bob.id = "CJIAPEJ-VO74RR4-F75VU6M-QNZAMYG-FYUJG7Y-6AT62HJ-355PRPL-PJFETAZ"; # cspell:disable-line
brain.id = "SSCGIPI-IV3VYKB-TRNIJE3-COV4T2H-CDBER7F-I2CGHYA-NWOEUDU-3T5QAAN"; # cspell:disable-line
ipad.id = "KI76T3X-SFUGV2L-VSNYTKR-TSIUV5L-SHWD3HE-GQRGRCN-GY4UFMD-CW6Z6AX"; # cspell:disable-line
jeeves.id = "ICRHXZW-ECYJCUZ-I4CZ64R-3XRK7CG-LL2HAAK-FGOHD22-BQA4AI6-5OAL6AG"; # cspell:disable-line
phone.id = "TBRULKD-7DZPGGZ-F6LLB7J-MSO54AY-7KLPBIN-QOFK6PX-W2HBEWI-PHM2CQI"; # cspell:disable-line
rhapsody-in-green.id = "ASL3KC4-3XEN6PA-7BQBRKE-A7JXLI6-DJT43BY-Q4WPOER-7UALUAZ-VTPQ6Q4"; # cspell:disable-line
settings = {
devices = {
phone.id = "LTGPLAE-M4ZDJTM-TZ3DJGY-SLLAVWF-CQDVEVS-RGCS75T-GAPZYK3-KUM6LA5"; # cspell:disable-line
jeeves.id = "ICRHXZW-ECYJCUZ-I4CZ64R-3XRK7CG-LL2HAAK-FGOHD22-BQA4AI6-5OAL6AG"; # cspell:disable-line
ipad.id = "KI76T3X-SFUGV2L-VSNYTKR-TSIUV5L-SHWD3HE-GQRGRCN-GY4UFMD-CW6Z6AX"; # cspell:disable-line
bob.id = "CJIAPEJ-VO74RR4-F75VU6M-QNZAMYG-FYUJG7Y-6AT62HJ-355PRPL-PJFETAZ"; # cspell:disable-line
rhapsody-in-green.id = "ASL3KC4-3XEN6PA-7BQBRKE-A7JXLI6-DJT43BY-Q4WPOER-7UALUAZ-VTPQ6Q4"; # cspell:disable-line
};
};
};
}

View File

@@ -1,7 +0,0 @@
{
nix.settings = {
trusted-substituters = [ "http://cache.tmmworkshop.com" ];
substituters = [ "http://cache.tmmworkshop.com/?priority=1&want-mass-query=true" ];
trusted-public-keys = [ "cache.tmmworkshop.com:jHffkpgbmEdstQPoihJPYW9TQe6jnQbWR2LqkNGV3iA=" ];
};
}

View File

@@ -1,11 +1,32 @@
{ lib, pkgs, ... }:
{
systemd = {
services."autopull@dotfiles" = {
requires = [ "multi-user.target" ];
after = [ "multi-user.target" ];
description = "Pull the latest data for dotfiles";
serviceConfig = {
Type = "oneshot";
User = "root";
WorkingDirectory = /root/dotfiles;
ExecStart = "${pkgs.git}/bin/git pull --all --prune";
};
};
timers."autopull@dotfiles" = {
wantedBy = [ "timers.target" ];
timerConfig = {
OnBootSec = "1h";
OnUnitActiveSec = "1h";
Unit = "autopull@dotfiles.service";
};
};
};
system.autoUpgrade = {
enable = true;
enable = lib.mkDefault true;
flags = [ "--accept-flake-config" ];
randomizedDelaySec = "1h";
persistent = true;
flake = "github:RichieCahill/dotfiles";
allowReboot = true;
dates = "Sat *-*-* 06:00:00";
flake = "github:RAD-Development/nix-dotfiles";
};
}

View File

@@ -1,4 +0,0 @@
source "https://rubygems.org"
# The github-pages gem pins all compatible versions of Jekyll and its plugins
gem "github-pages", group: :jekyll_plugins

View File

@@ -1,23 +0,0 @@
title: "Richie Cahill"
description: "ALL THE CHAOS THAT I CANT DO AT WORK"
baseurl: "/dotfiles"
url: "https://richiecahill.github.io"
remote_theme: pages-themes/hacker@v0.2.0
plugins:
- jekyll-feed
- jekyll-remote-theme
- jekyll-seo-tag
- jekyll-sitemap
- jekyll-paginate
paginate: 5
paginate_path: "/page:num"
author:
name: "Richie Cahill"
email: "richie@tmmworkshop.com"
social_links:
github: "RichieCahill"
website: "https://tmmworkshop.com"

View File

@@ -1,13 +0,0 @@
# The MONOREPO experiment
Im testing a [MONOREPO](https://en.wikipedia.org/wiki/Monorepo) because Phil said this was a bad idea. To that i say hold my beer.
In all seriousness, I Think that for a small dev team/solo dev. The simplicity is worth higher barer to entry. One of my most annoying processes was updating my system tools. I had to build my update in a feature branch and then merge it into my main branch. then go to my dotfiles create a feature branch update the system tools merge it into main.
It will be starting with my Nix Dotfiles Python tools and now my blog.
I will be reaching ot to phil on 2030-10-31 and 2035-10-31 to give him updates on the progress.
Known Issues:
- the python tests are running on the current derivation not the one the derivation im updating to.

View File

@@ -1,17 +0,0 @@
---
layout: default
title: "Welcome"
---
Welcome to my build logs, notes, and experiments.
You can read my latest posts below
<ul>
{% for post in site.posts %}
<li>
<a href="{{ post.url | relative_url }}">{{ post.title }}</a>
<small>— {{ post.date | date: "%Y-%m-%d" }}</small>
</li>
{% endfor %}
</ul>

3
esphome/.gitignore vendored
View File

@@ -1,3 +0,0 @@
# esphome
/.esphome/
/secrets.yaml

View File

@@ -1,132 +0,0 @@
esphome:
name: batteries
friendly_name: batteries
esp32:
board: esp32dev
framework:
type: arduino
logger:
api:
encryption:
key: !secret api_key
external_components:
- source: github://syssi/esphome-jk-bms@main
ota:
- platform: esphome
password: !secret ota_password
wifi:
ssid: !secret wifi_ssid
password: !secret wifi_password
fast_connect: on
captive_portal:
esp32_ble_tracker:
scan_parameters:
interval: 1100ms
window: 1100ms
active: true
ble_client:
- mac_address: "C8:47:80:29:0F:DB"
id: jk_ble0
jk_bms_ble:
- ble_client_id: jk_ble0
protocol_version: JK02_32S
throttle: 1s
id: jk_bms0
button:
- platform: jk_bms_ble
retrieve_settings:
name: "JK0 retrieve settings"
retrieve_device_info:
name: "JK0 retrieve device info"
sensor:
- platform: jk_bms_ble
jk_bms_ble_id: jk_bms0
total_voltage:
name: "JK0 Total Voltage"
state_of_charge:
name: "JK0 SoC"
charging_power:
name: "JK0 charging power"
discharging_power:
name: "JK0 discharging power"
temperature_sensor_1:
name: "JK0 Temp 1"
temperature_sensor_2:
name: "JK0 Temp 2"
balancing:
name: "JK0 balancing"
total_runtime:
name: "JK0 total runtime"
balancing_current:
name: "JK0 balancing current"
delta_cell_voltage:
name: "JK0 cell delta voltage"
average_cell_voltage:
name: "JK0 cell average voltage"
cell_voltage_1:
name: "JK0 cell voltage 1"
cell_voltage_2:
name: "JK0 cell voltage 2"
cell_voltage_3:
name: "JK0 cell voltage 3"
cell_voltage_4:
name: "JK0 cell voltage 4"
cell_voltage_5:
name: "JK0 cell voltage 5"
cell_voltage_6:
name: "JK0 cell voltage 6"
cell_voltage_7:
name: "JK0 cell voltage 7"
cell_voltage_8:
name: "JK0 cell voltage 8"
cell_resistance_1:
name: "JK0 cell resistance 1"
cell_resistance_2:
name: "JK0 cell resistance 2"
cell_resistance_3:
name: "JK0 cell resistance 3"
cell_resistance_4:
name: "JK0 cell resistance 4"
cell_resistance_5:
name: "JK0 cell resistance 5"
cell_resistance_6:
name: "JK0 cell resistance 6"
cell_resistance_7:
name: "JK0 cell resistance 7"
cell_resistance_8:
name: "JK0 cell resistance 8"
total_charging_cycle_capacity:
name: "JK0 total charging cycle capacity"
text_sensor:
- platform: jk_bms_ble
jk_bms_ble_id: jk_bms0
errors:
name: "JK0 Errors"
switch:
- platform: jk_bms_ble
jk_bms_ble_id: jk_bms0
charging:
name: "JK0 Charging"
discharging:
name: "JK0 Discharging"
balancer:
name: "JK0 Balancing"
- platform: ble_client
ble_client_id: jk_ble0
name: "JK0 enable bluetooth connection"
id: ble_client_switch0

View File

@@ -1,132 +0,0 @@
esphome:
name: battery1
friendly_name: battery1
esp32:
board: esp32dev
framework:
type: arduino
logger:
api:
encryption:
key: !secret api_key
external_components:
- source: github://syssi/esphome-jk-bms@main
ota:
- platform: esphome
password: !secret ota_password
wifi:
ssid: !secret wifi_ssid
password: !secret wifi_password
fast_connect: on
captive_portal:
esp32_ble_tracker:
scan_parameters:
interval: 1100ms
window: 1100ms
active: true
ble_client:
- mac_address: "C8:47:80:37:9D:DD"
id: jk_ble1
jk_bms_ble:
- ble_client_id: jk_ble1
protocol_version: JK02_32S
throttle: 1s
id: jk_bms1
button:
- platform: jk_bms_ble
retrieve_settings:
name: "JK1 retrieve settings"
retrieve_device_info:
name: "JK1 retrieve device info"
sensor:
- platform: jk_bms_ble
jk_bms_ble_id: jk_bms1
total_voltage:
name: "JK1 Total Voltage"
state_of_charge:
name: "JK1 SoC"
charging_power:
name: "JK1 charging power"
discharging_power:
name: "JK1 discharging power"
temperature_sensor_1:
name: "JK1 Temp 1"
temperature_sensor_2:
name: "JK1 Temp 2"
balancing:
name: "JK1 balancing"
total_runtime:
name: "JK1 total runtime"
balancing_current:
name: "JK1 balancing current"
delta_cell_voltage:
name: "JK1 cell delta voltage"
average_cell_voltage:
name: "JK1 cell average voltage"
cell_voltage_1:
name: "JK1 cell voltage 1"
cell_voltage_2:
name: "JK1 cell voltage 2"
cell_voltage_3:
name: "JK1 cell voltage 3"
cell_voltage_4:
name: "JK1 cell voltage 4"
cell_voltage_5:
name: "JK1 cell voltage 5"
cell_voltage_6:
name: "JK1 cell voltage 6"
cell_voltage_7:
name: "JK1 cell voltage 7"
cell_voltage_8:
name: "JK1 cell voltage 8"
cell_resistance_1:
name: "JK1 cell resistance 1"
cell_resistance_2:
name: "JK1 cell resistance 2"
cell_resistance_3:
name: "JK1 cell resistance 3"
cell_resistance_4:
name: "JK1 cell resistance 4"
cell_resistance_5:
name: "JK1 cell resistance 5"
cell_resistance_6:
name: "JK1 cell resistance 6"
cell_resistance_7:
name: "JK1 cell resistance 7"
cell_resistance_8:
name: "JK1 cell resistance 8"
total_charging_cycle_capacity:
name: "JK1 total charging cycle capacity"
text_sensor:
- platform: jk_bms_ble
jk_bms_ble_id: jk_bms1
errors:
name: "JK1 Errors"
switch:
- platform: jk_bms_ble
jk_bms_ble_id: jk_bms1
charging:
name: "JK1 Charging"
discharging:
name: "JK1 Discharging"
balancer:
name: "JK1 Balancing"
- platform: ble_client
ble_client_id: jk_ble1
name: "JK1 enable bluetooth connection"
id: ble_client_switch0

View File

@@ -1,48 +0,0 @@
esphome:
name: "environment"
friendly_name: "environment"
esp32:
board: esp32dev
framework:
type: arduino
i2c:
sda: GPIO21
scl: GPIO22
scan: True
id: bus_a
sensor:
- platform: aht10
i2c_id: bus_a
address: 0x38
variant: AHT20
temperature:
name: "environment Temperature"
id: aht10_temperature
humidity:
name: "environment Humidity"
id: aht10_humidity
update_interval: 5s
web_server:
port: 80
logger:
level: DEBUG
api:
encryption:
key: !secret api_key
ota:
- platform: esphome
password: !secret ota_password
wifi:
ssid: !secret wifi_ssid
password: !secret wifi_password
fast_connect: on
captive_portal:

File diff suppressed because one or more lines are too long

275
flake.lock generated
View File

@@ -2,17 +2,18 @@
"nodes": {
"firefox-addons": {
"inputs": {
"flake-utils": "flake-utils",
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"dir": "pkgs/firefox-addons",
"lastModified": 1772824881,
"narHash": "sha256-NqX+JCA8hRV3GoYrsqnHB2IWKte1eQ8NK2WVbJkORcw=",
"lastModified": 1736279099,
"narHash": "sha256-WXbwHw+JeIOo3KvGDjrfsuwj8JWBUkFFxmlm07w9i7Q=",
"owner": "rycee",
"repo": "nur-expressions",
"rev": "07e1616c9b13fe4794dad4bcc33cd7088c554465",
"rev": "0ad9cf75bae6a9290e9c83cef19f60595f189979",
"type": "gitlab"
},
"original": {
@@ -22,6 +23,55 @@
"type": "gitlab"
}
},
"flake-compat": {
"flake": false,
"locked": {
"lastModified": 1717312683,
"narHash": "sha256-FrlieJH50AuvagamEvWMIE6D2OAnERuDboFDYAED/dE=",
"owner": "nix-community",
"repo": "flake-compat",
"rev": "38fd3954cf65ce6faf3d0d45cd26059e059f07ea",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "flake-compat",
"type": "github"
}
},
"flake-utils": {
"locked": {
"lastModified": 1629284811,
"narHash": "sha256-JHgasjPR0/J1J3DRm4KxM4zTyAj4IOJY8vIl75v/kPI=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "c5d161cc0af116a2e17f54316f0bf43f0819785c",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "flake-utils",
"type": "github"
}
},
"flake-utils_2": {
"inputs": {
"systems": "systems"
},
"locked": {
"lastModified": 1731533236,
"narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "flake-utils",
"type": "github"
}
},
"home-manager": {
"inputs": {
"nixpkgs": [
@@ -29,11 +79,11 @@
]
},
"locked": {
"lastModified": 1772807318,
"narHash": "sha256-Qjw6ILt8cb2HQQpCmWNLMZZ63wEo1KjTQt+1BcQBr7k=",
"lastModified": 1736277415,
"narHash": "sha256-kPDXF6cIPsVqSK08XF5EC6KM7BdMnM9vtJDzsnf+lLU=",
"owner": "nix-community",
"repo": "home-manager",
"rev": "daa2c221320809f5514edde74d0ad0193ad54ed8",
"rev": "5c4302313d9207f7ec0886d68f8ff4a3c71209a1",
"type": "github"
},
"original": {
@@ -42,13 +92,58 @@
"type": "github"
}
},
"nix-github-actions": {
"inputs": {
"nixpkgs": [
"system_tools",
"poetry2nix",
"nixpkgs"
]
},
"locked": {
"lastModified": 1729742964,
"narHash": "sha256-B4mzTcQ0FZHdpeWcpDYPERtyjJd/NIuaQ9+BV1h+MpA=",
"owner": "nix-community",
"repo": "nix-github-actions",
"rev": "e04df33f62cdcf93d73e9a04142464753a16db67",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "nix-github-actions",
"type": "github"
}
},
"nixos-cosmic": {
"inputs": {
"flake-compat": "flake-compat",
"nixpkgs": [
"nixpkgs"
],
"nixpkgs-stable": "nixpkgs-stable",
"rust-overlay": "rust-overlay"
},
"locked": {
"lastModified": 1736214624,
"narHash": "sha256-Pi70vbASZ1O9cR8RO5d2hBiNjIJBKKLoABl4sxWyOgg=",
"owner": "lilyinstarlight",
"repo": "nixos-cosmic",
"rev": "0830abeebf3b2d1bae44652ffb2c89cf0d56ddaa",
"type": "github"
},
"original": {
"owner": "lilyinstarlight",
"repo": "nixos-cosmic",
"type": "github"
}
},
"nixos-hardware": {
"locked": {
"lastModified": 1771969195,
"narHash": "sha256-qwcDBtrRvJbrrnv1lf/pREQi8t2hWZxVAyeMo7/E9sw=",
"lastModified": 1736283893,
"narHash": "sha256-BG1FfTexFwNty5VhYjaQLMR6CMPfI3QRcaZrFQYu2EM=",
"owner": "nixos",
"repo": "nixos-hardware",
"rev": "41c6b421bdc301b2624486e11905c9af7b8ec68e",
"rev": "4f339f6be2b61662f957c2ee9eda0fa597d8a6d6",
"type": "github"
},
"original": {
@@ -60,11 +155,11 @@
},
"nixpkgs": {
"locked": {
"lastModified": 1772624091,
"narHash": "sha256-QKyJ0QGWBn6r0invrMAK8dmJoBYWoOWy7lN+UHzW1jc=",
"lastModified": 1736012469,
"narHash": "sha256-/qlNWm/IEVVH7GfgAIyP6EsVZI6zjAx1cV5zNyrs+rI=",
"owner": "nixos",
"repo": "nixpkgs",
"rev": "80bdc1e5ce51f56b19791b52b2901187931f5353",
"rev": "8f3e1f807051e32d8c95cd12b9b421623850a34d",
"type": "github"
},
"original": {
@@ -76,11 +171,11 @@
},
"nixpkgs-master": {
"locked": {
"lastModified": 1772842888,
"narHash": "sha256-bQRYIwRb9xuEMHTLd5EzjHhYMKzbUbIo7abFV84iUjM=",
"lastModified": 1736294919,
"narHash": "sha256-6Epm0TQkXt0DcS3akYSeT0PzYMpgkZf0V14+5UGe6oE=",
"owner": "nixos",
"repo": "nixpkgs",
"rev": "af5157af67f118e13172750f63012f199b61e3a1",
"rev": "ed6183b173fe27e1d7dceef1ddca2aa64d07aad4",
"type": "github"
},
"original": {
@@ -91,6 +186,22 @@
}
},
"nixpkgs-stable": {
"locked": {
"lastModified": 1736061677,
"narHash": "sha256-DjkQPnkAfd7eB522PwnkGhOMuT9QVCZspDpJJYyOj60=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "cbd8ec4de4469333c82ff40d057350c30e9f7d36",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-24.11",
"repo": "nixpkgs",
"type": "github"
}
},
"nixpkgs-stable_2": {
"locked": {
"lastModified": 1735563628,
"narHash": "sha256-OnSAY7XDSx7CtDoqNh8jwVwh4xNL/2HaJxGjryLWzX8=",
@@ -106,16 +217,67 @@
"type": "github"
}
},
"poetry2nix": {
"inputs": {
"flake-utils": [
"system_tools",
"flake-utils"
],
"nix-github-actions": "nix-github-actions",
"nixpkgs": [
"system_tools",
"nixpkgs"
],
"systems": "systems_2",
"treefmt-nix": "treefmt-nix"
},
"locked": {
"lastModified": 1731205797,
"narHash": "sha256-F7N1mxH1VrkVNHR3JGNMRvp9+98KYO4b832KS8Gl2xI=",
"owner": "nix-community",
"repo": "poetry2nix",
"rev": "f554d27c1544d9c56e5f1f8e2b8aff399803674e",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "poetry2nix",
"type": "github"
}
},
"root": {
"inputs": {
"firefox-addons": "firefox-addons",
"home-manager": "home-manager",
"nixos-cosmic": "nixos-cosmic",
"nixos-hardware": "nixos-hardware",
"nixpkgs": "nixpkgs",
"nixpkgs-master": "nixpkgs-master",
"nixpkgs-stable": "nixpkgs-stable",
"nixpkgs-stable": "nixpkgs-stable_2",
"sops-nix": "sops-nix",
"systems": "systems"
"system_tools": "system_tools",
"systems": "systems_3"
}
},
"rust-overlay": {
"inputs": {
"nixpkgs": [
"nixos-cosmic",
"nixpkgs"
]
},
"locked": {
"lastModified": 1736130662,
"narHash": "sha256-z+WGez9oTR2OsiUWE5ZhIpETqM1ogrv6Xcd24WFi6KQ=",
"owner": "oxalica",
"repo": "rust-overlay",
"rev": "2f5d4d9cd31cc02c36e51cb2e21c4b25c4f78c52",
"type": "github"
},
"original": {
"owner": "oxalica",
"repo": "rust-overlay",
"type": "github"
}
},
"sops-nix": {
@@ -125,11 +287,11 @@
]
},
"locked": {
"lastModified": 1772495394,
"narHash": "sha256-hmIvE/slLKEFKNEJz27IZ8BKlAaZDcjIHmkZ7GCEjfw=",
"lastModified": 1736203741,
"narHash": "sha256-eSjkBwBdQk+TZWFlLbclF2rAh4JxbGg8az4w/Lfe7f4=",
"owner": "Mic92",
"repo": "sops-nix",
"rev": "1d9b98a29a45abe9c4d3174bd36de9f28755e3ff",
"rev": "c9c88f08e3ee495e888b8d7c8624a0b2519cb773",
"type": "github"
},
"original": {
@@ -138,7 +300,58 @@
"type": "github"
}
},
"system_tools": {
"inputs": {
"flake-utils": "flake-utils_2",
"nixpkgs": [
"nixpkgs"
],
"poetry2nix": "poetry2nix"
},
"locked": {
"lastModified": 1733083480,
"narHash": "sha256-B13faNyBbA3MeI7Jp6pFVbp58rI2Rx5Uvd83csW2p48=",
"owner": "RichieCahill",
"repo": "system_tools",
"rev": "bc357d8fabd83c1423611829091e5b1d86dd913c",
"type": "github"
},
"original": {
"owner": "RichieCahill",
"repo": "system_tools",
"type": "github"
}
},
"systems": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"type": "github"
}
},
"systems_2": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"id": "systems",
"type": "indirect"
}
},
"systems_3": {
"locked": {
"lastModified": 1689347949,
"narHash": "sha256-12tWmuL2zgBgZkdoB6qXZsgJEH9LR3oUgpaQq2RbI80=",
@@ -152,6 +365,28 @@
"repo": "default-linux",
"type": "github"
}
},
"treefmt-nix": {
"inputs": {
"nixpkgs": [
"system_tools",
"poetry2nix",
"nixpkgs"
]
},
"locked": {
"lastModified": 1730120726,
"narHash": "sha256-LqHYIxMrl/1p3/kvm2ir925tZ8DkI0KA10djk8wecSk=",
"owner": "numtide",
"repo": "treefmt-nix",
"rev": "9ef337e492a5555d8e17a51c911ff1f02635be15",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "treefmt-nix",
"type": "github"
}
}
},
"root": "root",

121
flake.nix
View File

@@ -4,12 +4,14 @@
nixConfig = {
extra-substituters = [
"https://cache.nixos.org/?priority=2&want-mass-query=true"
"https://cache.tmmworkshop.com/?priority=2&want-mass-query=true"
"https://nix-community.cachix.org/?priority=10&want-mass-query=true"
];
extra-trusted-public-keys = [
"cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=" # cspell:disable-line
"nix-community.cachix.org-1:mB9FSh9qf2dCimDSUo8Zy7bkq5CX+/rkCWyvRCYg3Fs=" # cspell:disable-line
"cache-nix-dot:Od9KN34LXc6Lu7y1ozzV1kIXZa8coClozgth/SYE7dU=" # cspell:disable-line
"cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY="
"cache.tmmworkshop.com:jHffkpgbmEdstQPoihJPYW9TQe6jnQbWR2LqkNGV3iA="
"nix-community.cachix.org-1:mB9FSh9qf2dCimDSUo8Zy7bkq5CX+/rkCWyvRCYg3Fs="
"cache-nix-dot:Od9KN34LXc6Lu7y1ozzV1kIXZa8coClozgth/SYE7dU="
];
};
@@ -31,72 +33,77 @@
inputs.nixpkgs.follows = "nixpkgs";
};
system_tools = {
url = "github:RichieCahill/system_tools";
inputs.nixpkgs.follows = "nixpkgs";
};
nixos-cosmic = {
url = "github:lilyinstarlight/nixos-cosmic";
inputs.nixpkgs.follows = "nixpkgs";
};
sops-nix = {
url = "github:Mic92/sops-nix";
inputs.nixpkgs.follows = "nixpkgs";
};
};
outputs =
{
self,
nixpkgs,
home-manager,
systems,
sops-nix,
...
}@inputs:
let
inherit (self) outputs;
lib = nixpkgs.lib // home-manager.lib;
forEachSystem = f: lib.genAttrs (import systems) (system: f pkgsFor.${system});
pkgsFor = lib.genAttrs (import systems) (
system:
outputs = {
self,
nixpkgs,
home-manager,
systems,
nixos-cosmic,
sops-nix,
...
} @ inputs: let
inherit (self) outputs;
lib = nixpkgs.lib // home-manager.lib;
forEachSystem = f: lib.genAttrs (import systems) (system: f pkgsFor.${system});
pkgsFor = lib.genAttrs (import systems) (
system:
import nixpkgs {
inherit system;
overlays = builtins.attrValues outputs.overlays;
config.allowUnfree = true;
}
);
in
{
inherit lib;
overlays = import ./overlays { inherit inputs outputs; };
);
in {
inherit lib;
overlays = import ./overlays {inherit inputs outputs;};
devShells = forEachSystem (pkgs: import ./shell.nix { inherit pkgs; });
formatter = forEachSystem (pkgs: pkgs.treefmt);
devShells = forEachSystem (pkgs: import ./shell.nix {inherit pkgs;});
formatter = forEachSystem (pkgs: pkgs.alejandra);
nixosConfigurations = {
bob = lib.nixosSystem {
modules = [
./systems/bob
];
specialArgs = { inherit inputs outputs; };
};
brain = lib.nixosSystem {
modules = [
./systems/brain
];
specialArgs = { inherit inputs outputs; };
};
jeeves = lib.nixosSystem {
modules = [
./systems/jeeves
];
specialArgs = { inherit inputs outputs; };
};
rhapsody-in-green = lib.nixosSystem {
modules = [
./systems/rhapsody-in-green
];
specialArgs = { inherit inputs outputs; };
};
leviathan = lib.nixosSystem {
modules = [
./systems/leviathan
];
specialArgs = { inherit inputs outputs; };
};
nixosConfigurations = {
bob = lib.nixosSystem {
modules = [
./systems/bob
];
specialArgs = {inherit inputs outputs;};
};
jeeves = lib.nixosSystem {
modules = [
./systems/jeeves
];
specialArgs = {inherit inputs outputs;};
};
rhapsody-in-green = lib.nixosSystem {
modules = [
./systems/rhapsody-in-green
];
specialArgs = {inherit inputs outputs;};
};
muninn = lib.nixosSystem {
modules = [
./systems/muninn
];
specialArgs = {inherit inputs outputs;};
};
router = lib.nixosSystem {
modules = [./systems/router];
specialArgs = {inherit inputs outputs;};
};
};
};
}

24
frontend/.gitignore vendored
View File

@@ -1,24 +0,0 @@
# Logs
logs
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
pnpm-debug.log*
lerna-debug.log*
node_modules
dist
dist-ssr
*.local
# Editor directories and files
.vscode/*
!.vscode/extensions.json
.idea
.DS_Store
*.suo
*.ntvs*
*.njsproj
*.sln
*.sw?

View File

@@ -1,73 +0,0 @@
# React + TypeScript + Vite
This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
Currently, two official plugins are available:
- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Babel](https://babeljs.io/) (or [oxc](https://oxc.rs) when used in [rolldown-vite](https://vite.dev/guide/rolldown)) for Fast Refresh
- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/) for Fast Refresh
## React Compiler
The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
## Expanding the ESLint configuration
If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
```js
export default defineConfig([
globalIgnores(['dist']),
{
files: ['**/*.{ts,tsx}'],
extends: [
// Other configs...
// Remove tseslint.configs.recommended and replace with this
tseslint.configs.recommendedTypeChecked,
// Alternatively, use this for stricter rules
tseslint.configs.strictTypeChecked,
// Optionally, add this for stylistic rules
tseslint.configs.stylisticTypeChecked,
// Other configs...
],
languageOptions: {
parserOptions: {
project: ['./tsconfig.node.json', './tsconfig.app.json'],
tsconfigRootDir: import.meta.dirname,
},
// other options...
},
},
])
```
You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
```js
// eslint.config.js
import reactX from 'eslint-plugin-react-x'
import reactDom from 'eslint-plugin-react-dom'
export default defineConfig([
globalIgnores(['dist']),
{
files: ['**/*.{ts,tsx}'],
extends: [
// Other configs...
// Enable lint rules for React
reactX.configs['recommended-typescript'],
// Enable lint rules for React DOM
reactDom.configs.recommended,
],
languageOptions: {
parserOptions: {
project: ['./tsconfig.node.json', './tsconfig.app.json'],
tsconfigRootDir: import.meta.dirname,
},
// other options...
},
},
])
```

View File

@@ -1,23 +0,0 @@
import js from '@eslint/js'
import globals from 'globals'
import reactHooks from 'eslint-plugin-react-hooks'
import reactRefresh from 'eslint-plugin-react-refresh'
import tseslint from 'typescript-eslint'
import { defineConfig, globalIgnores } from 'eslint/config'
export default defineConfig([
globalIgnores(['dist']),
{
files: ['**/*.{ts,tsx}'],
extends: [
js.configs.recommended,
tseslint.configs.recommended,
reactHooks.configs.flat.recommended,
reactRefresh.configs.vite,
],
languageOptions: {
ecmaVersion: 2020,
globals: globals.browser,
},
},
])

View File

@@ -1,13 +0,0 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<link rel="icon" type="image/svg+xml" href="/vite.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>frontend</title>
</head>
<body>
<div id="root"></div>
<script type="module" src="/src/main.tsx"></script>
</body>
</html>

File diff suppressed because it is too large Load Diff

View File

@@ -1,31 +0,0 @@
{
"name": "frontend",
"private": true,
"version": "0.0.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "tsc -b && vite build",
"lint": "eslint .",
"preview": "vite preview"
},
"dependencies": {
"react": "^19.2.0",
"react-dom": "^19.2.0",
"react-router-dom": "^7.12.0"
},
"devDependencies": {
"@eslint/js": "^9.39.1",
"@types/node": "^24.10.1",
"@types/react": "^19.2.5",
"@types/react-dom": "^19.2.3",
"@vitejs/plugin-react": "^5.1.1",
"eslint": "^9.39.1",
"eslint-plugin-react-hooks": "^7.0.1",
"eslint-plugin-react-refresh": "^0.4.24",
"globals": "^16.5.0",
"typescript": "~5.9.3",
"typescript-eslint": "^8.46.4",
"vite": "^7.2.4"
}
}

View File

@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="31.88" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 257"><defs><linearGradient id="IconifyId1813088fe1fbc01fb466" x1="-.828%" x2="57.636%" y1="7.652%" y2="78.411%"><stop offset="0%" stop-color="#41D1FF"></stop><stop offset="100%" stop-color="#BD34FE"></stop></linearGradient><linearGradient id="IconifyId1813088fe1fbc01fb467" x1="43.376%" x2="50.316%" y1="2.242%" y2="89.03%"><stop offset="0%" stop-color="#FFEA83"></stop><stop offset="8.333%" stop-color="#FFDD35"></stop><stop offset="100%" stop-color="#FFA800"></stop></linearGradient></defs><path fill="url(#IconifyId1813088fe1fbc01fb466)" d="M255.153 37.938L134.897 252.976c-2.483 4.44-8.862 4.466-11.382.048L.875 37.958c-2.746-4.814 1.371-10.646 6.827-9.67l120.385 21.517a6.537 6.537 0 0 0 2.322-.004l117.867-21.483c5.438-.991 9.574 4.796 6.877 9.62Z"></path><path fill="url(#IconifyId1813088fe1fbc01fb467)" d="M185.432.063L96.44 17.501a3.268 3.268 0 0 0-2.634 3.014l-5.474 92.456a3.268 3.268 0 0 0 3.997 3.378l24.777-5.718c2.318-.535 4.413 1.507 3.936 3.838l-7.361 36.047c-.495 2.426 1.782 4.5 4.151 3.78l15.304-4.649c2.372-.72 4.652 1.36 4.15 3.788l-11.698 56.621c-.732 3.542 3.979 5.473 5.943 2.437l1.313-2.028l72.516-144.72c1.215-2.423-.88-5.186-3.54-4.672l-25.505 4.922c-2.396.462-4.435-1.77-3.759-4.114l16.646-57.705c.677-2.35-1.37-4.583-3.769-4.113Z"></path></svg>

Before

Width:  |  Height:  |  Size: 1.5 KiB

View File

@@ -1,654 +0,0 @@
* {
box-sizing: border-box;
}
body {
margin: 0;
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
background: var(--color-bg);
color: var(--color-text);
}
.app {
max-width: 1000px;
margin: 0 auto;
padding: 20px;
}
nav {
display: flex;
align-items: center;
gap: 20px;
padding: 15px 0;
border-bottom: 1px solid var(--color-border);
margin-bottom: 20px;
}
.theme-toggle {
margin-left: auto;
}
nav a {
color: var(--color-primary);
text-decoration: none;
font-weight: 500;
}
nav a:hover {
text-decoration: underline;
}
main {
background: var(--color-bg-card);
padding: 20px;
border-radius: 8px;
box-shadow: var(--shadow);
}
.header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 20px;
}
.header h1 {
margin: 0;
}
.btn {
display: inline-block;
padding: 8px 16px;
border: 1px solid var(--color-border);
border-radius: 4px;
background: var(--color-bg-card);
color: var(--color-text);
text-decoration: none;
cursor: pointer;
font-size: 14px;
margin-left: 8px;
}
.btn:hover {
background: var(--color-bg-hover);
}
.btn-primary {
background: var(--color-primary);
border-color: var(--color-primary);
color: white;
}
.btn-primary:hover {
background: var(--color-primary-hover);
}
.btn-danger {
background: var(--color-danger);
border-color: var(--color-danger);
color: white;
}
.btn-danger:hover {
background: var(--color-danger-hover);
}
.btn-small {
padding: 4px 8px;
font-size: 12px;
}
.btn:disabled {
opacity: 0.6;
cursor: not-allowed;
}
table {
width: 100%;
border-collapse: collapse;
}
th,
td {
padding: 12px;
text-align: left;
border-bottom: 1px solid var(--color-border-light);
}
th {
font-weight: 600;
background: var(--color-bg-muted);
}
tr:hover {
background: var(--color-bg-muted);
}
.error {
background: var(--color-bg-error);
color: var(--color-text-error);
padding: 10px;
border-radius: 4px;
margin-bottom: 20px;
}
.info-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(200px, 1fr));
gap: 10px;
margin-bottom: 20px;
}
.section {
margin-top: 30px;
padding-top: 20px;
border-top: 1px solid var(--color-border-light);
}
.section h3 {
margin-top: 0;
margin-bottom: 15px;
}
.section h4 {
margin: 15px 0 10px;
font-size: 14px;
color: var(--color-text-muted);
}
.section ul {
list-style: none;
padding: 0;
margin: 0;
}
.section li {
display: flex;
align-items: center;
gap: 10px;
padding: 8px 0;
border-bottom: 1px solid var(--color-border-lighter);
}
.tag {
display: inline-block;
background: var(--color-tag-bg);
padding: 2px 8px;
border-radius: 12px;
font-size: 12px;
color: var(--color-text-muted);
}
.add-form {
display: flex;
gap: 10px;
margin-top: 15px;
flex-wrap: wrap;
}
.add-form select,
.add-form input {
padding: 8px;
border: 1px solid var(--color-border);
border-radius: 4px;
min-width: 200px;
background: var(--color-bg-card);
color: var(--color-text);
}
.form-group {
margin-bottom: 20px;
}
.form-group label {
display: block;
font-weight: 500;
margin-bottom: 5px;
}
.form-group input,
.form-group textarea,
.form-group select {
width: 100%;
padding: 10px;
border: 1px solid var(--color-border);
border-radius: 4px;
font-size: 14px;
background: var(--color-bg-card);
color: var(--color-text);
}
.form-group textarea {
resize: vertical;
}
.form-row {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 20px;
}
.checkbox-group {
display: flex;
flex-wrap: wrap;
gap: 15px;
}
.checkbox-label {
display: flex;
align-items: center;
gap: 5px;
cursor: pointer;
}
.form-actions {
display: flex;
gap: 10px;
margin-top: 30px;
padding-top: 20px;
border-top: 1px solid var(--color-border-light);
}
.need-list .header {
margin-bottom: 20px;
}
.need-form {
background: var(--color-bg-muted);
padding: 20px;
border-radius: 4px;
margin-bottom: 20px;
}
.need-items {
list-style: none;
padding: 0;
}
.need-items li {
display: flex;
justify-content: space-between;
align-items: flex-start;
padding: 15px;
border: 1px solid var(--color-border-light);
border-radius: 4px;
margin-bottom: 10px;
}
.need-info p {
margin: 5px 0 0;
color: var(--color-text-muted);
font-size: 14px;
}
a {
color: var(--color-primary);
}
a:hover {
text-decoration: underline;
}
/* Graph styles */
.graph-container {
width: 100%;
}
.graph-hint {
color: var(--color-text-muted);
font-size: 14px;
margin-bottom: 15px;
}
.selected-info {
margin-top: 15px;
padding: 15px;
background: var(--color-bg-muted);
border-radius: 8px;
}
.selected-info h3 {
margin: 0 0 10px;
}
.selected-info p {
margin: 5px 0;
color: var(--color-text-muted);
}
.legend {
margin-top: 20px;
padding: 15px;
background: var(--color-bg-muted);
border-radius: 8px;
}
.legend h4 {
margin: 0 0 10px;
font-size: 14px;
}
.legend-items {
display: flex;
flex-wrap: wrap;
gap: 15px;
}
.legend-item {
display: flex;
align-items: center;
gap: 8px;
font-size: 12px;
color: var(--color-text-muted);
}
.legend-line {
width: 30px;
border-radius: 2px;
}
/* Weight control styles */
.weight-control {
display: flex;
align-items: center;
gap: 8px;
font-size: 12px;
color: var(--color-text-muted);
}
.weight-control input[type="range"] {
width: 80px;
cursor: pointer;
}
.weight-value {
min-width: 20px;
text-align: center;
font-weight: 600;
}
.weight-display {
font-size: 12px;
color: var(--color-text-muted);
margin-left: auto;
}
/* ID Card Styles */
.id-card {
width: 100%;
}
.id-card-inner {
background: linear-gradient(135deg, #0a0a0f 0%, #1a1a2e 50%, #0a0a0f 100%);
background-image:
radial-gradient(white 1px, transparent 1px),
linear-gradient(135deg, #0a0a0f 0%, #1a1a2e 50%, #0a0a0f 100%);
background-size: 50px 50px, 100% 100%;
background-position: 0 0, 0 0;
color: #fff;
border-radius: 12px;
padding: 25px;
min-height: 500px;
position: relative;
overflow: hidden;
}
.id-card-header {
display: flex;
justify-content: space-between;
align-items: flex-start;
margin-bottom: 15px;
}
.id-card-header-left {
flex: 1;
}
.id-card-header-right {
display: flex;
flex-direction: column;
align-items: flex-end;
gap: 10px;
}
.id-card-title {
font-size: 2.5rem;
font-weight: 700;
margin: 0;
color: #fff;
text-shadow: 2px 2px 4px rgba(0,0,0,0.5);
}
.id-profile-pic {
width: 80px;
height: 80px;
border-radius: 8px;
object-fit: cover;
border: 2px solid rgba(255,255,255,0.3);
}
.id-profile-placeholder {
width: 80px;
height: 80px;
border-radius: 8px;
background: linear-gradient(135deg, #4ecdc4 0%, #44a8a0 100%);
display: flex;
align-items: center;
justify-content: center;
border: 2px solid rgba(255,255,255,0.3);
}
.id-profile-placeholder span {
font-size: 2rem;
font-weight: 700;
color: #fff;
text-shadow: 1px 1px 2px rgba(0,0,0,0.3);
}
.id-card-actions {
display: flex;
gap: 8px;
}
.id-card-actions .btn {
background: rgba(255,255,255,0.1);
border-color: rgba(255,255,255,0.3);
color: #fff;
}
.id-card-actions .btn:hover {
background: rgba(255,255,255,0.2);
}
.id-card-body {
display: grid;
grid-template-columns: 1fr 1.5fr;
gap: 30px;
}
.id-card-left {
display: flex;
flex-direction: column;
gap: 8px;
}
.id-field {
font-size: 1rem;
line-height: 1.4;
}
.id-field-block {
margin-top: 15px;
font-size: 0.95rem;
line-height: 1.5;
}
.id-label {
color: #4ecdc4;
font-weight: 500;
}
.id-card-right {
display: flex;
flex-direction: column;
gap: 20px;
}
.id-bio {
font-size: 0.9rem;
line-height: 1.6;
color: #e0e0e0;
}
.id-relationships {
margin-top: 10px;
}
.id-section-title {
font-size: 1.5rem;
margin: 0 0 15px;
color: #fff;
border-bottom: 1px solid rgba(255,255,255,0.2);
padding-bottom: 8px;
}
.id-rel-group {
margin-bottom: 12px;
font-size: 0.9rem;
line-height: 1.6;
}
.id-rel-label {
color: #a0a0a0;
}
.id-rel-group a {
color: #4ecdc4;
text-decoration: none;
}
.id-rel-group a:hover {
text-decoration: underline;
}
.id-rel-type {
color: #888;
font-size: 0.85em;
}
.id-card-warnings {
margin-top: 30px;
padding-top: 20px;
border-top: 1px solid rgba(255,255,255,0.2);
display: flex;
flex-wrap: wrap;
gap: 20px;
}
.id-warning {
display: flex;
align-items: center;
gap: 8px;
font-size: 0.9rem;
color: #ff6b6b;
}
.warning-dot {
width: 8px;
height: 8px;
background: #ff6b6b;
border-radius: 50%;
flex-shrink: 0;
}
.warning-desc {
color: #ccc;
}
/* Management section */
.id-card-manage {
margin-top: 20px;
background: var(--color-bg-muted);
border-radius: 8px;
padding: 15px;
}
.id-card-manage summary {
cursor: pointer;
font-weight: 600;
font-size: 1.1rem;
padding: 5px 0;
}
.id-card-manage[open] summary {
margin-bottom: 15px;
border-bottom: 1px solid var(--color-border-light);
padding-bottom: 10px;
}
.manage-section {
margin-bottom: 25px;
}
.manage-section h3 {
margin: 0 0 15px;
font-size: 1rem;
}
.manage-relationships {
display: flex;
flex-direction: column;
gap: 10px;
margin-bottom: 15px;
}
.manage-rel-item {
display: flex;
align-items: center;
gap: 12px;
padding: 10px;
background: var(--color-bg-card);
border-radius: 6px;
flex-wrap: wrap;
}
.manage-rel-item a {
font-weight: 500;
min-width: 120px;
}
.manage-needs-list {
list-style: none;
padding: 0;
margin: 0 0 15px;
}
.manage-needs-list li {
display: flex;
align-items: center;
gap: 12px;
padding: 10px;
background: var(--color-bg-card);
border-radius: 6px;
margin-bottom: 8px;
}
.manage-needs-list li .btn {
margin-left: auto;
}
/* Responsive adjustments */
@media (max-width: 768px) {
.id-card-body {
grid-template-columns: 1fr;
}
.id-card-title {
font-size: 1.8rem;
}
.id-card-header {
flex-direction: column;
gap: 15px;
}
}

View File

@@ -1,50 +0,0 @@
import { useEffect, useState } from "react";
import { Link, Route, Routes } from "react-router-dom";
import { ContactDetail } from "./components/ContactDetail";
import { ContactForm } from "./components/ContactForm";
import { ContactList } from "./components/ContactList";
import { NeedList } from "./components/NeedList";
import { RelationshipGraph } from "./components/RelationshipGraph";
import "./App.css";
function App() {
const [theme, setTheme] = useState<"light" | "dark">(() => {
return (localStorage.getItem("theme") as "light" | "dark") || "light";
});
useEffect(() => {
document.documentElement.setAttribute("data-theme", theme);
localStorage.setItem("theme", theme);
}, [theme]);
const toggleTheme = () => {
setTheme((prev) => (prev === "light" ? "dark" : "light"));
};
return (
<div className="app">
<nav>
<Link to="/contacts">Contacts</Link>
<Link to="/graph">Graph</Link>
<Link to="/needs">Needs</Link>
<button className="btn btn-small theme-toggle" onClick={toggleTheme}>
{theme === "light" ? "Dark" : "Light"}
</button>
</nav>
<main>
<Routes>
<Route path="/" element={<ContactList />} />
<Route path="/contacts" element={<ContactList />} />
<Route path="/contacts/new" element={<ContactForm />} />
<Route path="/contacts/:id" element={<ContactDetail />} />
<Route path="/contacts/:id/edit" element={<ContactForm />} />
<Route path="/graph" element={<RelationshipGraph />} />
<Route path="/needs" element={<NeedList />} />
</Routes>
</main>
</div>
);
}
export default App;

View File

@@ -1,105 +0,0 @@
import type {
Contact,
ContactCreate,
ContactListItem,
ContactRelationship,
ContactRelationshipCreate,
ContactRelationshipUpdate,
ContactUpdate,
GraphData,
Need,
NeedCreate,
} from "../types";
const API_BASE = "";
async function request<T>(
endpoint: string,
options?: RequestInit
): Promise<T> {
const response = await fetch(`${API_BASE}${endpoint}`, {
...options,
headers: {
"Content-Type": "application/json",
...options?.headers,
},
});
if (!response.ok) {
const error = await response.json().catch(() => ({}));
throw new Error(error.detail || `HTTP ${response.status}`);
}
return response.json();
}
export const api = {
// Needs
needs: {
list: () => request<Need[]>("/api/needs"),
get: (id: number) => request<Need>(`/api/needs/${id}`),
create: (data: NeedCreate) =>
request<Need>("/api/needs", {
method: "POST",
body: JSON.stringify(data),
}),
delete: (id: number) =>
request<{ deleted: boolean }>(`/api/needs/${id}`, { method: "DELETE" }),
},
// Contacts
contacts: {
list: (skip = 0, limit = 100) =>
request<ContactListItem[]>(`/api/contacts?skip=${skip}&limit=${limit}`),
get: (id: number) => request<Contact>(`/api/contacts/${id}`),
create: (data: ContactCreate) =>
request<Contact>("/api/contacts", {
method: "POST",
body: JSON.stringify(data),
}),
update: (id: number, data: ContactUpdate) =>
request<Contact>(`/api/contacts/${id}`, {
method: "PATCH",
body: JSON.stringify(data),
}),
delete: (id: number) =>
request<{ deleted: boolean }>(`/api/contacts/${id}`, { method: "DELETE" }),
// Contact-Need relationships
addNeed: (contactId: number, needId: number) =>
request<{ added: boolean }>(`/api/contacts/${contactId}/needs/${needId}`, {
method: "POST",
}),
removeNeed: (contactId: number, needId: number) =>
request<{ removed: boolean }>(`/api/contacts/${contactId}/needs/${needId}`, {
method: "DELETE",
}),
// Contact-Contact relationships
getRelationships: (contactId: number) =>
request<ContactRelationship[]>(`/api/contacts/${contactId}/relationships`),
addRelationship: (contactId: number, data: ContactRelationshipCreate) =>
request<ContactRelationship>(`/api/contacts/${contactId}/relationships`, {
method: "POST",
body: JSON.stringify(data),
}),
updateRelationship: (contactId: number, relatedContactId: number, data: ContactRelationshipUpdate) =>
request<ContactRelationship>(
`/api/contacts/${contactId}/relationships/${relatedContactId}`,
{
method: "PATCH",
body: JSON.stringify(data),
}
),
removeRelationship: (contactId: number, relatedContactId: number) =>
request<{ deleted: boolean }>(
`/api/contacts/${contactId}/relationships/${relatedContactId}`,
{ method: "DELETE" }
),
},
// Graph
graph: {
get: () => request<GraphData>("/api/graph"),
},
};

View File

@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="35.93" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 228"><path fill="#00D8FF" d="M210.483 73.824a171.49 171.49 0 0 0-8.24-2.597c.465-1.9.893-3.777 1.273-5.621c6.238-30.281 2.16-54.676-11.769-62.708c-13.355-7.7-35.196.329-57.254 19.526a171.23 171.23 0 0 0-6.375 5.848a155.866 155.866 0 0 0-4.241-3.917C100.759 3.829 77.587-4.822 63.673 3.233C50.33 10.957 46.379 33.89 51.995 62.588a170.974 170.974 0 0 0 1.892 8.48c-3.28.932-6.445 1.924-9.474 2.98C17.309 83.498 0 98.307 0 113.668c0 15.865 18.582 31.778 46.812 41.427a145.52 145.52 0 0 0 6.921 2.165a167.467 167.467 0 0 0-2.01 9.138c-5.354 28.2-1.173 50.591 12.134 58.266c13.744 7.926 36.812-.22 59.273-19.855a145.567 145.567 0 0 0 5.342-4.923a168.064 168.064 0 0 0 6.92 6.314c21.758 18.722 43.246 26.282 56.54 18.586c13.731-7.949 18.194-32.003 12.4-61.268a145.016 145.016 0 0 0-1.535-6.842c1.62-.48 3.21-.974 4.76-1.488c29.348-9.723 48.443-25.443 48.443-41.52c0-15.417-17.868-30.326-45.517-39.844Zm-6.365 70.984c-1.4.463-2.836.91-4.3 1.345c-3.24-10.257-7.612-21.163-12.963-32.432c5.106-11 9.31-21.767 12.459-31.957c2.619.758 5.16 1.557 7.61 2.4c23.69 8.156 38.14 20.213 38.14 29.504c0 9.896-15.606 22.743-40.946 31.14Zm-10.514 20.834c2.562 12.94 2.927 24.64 1.23 33.787c-1.524 8.219-4.59 13.698-8.382 15.893c-8.067 4.67-25.32-1.4-43.927-17.412a156.726 156.726 0 0 1-6.437-5.87c7.214-7.889 14.423-17.06 21.459-27.246c12.376-1.098 24.068-2.894 34.671-5.345a134.17 134.17 0 0 1 1.386 6.193ZM87.276 214.515c-7.882 2.783-14.16 2.863-17.955.675c-8.075-4.657-11.432-22.636-6.853-46.752a156.923 156.923 0 0 1 1.869-8.499c10.486 2.32 22.093 3.988 34.498 4.994c7.084 9.967 14.501 19.128 21.976 27.15a134.668 134.668 0 0 1-4.877 4.492c-9.933 8.682-19.886 14.842-28.658 17.94ZM50.35 144.747c-12.483-4.267-22.792-9.812-29.858-15.863c-6.35-5.437-9.555-10.836-9.555-15.216c0-9.322 13.897-21.212 37.076-29.293c2.813-.98 5.757-1.905 8.812-2.773c3.204 10.42 7.406 21.315 12.477 32.332c-5.137 11.18-9.399 22.249-12.634 32.792a134.718 134.718 0 0 1-6.318-1.979Zm12.378-84.26c-4.811-24.587-1.616-43.134 6.425-47.789c8.564-4.958 27.502 2.111 47.463 19.835a144.318 144.318 0 0 1 3.841 3.545c-7.438 7.987-14.787 17.08-21.808 26.988c-12.04 1.116-23.565 2.908-34.161 5.309a160.342 160.342 0 0 1-1.76-7.887Zm110.427 27.268a347.8 347.8 0 0 0-7.785-12.803c8.168 1.033 15.994 2.404 23.343 4.08c-2.206 7.072-4.956 14.465-8.193 22.045a381.151 381.151 0 0 0-7.365-13.322Zm-45.032-43.861c5.044 5.465 10.096 11.566 15.065 18.186a322.04 322.04 0 0 0-30.257-.006c4.974-6.559 10.069-12.652 15.192-18.18ZM82.802 87.83a323.167 323.167 0 0 0-7.227 13.238c-3.184-7.553-5.909-14.98-8.134-22.152c7.304-1.634 15.093-2.97 23.209-3.984a321.524 321.524 0 0 0-7.848 12.897Zm8.081 65.352c-8.385-.936-16.291-2.203-23.593-3.793c2.26-7.3 5.045-14.885 8.298-22.6a321.187 321.187 0 0 0 7.257 13.246c2.594 4.48 5.28 8.868 8.038 13.147Zm37.542 31.03c-5.184-5.592-10.354-11.779-15.403-18.433c4.902.192 9.899.29 14.978.29c5.218 0 10.376-.117 15.453-.343c-4.985 6.774-10.018 12.97-15.028 18.486Zm52.198-57.817c3.422 7.8 6.306 15.345 8.596 22.52c-7.422 1.694-15.436 3.058-23.88 4.071a382.417 382.417 0 0 0 7.859-13.026a347.403 347.403 0 0 0 7.425-13.565Zm-16.898 8.101a358.557 358.557 0 0 1-12.281 19.815a329.4 329.4 0 0 1-23.444.823c-7.967 0-15.716-.248-23.178-.732a310.202 310.202 0 0 1-12.513-19.846h.001a307.41 307.41 0 0 1-10.923-20.627a310.278 310.278 0 0 1 10.89-20.637l-.001.001a307.318 307.318 0 0 1 12.413-19.761c7.613-.576 15.42-.876 23.31-.876H128c7.926 0 15.743.303 23.354.883a329.357 329.357 0 0 1 12.335 19.695a358.489 358.489 0 0 1 11.036 20.54a329.472 329.472 0 0 1-11 20.722Zm22.56-122.124c8.572 4.944 11.906 24.881 6.52 51.026c-.344 1.668-.73 3.367-1.15 5.09c-10.622-2.452-22.155-4.275-34.23-5.408c-7.034-10.017-14.323-19.124-21.64-27.008a160.789 160.789 0 0 1 5.888-5.4c18.9-16.447 36.564-22.941 44.612-18.3ZM128 90.808c12.625 0 22.86 10.235 22.86 22.86s-10.235 22.86-22.86 22.86s-22.86-10.235-22.86-22.86s10.235-22.86 22.86-22.86Z"></path></svg>

Before

Width:  |  Height:  |  Size: 4.0 KiB

View File

@@ -1,456 +0,0 @@
import { useEffect, useState } from "react";
import { Link, useParams } from "react-router-dom";
import { api } from "../api/client";
import type { Contact, ContactListItem, Need, RelationshipTypeValue } from "../types";
import { RELATIONSHIP_TYPES } from "../types";
export function ContactDetail() {
const { id } = useParams<{ id: string }>();
const [contact, setContact] = useState<Contact | null>(null);
const [allNeeds, setAllNeeds] = useState<Need[]>([]);
const [allContacts, setAllContacts] = useState<ContactListItem[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const [newNeedId, setNewNeedId] = useState<number | "">("");
const [newRelContactId, setNewRelContactId] = useState<number | "">("");
const [newRelType, setNewRelType] = useState<RelationshipTypeValue | "">("");
useEffect(() => {
if (!id) return;
Promise.all([
api.contacts.get(Number(id)),
api.needs.list(),
api.contacts.list(),
])
.then(([c, n, contacts]) => {
setContact(c);
setAllNeeds(n);
setAllContacts(contacts.filter((ct) => ct.id !== Number(id)));
})
.catch((err) => setError(err.message))
.finally(() => setLoading(false));
}, [id]);
const handleAddNeed = async () => {
if (!contact || newNeedId === "") return;
try {
await api.contacts.addNeed(contact.id, Number(newNeedId));
const updated = await api.contacts.get(contact.id);
setContact(updated);
setNewNeedId("");
} catch (err) {
setError(err instanceof Error ? err.message : "Failed to add need");
}
};
const handleRemoveNeed = async (needId: number) => {
if (!contact) return;
try {
await api.contacts.removeNeed(contact.id, needId);
const updated = await api.contacts.get(contact.id);
setContact(updated);
} catch (err) {
setError(err instanceof Error ? err.message : "Failed to remove need");
}
};
const handleAddRelationship = async () => {
if (!contact || newRelContactId === "" || newRelType === "") return;
try {
await api.contacts.addRelationship(contact.id, {
related_contact_id: Number(newRelContactId),
relationship_type: newRelType,
});
const updated = await api.contacts.get(contact.id);
setContact(updated);
setNewRelContactId("");
setNewRelType("");
} catch (err) {
setError(
err instanceof Error ? err.message : "Failed to add relationship"
);
}
};
const handleRemoveRelationship = async (relatedContactId: number) => {
if (!contact) return;
try {
await api.contacts.removeRelationship(contact.id, relatedContactId);
const updated = await api.contacts.get(contact.id);
setContact(updated);
} catch (err) {
setError(
err instanceof Error ? err.message : "Failed to remove relationship"
);
}
};
const handleUpdateWeight = async (relatedContactId: number, newWeight: number) => {
if (!contact) return;
try {
await api.contacts.updateRelationship(contact.id, relatedContactId, {
closeness_weight: newWeight,
});
const updated = await api.contacts.get(contact.id);
setContact(updated);
} catch (err) {
setError(
err instanceof Error ? err.message : "Failed to update weight"
);
}
};
if (loading) return <div>Loading...</div>;
if (error) return <div className="error">Error: {error}</div>;
if (!contact) return <div>Contact not found</div>;
const availableNeeds = allNeeds.filter(
(n) => !contact.needs.some((cn) => cn.id === n.id)
);
const getContactName = (contactId: number) => {
const c = allContacts.find((ct) => ct.id === contactId);
return c?.name || `Contact #${contactId}`;
};
const getRelationshipDisplayName = (type: string) => {
const rt = RELATIONSHIP_TYPES.find((r) => r.value === type);
return rt?.displayName || type;
};
// Group relationships by category for display
const groupRelationships = () => {
const familial: typeof contact.related_to = [];
const friends: typeof contact.related_to = [];
const partners: typeof contact.related_to = [];
const professional: typeof contact.related_to = [];
const other: typeof contact.related_to = [];
const familialTypes = ['parent', 'child', 'sibling', 'grandparent', 'grandchild', 'aunt_uncle', 'niece_nephew', 'cousin', 'in_law'];
const friendTypes = ['best_friend', 'close_friend', 'friend', 'acquaintance', 'neighbor'];
const partnerTypes = ['spouse', 'partner'];
const professionalTypes = ['mentor', 'mentee', 'business_partner', 'colleague', 'manager', 'direct_report', 'client'];
for (const rel of contact.related_to) {
if (familialTypes.includes(rel.relationship_type)) {
familial.push(rel);
} else if (friendTypes.includes(rel.relationship_type)) {
friends.push(rel);
} else if (partnerTypes.includes(rel.relationship_type)) {
partners.push(rel);
} else if (professionalTypes.includes(rel.relationship_type)) {
professional.push(rel);
} else {
other.push(rel);
}
}
return { familial, friends, partners, professional, other };
};
const relationshipGroups = groupRelationships();
return (
<div className="id-card">
<div className="id-card-inner">
{/* Header with name and profile pic */}
<div className="id-card-header">
<div className="id-card-header-left">
<h1 className="id-card-title">I.D.: {contact.name}</h1>
</div>
<div className="id-card-header-right">
{contact.profile_pic ? (
<img
src={contact.profile_pic}
alt={`${contact.name}'s profile`}
className="id-profile-pic"
/>
) : (
<div className="id-profile-placeholder">
<span>{contact.name.charAt(0).toUpperCase()}</span>
</div>
)}
<div className="id-card-actions">
<Link to={`/contacts/${contact.id}/edit`} className="btn btn-small">
Edit
</Link>
<Link to="/contacts" className="btn btn-small">
Back
</Link>
</div>
</div>
</div>
<div className="id-card-body">
{/* Left column - Basic info */}
<div className="id-card-left">
{contact.legal_name && (
<div className="id-field">Legal name: {contact.legal_name}</div>
)}
{contact.suffix && (
<div className="id-field">Suffix: {contact.suffix}</div>
)}
{contact.gender && (
<div className="id-field">Gender: {contact.gender}</div>
)}
{contact.age && (
<div className="id-field">Age: {contact.age}</div>
)}
{contact.current_job && (
<div className="id-field">Job: {contact.current_job}</div>
)}
{contact.social_structure_style && (
<div className="id-field">Social style: {contact.social_structure_style}</div>
)}
{contact.self_sufficiency_score !== null && (
<div className="id-field">Self-Sufficiency: {contact.self_sufficiency_score}</div>
)}
{contact.timezone && (
<div className="id-field">Timezone: {contact.timezone}</div>
)}
{contact.safe_conversation_starters && (
<div className="id-field-block">
<span className="id-label">Safe con starters:</span> {contact.safe_conversation_starters}
</div>
)}
{contact.topics_to_avoid && (
<div className="id-field-block">
<span className="id-label">Topics to avoid:</span> {contact.topics_to_avoid}
</div>
)}
{contact.goals && (
<div className="id-field-block">
<span className="id-label">Goals:</span> {contact.goals}
</div>
)}
</div>
{/* Right column - Bio and Relationships */}
<div className="id-card-right">
{contact.bio && (
<div className="id-bio">
<span className="id-label">Bio:</span> {contact.bio}
</div>
)}
<div className="id-relationships">
<h2 className="id-section-title">Relationships</h2>
{relationshipGroups.familial.length > 0 && (
<div className="id-rel-group">
<span className="id-rel-label">Familial:</span>{" "}
{relationshipGroups.familial.map((rel, i) => (
<span key={rel.related_contact_id}>
<Link to={`/contacts/${rel.related_contact_id}`}>
{getContactName(rel.related_contact_id)}
</Link>
<span className="id-rel-type">({getRelationshipDisplayName(rel.relationship_type)})</span>
{i < relationshipGroups.familial.length - 1 && ", "}
</span>
))}
</div>
)}
{relationshipGroups.partners.length > 0 && (
<div className="id-rel-group">
<span className="id-rel-label">Partners:</span>{" "}
{relationshipGroups.partners.map((rel, i) => (
<span key={rel.related_contact_id}>
<Link to={`/contacts/${rel.related_contact_id}`}>
{getContactName(rel.related_contact_id)}
</Link>
{i < relationshipGroups.partners.length - 1 && ", "}
</span>
))}
</div>
)}
{relationshipGroups.friends.length > 0 && (
<div className="id-rel-group">
<span className="id-rel-label">Friends:</span>{" "}
{relationshipGroups.friends.map((rel, i) => (
<span key={rel.related_contact_id}>
<Link to={`/contacts/${rel.related_contact_id}`}>
{getContactName(rel.related_contact_id)}
</Link>
{i < relationshipGroups.friends.length - 1 && ", "}
</span>
))}
</div>
)}
{relationshipGroups.professional.length > 0 && (
<div className="id-rel-group">
<span className="id-rel-label">Professional:</span>{" "}
{relationshipGroups.professional.map((rel, i) => (
<span key={rel.related_contact_id}>
<Link to={`/contacts/${rel.related_contact_id}`}>
{getContactName(rel.related_contact_id)}
</Link>
<span className="id-rel-type">({getRelationshipDisplayName(rel.relationship_type)})</span>
{i < relationshipGroups.professional.length - 1 && ", "}
</span>
))}
</div>
)}
{relationshipGroups.other.length > 0 && (
<div className="id-rel-group">
<span className="id-rel-label">Other:</span>{" "}
{relationshipGroups.other.map((rel, i) => (
<span key={rel.related_contact_id}>
<Link to={`/contacts/${rel.related_contact_id}`}>
{getContactName(rel.related_contact_id)}
</Link>
<span className="id-rel-type">({getRelationshipDisplayName(rel.relationship_type)})</span>
{i < relationshipGroups.other.length - 1 && ", "}
</span>
))}
</div>
)}
{contact.related_from.length > 0 && (
<div className="id-rel-group">
<span className="id-rel-label">Known by:</span>{" "}
{contact.related_from.map((rel, i) => (
<span key={rel.contact_id}>
<Link to={`/contacts/${rel.contact_id}`}>
{getContactName(rel.contact_id)}
</Link>
{i < contact.related_from.length - 1 && ", "}
</span>
))}
</div>
)}
</div>
</div>
</div>
{/* Needs/Warnings at bottom */}
{contact.needs.length > 0 && (
<div className="id-card-warnings">
{contact.needs.map((need) => (
<div key={need.id} className="id-warning">
<span className="warning-dot"></span>
Warning: {need.name}
{need.description && <span className="warning-desc"> - {need.description}</span>}
</div>
))}
</div>
)}
</div>
{/* Management section (expandable) */}
<details className="id-card-manage">
<summary>Manage Contact</summary>
<div className="manage-section">
<h3>Manage Relationships</h3>
<div className="manage-relationships">
{contact.related_to.map((rel) => (
<div key={rel.related_contact_id} className="manage-rel-item">
<Link to={`/contacts/${rel.related_contact_id}`}>
{getContactName(rel.related_contact_id)}
</Link>
<span className="tag">{getRelationshipDisplayName(rel.relationship_type)}</span>
<label className="weight-control">
<span>Closeness:</span>
<input
type="range"
min="1"
max="10"
value={rel.closeness_weight}
onChange={(e) => handleUpdateWeight(rel.related_contact_id, Number(e.target.value))}
/>
<span className="weight-value">{rel.closeness_weight}</span>
</label>
<button
onClick={() => handleRemoveRelationship(rel.related_contact_id)}
className="btn btn-small btn-danger"
>
Remove
</button>
</div>
))}
</div>
{allContacts.length > 0 && (
<div className="add-form">
<select
value={newRelContactId}
onChange={(e) =>
setNewRelContactId(
e.target.value ? Number(e.target.value) : ""
)
}
>
<option value="">Select contact...</option>
{allContacts.map((c) => (
<option key={c.id} value={c.id}>
{c.name}
</option>
))}
</select>
<select
value={newRelType}
onChange={(e) => setNewRelType(e.target.value as RelationshipTypeValue | "")}
>
<option value="">Select relationship type...</option>
{RELATIONSHIP_TYPES.map((rt) => (
<option key={rt.value} value={rt.value}>
{rt.displayName}
</option>
))}
</select>
<button onClick={handleAddRelationship} className="btn btn-primary">
Add Relationship
</button>
</div>
)}
</div>
<div className="manage-section">
<h3>Manage Needs/Warnings</h3>
<ul className="manage-needs-list">
{contact.needs.map((need) => (
<li key={need.id}>
<strong>{need.name}</strong>
{need.description && <span> - {need.description}</span>}
<button
onClick={() => handleRemoveNeed(need.id)}
className="btn btn-small btn-danger"
>
Remove
</button>
</li>
))}
</ul>
{availableNeeds.length > 0 && (
<div className="add-form">
<select
value={newNeedId}
onChange={(e) =>
setNewNeedId(e.target.value ? Number(e.target.value) : "")
}
>
<option value="">Select a need...</option>
{availableNeeds.map((n) => (
<option key={n.id} value={n.id}>
{n.name}
</option>
))}
</select>
<button onClick={handleAddNeed} className="btn btn-primary">
Add Need
</button>
</div>
)}
</div>
</details>
</div>
);
}

View File

@@ -1,325 +0,0 @@
import { useEffect, useState } from "react";
import { useNavigate, useParams } from "react-router-dom";
import { api } from "../api/client";
import type { ContactCreate, Need } from "../types";
export function ContactForm() {
const { id } = useParams<{ id: string }>();
const navigate = useNavigate();
const isEdit = Boolean(id);
const [allNeeds, setAllNeeds] = useState<Need[]>([]);
const [loading, setLoading] = useState(isEdit);
const [error, setError] = useState<string | null>(null);
const [submitting, setSubmitting] = useState(false);
const [form, setForm] = useState<ContactCreate>({
name: "",
age: null,
bio: null,
current_job: null,
gender: null,
goals: null,
legal_name: null,
profile_pic: null,
safe_conversation_starters: null,
self_sufficiency_score: null,
social_structure_style: null,
ssn: null,
suffix: null,
timezone: null,
topics_to_avoid: null,
need_ids: [],
});
useEffect(() => {
const loadData = async () => {
try {
const needs = await api.needs.list();
setAllNeeds(needs);
if (id) {
const contact = await api.contacts.get(Number(id));
setForm({
name: contact.name,
age: contact.age,
bio: contact.bio,
current_job: contact.current_job,
gender: contact.gender,
goals: contact.goals,
legal_name: contact.legal_name,
profile_pic: contact.profile_pic,
safe_conversation_starters: contact.safe_conversation_starters,
self_sufficiency_score: contact.self_sufficiency_score,
social_structure_style: contact.social_structure_style,
ssn: contact.ssn,
suffix: contact.suffix,
timezone: contact.timezone,
topics_to_avoid: contact.topics_to_avoid,
need_ids: contact.needs.map((n) => n.id),
});
}
} catch (err) {
setError(err instanceof Error ? err.message : "Failed to load data");
} finally {
setLoading(false);
}
};
loadData();
}, [id]);
const handleSubmit = async (e: React.FormEvent) => {
e.preventDefault();
setSubmitting(true);
setError(null);
try {
if (isEdit) {
await api.contacts.update(Number(id), form);
navigate(`/contacts/${id}`);
} else {
const created = await api.contacts.create(form);
navigate(`/contacts/${created.id}`);
}
} catch (err) {
setError(err instanceof Error ? err.message : "Save failed");
setSubmitting(false);
}
};
const updateField = <K extends keyof ContactCreate>(
field: K,
value: ContactCreate[K]
) => {
setForm((prev) => ({ ...prev, [field]: value }));
};
const toggleNeed = (needId: number) => {
setForm((prev) => ({
...prev,
need_ids: prev.need_ids?.includes(needId)
? prev.need_ids.filter((id) => id !== needId)
: [...(prev.need_ids || []), needId],
}));
};
if (loading) return <div>Loading...</div>;
return (
<div className="contact-form">
<h1>{isEdit ? "Edit Contact" : "New Contact"}</h1>
{error && <div className="error">{error}</div>}
<form onSubmit={handleSubmit}>
<div className="form-group">
<label htmlFor="name">Name *</label>
<input
id="name"
type="text"
value={form.name}
onChange={(e) => updateField("name", e.target.value)}
required
/>
</div>
<div className="form-row">
<div className="form-group">
<label htmlFor="legal_name">Legal Name</label>
<input
id="legal_name"
type="text"
value={form.legal_name || ""}
onChange={(e) =>
updateField("legal_name", e.target.value || null)
}
/>
</div>
<div className="form-group">
<label htmlFor="suffix">Suffix</label>
<input
id="suffix"
type="text"
value={form.suffix || ""}
onChange={(e) => updateField("suffix", e.target.value || null)}
/>
</div>
</div>
<div className="form-row">
<div className="form-group">
<label htmlFor="age">Age</label>
<input
id="age"
type="number"
value={form.age ?? ""}
onChange={(e) =>
updateField("age", e.target.value ? Number(e.target.value) : null)
}
/>
</div>
<div className="form-group">
<label htmlFor="gender">Gender</label>
<input
id="gender"
type="text"
value={form.gender || ""}
onChange={(e) => updateField("gender", e.target.value || null)}
/>
</div>
</div>
<div className="form-group">
<label htmlFor="current_job">Current Job</label>
<input
id="current_job"
type="text"
value={form.current_job || ""}
onChange={(e) =>
updateField("current_job", e.target.value || null)
}
/>
</div>
<div className="form-group">
<label htmlFor="timezone">Timezone</label>
<input
id="timezone"
type="text"
value={form.timezone || ""}
onChange={(e) => updateField("timezone", e.target.value || null)}
/>
</div>
<div className="form-group">
<label htmlFor="profile_pic">Profile Picture URL</label>
<input
id="profile_pic"
type="url"
placeholder="https://example.com/photo.jpg"
value={form.profile_pic || ""}
onChange={(e) => updateField("profile_pic", e.target.value || null)}
/>
</div>
<div className="form-group">
<label htmlFor="bio">Bio</label>
<textarea
id="bio"
value={form.bio || ""}
onChange={(e) => updateField("bio", e.target.value || null)}
rows={3}
/>
</div>
<div className="form-group">
<label htmlFor="goals">Goals</label>
<textarea
id="goals"
value={form.goals || ""}
onChange={(e) => updateField("goals", e.target.value || null)}
rows={3}
/>
</div>
<div className="form-group">
<label htmlFor="social_structure_style">Social Structure Style</label>
<input
id="social_structure_style"
type="text"
value={form.social_structure_style || ""}
onChange={(e) =>
updateField("social_structure_style", e.target.value || null)
}
/>
</div>
<div className="form-group">
<label htmlFor="self_sufficiency_score">
Self-Sufficiency Score (1-10)
</label>
<input
id="self_sufficiency_score"
type="number"
min="1"
max="10"
value={form.self_sufficiency_score ?? ""}
onChange={(e) =>
updateField(
"self_sufficiency_score",
e.target.value ? Number(e.target.value) : null
)
}
/>
</div>
<div className="form-group">
<label htmlFor="safe_conversation_starters">
Safe Conversation Starters
</label>
<textarea
id="safe_conversation_starters"
value={form.safe_conversation_starters || ""}
onChange={(e) =>
updateField("safe_conversation_starters", e.target.value || null)
}
rows={2}
/>
</div>
<div className="form-group">
<label htmlFor="topics_to_avoid">Topics to Avoid</label>
<textarea
id="topics_to_avoid"
value={form.topics_to_avoid || ""}
onChange={(e) =>
updateField("topics_to_avoid", e.target.value || null)
}
rows={2}
/>
</div>
<div className="form-group">
<label htmlFor="ssn">SSN</label>
<input
id="ssn"
type="text"
value={form.ssn || ""}
onChange={(e) => updateField("ssn", e.target.value || null)}
/>
</div>
{allNeeds.length > 0 && (
<div className="form-group">
<label>Needs/Accommodations</label>
<div className="checkbox-group">
{allNeeds.map((need) => (
<label key={need.id} className="checkbox-label">
<input
type="checkbox"
checked={form.need_ids?.includes(need.id) || false}
onChange={() => toggleNeed(need.id)}
/>
{need.name}
</label>
))}
</div>
</div>
)}
<div className="form-actions">
<button type="submit" className="btn btn-primary" disabled={submitting}>
{submitting ? "Saving..." : "Save"}
</button>
<button
type="button"
className="btn"
onClick={() => navigate(isEdit ? `/contacts/${id}` : "/contacts")}
>
Cancel
</button>
</div>
</form>
</div>
);
}

View File

@@ -1,79 +0,0 @@
import { useEffect, useState } from "react";
import { Link } from "react-router-dom";
import { api } from "../api/client";
import type { ContactListItem } from "../types";
export function ContactList() {
const [contacts, setContacts] = useState<ContactListItem[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
useEffect(() => {
api.contacts
.list()
.then(setContacts)
.catch((err) => setError(err.message))
.finally(() => setLoading(false));
}, []);
const handleDelete = async (id: number) => {
if (!confirm("Delete this contact?")) return;
try {
await api.contacts.delete(id);
setContacts((prev) => prev.filter((c) => c.id !== id));
} catch (err) {
setError(err instanceof Error ? err.message : "Delete failed");
}
};
if (loading) return <div>Loading...</div>;
if (error) return <div className="error">Error: {error}</div>;
return (
<div className="contact-list">
<div className="header">
<h1>Contacts</h1>
<Link to="/contacts/new" className="btn btn-primary">
Add Contact
</Link>
</div>
{contacts.length === 0 ? (
<p>No contacts yet.</p>
) : (
<table>
<thead>
<tr>
<th>Name</th>
<th>Job</th>
<th>Timezone</th>
<th>Actions</th>
</tr>
</thead>
<tbody>
{contacts.map((contact) => (
<tr key={contact.id}>
<td>
<Link to={`/contacts/${contact.id}`}>{contact.name}</Link>
</td>
<td>{contact.current_job || "-"}</td>
<td>{contact.timezone || "-"}</td>
<td>
<Link to={`/contacts/${contact.id}/edit`} className="btn">
Edit
</Link>
<button
onClick={() => handleDelete(contact.id)}
className="btn btn-danger"
>
Delete
</button>
</td>
</tr>
))}
</tbody>
</table>
)}
</div>
);
}

View File

@@ -1,117 +0,0 @@
import { useEffect, useState } from "react";
import { api } from "../api/client";
import type { Need, NeedCreate } from "../types";
export function NeedList() {
const [needs, setNeeds] = useState<Need[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const [showForm, setShowForm] = useState(false);
const [form, setForm] = useState<NeedCreate>({ name: "", description: null });
const [submitting, setSubmitting] = useState(false);
useEffect(() => {
api.needs
.list()
.then(setNeeds)
.catch((err) => setError(err.message))
.finally(() => setLoading(false));
}, []);
const handleSubmit = async (e: React.FormEvent) => {
e.preventDefault();
if (!form.name.trim()) return;
setSubmitting(true);
try {
const created = await api.needs.create(form);
setNeeds((prev) => [...prev, created]);
setForm({ name: "", description: null });
setShowForm(false);
} catch (err) {
setError(err instanceof Error ? err.message : "Create failed");
} finally {
setSubmitting(false);
}
};
const handleDelete = async (id: number) => {
if (!confirm("Delete this need?")) return;
try {
await api.needs.delete(id);
setNeeds((prev) => prev.filter((n) => n.id !== id));
} catch (err) {
setError(err instanceof Error ? err.message : "Delete failed");
}
};
if (loading) return <div>Loading...</div>;
return (
<div className="need-list">
<div className="header">
<h1>Needs / Accommodations</h1>
<button
onClick={() => setShowForm(!showForm)}
className="btn btn-primary"
>
{showForm ? "Cancel" : "Add Need"}
</button>
</div>
{error && <div className="error">{error}</div>}
{showForm && (
<form onSubmit={handleSubmit} className="need-form">
<div className="form-group">
<label htmlFor="name">Name *</label>
<input
id="name"
type="text"
value={form.name}
onChange={(e) => setForm({ ...form, name: e.target.value })}
placeholder="e.g., Light Sensitive, ADHD"
required
/>
</div>
<div className="form-group">
<label htmlFor="description">Description</label>
<textarea
id="description"
value={form.description || ""}
onChange={(e) =>
setForm({ ...form, description: e.target.value || null })
}
placeholder="Optional description..."
rows={2}
/>
</div>
<button type="submit" className="btn btn-primary" disabled={submitting}>
{submitting ? "Creating..." : "Create"}
</button>
</form>
)}
{needs.length === 0 ? (
<p>No needs defined yet.</p>
) : (
<ul className="need-items">
{needs.map((need) => (
<li key={need.id}>
<div className="need-info">
<strong>{need.name}</strong>
{need.description && <p>{need.description}</p>}
</div>
<button
onClick={() => handleDelete(need.id)}
className="btn btn-danger"
>
Delete
</button>
</li>
))}
</ul>
)}
</div>
);
}

View File

@@ -1,330 +0,0 @@
import { useEffect, useRef, useState } from "react";
import { api } from "../api/client";
import type { GraphData, GraphEdge, GraphNode } from "../types";
import { RELATIONSHIP_TYPES } from "../types";
interface SimNode extends GraphNode {
x: number;
y: number;
vx: number;
vy: number;
}
interface SimEdge extends GraphEdge {
sourceNode: SimNode;
targetNode: SimNode;
}
export function RelationshipGraph() {
const canvasRef = useRef<HTMLCanvasElement>(null);
const [data, setData] = useState<GraphData | null>(null);
const [error, setError] = useState<string | null>(null);
const [loading, setLoading] = useState(true);
const [selectedNode, setSelectedNode] = useState<SimNode | null>(null);
const nodesRef = useRef<SimNode[]>([]);
const edgesRef = useRef<SimEdge[]>([]);
const dragNodeRef = useRef<SimNode | null>(null);
const animationRef = useRef<number>(0);
useEffect(() => {
api.graph.get()
.then(setData)
.catch((err) => setError(err.message))
.finally(() => setLoading(false));
}, []);
useEffect(() => {
if (!data || !canvasRef.current) return;
const canvas = canvasRef.current;
const maybeCtx = canvas.getContext("2d");
if (!maybeCtx) return;
const ctx: CanvasRenderingContext2D = maybeCtx;
const width = canvas.width;
const height = canvas.height;
const centerX = width / 2;
const centerY = height / 2;
// Initialize nodes with random positions
const nodes: SimNode[] = data.nodes.map((node) => ({
...node,
x: centerX + (Math.random() - 0.5) * 300,
y: centerY + (Math.random() - 0.5) * 300,
vx: 0,
vy: 0,
}));
nodesRef.current = nodes;
const nodeMap = new Map(nodes.map((n) => [n.id, n]));
// Create edges with node references
const edges: SimEdge[] = data.edges
.map((edge) => {
const sourceNode = nodeMap.get(edge.source);
const targetNode = nodeMap.get(edge.target);
if (!sourceNode || !targetNode) return null;
return { ...edge, sourceNode, targetNode };
})
.filter((e): e is SimEdge => e !== null);
edgesRef.current = edges;
// Force simulation parameters
const repulsion = 5000;
const springStrength = 0.05;
const baseSpringLength = 150;
const damping = 0.9;
const centerPull = 0.01;
function simulate() {
const nodes = nodesRef.current;
const edges = edgesRef.current;
// Reset forces
for (const node of nodes) {
node.vx = 0;
node.vy = 0;
}
// Repulsion between all nodes
for (let i = 0; i < nodes.length; i++) {
for (let j = i + 1; j < nodes.length; j++) {
const dx = nodes[j].x - nodes[i].x;
const dy = nodes[j].y - nodes[i].y;
const dist = Math.sqrt(dx * dx + dy * dy) || 1;
const force = repulsion / (dist * dist);
const fx = (dx / dist) * force;
const fy = (dy / dist) * force;
nodes[i].vx -= fx;
nodes[i].vy -= fy;
nodes[j].vx += fx;
nodes[j].vy += fy;
}
}
// Spring forces for edges - closer relationships = shorter springs
// Weight is 1-10, normalize to 0-1 for calculations
for (const edge of edges) {
const dx = edge.targetNode.x - edge.sourceNode.x;
const dy = edge.targetNode.y - edge.sourceNode.y;
const dist = Math.sqrt(dx * dx + dy * dy) || 1;
// Higher weight (1-10) = shorter ideal length
// Normalize: weight 10 -> 0.5x length, weight 1 -> 1.4x length
const normalizedWeight = edge.closeness_weight / 10;
const idealLength = baseSpringLength * (1.5 - normalizedWeight);
const displacement = dist - idealLength;
const force = springStrength * displacement;
const fx = (dx / dist) * force;
const fy = (dy / dist) * force;
edge.sourceNode.vx += fx;
edge.sourceNode.vy += fy;
edge.targetNode.vx -= fx;
edge.targetNode.vy -= fy;
}
// Pull toward center
for (const node of nodes) {
node.vx += (centerX - node.x) * centerPull;
node.vy += (centerY - node.y) * centerPull;
}
// Apply velocities with damping (skip dragged node)
for (const node of nodes) {
if (node === dragNodeRef.current) continue;
node.x += node.vx * damping;
node.y += node.vy * damping;
// Keep within bounds
node.x = Math.max(30, Math.min(width - 30, node.x));
node.y = Math.max(30, Math.min(height - 30, node.y));
}
}
function getEdgeColor(weight: number): string {
// Interpolate from light gray (distant) to dark blue (close)
// weight is 1-10, normalize to 0-1
const normalized = weight / 10;
const hue = 220;
const saturation = 70;
const lightness = 80 - normalized * 40;
return `hsl(${hue}, ${saturation}%, ${lightness}%)`;
}
function draw(context: CanvasRenderingContext2D) {
const nodes = nodesRef.current;
const edges = edgesRef.current;
context.clearRect(0, 0, width, height);
// Draw edges
for (const edge of edges) {
// Weight is 1-10, scale line width accordingly
const lineWidth = 1 + (edge.closeness_weight / 10) * 3;
context.strokeStyle = getEdgeColor(edge.closeness_weight);
context.lineWidth = lineWidth;
context.beginPath();
context.moveTo(edge.sourceNode.x, edge.sourceNode.y);
context.lineTo(edge.targetNode.x, edge.targetNode.y);
context.stroke();
// Draw relationship type label at midpoint
const midX = (edge.sourceNode.x + edge.targetNode.x) / 2;
const midY = (edge.sourceNode.y + edge.targetNode.y) / 2;
context.fillStyle = "#666";
context.font = "10px sans-serif";
context.textAlign = "center";
const typeInfo = RELATIONSHIP_TYPES.find(t => t.value === edge.relationship_type);
const label = typeInfo?.displayName || edge.relationship_type;
context.fillText(label, midX, midY - 5);
}
// Draw nodes
for (const node of nodes) {
const isSelected = node === selectedNode;
const radius = isSelected ? 25 : 20;
// Node circle
context.beginPath();
context.arc(node.x, node.y, radius, 0, Math.PI * 2);
context.fillStyle = isSelected ? "#0066cc" : "#fff";
context.fill();
context.strokeStyle = "#0066cc";
context.lineWidth = 2;
context.stroke();
// Node label
context.fillStyle = isSelected ? "#fff" : "#333";
context.font = "12px sans-serif";
context.textAlign = "center";
context.textBaseline = "middle";
const name = node.name.length > 10 ? node.name.slice(0, 9) + "…" : node.name;
context.fillText(name, node.x, node.y);
}
}
function animate() {
simulate();
draw(ctx);
animationRef.current = requestAnimationFrame(animate);
}
animate();
return () => {
cancelAnimationFrame(animationRef.current);
};
}, [data, selectedNode]);
// Mouse interaction handlers
useEffect(() => {
const canvas = canvasRef.current;
if (!canvas) return;
function getNodeAtPosition(x: number, y: number): SimNode | null {
for (const node of nodesRef.current) {
const dx = x - node.x;
const dy = y - node.y;
if (dx * dx + dy * dy < 400) {
return node;
}
}
return null;
}
function handleMouseDown(e: MouseEvent) {
const rect = canvas!.getBoundingClientRect();
const x = e.clientX - rect.left;
const y = e.clientY - rect.top;
const node = getNodeAtPosition(x, y);
if (node) {
dragNodeRef.current = node;
setSelectedNode(node);
}
}
function handleMouseMove(e: MouseEvent) {
if (!dragNodeRef.current) return;
const rect = canvas!.getBoundingClientRect();
dragNodeRef.current.x = e.clientX - rect.left;
dragNodeRef.current.y = e.clientY - rect.top;
}
function handleMouseUp() {
dragNodeRef.current = null;
}
canvas.addEventListener("mousedown", handleMouseDown);
canvas.addEventListener("mousemove", handleMouseMove);
canvas.addEventListener("mouseup", handleMouseUp);
canvas.addEventListener("mouseleave", handleMouseUp);
return () => {
canvas.removeEventListener("mousedown", handleMouseDown);
canvas.removeEventListener("mousemove", handleMouseMove);
canvas.removeEventListener("mouseup", handleMouseUp);
canvas.removeEventListener("mouseleave", handleMouseUp);
};
}, []);
if (loading) return <p>Loading graph...</p>;
if (error) return <div className="error">{error}</div>;
if (!data) return <p>No data available</p>;
return (
<div className="graph-container">
<div className="header">
<h1>Relationship Graph</h1>
</div>
<p className="graph-hint">
Drag nodes to reposition. Closer relationships have shorter, darker edges.
</p>
<canvas
ref={canvasRef}
width={900}
height={600}
style={{
border: "1px solid var(--color-border)",
borderRadius: "8px",
background: "var(--color-bg)",
cursor: "grab",
}}
/>
{selectedNode && (
<div className="selected-info">
<h3>{selectedNode.name}</h3>
{selectedNode.current_job && <p>Job: {selectedNode.current_job}</p>}
<a href={`/contacts/${selectedNode.id}`}>View details</a>
</div>
)}
<div className="legend">
<h4>Relationship Closeness (1-10)</h4>
<div className="legend-items">
<div className="legend-item">
<span className="legend-line" style={{ background: getEdgeColorCSS(10), height: "4px" }}></span>
<span>10 - Very Close (Spouse, Partner)</span>
</div>
<div className="legend-item">
<span className="legend-line" style={{ background: getEdgeColorCSS(7), height: "3px" }}></span>
<span>7 - Close (Family, Best Friend)</span>
</div>
<div className="legend-item">
<span className="legend-line" style={{ background: getEdgeColorCSS(4), height: "2px" }}></span>
<span>4 - Moderate (Friend, Colleague)</span>
</div>
<div className="legend-item">
<span className="legend-line" style={{ background: getEdgeColorCSS(2), height: "1px" }}></span>
<span>2 - Distant (Acquaintance)</span>
</div>
</div>
</div>
</div>
);
}
function getEdgeColorCSS(weight: number): string {
// weight is 1-10, normalize to 0-1
const normalized = weight / 10;
const hue = 220;
const saturation = 70;
const lightness = 80 - normalized * 40;
return `hsl(${hue}, ${saturation}%, ${lightness}%)`;
}

View File

@@ -1,62 +0,0 @@
:root {
/* Light theme (default) */
--color-bg: #f5f5f5;
--color-bg-card: #ffffff;
--color-bg-hover: #f0f0f0;
--color-bg-muted: #f9f9f9;
--color-bg-error: #ffe0e0;
--color-text: #333333;
--color-text-muted: #666666;
--color-text-error: #cc0000;
--color-border: #dddddd;
--color-border-light: #eeeeee;
--color-border-lighter: #f0f0f0;
--color-primary: #0066cc;
--color-primary-hover: #0055aa;
--color-danger: #cc3333;
--color-danger-hover: #aa2222;
--color-tag-bg: #e0e0e0;
--shadow: 0 1px 3px rgba(0, 0, 0, 0.1);
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
line-height: 1.5;
font-weight: 400;
color: var(--color-text);
background-color: var(--color-bg);
font-synthesis: none;
text-rendering: optimizeLegibility;
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
[data-theme="dark"] {
--color-bg: #1a1a1a;
--color-bg-card: #2d2d2d;
--color-bg-hover: #3d3d3d;
--color-bg-muted: #252525;
--color-bg-error: #4a2020;
--color-text: #e0e0e0;
--color-text-muted: #a0a0a0;
--color-text-error: #ff6b6b;
--color-border: #404040;
--color-border-light: #353535;
--color-border-lighter: #303030;
--color-primary: #4da6ff;
--color-primary-hover: #7dbfff;
--color-danger: #ff6b6b;
--color-danger-hover: #ff8a8a;
--color-tag-bg: #404040;
--shadow: 0 1px 3px rgba(0, 0, 0, 0.3);
}

View File

@@ -1,13 +0,0 @@
import { StrictMode } from "react";
import { createRoot } from "react-dom/client";
import { BrowserRouter } from "react-router-dom";
import App from "./App.tsx";
import "./index.css";
createRoot(document.getElementById("root")!).render(
<StrictMode>
<BrowserRouter>
<App />
</BrowserRouter>
</StrictMode>
);

View File

@@ -1,155 +0,0 @@
export interface Need {
id: number;
name: string;
description: string | null;
}
export interface NeedCreate {
name: string;
description?: string | null;
}
export const RELATIONSHIP_TYPES = [
{ value: 'spouse', displayName: 'Spouse', defaultWeight: 10 },
{ value: 'partner', displayName: 'Partner', defaultWeight: 10 },
{ value: 'parent', displayName: 'Parent', defaultWeight: 9 },
{ value: 'child', displayName: 'Child', defaultWeight: 9 },
{ value: 'sibling', displayName: 'Sibling', defaultWeight: 9 },
{ value: 'best_friend', displayName: 'Best Friend', defaultWeight: 8 },
{ value: 'grandparent', displayName: 'Grandparent', defaultWeight: 7 },
{ value: 'grandchild', displayName: 'Grandchild', defaultWeight: 7 },
{ value: 'aunt_uncle', displayName: 'Aunt/Uncle', defaultWeight: 7 },
{ value: 'niece_nephew', displayName: 'Niece/Nephew', defaultWeight: 7 },
{ value: 'cousin', displayName: 'Cousin', defaultWeight: 7 },
{ value: 'in_law', displayName: 'In-Law', defaultWeight: 7 },
{ value: 'close_friend', displayName: 'Close Friend', defaultWeight: 6 },
{ value: 'friend', displayName: 'Friend', defaultWeight: 6 },
{ value: 'mentor', displayName: 'Mentor', defaultWeight: 5 },
{ value: 'mentee', displayName: 'Mentee', defaultWeight: 5 },
{ value: 'business_partner', displayName: 'Business Partner', defaultWeight: 5 },
{ value: 'colleague', displayName: 'Colleague', defaultWeight: 4 },
{ value: 'manager', displayName: 'Manager', defaultWeight: 4 },
{ value: 'direct_report', displayName: 'Direct Report', defaultWeight: 4 },
{ value: 'client', displayName: 'Client', defaultWeight: 4 },
{ value: 'acquaintance', displayName: 'Acquaintance', defaultWeight: 3 },
{ value: 'neighbor', displayName: 'Neighbor', defaultWeight: 3 },
{ value: 'ex', displayName: 'Ex', defaultWeight: 2 },
{ value: 'other', displayName: 'Other', defaultWeight: 2 },
] as const;
export type RelationshipTypeValue = typeof RELATIONSHIP_TYPES[number]['value'];
export interface ContactRelationship {
contact_id: number;
related_contact_id: number;
relationship_type: string;
closeness_weight: number;
}
export interface ContactRelationshipCreate {
related_contact_id: number;
relationship_type: RelationshipTypeValue;
closeness_weight?: number;
}
export interface ContactRelationshipUpdate {
relationship_type?: RelationshipTypeValue;
closeness_weight?: number;
}
export interface GraphNode {
id: number;
name: string;
current_job: string | null;
}
export interface GraphEdge {
source: number;
target: number;
relationship_type: string;
closeness_weight: number;
}
export interface GraphData {
nodes: GraphNode[];
edges: GraphEdge[];
}
export interface Contact {
id: number;
name: string;
age: number | null;
bio: string | null;
current_job: string | null;
gender: string | null;
goals: string | null;
legal_name: string | null;
profile_pic: string | null;
safe_conversation_starters: string | null;
self_sufficiency_score: number | null;
social_structure_style: string | null;
ssn: string | null;
suffix: string | null;
timezone: string | null;
topics_to_avoid: string | null;
needs: Need[];
related_to: ContactRelationship[];
related_from: ContactRelationship[];
}
export interface ContactListItem {
id: number;
name: string;
age: number | null;
bio: string | null;
current_job: string | null;
gender: string | null;
goals: string | null;
legal_name: string | null;
profile_pic: string | null;
safe_conversation_starters: string | null;
self_sufficiency_score: number | null;
social_structure_style: string | null;
ssn: string | null;
suffix: string | null;
timezone: string | null;
topics_to_avoid: string | null;
}
export interface ContactCreate {
name: string;
age?: number | null;
bio?: string | null;
current_job?: string | null;
gender?: string | null;
goals?: string | null;
legal_name?: string | null;
profile_pic?: string | null;
safe_conversation_starters?: string | null;
self_sufficiency_score?: number | null;
social_structure_style?: string | null;
ssn?: string | null;
suffix?: string | null;
timezone?: string | null;
topics_to_avoid?: string | null;
need_ids?: number[];
}
export interface ContactUpdate {
name?: string | null;
age?: number | null;
bio?: string | null;
current_job?: string | null;
gender?: string | null;
goals?: string | null;
legal_name?: string | null;
profile_pic?: string | null;
safe_conversation_starters?: string | null;
self_sufficiency_score?: number | null;
social_structure_style?: string | null;
ssn?: string | null;
suffix?: string | null;
timezone?: string | null;
topics_to_avoid?: string | null;
need_ids?: number[] | null;
}

View File

@@ -1,28 +0,0 @@
{
"compilerOptions": {
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.app.tsbuildinfo",
"target": "ES2022",
"useDefineForClassFields": true,
"lib": ["ES2022", "DOM", "DOM.Iterable"],
"module": "ESNext",
"types": ["vite/client"],
"skipLibCheck": true,
/* Bundler mode */
"moduleResolution": "bundler",
"allowImportingTsExtensions": true,
"verbatimModuleSyntax": true,
"moduleDetection": "force",
"noEmit": true,
"jsx": "react-jsx",
/* Linting */
"strict": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"erasableSyntaxOnly": true,
"noFallthroughCasesInSwitch": true,
"noUncheckedSideEffectImports": true
},
"include": ["src"]
}

View File

@@ -1,7 +0,0 @@
{
"files": [],
"references": [
{ "path": "./tsconfig.app.json" },
{ "path": "./tsconfig.node.json" }
]
}

View File

@@ -1,26 +0,0 @@
{
"compilerOptions": {
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.node.tsbuildinfo",
"target": "ES2023",
"lib": ["ES2023"],
"module": "ESNext",
"types": ["node"],
"skipLibCheck": true,
/* Bundler mode */
"moduleResolution": "bundler",
"allowImportingTsExtensions": true,
"verbatimModuleSyntax": true,
"moduleDetection": "force",
"noEmit": true,
/* Linting */
"strict": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"erasableSyntaxOnly": true,
"noFallthroughCasesInSwitch": true,
"noUncheckedSideEffectImports": true
},
"include": ["vite.config.ts"]
}

View File

@@ -1,11 +0,0 @@
import { defineConfig } from "vite";
import react from "@vitejs/plugin-react";
export default defineConfig({
plugins: [react()],
server: {
proxy: {
"/api": "http://localhost:8000",
},
},
});

View File

@@ -1,48 +1,16 @@
{ inputs, ... }:
{
{inputs, ...}: {
# When applied, the stable nixpkgs set (declared in the flake inputs) will be accessible through 'pkgs.stable'
stable = final: _prev: {
stable = import inputs.nixpkgs-stable {
system = final.stdenv.hostPlatform.system;
system = final.system;
config.allowUnfree = true;
};
};
# When applied, the master nixpkgs set (declared in the flake inputs) will be accessible through 'pkgs.master'
master = final: _prev: {
master = import inputs.nixpkgs-master {
system = final.stdenv.hostPlatform.system;
system = final.system;
config.allowUnfree = true;
};
};
python-env = final: _prev: {
my_python = final.python314.withPackages (
ps: with ps; [
alembic
apprise
apscheduler
fastapi
fastapi-cli
httpx
mypy
polars
psycopg
pydantic
pyfakefs
pytest
pytest-cov
pytest-mock
pytest-xdist
requests
ruff
scalene
sqlalchemy
sqlalchemy
textual
tinytuya
typer
types-requests
]
);
};
}

View File

@@ -1,90 +0,0 @@
[project]
name = "system_tools"
version = "0.1.0"
description = ""
authors = [{ name = "Richie Cahill", email = "richie@tmmworkshop.com" }]
requires-python = "~=3.13.0"
readme = "README.md"
license = "MIT"
# these dependencies are a best effort and aren't guaranteed to work
dependencies = ["apprise", "apscheduler", "httpx", "polars", "pydantic", "pyyaml", "requests", "typer"]
[dependency-groups]
dev = [
"mypy",
"pyfakefs",
"pytest-cov",
"pytest-mock",
"pytest-xdist",
"pytest",
"ruff",
"types-requests",
]
[tool.ruff]
target-version = "py313"
line-length = 120
lint.select = ["ALL"]
lint.ignore = [
"G004", # (PERM) This is a performers nit
"COM812", # (TEMP) conflicts when used with the formatter
"ISC001", # (TEMP) conflicts when used with the formatter
"S603", # (PERM) This is known to cause a false positive
]
[tool.ruff.lint.per-file-ignores]
"tests/**" = [
"S101", # (perm) pytest needs asserts
]
"python/stuff/**" = [
"T201", # (perm) I don't care about print statements dir
]
"python/testing/**" = [
"T201", # (perm) I don't care about print statements dir
"ERA001", # (perm) I don't care about print statements dir
]
"python/splendor/**" = [
"S311", # (perm) there is no security issue here
"T201", # (perm) I don't care about print statements dir
"PLR2004", # (temps) need to think about this
]
"python/orm/**" = [
"TC003", # (perm) this creates issues because sqlalchemy uses these at runtime
]
"python/congress_tracker/**" = [
"TC003", # (perm) this creates issues because sqlalchemy uses these at runtime
]
"python/eval_warnings/**" = [
"S607", # (perm) gh and git are expected on PATH in the runner environment
]
"python/alembic/**" = [
"INP001", # (perm) this creates LSP issues for alembic
]
[tool.ruff.lint.pydocstyle]
convention = "google"
[tool.ruff.lint.flake8-builtins]
builtins-ignorelist = ["id"]
[tool.ruff.lint.pylint]
max-args = 9
[tool.coverage.run]
source = ["system_tools"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"raise NotImplementedError",
"if __name__ == \"__main__\":",
]
[tool.pytest.ini_options]
addopts = "-n auto -ra"
# --cov=system_tools --cov-report=term-missing --cov-report=xml --cov-report=html --cov-branch

View File

@@ -1 +0,0 @@
"""Server Tools."""

View File

@@ -1,109 +0,0 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
# Use forward slashes (/) also on windows to provide an os agnostic path
script_location = python/alembic
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
file_template = %%(year)d_%%(month).2d_%%(day).2d-%%(slug)s_%%(rev)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python>=3.9 or backports.zoneinfo library and tzdata library.
# Any required deps can installed by adding `alembic[tz]` to the pip requirements
# string value is passed to ZoneInfo()
# leave blank for localtime
# timezone =
# max length of characters to apply to the "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to alembic/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "version_path_separator" below.
# version_locations = %(here)s/bar:%(here)s/bat:alembic/versions
# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
# version_path_separator = newline
#
# Use os.pathsep. Default configuration used for new projects.
version_path_separator = os
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
sqlalchemy.url = driver://user:pass@localhost/dbname
revision_environment = true
[post_write_hooks]
hooks = dynamic_schema,ruff
dynamic_schema.type = dynamic_schema
ruff.type = ruff
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

View File

@@ -1,93 +0,0 @@
"""Alembic."""
from __future__ import annotations
import logging
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
from alembic import context
from alembic.script import write_hooks
from python.common import bash_wrapper
from python.orm import RichieBase
from python.orm.base import get_postgres_engine
if TYPE_CHECKING:
from collections.abc import MutableMapping
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
target_metadata = RichieBase.metadata
logging.basicConfig(
level="DEBUG",
datefmt="%Y-%m-%dT%H:%M:%S%z",
format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s",
handlers=[logging.StreamHandler(sys.stdout)],
)
@write_hooks.register("dynamic_schema")
def dynamic_schema(filename: str, _options: dict[Any, Any]) -> None:
"""Dynamic schema."""
original_file = Path(filename).read_text()
dynamic_schema_file_part1 = original_file.replace(f"schema='{RichieBase.schema_name}'", "schema=schema")
dynamic_schema_file = dynamic_schema_file_part1.replace(f"'{RichieBase.schema_name}.", "f'{schema}.")
Path(filename).write_text(dynamic_schema_file)
@write_hooks.register("ruff")
def ruff_check_and_format(filename: str, _options: dict[Any, Any]) -> None:
"""Docstring for ruff_check_and_format."""
bash_wrapper(f"ruff check --fix {filename}")
bash_wrapper(f"ruff format {filename}")
def include_name(
name: str | None,
type_: Literal["schema", "table", "column", "index", "unique_constraint", "foreign_key_constraint"],
_parent_names: MutableMapping[Literal["schema_name", "table_name", "schema_qualified_table_name"], str | None],
) -> bool:
"""This filter table to be included in the migration.
Args:
name (str): The name of the table.
type_ (str): The type of the table.
parent_names (list[str]): The names of the parent tables.
Returns:
bool: True if the table should be included, False otherwise.
"""
if type_ == "schema":
return name == target_metadata.schema
return True
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = get_postgres_engine()
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
include_schemas=True,
version_table_schema=RichieBase.schema_name,
include_name=include_name,
)
with context.begin_transaction():
context.run_migrations()
run_migrations_online()

View File

@@ -1,36 +0,0 @@
"""${message}.
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import sqlalchemy as sa
from alembic import op
from python.orm import RichieBase
if TYPE_CHECKING:
from collections.abc import Sequence
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: str | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
schema=RichieBase.schema_name
def upgrade() -> None:
"""Upgrade."""
${upgrades if upgrades else "pass"}
def downgrade() -> None:
"""Downgrade."""
${downgrades if downgrades else "pass"}

View File

@@ -1,113 +0,0 @@
"""created contact api.
Revision ID: edd7dd61a3d2
Revises:
Create Date: 2026-01-11 15:45:59.909266
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import sqlalchemy as sa
from alembic import op
from python.orm import RichieBase
if TYPE_CHECKING:
from collections.abc import Sequence
# revision identifiers, used by Alembic.
revision: str = "edd7dd61a3d2"
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
schema = RichieBase.schema_name
def upgrade() -> None:
"""Upgrade."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"contact",
sa.Column("name", sa.String(), nullable=False),
sa.Column("age", sa.Integer(), nullable=True),
sa.Column("bio", sa.String(), nullable=True),
sa.Column("current_job", sa.String(), nullable=True),
sa.Column("gender", sa.String(), nullable=True),
sa.Column("goals", sa.String(), nullable=True),
sa.Column("legal_name", sa.String(), nullable=True),
sa.Column("profile_pic", sa.String(), nullable=True),
sa.Column("safe_conversation_starters", sa.String(), nullable=True),
sa.Column("self_sufficiency_score", sa.Integer(), nullable=True),
sa.Column("social_structure_style", sa.String(), nullable=True),
sa.Column("ssn", sa.String(), nullable=True),
sa.Column("suffix", sa.String(), nullable=True),
sa.Column("timezone", sa.String(), nullable=True),
sa.Column("topics_to_avoid", sa.String(), nullable=True),
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.PrimaryKeyConstraint("id", name=op.f("pk_contact")),
schema=schema,
)
op.create_table(
"need",
sa.Column("name", sa.String(), nullable=False),
sa.Column("description", sa.String(), nullable=True),
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.PrimaryKeyConstraint("id", name=op.f("pk_need")),
schema=schema,
)
op.create_table(
"contact_need",
sa.Column("contact_id", sa.Integer(), nullable=False),
sa.Column("need_id", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(
["contact_id"],
[f"{schema}.contact.id"],
name=op.f("fk_contact_need_contact_id_contact"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["need_id"], [f"{schema}.need.id"], name=op.f("fk_contact_need_need_id_need"), ondelete="CASCADE"
),
sa.PrimaryKeyConstraint("contact_id", "need_id", name=op.f("pk_contact_need")),
schema=schema,
)
op.create_table(
"contact_relationship",
sa.Column("contact_id", sa.Integer(), nullable=False),
sa.Column("related_contact_id", sa.Integer(), nullable=False),
sa.Column("relationship_type", sa.String(length=100), nullable=False),
sa.Column("closeness_weight", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(
["contact_id"],
[f"{schema}.contact.id"],
name=op.f("fk_contact_relationship_contact_id_contact"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["related_contact_id"],
[f"{schema}.contact.id"],
name=op.f("fk_contact_relationship_related_contact_id_contact"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("contact_id", "related_contact_id", name=op.f("pk_contact_relationship")),
schema=schema,
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("contact_relationship", schema=schema)
op.drop_table("contact_need", schema=schema)
op.drop_table("need", schema=schema)
op.drop_table("contact", schema=schema)
# ### end Alembic commands ###

View File

@@ -1 +0,0 @@
"""FastAPI applications."""

View File

@@ -1,16 +0,0 @@
"""FastAPI dependencies."""
from collections.abc import Iterator
from typing import Annotated
from fastapi import Depends, Request
from sqlalchemy.orm import Session
def get_db(request: Request) -> Iterator[Session]:
"""Get database session from app state."""
with Session(request.app.state.engine) as session:
yield session
DbSession = Annotated[Session, Depends(get_db)]

View File

@@ -1,117 +0,0 @@
"""FastAPI interface for Contact database."""
import logging
import shutil
import subprocess
import tempfile
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from os import environ
from pathlib import Path
from typing import Annotated
import typer
import uvicorn
from fastapi import FastAPI
from python.api.routers import contact_router, create_frontend_router
from python.common import configure_logger
from python.orm.base import get_postgres_engine
logger = logging.getLogger(__name__)
def create_app(frontend_dir: Path | None = None) -> FastAPI:
"""Create and configure the FastAPI application."""
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
"""Manage application lifespan."""
app.state.engine = get_postgres_engine()
yield
app.state.engine.dispose()
app = FastAPI(title="Contact Database API", lifespan=lifespan)
app.include_router(contact_router)
if frontend_dir:
logger.info(f"Serving frontend from {frontend_dir}")
frontend_router = create_frontend_router(frontend_dir)
app.include_router(frontend_router)
return app
def build_frontend(source_dir: Path | None, cache_dir: Path | None = None) -> Path | None:
"""Run npm build and copy output to a temp directory.
Works even if source_dir is read-only by copying to a temp directory first.
Args:
source_dir: Frontend source directory.
cache_dir: Optional npm cache directory for faster repeated builds.
Returns:
Path to frontend build directory, or None if no source_dir provided.
"""
if not source_dir:
return None
if not source_dir.exists():
error = f"Frontend directory {source_dir} does not exist"
raise FileExistsError(error)
logger.info("Building frontend from %s...", source_dir)
# Copy source to a writable temp directory
build_dir = Path(tempfile.mkdtemp(prefix="contact_frontend_build_"))
shutil.copytree(source_dir, build_dir, dirs_exist_ok=True)
env = dict(environ)
if cache_dir:
cache_dir.mkdir(parents=True, exist_ok=True)
env["npm_config_cache"] = str(cache_dir)
subprocess.run(["npm", "install"], cwd=build_dir, env=env, check=True) # noqa: S607
subprocess.run(["npm", "run", "build"], cwd=build_dir, env=env, check=True) # noqa: S607
dist_dir = build_dir / "dist"
if not dist_dir.exists():
error = f"Build output not found at {dist_dir}"
raise FileNotFoundError(error)
output_dir = Path(tempfile.mkdtemp(prefix="contact_frontend_"))
shutil.copytree(dist_dir, output_dir, dirs_exist_ok=True)
logger.info(f"Frontend built and copied to {output_dir}")
shutil.rmtree(build_dir)
return output_dir
def serve(
host: Annotated[str, typer.Option("--host", "-h", help="Host to bind to")],
frontend_dir: Annotated[
Path | None,
typer.Option(
"--frontend-dir",
"-f",
help="Frontend source directory. If provided, runs npm build and serves from temp dir.",
),
] = None,
port: Annotated[int, typer.Option("--port", "-p", help="Port to bind to")] = 8000,
log_level: Annotated[str, typer.Option("--log-level", "-l", help="Log level")] = "INFO",
) -> None:
"""Start the Contact API server."""
configure_logger(log_level)
cache_dir = Path(environ["HOME"]) / ".npm"
serve_dir = build_frontend(frontend_dir, cache_dir=cache_dir)
app = create_app(frontend_dir=serve_dir)
uvicorn.run(app, host=host, port=port)
if __name__ == "__main__":
typer.run(serve)

View File

@@ -1,6 +0,0 @@
"""API routers."""
from python.api.routers.contact import router as contact_router
from python.api.routers.frontend import create_frontend_router
__all__ = ["contact_router", "create_frontend_router"]

View File

@@ -1,459 +0,0 @@
"""Contact API router."""
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from python.api.dependencies import DbSession
from python.orm.contact import Contact, ContactRelationship, Need, RelationshipType
class NeedBase(BaseModel):
"""Base schema for Need."""
name: str
description: str | None = None
class NeedCreate(NeedBase):
"""Schema for creating a Need."""
class NeedResponse(NeedBase):
"""Schema for Need response."""
id: int
model_config = {"from_attributes": True}
class ContactRelationshipCreate(BaseModel):
"""Schema for creating a contact relationship."""
related_contact_id: int
relationship_type: RelationshipType
closeness_weight: int | None = None
class ContactRelationshipUpdate(BaseModel):
"""Schema for updating a contact relationship."""
relationship_type: RelationshipType | None = None
closeness_weight: int | None = None
class ContactRelationshipResponse(BaseModel):
"""Schema for contact relationship response."""
contact_id: int
related_contact_id: int
relationship_type: str
closeness_weight: int
model_config = {"from_attributes": True}
class RelationshipTypeInfo(BaseModel):
"""Information about a relationship type."""
value: str
display_name: str
default_weight: int
class GraphNode(BaseModel):
"""Node in the relationship graph."""
id: int
name: str
current_job: str | None = None
class GraphEdge(BaseModel):
"""Edge in the relationship graph."""
source: int
target: int
relationship_type: str
closeness_weight: int
class GraphData(BaseModel):
"""Complete graph data for visualization."""
nodes: list[GraphNode]
edges: list[GraphEdge]
class ContactBase(BaseModel):
"""Base schema for Contact."""
name: str
age: int | None = None
bio: str | None = None
current_job: str | None = None
gender: str | None = None
goals: str | None = None
legal_name: str | None = None
profile_pic: str | None = None
safe_conversation_starters: str | None = None
self_sufficiency_score: int | None = None
social_structure_style: str | None = None
ssn: str | None = None
suffix: str | None = None
timezone: str | None = None
topics_to_avoid: str | None = None
class ContactCreate(ContactBase):
"""Schema for creating a Contact."""
need_ids: list[int] = []
class ContactUpdate(BaseModel):
"""Schema for updating a Contact."""
name: str | None = None
age: int | None = None
bio: str | None = None
current_job: str | None = None
gender: str | None = None
goals: str | None = None
legal_name: str | None = None
profile_pic: str | None = None
safe_conversation_starters: str | None = None
self_sufficiency_score: int | None = None
social_structure_style: str | None = None
ssn: str | None = None
suffix: str | None = None
timezone: str | None = None
topics_to_avoid: str | None = None
need_ids: list[int] | None = None
class ContactResponse(ContactBase):
"""Schema for Contact response with relationships."""
id: int
needs: list[NeedResponse] = []
related_to: list[ContactRelationshipResponse] = []
related_from: list[ContactRelationshipResponse] = []
model_config = {"from_attributes": True}
class ContactListResponse(ContactBase):
"""Schema for Contact list response."""
id: int
model_config = {"from_attributes": True}
router = APIRouter(prefix="/api", tags=["contacts"])
@router.post("/needs", response_model=NeedResponse)
def create_need(need: NeedCreate, db: DbSession) -> Need:
"""Create a new need."""
db_need = Need(name=need.name, description=need.description)
db.add(db_need)
db.commit()
db.refresh(db_need)
return db_need
@router.get("/needs", response_model=list[NeedResponse])
def list_needs(db: DbSession) -> list[Need]:
"""List all needs."""
return list(db.scalars(select(Need)).all())
@router.get("/needs/{need_id}", response_model=NeedResponse)
def get_need(need_id: int, db: DbSession) -> Need:
"""Get a need by ID."""
need = db.get(Need, need_id)
if not need:
raise HTTPException(status_code=404, detail="Need not found")
return need
@router.delete("/needs/{need_id}")
def delete_need(need_id: int, db: DbSession) -> dict[str, bool]:
"""Delete a need by ID."""
need = db.get(Need, need_id)
if not need:
raise HTTPException(status_code=404, detail="Need not found")
db.delete(need)
db.commit()
return {"deleted": True}
@router.post("/contacts", response_model=ContactResponse)
def create_contact(contact: ContactCreate, db: DbSession) -> Contact:
"""Create a new contact."""
need_ids = contact.need_ids
contact_data = contact.model_dump(exclude={"need_ids"})
db_contact = Contact(**contact_data)
if need_ids:
needs = list(db.scalars(select(Need).where(Need.id.in_(need_ids))).all())
db_contact.needs = needs
db.add(db_contact)
db.commit()
db.refresh(db_contact)
return db_contact
@router.get("/contacts", response_model=list[ContactListResponse])
def list_contacts(
db: DbSession,
skip: int = 0,
limit: int = 100,
) -> list[Contact]:
"""List all contacts with pagination."""
return list(db.scalars(select(Contact).offset(skip).limit(limit)).all())
@router.get("/contacts/{contact_id}", response_model=ContactResponse)
def get_contact(contact_id: int, db: DbSession) -> Contact:
"""Get a contact by ID with all relationships."""
contact = db.scalar(
select(Contact)
.where(Contact.id == contact_id)
.options(
selectinload(Contact.needs),
selectinload(Contact.related_to),
selectinload(Contact.related_from),
)
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
return contact
@router.patch("/contacts/{contact_id}", response_model=ContactResponse)
def update_contact(
contact_id: int,
contact: ContactUpdate,
db: DbSession,
) -> Contact:
"""Update a contact by ID."""
db_contact = db.get(Contact, contact_id)
if not db_contact:
raise HTTPException(status_code=404, detail="Contact not found")
update_data = contact.model_dump(exclude_unset=True)
need_ids = update_data.pop("need_ids", None)
for key, value in update_data.items():
setattr(db_contact, key, value)
if need_ids is not None:
needs = list(db.scalars(select(Need).where(Need.id.in_(need_ids))).all())
db_contact.needs = needs
db.commit()
db.refresh(db_contact)
return db_contact
@router.delete("/contacts/{contact_id}")
def delete_contact(contact_id: int, db: DbSession) -> dict[str, bool]:
"""Delete a contact by ID."""
contact = db.get(Contact, contact_id)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
db.delete(contact)
db.commit()
return {"deleted": True}
@router.post("/contacts/{contact_id}/needs/{need_id}")
def add_need_to_contact(
contact_id: int,
need_id: int,
db: DbSession,
) -> dict[str, bool]:
"""Add a need to a contact."""
contact = db.get(Contact, contact_id)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
need = db.get(Need, need_id)
if not need:
raise HTTPException(status_code=404, detail="Need not found")
if need not in contact.needs:
contact.needs.append(need)
db.commit()
return {"added": True}
@router.delete("/contacts/{contact_id}/needs/{need_id}")
def remove_need_from_contact(
contact_id: int,
need_id: int,
db: DbSession,
) -> dict[str, bool]:
"""Remove a need from a contact."""
contact = db.get(Contact, contact_id)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
need = db.get(Need, need_id)
if not need:
raise HTTPException(status_code=404, detail="Need not found")
if need in contact.needs:
contact.needs.remove(need)
db.commit()
return {"removed": True}
@router.post(
"/contacts/{contact_id}/relationships",
response_model=ContactRelationshipResponse,
)
def add_contact_relationship(
contact_id: int,
relationship: ContactRelationshipCreate,
db: DbSession,
) -> ContactRelationship:
"""Add a relationship between two contacts."""
contact = db.get(Contact, contact_id)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
related_contact = db.get(Contact, relationship.related_contact_id)
if not related_contact:
raise HTTPException(status_code=404, detail="Related contact not found")
if contact_id == relationship.related_contact_id:
raise HTTPException(status_code=400, detail="Cannot relate contact to itself")
# Use provided weight or default from relationship type
weight = relationship.closeness_weight
if weight is None:
weight = relationship.relationship_type.default_weight
db_relationship = ContactRelationship(
contact_id=contact_id,
related_contact_id=relationship.related_contact_id,
relationship_type=relationship.relationship_type.value,
closeness_weight=weight,
)
db.add(db_relationship)
db.commit()
db.refresh(db_relationship)
return db_relationship
@router.get(
"/contacts/{contact_id}/relationships",
response_model=list[ContactRelationshipResponse],
)
def get_contact_relationships(
contact_id: int,
db: DbSession,
) -> list[ContactRelationship]:
"""Get all relationships for a contact."""
contact = db.get(Contact, contact_id)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
outgoing = list(db.scalars(select(ContactRelationship).where(ContactRelationship.contact_id == contact_id)).all())
incoming = list(
db.scalars(select(ContactRelationship).where(ContactRelationship.related_contact_id == contact_id)).all()
)
return outgoing + incoming
@router.patch(
"/contacts/{contact_id}/relationships/{related_contact_id}",
response_model=ContactRelationshipResponse,
)
def update_contact_relationship(
contact_id: int,
related_contact_id: int,
update: ContactRelationshipUpdate,
db: DbSession,
) -> ContactRelationship:
"""Update a relationship between two contacts."""
relationship = db.scalar(
select(ContactRelationship).where(
ContactRelationship.contact_id == contact_id,
ContactRelationship.related_contact_id == related_contact_id,
)
)
if not relationship:
raise HTTPException(status_code=404, detail="Relationship not found")
if update.relationship_type is not None:
relationship.relationship_type = update.relationship_type.value
if update.closeness_weight is not None:
relationship.closeness_weight = update.closeness_weight
db.commit()
db.refresh(relationship)
return relationship
@router.delete("/contacts/{contact_id}/relationships/{related_contact_id}")
def remove_contact_relationship(
contact_id: int,
related_contact_id: int,
db: DbSession,
) -> dict[str, bool]:
"""Remove a relationship between two contacts."""
relationship = db.scalar(
select(ContactRelationship).where(
ContactRelationship.contact_id == contact_id,
ContactRelationship.related_contact_id == related_contact_id,
)
)
if not relationship:
raise HTTPException(status_code=404, detail="Relationship not found")
db.delete(relationship)
db.commit()
return {"deleted": True}
@router.get("/relationship-types")
def list_relationship_types() -> list[RelationshipTypeInfo]:
"""List all available relationship types with their default weights."""
return [
RelationshipTypeInfo(
value=rt.value,
display_name=rt.display_name,
default_weight=rt.default_weight,
)
for rt in RelationshipType
]
@router.get("/graph")
def get_relationship_graph(db: DbSession) -> GraphData:
"""Get all contacts and relationships as graph data for visualization."""
contacts = list(db.scalars(select(Contact)).all())
relationships = list(db.scalars(select(ContactRelationship)).all())
nodes = [GraphNode(id=c.id, name=c.name, current_job=c.current_job) for c in contacts]
edges = [
GraphEdge(
source=rel.contact_id,
target=rel.related_contact_id,
relationship_type=rel.relationship_type,
closeness_weight=rel.closeness_weight,
)
for rel in relationships
]
return GraphData(nodes=nodes, edges=edges)

View File

@@ -1,24 +0,0 @@
"""Frontend SPA router."""
from pathlib import Path
from fastapi import APIRouter
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
def create_frontend_router(frontend_dir: Path) -> APIRouter:
"""Create a router for serving the frontend SPA."""
router = APIRouter(tags=["frontend"])
router.mount("/assets", StaticFiles(directory=frontend_dir / "assets"), name="assets")
@router.get("/{full_path:path}")
async def serve_spa(full_path: str) -> FileResponse:
"""Serve React SPA for all non-API routes."""
file_path = frontend_dir / full_path
if file_path.is_file():
return FileResponse(file_path)
return FileResponse(frontend_dir / "index.html")
return router

View File

@@ -1,72 +0,0 @@
"""common."""
from __future__ import annotations
import logging
import sys
from datetime import UTC, datetime
from os import getenv
from subprocess import PIPE, Popen
from apprise import Apprise
logger = logging.getLogger(__name__)
def configure_logger(level: str = "INFO") -> None:
"""Configure the logger.
Args:
level (str, optional): The logging level. Defaults to "INFO".
"""
logging.basicConfig(
level=level,
datefmt="%Y-%m-%dT%H:%M:%S%z",
format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s",
handlers=[logging.StreamHandler(sys.stdout)],
)
def bash_wrapper(command: str) -> tuple[str, int]:
"""Execute a bash command and capture the output.
Args:
command (str): The bash command to be executed.
Returns:
Tuple[str, int]: A tuple containing the output of the command (stdout) as a string,
the error output (stderr) as a string (optional), and the return code as an integer.
"""
# This is a acceptable risk
process = Popen(command.split(), stdout=PIPE, stderr=PIPE)
output, error = process.communicate()
if error:
logger.error(f"{error=}")
return error.decode(), process.returncode
return output.decode(), process.returncode
def signal_alert(body: str, title: str = "") -> None:
"""Send a signal alert.
Args:
body (str): The body of the alert.
title (str, optional): The title of the alert. Defaults to "".
"""
apprise_client = Apprise()
from_phone = getenv("SIGNAL_ALERT_FROM_PHONE")
to_phone = getenv("SIGNAL_ALERT_TO_PHONE")
if not from_phone or not to_phone:
logger.info("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set")
return
apprise_client.add(f"signal://localhost:8989/{from_phone}/{to_phone}")
apprise_client.notify(title=title, body=body)
def utcnow() -> datetime:
"""Get the current UTC time."""
return datetime.now(tz=UTC)

View File

@@ -1,59 +0,0 @@
"""database."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from sqlalchemy import inspect
from sqlalchemy.exc import NoInspectionAvailable
if TYPE_CHECKING:
from collections.abc import Sequence
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
def safe_insert(orm_objects: Sequence[object], session: Session) -> list[tuple[Exception, object]]:
"""Safer insert at allows for partial rollbacks.
Args:
orm_objects (Sequence[object]): Tables to insert.
session (Session): Database session.
"""
if unmapped := [orm_object for orm_object in orm_objects if not _is_mapped_instance(orm_object)]:
error = f"safe_insert expects ORM-mapped instances {unmapped}"
raise TypeError(error)
return _safe_insert(orm_objects, session)
def _safe_insert(objects: Sequence[object], session: Session) -> list[tuple[Exception, object]]:
exceptions: list[tuple[Exception, object]] = []
try:
session.add_all(objects)
session.commit()
except Exception as error:
session.rollback()
objects_len = len(objects)
if objects_len == 1:
logger.exception(objects)
return [(error, objects[0])]
middle = objects_len // 2
exceptions.extend(_safe_insert(objects=objects[:middle], session=session))
exceptions.extend(_safe_insert(objects=objects[middle:], session=session))
return exceptions
def _is_mapped_instance(obj: object) -> bool:
"""Return True if `obj` is a SQLAlchemy ORM-mapped instance."""
try:
inspect(obj) # raises NoInspectionAvailable if not mapped
except NoInspectionAvailable:
return False
else:
return True

View File

@@ -1 +0,0 @@
"""Detect Nix evaluation warnings from build logs and create PRs with LLM-suggested fixes."""

View File

@@ -1,449 +0,0 @@
"""Detect Nix evaluation warnings and create PRs with LLM-suggested fixes."""
from __future__ import annotations
import hashlib
import logging
import re
import subprocess
from dataclasses import dataclass
from io import BytesIO
from pathlib import Path
from typing import Annotated
from zipfile import ZipFile
import typer
from httpx import HTTPError, post
from python.common import configure_logger
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class EvalWarning:
"""A single Nix evaluation warning."""
system: str
message: str
@dataclass
class FileChange:
"""A file change suggested by the LLM."""
file_path: str
original: str
fixed: str
def run_cmd(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
"""Run a subprocess command and return the result.
Args:
cmd: Command and arguments.
check: Whether to raise on non-zero exit.
Returns:
CompletedProcess with captured stdout/stderr.
"""
logger.debug("Running: %s", " ".join(cmd))
return subprocess.run(cmd, capture_output=True, text=True, check=check)
def download_logs(run_id: str, repo: str) -> dict[str, str]:
"""Download build logs for a GitHub Actions run.
Args:
run_id: The workflow run ID.
repo: The GitHub repository (owner/repo).
Returns:
Dict mapping zip entry names to their text content, filtered to build log files.
Raises:
RuntimeError: If log download fails.
"""
result = subprocess.run(
["gh", "api", f"repos/{repo}/actions/runs/{run_id}/logs"],
capture_output=True,
check=False,
)
if result.returncode != 0:
msg = f"Failed to download logs: {result.stderr.decode(errors='replace')}"
raise RuntimeError(msg)
logs: dict[str, str] = {}
with ZipFile(BytesIO(result.stdout)) as zip_file:
for name in zip_file.namelist():
if name.startswith("build-") and name.endswith(".txt"):
logs[name] = zip_file.read(name).decode(errors="replace")
return logs
def parse_warnings(logs: dict[str, str]) -> set[EvalWarning]:
"""Parse Nix evaluation warnings from build log contents.
Args:
logs: Dict mapping zip entry names (e.g. "build-bob/2_Build.txt") to their text.
Returns:
Deduplicated set of warnings.
"""
warnings: set[EvalWarning] = set()
warning_pattern = re.compile(r"(?:^[\d\-T:.Z]+ )?(warning:|trace: warning:)")
timestamp_prefix = re.compile(r"^[\d\-T:.Z]+ ")
for name, content in sorted(logs.items()):
system = name.split("/")[0].removeprefix("build-")
for line in content.splitlines():
if warning_pattern.search(line):
message = timestamp_prefix.sub("", line).strip()
if message.startswith("warning: ignoring untrusted flake configuration setting"):
continue
logger.debug(f"Found warning: {line}")
warnings.add(EvalWarning(system=system, message=message))
logger.info("Found %d unique warnings", len(warnings))
return warnings
def extract_referenced_files(warnings: set[EvalWarning]) -> dict[str, str]:
"""Extract file paths referenced in warnings and read their contents.
Args:
warnings: List of parsed warnings.
Returns:
Dict mapping repo-relative file paths to their contents.
"""
paths: set[str] = set()
warning_text = "\n".join(w.message for w in warnings)
nix_store_path = re.compile(r"/nix/store/[^/]+-source/([^:]+\.nix)")
for match in nix_store_path.finditer(warning_text):
paths.add(match.group(1))
repo_relative_path = re.compile(r"(?<![/\w])(systems|common|users|overlays)/[^:\s]+\.nix")
for match in repo_relative_path.finditer(warning_text):
paths.add(match.group(0))
files: dict[str, str] = {}
for path_str in sorted(paths):
path = Path(path_str)
if path.is_file():
files[path_str] = path.read_text()
if not files and Path("flake.nix").is_file():
files["flake.nix"] = Path("flake.nix").read_text()
logger.info("Extracted %d referenced files", len(files))
return files
def compute_warning_hash(warnings: set[EvalWarning]) -> str:
"""Compute a short hash of the warning set for deduplication.
Args:
warnings: List of warnings.
Returns:
8-character hex hash.
"""
text = "\n".join(sorted(f"[{w.system}] {w.message}" for w in warnings))
return hashlib.sha256(text.encode()).hexdigest()[:8]
def check_duplicate_pr(warning_hash: str) -> bool:
"""Check if an open PR already exists for this warning hash.
Args:
warning_hash: The hash to check.
Returns:
True if a duplicate PR exists.
Raises:
RuntimeError: If the gh CLI call fails.
"""
result = run_cmd(
[
"gh",
"pr",
"list",
"--state",
"open",
"--label",
"eval-warning-fix",
"--json",
"title",
"--jq",
".[].title",
],
check=False,
)
if result.returncode != 0:
msg = f"Failed to check for duplicate PRs: {result.stderr}"
raise RuntimeError(msg)
for title in result.stdout.splitlines():
if warning_hash in title:
logger.info("Duplicate PR found for hash %s", warning_hash)
return True
return False
def query_ollama(
warnings: set[EvalWarning],
files: dict[str, str],
ollama_url: str,
) -> str | None:
"""Query Ollama for a fix suggestion.
Args:
warnings: List of warnings.
files: Referenced file contents.
ollama_url: Ollama API base URL.
Returns:
LLM response text, or None on failure.
"""
warning_text = "\n".join(f"[{w.system}] {w.message}" for w in warnings)
file_context = "\n".join(f"--- FILE: {path} ---\n{content}\n--- END FILE ---" for path, content in files.items())
prompt = f"""You are a NixOS configuration expert. \
Analyze the following Nix evaluation warnings and suggest fixes.
## Warnings
{warning_text}
## Referenced Files
{file_context}
## Instructions
- Identify the root cause of each warning
- Provide the exact file changes needed to fix the warnings
- Output your response in two clearly separated sections:
1. **REASONING**: Brief explanation of what causes each warning and how to fix it
2. **CHANGES**: For each file that needs changes, output a block like:
FILE: path/to/file.nix
<<<<<<< ORIGINAL
the original lines to replace
=======
the replacement lines
>>>>>>> FIXED
- Only suggest changes for files that exist in the repository
- Do not add unnecessary complexity
- Preserve the existing code style
- If a warning comes from upstream nixpkgs and cannot be fixed in this repo, \
say so in REASONING and do not suggest changes"""
try:
response = post(
f"{ollama_url}/api/generate",
json={
"model": "qwen3-coder:30b",
"prompt": prompt,
"stream": False,
"options": {"num_predict": 4096},
},
timeout=300,
)
response.raise_for_status()
except HTTPError:
logger.exception("Ollama request failed")
return None
return response.json().get("response")
def parse_changes(response: str) -> list[FileChange]:
"""Parse file changes from the **CHANGES** section of the LLM response.
Expects blocks in the format:
FILE: path/to/file.nix
<<<<<<< ORIGINAL
...
=======
...
>>>>>>> FIXED
Args:
response: Raw LLM response text.
Returns:
List of parsed file changes.
"""
if "**CHANGES**" not in response:
logger.warning("LLM response missing **CHANGES** section")
return []
changes_section = response.split("**CHANGES**", 1)[1]
changes: list[FileChange] = []
current_file = ""
section: str | None = None
original_lines: list[str] = []
fixed_lines: list[str] = []
for line in changes_section.splitlines():
stripped = line.strip()
if stripped.startswith("FILE:"):
current_file = stripped.removeprefix("FILE:").strip()
elif stripped == "<<<<<<< ORIGINAL":
section = "original"
original_lines = []
elif stripped == "=======" and section == "original":
section = "fixed"
fixed_lines = []
elif stripped == ">>>>>>> FIXED" and section == "fixed":
section = None
if current_file:
changes.append(FileChange(current_file, "\n".join(original_lines), "\n".join(fixed_lines)))
elif section == "original":
original_lines.append(line)
elif section == "fixed":
fixed_lines.append(line)
logger.info("Parsed %d file changes", len(changes))
return changes
def apply_changes(changes: list[FileChange]) -> int:
"""Apply file changes to the working directory.
Args:
changes: List of changes to apply.
Returns:
Number of changes successfully applied.
"""
applied = 0
cwd = Path.cwd().resolve()
for change in changes:
path = Path(change.file_path).resolve()
if not path.is_relative_to(cwd):
logger.warning("Path traversal blocked: %s", change.file_path)
continue
if not path.is_file():
logger.warning("File not found: %s", change.file_path)
continue
content = path.read_text()
if change.original not in content:
logger.warning("Original text not found in %s", change.file_path)
continue
path.write_text(content.replace(change.original, change.fixed, 1))
logger.info("Applied fix to %s", change.file_path)
applied += 1
return applied
def create_pr(
warning_hash: str,
warnings: set[EvalWarning],
llm_response: str,
run_url: str,
) -> None:
"""Create a git branch and PR with the applied fixes.
Args:
warning_hash: Short hash for branch naming and deduplication.
warnings: Original warnings for the PR body.
llm_response: Full LLM response for extracting reasoning.
run_url: URL to the triggering build run.
"""
branch = f"fix/eval-warning-{warning_hash}"
warning_text = "\n".join(f"[{w.system}] {w.message}" for w in warnings)
if "**REASONING**" not in llm_response:
logger.warning("LLM response missing **REASONING** section")
reasoning = ""
else:
_, after = llm_response.split("**REASONING**", 1)
reasoning = "\n".join(after.split("**CHANGES**", 1)[0].strip().splitlines()[:50])
run_cmd(["git", "config", "user.name", "github-actions[bot]"])
run_cmd(["git", "config", "user.email", "github-actions[bot]@users.noreply.github.com"])
run_cmd(["git", "checkout", "-b", branch])
run_cmd(["git", "add", "-A"])
diff_result = run_cmd(["git", "diff", "--cached", "--quiet"], check=False)
if diff_result.returncode == 0:
logger.info("No file changes to commit")
return
run_cmd(["git", "commit", "-m", f"fix: resolve nix evaluation warnings ({warning_hash})"])
run_cmd(["git", "push", "origin", branch, "--force"])
body = f"""## Nix Evaluation Warnings
Detected in [build_systems run]({run_url}):
```
{warning_text}
```
## LLM Analysis (qwen3-coder:30b)
{reasoning}
---
*Auto-generated by fix_eval_warnings. Review carefully before merging.*"""
run_cmd(
[
"gh",
"pr",
"create",
"--title",
f"fix: resolve nix eval warnings ({warning_hash})",
"--label",
"automated",
"--label",
"eval-warning-fix",
"--body",
body,
]
)
logger.info("PR created on branch %s", branch)
def main(
run_id: Annotated[str, typer.Option("--run-id", help="GitHub Actions run ID")],
repo: Annotated[str, typer.Option("--repo", help="GitHub repository (owner/repo)")],
ollama_url: Annotated[str, typer.Option("--ollama-url", help="Ollama API base URL")],
run_url: Annotated[str, typer.Option("--run-url", help="URL to the triggering build run")],
log_level: Annotated[str, typer.Option("--log-level", "-l", help="Log level")] = "INFO",
) -> None:
"""Detect Nix evaluation warnings and create PRs with LLM-suggested fixes."""
configure_logger(log_level)
logs = download_logs(run_id, repo)
warnings = parse_warnings(logs)
if not warnings:
return
warning_hash = compute_warning_hash(warnings)
if check_duplicate_pr(warning_hash):
return
files = extract_referenced_files(warnings)
llm_response = query_ollama(warnings, files, ollama_url)
if not llm_response:
return
changes = parse_changes(llm_response)
applied = apply_changes(changes)
if applied == 0:
logger.info("No changes could be applied")
return
create_pr(warning_hash, warnings, llm_response, run_url)
if __name__ == "__main__":
typer.run(main)

View File

@@ -1 +0,0 @@
"""Tuya heater control service."""

View File

@@ -1,69 +0,0 @@
"""TinyTuya device controller for heater."""
import logging
import tinytuya
from python.heater.models import ActionResult, DeviceConfig, HeaterStatus
logger = logging.getLogger(__name__)
# DPS mapping for heater
DPS_POWER = "1" # bool: on/off
DPS_SETPOINT = "101" # int: target temp (read-only)
DPS_STATE = "102" # str: "Stop", "Heat", etc.
DPS_UNKNOWN = "104" # int: unknown
DPS_ERROR = "108" # int: last error code
class HeaterController:
"""Controls a Tuya heater device via local network."""
def __init__(self, config: DeviceConfig) -> None:
"""Initialize the controller."""
self.device = tinytuya.Device(config.device_id, config.ip, config.local_key)
self.device.set_version(config.version)
self.device.set_socketTimeout(0.5)
self.device.set_socketRetryLimit(1)
def status(self) -> HeaterStatus:
"""Get current heater status."""
data = self.device.status()
if "Error" in data:
logger.error("Device error: %s", data)
return HeaterStatus(power=False, raw_dps={"error": data["Error"]})
dps = data.get("dps", {})
return HeaterStatus(
power=bool(dps.get(DPS_POWER, False)),
setpoint=dps.get(DPS_SETPOINT),
state=dps.get(DPS_STATE),
error_code=dps.get(DPS_ERROR),
raw_dps=dps,
)
def turn_on(self) -> ActionResult:
"""Turn heater on."""
try:
self.device.set_value(index=DPS_POWER, value=True)
return ActionResult(success=True, action="on", power=True)
except Exception as error:
logger.exception("Failed to turn on")
return ActionResult(success=False, action="on", error=str(error))
def turn_off(self) -> ActionResult:
"""Turn heater off."""
try:
self.device.set_value(index=DPS_POWER, value=False)
return ActionResult(success=True, action="off", power=False)
except Exception as error:
logger.exception("Failed to turn off")
return ActionResult(success=False, action="off", error=str(error))
def toggle(self) -> ActionResult:
"""Toggle heater power state."""
status = self.status()
if status.power:
return self.turn_off()
return self.turn_on()

View File

@@ -1,85 +0,0 @@
"""FastAPI heater control service."""
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated
import typer
import uvicorn
from fastapi import FastAPI, HTTPException
from python.common import configure_logger
from python.heater.controller import HeaterController
from python.heater.models import ActionResult, DeviceConfig, HeaterStatus
logger = logging.getLogger(__name__)
def create_app(config: DeviceConfig) -> FastAPI:
"""Create FastAPI application."""
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
app.state.controller = HeaterController(config)
yield
app = FastAPI(
title="Heater Control API",
description="Fast local control for Tuya heater",
lifespan=lifespan,
)
@app.get("/status")
def get_status() -> HeaterStatus:
return app.state.controller.status()
@app.post("/on")
def heater_on() -> ActionResult:
result = app.state.controller.turn_on()
if not result.success:
raise HTTPException(status_code=500, detail=result.error)
return result
@app.post("/off")
def heater_off() -> ActionResult:
result = app.state.controller.turn_off()
if not result.success:
raise HTTPException(status_code=500, detail=result.error)
return result
@app.post("/toggle")
def heater_toggle() -> ActionResult:
result = app.state.controller.toggle()
if not result.success:
raise HTTPException(status_code=500, detail=result.error)
return result
return app
def serve(
host: Annotated[str, typer.Option("--host", "-h", help="Host to bind to")],
port: Annotated[int, typer.Option("--port", "-p", help="Port to bind to")] = 8124,
log_level: Annotated[str, typer.Option("--log-level", "-l", help="Log level")] = "INFO",
device_id: Annotated[str | None, typer.Option("--device-id", envvar="TUYA_DEVICE_ID")] = None,
device_ip: Annotated[str | None, typer.Option("--device-ip", envvar="TUYA_DEVICE_IP")] = None,
local_key: Annotated[str | None, typer.Option("--local-key", envvar="TUYA_LOCAL_KEY")] = None,
) -> None:
"""Start the heater control API server."""
configure_logger(log_level)
logger.info("Starting heater control API server")
if not device_id or not device_ip or not local_key:
error = "Must provide device ID, IP, and local key"
raise typer.Exit(error)
config = DeviceConfig(device_id=device_id, ip=device_ip, local_key=local_key)
app = create_app(config)
uvicorn.run(app, host=host, port=port)
if __name__ == "__main__":
typer.run(serve)

View File

@@ -1,31 +0,0 @@
"""Pydantic models for heater API."""
from pydantic import BaseModel, Field
class DeviceConfig(BaseModel):
"""Tuya device configuration."""
device_id: str
ip: str
local_key: str
version: float = 3.5
class HeaterStatus(BaseModel):
"""Current heater status."""
power: bool
setpoint: int | None = None
state: str | None = None # "Stop", "Heat", etc.
error_code: int | None = None
raw_dps: dict[str, object] = Field(default_factory=dict)
class ActionResult(BaseModel):
"""Result of a heater action."""
success: bool
action: str
power: bool | None = None
error: str | None = None

View File

@@ -1 +0,0 @@
"""installer."""

View File

@@ -1,308 +0,0 @@
"""Install NixOS on a ZFS pool."""
from __future__ import annotations
import curses
import logging
import sys
from os import getenv
from pathlib import Path
from random import getrandbits
from subprocess import PIPE, Popen, run
from time import sleep
from typing import TYPE_CHECKING
from python.common import configure_logger
from python.installer.tui import draw_menu
if TYPE_CHECKING:
from collections.abc import Sequence
logger = logging.getLogger(__name__)
def bash_wrapper(command: str) -> str:
"""Execute a bash command and capture the output.
Args:
command (str): The bash command to be executed.
Returns:
Tuple[str, int]: A tuple containing the output of the command (stdout) as a string,
the error output (stderr) as a string (optional), and the return code as an integer.
"""
logger.debug(f"running {command=}")
# This is a acceptable risk
process = Popen(command.split(), stdout=PIPE, stderr=PIPE)
output, _ = process.communicate()
if process.returncode != 0:
error = f"Failed to run command {command=} return code {process.returncode=}"
raise RuntimeError(error)
return output.decode()
def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None:
"""Partition a disk.
Args:
disk (str): The disk to partition.
swap_size (int): The size of the swap partition in GB.
minimum value is 1.
reserve (int, optional): The size of the reserve partition in GB. Defaults to 0.
minimum value is 0.
"""
logger.info(f"partitioning {disk=}")
swap_size = max(swap_size, 1)
reserve = max(reserve, 0)
bash_wrapper(f"blkdiscard -f {disk}")
if reserve > 0:
msg = f"Creating swap partition on {disk=} with size {swap_size=}GiB and reserve {reserve=}GiB"
logger.info(msg)
swap_start = swap_size + reserve
swap_partition = f"mkpart swap -{swap_start}GiB -{reserve}GiB "
else:
logger.info(f"Creating swap partition on {disk=} with size {swap_size=}GiB")
swap_start = swap_size
swap_partition = f"mkpart swap -{swap_start}GiB 100% "
logger.debug(f"{swap_partition=}")
create_partitions = (
f"parted --script --align=optimal {disk} -- "
"mklabel gpt "
"mkpart EFI 1MiB 4GiB "
f"mkpart root_pool 4GiB -{swap_start}GiB "
f"{swap_partition}"
"set 1 esp on"
)
bash_wrapper(create_partitions)
logger.info(f"{disk=} successfully partitioned")
def create_zfs_pool(pool_disks: Sequence[str], mnt_dir: str) -> None:
"""Create a ZFS pool.
Args:
pool_disks (Sequence[str]): A tuple of disks to use for the pool.
mnt_dir (str): The mount directory.
"""
if len(pool_disks) <= 0:
error = "disks must be a tuple of at least length 1"
raise ValueError(error)
zpool_create = (
"zpool create "
"-o ashift=12 "
"-o autotrim=on "
f"-R {mnt_dir} "
"-O acltype=posixacl "
"-O canmount=off "
"-O dnodesize=auto "
"-O normalization=formD "
"-O relatime=on "
"-O xattr=sa "
"-O mountpoint=legacy "
"-O compression=zstd "
"-O atime=off "
"root_pool "
)
if len(pool_disks) == 1:
zpool_create += pool_disks[0]
else:
zpool_create += "mirror "
zpool_create += " ".join(pool_disks)
bash_wrapper(zpool_create)
zpools = bash_wrapper("zpool list -o name")
if "root_pool" not in zpools.splitlines():
logger.critical("Failed to create root_pool")
sys.exit(1)
def create_zfs_datasets() -> None:
"""Create ZFS datasets."""
bash_wrapper("zfs create -o canmount=noauto -o reservation=10G root_pool/root")
bash_wrapper("zfs create root_pool/home")
bash_wrapper("zfs create root_pool/var -o reservation=1G")
bash_wrapper("zfs create -o compression=zstd-9 -o reservation=10G root_pool/nix")
datasets = bash_wrapper("zfs list -o name")
expected_datasets = {
"root_pool/root",
"root_pool/home",
"root_pool/var",
"root_pool/nix",
}
missing_datasets = expected_datasets.difference(datasets.splitlines())
if missing_datasets:
logger.critical(f"Failed to create pools {missing_datasets}")
sys.exit(1)
def get_cpu_manufacturer() -> str:
"""Get the CPU manufacturer."""
output = bash_wrapper("cat /proc/cpuinfo")
id_vendor = {"AuthenticAMD": "amd", "GenuineIntel": "intel"}
for line in output.splitlines():
if "vendor_id" in line:
return id_vendor[line.split(": ")[1].strip()]
error = "Failed to get CPU manufacturer"
raise RuntimeError(error)
def get_boot_drive_id(disk: str) -> str:
"""Get the boot drive ID."""
output = bash_wrapper(f"lsblk -o UUID {disk}-part1")
return output.splitlines()[1]
def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], encrypt: str | None) -> None:
"""Create a NixOS hardware file."""
cpu_manufacturer = get_cpu_manufacturer()
devices = ""
if encrypt:
disk = disks[0]
devices = (
f' luks.devices."luks-root-pool-{disk.split("/")[-1]}-part2"'
"= {\n"
f' device = "{disk}-part2";\n'
" bypassWorkqueues = true;\n"
" allowDiscards = true;\n"
" };\n"
)
host_id = format(getrandbits(32), "08x")
nix_hardware = (
"{ config, lib, modulesPath, ... }:\n"
"{\n"
' imports = [ (modulesPath + "/installer/scan/not-detected.nix") ];\n\n'
" boot = {\n"
" initrd = {\n"
' availableKernelModules = [ \n "ahci"\n "ehci_pci"\n "nvme"\n "sd_mod"\n'
' "usb_storage"\n "usbhid"\n "xhci_pci"\n ];\n'
" kernelModules = [ ];\n"
f" {devices}"
" };\n"
f' kernelModules = [ "kvm-{cpu_manufacturer}" ];\n'
" extraModulePackages = [ ];\n"
" };\n\n"
" fileSystems = {\n"
' "/" = lib.mkDefault {\n device = "root_pool/root";\n fsType = "zfs";\n };\n\n'
' "/home" = {\n device = "root_pool/home";\n fsType = "zfs";\n };\n\n'
' "/var" = {\n device = "root_pool/var";\n fsType = "zfs";\n };\n\n'
' "/nix" = {\n device = "root_pool/nix";\n fsType = "zfs";\n };\n\n'
' "/boot" = {\n'
f' device = "/dev/disk/by-uuid/{get_boot_drive_id(disks[0])}";\n'
' fsType = "vfat";\n options = [\n "fmask=0077"\n'
' "dmask=0077"\n ];\n };\n };\n\n'
" swapDevices = [ ];\n\n"
" networking.useDHCP = lib.mkDefault true;\n\n"
' nixpkgs.hostPlatform = lib.mkDefault "x86_64-linux";\n'
f" hardware.cpu.{cpu_manufacturer}.updateMicrocode = "
"lib.mkDefault config.hardware.enableRedistributableFirmware;\n"
f' networking.hostId = "{host_id}";\n'
"}\n"
)
Path(f"{mnt_dir}/etc/nixos/hardware-configuration.nix").write_text(nix_hardware)
def install_nixos(mnt_dir: str, disks: Sequence[str], encrypt: str | None) -> None:
"""Install NixOS."""
bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/root {mnt_dir}")
bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/home {mnt_dir}/home")
bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/var {mnt_dir}/var")
bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/nix {mnt_dir}/nix")
for disk in disks:
bash_wrapper(f"mkfs.vfat -n EFI {disk}-part1")
# set up mirroring afterwards if more than one disk
boot_partition = (
f"mount -t vfat -o fmask=0077,dmask=0077,iocharset=iso8859-1,X-mount.mkdir {disks[0]}-part1 {mnt_dir}/boot"
)
bash_wrapper(boot_partition)
bash_wrapper(f"nixos-generate-config --root {mnt_dir}")
create_nix_hardware_file(mnt_dir, disks, encrypt)
run(("nixos-install", "--root", mnt_dir), check=True)
def installer(
disks: Sequence[str],
swap_size: int,
reserve: int,
encrypt_key: str | None,
) -> None:
"""Main."""
logger.info("Starting installation")
for disk in disks:
partition_disk(disk, swap_size, reserve)
test = Popen(("printf", f"'{encrypt_key}'"), stdout=PIPE)
if encrypt_key:
sleep(1)
for command in (
f"cryptsetup luksFormat --type luks2 {disk}-part2 -",
f"cryptsetup luksOpen {disk}-part2 luks-root-pool-{disk.split('/')[-1]}-part2 -",
):
run(command, check=True, stdin=test.stdout)
mnt_dir = "/tmp/nix_install" # noqa: S108
Path(mnt_dir).mkdir(parents=True, exist_ok=True)
if encrypt_key:
pool_disks = [f"/dev/mapper/luks-root-pool-{disk.split('/')[-1]}-part2" for disk in disks]
else:
pool_disks = [f"{disk}-part2" for disk in disks]
create_zfs_pool(pool_disks, mnt_dir)
create_zfs_datasets()
install_nixos(mnt_dir, disks, encrypt_key)
logger.info("Installation complete")
def main() -> None:
"""Main."""
configure_logger("DEBUG")
state = curses.wrapper(draw_menu)
encrypt_key = getenv("ENCRYPT_KEY")
logger.info("installing_nixos")
logger.info(f"disks: {state.selected_device_ids}")
logger.info(f"swap_size: {state.swap_size}")
logger.info(f"reserve: {state.reserve_size}")
logger.info(f"encrypted: {bool(encrypt_key)}")
sleep(3)
installer(
disks=state.get_selected_devices(),
swap_size=state.swap_size,
reserve=state.reserve_size,
encrypt_key=encrypt_key,
)
if __name__ == "__main__":
main()

View File

@@ -1,498 +0,0 @@
"""TUI module."""
from __future__ import annotations
import curses
import logging
from collections import defaultdict
from subprocess import PIPE, Popen
logger = logging.getLogger(__name__)
def bash_wrapper(command: str) -> str:
"""Execute a bash command and capture the output.
Args:
command (str): The bash command to be executed.
Returns:
Tuple[str, int]: A tuple containing the output of the command (stdout) as a string,
the error output (stderr) as a string (optional), and the return code as an integer.
"""
logger.debug(f"running {command=}")
# This is a acceptable risk
process = Popen(command.split(), stdout=PIPE, stderr=PIPE)
output, _ = process.communicate()
if process.returncode != 0:
error = f"Failed to run command {command=} return code {process.returncode=}"
raise RuntimeError(error)
return output.decode()
class Cursor:
"""Cursor class."""
def __init__(self) -> None:
"""Initialize the Cursor class."""
self.x_position = 0
self.y_position = 0
self.height = 0
self.width = 0
def set_height(self, height: int) -> None:
"""Set height."""
self.height = height
def set_width(self, width: int) -> None:
"""Set width."""
self.width = width
def x_bounce_check(self, cursor: int) -> int:
"""X bounce check."""
cursor = max(0, cursor)
return min(self.width - 1, cursor)
def y_bounce_check(self, cursor: int) -> int:
"""Y bounce check."""
cursor = max(0, cursor)
return min(self.height - 1, cursor)
def set_x(self, x: int) -> None:
"""Set x."""
self.x_position = self.x_bounce_check(x)
def set_y(self, y: int) -> None:
"""Set y."""
self.y_position = self.y_bounce_check(y)
def get_x(self) -> int:
"""Get x."""
return self.x_position
def get_y(self) -> int:
"""Get y."""
return self.y_position
def move_up(self) -> None:
"""Move up."""
self.set_y(self.y_position - 1)
def move_down(self) -> None:
"""Move down."""
self.set_y(self.y_position + 1)
def move_left(self) -> None:
"""Move left."""
self.set_x(self.x_position - 1)
def move_right(self) -> None:
"""Move right."""
self.set_x(self.x_position + 1)
def navigation(self, key: int) -> None:
"""Navigation.
Args:
key (int): The key.
"""
action = {
curses.KEY_DOWN: self.move_down,
curses.KEY_UP: self.move_up,
curses.KEY_RIGHT: self.move_right,
curses.KEY_LEFT: self.move_left,
}
action.get(key, lambda: None)()
class State:
"""State class to store the state of the program."""
def __init__(self) -> None:
"""Initialize the State class."""
self.key = 0
self.cursor = Cursor()
self.swap_size = 0
self.show_swap_input = False
self.reserve_size = 0
self.show_reserve_input = False
self.selected_device_ids: set[str] = set()
def get_selected_devices(self) -> tuple[str, ...]:
"""Get selected devices."""
return tuple(self.selected_device_ids)
def get_device(raw_device: str) -> dict[str, str]:
"""Get a device.
Args:
raw_device (str): The raw device.
Returns:
dict[str, str]: The device.
"""
raw_device_components = raw_device.split(" ")
return {thing.split("=")[0].lower(): thing.split("=")[1].strip('"') for thing in raw_device_components}
def get_devices() -> list[dict[str, str]]:
"""Get a list of devices."""
# --bytes
raw_devices = bash_wrapper("lsblk --paths --pairs").splitlines()
return [get_device(raw_device) for raw_device in raw_devices]
def set_color() -> None:
"""Set the color."""
curses.start_color()
curses.use_default_colors()
for i in range(curses.COLORS):
curses.init_pair(i + 1, i, -1)
def debug_menu(std_screen: curses.window, key: int) -> None:
"""Debug menu.
Args:
std_screen (curses.window): The curses window.
key (int): The key.
"""
height, width = std_screen.getmaxyx()
std_screen.addstr(height - 4, 0, f"Width: {width}, Height: {height}", curses.color_pair(5))
key_pressed = f"Last key pressed: {key}"[: width - 1]
if key == 0:
key_pressed = "No key press detected..."[: width - 1]
std_screen.addstr(height - 3, 0, key_pressed)
for i in range(8):
std_screen.addstr(height - 2, i * 3, f"{i}██", curses.color_pair(i))
def get_text_input(std_screen: curses.window, prompt: str, y: int, x: int) -> str:
"""Get text input.
Args:
std_screen (curses.window): The curses window.
prompt (str): The prompt.
y (int): The y position.
x (int): The x position.
Returns:
str: The input string.
"""
esc_key = 27
curses.echo()
std_screen.addstr(y, x, prompt)
input_str = ""
while True:
key = std_screen.getch()
if key == ord("\n"):
break
if key == esc_key:
input_str = ""
break
if key in (curses.KEY_BACKSPACE, ord("\b"), 127):
input_str = input_str[:-1]
std_screen.addstr(y, x + len(prompt), input_str + " ")
else:
input_str += chr(key)
std_screen.refresh()
curses.noecho()
return input_str
def swap_size_input(
std_screen: curses.window,
state: State,
swap_offset: int,
) -> State:
"""Reserve size input.
Args:
std_screen (curses.window): The curses window.
state (State): The state object.
swap_offset (int): The swap offset.
Returns:
State: The updated state object.
"""
swap_size_text = "Swap size (GB): "
std_screen.addstr(swap_offset, 0, f"{swap_size_text}{state.swap_size}")
if state.key == ord("\n") and state.cursor.get_y() == swap_offset:
state.show_swap_input = True
if state.show_swap_input:
swap_size_str = get_text_input(std_screen, swap_size_text, swap_offset, 0)
try:
state.swap_size = int(swap_size_str)
state.show_swap_input = False
except ValueError:
std_screen.addstr(swap_offset, 0, "Invalid input. Press any key to continue.")
std_screen.getch()
state.show_swap_input = False
return state
def reserve_size_input(
std_screen: curses.window,
state: State,
reserve_offset: int,
) -> State:
"""Reserve size input.
Args:
std_screen (curses.window): The curses window.
state (State): The state object.
reserve_offset (int): The reserve offset.
Returns:
State: The updated state object.
"""
reserve_size_text = "reserve size (GB): "
std_screen.addstr(reserve_offset, 0, f"{reserve_size_text}{state.reserve_size}")
if state.key == ord("\n") and state.cursor.get_y() == reserve_offset:
state.show_reserve_input = True
if state.show_reserve_input:
reserve_size_str = get_text_input(std_screen, reserve_size_text, reserve_offset, 0)
try:
state.reserve_size = int(reserve_size_str)
state.show_reserve_input = False
except ValueError:
std_screen.addstr(reserve_offset, 0, "Invalid input. Press any key to continue.")
std_screen.getch()
state.show_reserve_input = False
return state
def status_bar(
std_screen: curses.window,
cursor: Cursor,
width: int,
height: int,
) -> None:
"""Draw the status bar.
Args:
std_screen (curses.window): The curses window.
cursor (Cursor): The cursor.
width (int): The width.
height (int): The height.
"""
std_screen.attron(curses.A_REVERSE)
std_screen.attron(curses.color_pair(3))
status_bar = f"Press 'q' to exit | STATUS BAR | Pos: {cursor.get_x()}, {cursor.get_y()}"
std_screen.addstr(height - 1, 0, status_bar)
std_screen.addstr(height - 1, len(status_bar), " " * (width - len(status_bar) - 1))
std_screen.attroff(curses.color_pair(3))
std_screen.attroff(curses.A_REVERSE)
def get_device_id_mapping() -> dict[str, set[str]]:
"""Get a list of device ids.
Returns:
list[str]: the list of device ids
"""
device_ids = bash_wrapper("find /dev/disk/by-id -type l").splitlines()
device_id_mapping: dict[str, set[str]] = defaultdict(set)
for device_id in device_ids:
device = bash_wrapper(f"readlink -f {device_id}").strip()
device_id_mapping[device].add(device_id)
return device_id_mapping
def calculate_device_menu_padding(devices: list[dict[str, str]], column: str, padding: int = 0) -> int:
"""Calculate the device menu padding.
Args:
devices (list[dict[str, str]]): The devices.
column (str): The column.
padding (int, optional): The padding. Defaults to 0.
Returns:
int: The calculated padding.
"""
return max(len(device[column]) for device in devices) + padding
def draw_device_ids(
state: State,
row_number: int,
menu_start_x: int,
std_screen: curses.window,
menu_width: list[int],
device_ids: set[str],
) -> tuple[State, int]:
"""Draw device IDs.
Args:
state (State): The state object.
row_number (int): The row number.
menu_start_x (int): The menu start x.
std_screen (curses.window): The curses window.
menu_width (list[int]): The menu width.
device_ids (set[str]): The device IDs.
Returns:
tuple[State, int]: The updated state object and the row number.
"""
for device_id in sorted(device_ids):
row_number = row_number + 1
if row_number == state.cursor.get_y() and state.cursor.get_x() in menu_width:
std_screen.attron(curses.A_BOLD)
if state.key == ord(" "):
if device_id not in state.selected_device_ids:
state.selected_device_ids.add(device_id)
else:
state.selected_device_ids.remove(device_id)
if device_id in state.selected_device_ids:
std_screen.attron(curses.color_pair(7))
std_screen.addstr(row_number, menu_start_x, f" {device_id}")
std_screen.attroff(curses.color_pair(7))
std_screen.attroff(curses.A_BOLD)
return state, row_number
def draw_device_menu(
std_screen: curses.window,
devices: list[dict[str, str]],
device_id_mapping: dict[str, set[str]],
state: State,
menu_start_y: int = 0,
menu_start_x: int = 0,
) -> tuple[State, int]:
"""Draw the device menu and handle user input.
Args:
std_screen (curses.window): the curses window to draw on
devices (list[dict[str, str]]): the list of devices to draw
device_id_mapping (dict[str, set[str]]): the list of device ids to draw
state (State): the state object to update
menu_start_y (int, optional): the y position to start drawing the menu. Defaults to 0.
menu_start_x (int, optional): the x position to start drawing the menu. Defaults to 0.
Returns:
State: the updated state object
"""
padding = 2
name_padding = calculate_device_menu_padding(devices, "name", padding)
size_padding = calculate_device_menu_padding(devices, "size", padding)
type_padding = calculate_device_menu_padding(devices, "type", padding)
mountpoints_padding = calculate_device_menu_padding(devices, "mountpoints", padding)
device_header = (
f"{'Name':{name_padding}}{'Size':{size_padding}}{'Type':{type_padding}}{'Mountpoints':{mountpoints_padding}}"
)
menu_width = list(range(menu_start_x, len(device_header) + menu_start_x))
std_screen.addstr(menu_start_y, menu_start_x, device_header, curses.color_pair(5))
devises_list_start = menu_start_y + 1
row_number = devises_list_start
for device in devices:
row_number = row_number + 1
device_name = device["name"]
device_row = (
f"{device_name:{name_padding}}"
f"{device['size']:{size_padding}}"
f"{device['type']:{type_padding}}"
f"{device['mountpoints']:{mountpoints_padding}}"
)
std_screen.addstr(row_number, menu_start_x, device_row)
state, row_number = draw_device_ids(
state=state,
row_number=row_number,
menu_start_x=menu_start_x,
std_screen=std_screen,
menu_width=menu_width,
device_ids=device_id_mapping[device_name],
)
return state, row_number
def draw_menu(std_screen: curses.window) -> State:
"""Draw the menu and handle user input.
Args:
std_screen (curses.window): the curses window to draw on
Returns:
State: the state object
"""
# Clear and refresh the screen for a blank canvas
std_screen.clear()
std_screen.refresh()
set_color()
state = State()
devices = get_devices()
device_id_mapping = get_device_id_mapping()
# Loop where k is the last character pressed
while state.key != ord("q"):
std_screen.clear()
height, width = std_screen.getmaxyx()
state.cursor.set_height(height)
state.cursor.set_width(width)
state.cursor.navigation(state.key)
state, device_menu_size = draw_device_menu(
std_screen=std_screen,
state=state,
devices=devices,
device_id_mapping=device_id_mapping,
)
swap_offset = device_menu_size + 2
swap_size_input(
std_screen=std_screen,
state=state,
swap_offset=swap_offset,
)
reserve_size_input(
std_screen=std_screen,
state=state,
reserve_offset=swap_offset + 1,
)
status_bar(std_screen, state.cursor, width, height)
debug_menu(std_screen, state.key)
std_screen.move(state.cursor.get_y(), state.cursor.get_x())
std_screen.refresh()
state.key = std_screen.getch()
return state

View File

@@ -1,22 +0,0 @@
"""ORM package exports."""
from __future__ import annotations
from python.orm.base import RichieBase, TableBase
from python.orm.contact import (
Contact,
ContactNeed,
ContactRelationship,
Need,
RelationshipType,
)
__all__ = [
"Contact",
"ContactNeed",
"ContactRelationship",
"Need",
"RelationshipType",
"RichieBase",
"TableBase",
]

View File

@@ -1,80 +0,0 @@
"""Base ORM definitions."""
from __future__ import annotations
from datetime import datetime
from os import getenv
from typing import cast
from sqlalchemy import DateTime, MetaData, create_engine, func
from sqlalchemy.engine import URL, Engine
from sqlalchemy.ext.declarative import AbstractConcreteBase
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class RichieBase(DeclarativeBase):
"""Base class for all ORM models."""
schema_name = "main"
metadata = MetaData(
schema=schema_name,
naming_convention={
"ix": "ix_%(table_name)s_%(column_0_name)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
},
)
class TableBase(AbstractConcreteBase, RichieBase):
"""Abstract concrete base for tables with IDs and timestamps."""
__abstract__ = True
id: Mapped[int] = mapped_column(primary_key=True)
created: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
)
updated: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)
def get_connection_info() -> tuple[str, str, str, str, str | None]:
"""Get connection info from environment variables."""
database = getenv("POSTGRES_DB")
host = getenv("POSTGRES_HOST")
port = getenv("POSTGRES_PORT")
username = getenv("POSTGRES_USER")
password = getenv("POSTGRES_PASSWORD")
if None in (database, host, port, username):
error = f"Missing environment variables for Postgres connection.\n{database=}\n{host=}\n{port=}\n{username=}\n"
raise ValueError(error)
return cast("tuple[str, str, str, str, str | None]", (database, host, port, username, password))
def get_postgres_engine(*, pool_pre_ping: bool = True) -> Engine:
"""Create a SQLAlchemy engine from environment variables."""
database, host, port, username, password = get_connection_info()
url = URL.create(
drivername="postgresql+psycopg",
username=username,
password=password,
host=host,
port=int(port),
database=database,
)
return create_engine(
url=url,
pool_pre_ping=pool_pre_ping,
pool_recycle=1800,
)

View File

@@ -1,168 +0,0 @@
"""Contact database models."""
from __future__ import annotations
from enum import Enum
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from python.orm.base import RichieBase, TableBase
class RelationshipType(str, Enum):
"""Relationship types with default closeness weights.
Default weight is an integer 1-10 where 10 = closest relationship.
Users can override this per-relationship in the UI.
"""
SPOUSE = "spouse"
PARTNER = "partner"
PARENT = "parent"
CHILD = "child"
SIBLING = "sibling"
BEST_FRIEND = "best_friend"
GRANDPARENT = "grandparent"
GRANDCHILD = "grandchild"
AUNT_UNCLE = "aunt_uncle"
NIECE_NEPHEW = "niece_nephew"
COUSIN = "cousin"
IN_LAW = "in_law"
CLOSE_FRIEND = "close_friend"
FRIEND = "friend"
MENTOR = "mentor"
MENTEE = "mentee"
BUSINESS_PARTNER = "business_partner"
COLLEAGUE = "colleague"
MANAGER = "manager"
DIRECT_REPORT = "direct_report"
CLIENT = "client"
ACQUAINTANCE = "acquaintance"
NEIGHBOR = "neighbor"
EX = "ex"
OTHER = "other"
@property
def default_weight(self) -> int:
"""Return the default closeness weight (1-10) for this relationship type."""
weights = {
RelationshipType.SPOUSE: 10,
RelationshipType.PARTNER: 10,
RelationshipType.PARENT: 9,
RelationshipType.CHILD: 9,
RelationshipType.SIBLING: 9,
RelationshipType.BEST_FRIEND: 8,
RelationshipType.GRANDPARENT: 7,
RelationshipType.GRANDCHILD: 7,
RelationshipType.AUNT_UNCLE: 7,
RelationshipType.NIECE_NEPHEW: 7,
RelationshipType.COUSIN: 7,
RelationshipType.IN_LAW: 7,
RelationshipType.CLOSE_FRIEND: 6,
RelationshipType.FRIEND: 6,
RelationshipType.MENTOR: 5,
RelationshipType.MENTEE: 5,
RelationshipType.BUSINESS_PARTNER: 5,
RelationshipType.COLLEAGUE: 4,
RelationshipType.MANAGER: 4,
RelationshipType.DIRECT_REPORT: 4,
RelationshipType.CLIENT: 4,
RelationshipType.ACQUAINTANCE: 3,
RelationshipType.NEIGHBOR: 3,
RelationshipType.EX: 2,
RelationshipType.OTHER: 2,
}
return weights.get(self, 5)
@property
def display_name(self) -> str:
"""Return a human-readable display name."""
return self.value.replace("_", " ").title()
class ContactNeed(RichieBase):
"""Association table: Contact <-> Need."""
__tablename__ = "contact_need"
contact_id: Mapped[int] = mapped_column(
ForeignKey("main.contact.id", ondelete="CASCADE"),
primary_key=True,
)
need_id: Mapped[int] = mapped_column(
ForeignKey("main.need.id", ondelete="CASCADE"),
primary_key=True,
)
class ContactRelationship(RichieBase):
"""Association table: Contact <-> Contact with relationship type and weight."""
__tablename__ = "contact_relationship"
contact_id: Mapped[int] = mapped_column(
ForeignKey("main.contact.id", ondelete="CASCADE"),
primary_key=True,
)
related_contact_id: Mapped[int] = mapped_column(
ForeignKey("main.contact.id", ondelete="CASCADE"),
primary_key=True,
)
relationship_type: Mapped[str] = mapped_column(String(100))
closeness_weight: Mapped[int] = mapped_column(default=5)
class Contact(TableBase):
"""Contact model."""
__tablename__ = "contact"
name: Mapped[str]
age: Mapped[int | None]
bio: Mapped[str | None]
current_job: Mapped[str | None]
gender: Mapped[str | None]
goals: Mapped[str | None]
legal_name: Mapped[str | None]
profile_pic: Mapped[str | None]
safe_conversation_starters: Mapped[str | None]
self_sufficiency_score: Mapped[int | None]
social_structure_style: Mapped[str | None]
ssn: Mapped[str | None]
suffix: Mapped[str | None]
timezone: Mapped[str | None]
topics_to_avoid: Mapped[str | None]
needs: Mapped[list[Need]] = relationship(
"Need",
secondary=ContactNeed.__table__,
back_populates="contacts",
)
related_to: Mapped[list[ContactRelationship]] = relationship(
"ContactRelationship",
foreign_keys=[ContactRelationship.contact_id],
cascade="all, delete-orphan",
)
related_from: Mapped[list[ContactRelationship]] = relationship(
"ContactRelationship",
foreign_keys=[ContactRelationship.related_contact_id],
cascade="all, delete-orphan",
)
class Need(TableBase):
"""Need/accommodation model (e.g., light sensitive, ADHD)."""
__tablename__ = "need"
name: Mapped[str]
description: Mapped[str | None]
contacts: Mapped[list[Contact]] = relationship(
"Contact",
secondary=ContactNeed.__table__,
back_populates="needs",
)

View File

@@ -1,155 +0,0 @@
"""Thing."""
from __future__ import annotations
import logging
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from dataclasses import dataclass
from multiprocessing import cpu_count
from typing import TYPE_CHECKING, Any, Literal, TypeVar
if TYPE_CHECKING:
from collections.abc import Callable, Mapping, Sequence
logger = logging.getLogger(__name__)
R = TypeVar("R")
modes = Literal["normal", "early_error"]
@dataclass
class ExecutorResults[R]:
"""Dataclass to store the results and exceptions of the parallel execution."""
results: list[R]
exceptions: list[BaseException]
def __repr__(self) -> str:
"""Return a string representation of the object."""
return f"results={self.results} exceptions={self.exceptions}"
def _parallelize_base[R](
executor_type: type[ThreadPoolExecutor | ProcessPoolExecutor],
func: Callable[..., R],
kwargs_list: Sequence[Mapping[str, Any]],
max_workers: int | None,
progress_tracker: int | None,
mode: modes,
) -> ExecutorResults:
total_work = len(kwargs_list)
with executor_type(max_workers=max_workers) as executor:
futures = [executor.submit(func, **kwarg) for kwarg in kwargs_list]
results = []
exceptions = []
for index, future in enumerate(futures, 1):
if exception := future.exception():
logger.error(f"{future} raised {exception.__class__.__name__}")
exceptions.append(exception)
if mode == "early_error":
executor.shutdown(wait=False)
raise exception
continue
results.append(future.result())
if progress_tracker and index % progress_tracker == 0:
logger.info(f"Progress: {index}/{total_work}")
return ExecutorResults(results, exceptions)
def parallelize_thread[R](
func: Callable[..., R],
kwargs_list: Sequence[Mapping[str, Any]],
max_workers: int | None = None,
progress_tracker: int | None = None,
mode: modes = "normal",
) -> ExecutorResults:
"""Generic function to run a function with multiple arguments in threads.
Args:
func (Callable[..., R]): Function to run in threads.
kwargs_list (Sequence[Mapping[str, Any]]): List of dictionaries with the arguments for the function.
max_workers (int, optional): Number of workers to use. Defaults to 8.
progress_tracker (int, optional): Number of tasks to complete before logging progress.
mode (modes, optional): Mode to use. Defaults to "normal".
Returns:
tuple[list[R], list[Exception]]: List with the results and a list with the exceptions.
"""
return _parallelize_base(
executor_type=ThreadPoolExecutor,
func=func,
kwargs_list=kwargs_list,
max_workers=max_workers,
progress_tracker=progress_tracker,
mode=mode,
)
def parallelize_process[R](
func: Callable[..., R],
kwargs_list: Sequence[Mapping[str, Any]],
max_workers: int | None = None,
progress_tracker: int | None = None,
mode: modes = "normal",
) -> ExecutorResults:
"""Generic function to run a function with multiple arguments in process.
Args:
func (Callable[..., R]): Function to run in process.
kwargs_list (Sequence[Mapping[str, Any]]): List of dictionaries with the arguments for the function.
max_workers (int, optional): Number of workers to use. Defaults to 4.
progress_tracker (int, optional): Number of tasks to complete before logging progress.
mode (modes, optional): Mode to use. Defaults to "normal".
Returns:
tuple[list[R], list[Exception]]: List with the results and a list with the exceptions.
"""
if max_workers and max_workers > cpu_count():
error = f"max_workers must be less than or equal to {cpu_count()}"
raise RuntimeError(error)
return process_executor_unchecked(
func=func,
kwargs_list=kwargs_list,
max_workers=max_workers,
progress_tracker=progress_tracker,
mode=mode,
)
def process_executor_unchecked[R](
func: Callable[..., R],
kwargs_list: Sequence[Mapping[str, Any]],
max_workers: int | None,
progress_tracker: int | None,
mode: modes = "normal",
) -> ExecutorResults:
"""Generic function to run a function with multiple arguments in parallel.
Note: this function does not check if the number of workers is greater than the number of CPUs.
This can cause the system to become unresponsive.
Args:
func (Callable[..., R]): Function to run in parallel.
kwargs_list (Sequence[Mapping[str, Any]]): List of dictionaries with the arguments for the function.
max_workers (int, optional): Number of workers to use. Defaults to 8.
progress_tracker (int, optional): Number of tasks to complete before logging progress.
mode (modes, optional): Mode to use. Defaults to "normal".
Returns:
tuple[list[R], list[Exception]]: List with the results and a list with the exceptions.
"""
return _parallelize_base(
executor_type=ProcessPoolExecutor,
func=func,
kwargs_list=kwargs_list,
max_workers=max_workers,
progress_tracker=progress_tracker,
mode=mode,
)

View File

@@ -1 +0,0 @@
game_data/

View File

@@ -1 +0,0 @@
"""init."""

View File

@@ -1,675 +0,0 @@
"""Base logic for the Splendor game."""
from __future__ import annotations
import itertools
import json
import random
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Literal, Protocol
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
GemColor = Literal["white", "blue", "green", "red", "black", "gold"]
GEM_COLORS: tuple[GemColor, ...] = (
"white",
"blue",
"green",
"red",
"black",
"gold",
)
BASE_COLORS: tuple[GemColor, ...] = (
"white",
"blue",
"green",
"red",
"black",
)
GEM_ORDER: list[GemColor] = list(GEM_COLORS)
GEM_INDEX: dict[GemColor, int] = {c: i for i, c in enumerate(GEM_ORDER)}
BASE_INDEX: dict[GemColor, int] = {c: i for i, c in enumerate(BASE_COLORS)}
@dataclass(frozen=True)
class Card:
"""Development card: gives points + a permanent gem discount."""
tier: int
points: int
color: GemColor
cost: dict[GemColor, int]
@dataclass(frozen=True)
class Noble:
"""Noble tile: gives points if you have enough bonuses."""
name: str
points: int
requirements: dict[GemColor, int]
@dataclass
class PlayerState:
"""State of a player in the game."""
strategy: Strategy
tokens: dict[GemColor, int] = field(default_factory=lambda: dict.fromkeys(GEM_COLORS, 0))
discounts: dict[GemColor, int] = field(default_factory=lambda: dict.fromkeys(GEM_COLORS, 0))
cards: list[Card] = field(default_factory=list)
reserved: list[Card] = field(default_factory=list)
nobles: list[Noble] = field(default_factory=list)
card_score: int = 0
noble_score: int = 0
def total_tokens(self) -> int:
"""Total tokens in player's bank."""
return sum(self.tokens.values())
def add_noble(self, noble: Noble) -> None:
"""Add a noble to the player."""
self.nobles.append(noble)
self.noble_score = sum(noble.points for noble in self.nobles)
def add_card(self, card: Card) -> None:
"""Add a card to the player."""
self.cards.append(card)
self.card_score = sum(card.points for card in self.cards)
@property
def score(self) -> int:
"""Total points in player's cards + nobles."""
return self.card_score + self.noble_score
def can_afford(self, card: Card) -> bool:
"""Check if player can afford card, using discounts + gold."""
missing = 0
gold = self.tokens["gold"]
for color, cost in card.cost.items():
missing += max(0, cost - self.discounts.get(color, 0) - self.tokens.get(color, 0))
if missing > gold:
return False
return True
def pay_for_card(self, card: Card) -> dict[GemColor, int]:
"""Pay tokens for card, move card to tableau, return payment for bank."""
if not self.can_afford(card):
msg = f"cannot afford card {card}"
raise ValueError(msg)
payment: dict[GemColor, int] = dict.fromkeys(GEM_COLORS, 0)
gold_available = self.tokens["gold"]
for color in BASE_COLORS:
cost = card.cost.get(color, 0)
effective_cost = max(0, cost - self.discounts.get(color, 0))
use = min(self.tokens[color], effective_cost)
self.tokens[color] -= use
payment[color] += use
remaining = effective_cost - use
if remaining > 0:
use_gold = min(gold_available, remaining)
gold_available -= use_gold
self.tokens["gold"] -= use_gold
payment["gold"] += use_gold
self.add_card(card)
self.discounts[card.color] += 1
return payment
def get_default_starting_tokens(player_count: int) -> dict[GemColor, int]:
"""get_default_starting_tokens."""
token_count = (player_count * player_count - 3 * player_count + 10) // 2
return {
"white": token_count,
"blue": token_count,
"green": token_count,
"red": token_count,
"black": token_count,
"gold": 5,
}
@dataclass
class GameConfig:
"""Game configuration: gems, bank, cards, nobles, etc."""
win_score: int = 15
table_cards_per_tier: int = 4
reserve_limit: int = 3
token_limit: int = 10
turn_limit: int = 1000
minimum_tokens_to_buy_2: int = 4
max_token_take: int = 3
cards: list[Card] = field(default_factory=list)
nobles: list[Noble] = field(default_factory=list)
class GameState:
"""Game state: players, bank, decks, table, available nobles, etc."""
def __init__(
self,
config: GameConfig,
players: list[PlayerState],
bank: dict[GemColor, int],
decks_by_tier: dict[int, list[Card]],
table_by_tier: dict[int, list[Card]],
available_nobles: list[Noble],
) -> None:
"""Game state."""
self.config = config
self.players = players
self.bank = bank
self.decks_by_tier = decks_by_tier
self.table_by_tier = table_by_tier
self.available_nobles = available_nobles
self.noble_min_requirements = 0
self.get_noble_min_requirements()
self.current_player_index = 0
self.finished = False
def get_noble_min_requirements(self) -> None:
"""Find the minimum requirement for all available nobles."""
test = 0
for noble in self.available_nobles:
test = max(test, min(foo for foo in noble.requirements.values()))
self.noble_min_requirements = test
def next_player(self) -> None:
"""Advance to the next player."""
self.current_player_index = (self.current_player_index + 1) % len(self.players)
@property
def current_player(self) -> PlayerState:
"""Current player."""
return self.players[self.current_player_index]
def refill_table(self) -> None:
"""Refill face-up cards from decks."""
for tier, deck in self.decks_by_tier.items():
table = self.table_by_tier[tier]
while len(table) < self.config.table_cards_per_tier and deck:
table.append(deck.pop())
def check_winner_simple(self) -> PlayerState | None:
"""Simplified: end immediately when someone hits win_score."""
eligible = [player for player in self.players if player.score >= self.config.win_score]
if not eligible:
return None
eligible.sort(
key=lambda p: (p.score, -len(p.cards)),
reverse=True,
)
self.finished = True
return eligible[0]
class Action:
"""Marker protocol for actions."""
@dataclass
class TakeDifferent(Action):
"""Take up to 3 different gem colors."""
colors: list[GemColor]
@dataclass
class TakeDouble(Action):
"""Take two of the same color."""
color: GemColor
@dataclass
class BuyCard(Action):
"""Buy a face-up card."""
tier: int
index: int
@dataclass
class BuyCardReserved(Action):
"""Buy a face-up card."""
index: int
@dataclass
class ReserveCard(Action):
"""Reserve a face-up card."""
tier: int
index: int | None = None
from_deck: bool = False
class Strategy(Protocol):
"""Implement this to make a bot or human controller."""
def __init__(self, name: str) -> None:
"""Initialize a strategy."""
self.name = name
def choose_action(self, game: GameState, player: PlayerState) -> Action | None:
"""Return an Action, or None to concede/end."""
raise NotImplementedError
def choose_discard(
self,
game: GameState, # noqa: ARG002
player: PlayerState,
excess: int,
) -> dict[GemColor, int]:
"""Called if player has more than token_limit tokens after an action.
Default: naive auto-discard.
"""
return auto_discard_tokens(player, excess)
def choose_noble(
self,
game: GameState, # noqa: ARG002
player: PlayerState, # noqa: ARG002
nobles: list[Noble],
) -> Noble:
"""Called if player qualifies for multiple nobles. Default: first."""
return nobles[0]
def auto_discard_tokens(player: PlayerState, excess: int) -> dict[GemColor, int]:
"""Very dumb discard logic: discard from colors you have the most of."""
to_discard: dict[GemColor, int] = dict.fromkeys(GEM_COLORS, 0)
remaining = excess
while remaining > 0:
color = max(player.tokens, key=lambda c: player.tokens[c])
if player.tokens[color] == 0:
break
player.tokens[color] -= 1
to_discard[color] += 1
remaining -= 1
return to_discard
def enforce_token_limit(
game: GameState,
strategy: Strategy,
player: PlayerState,
) -> None:
"""If player has more than token_limit tokens, force discards."""
limit = game.config.token_limit
total = player.total_tokens()
if total <= limit:
return
excess = total - limit
discards = strategy.choose_discard(game, player, excess)
for color, amount in discards.items():
available = player.tokens[color]
to_remove = min(amount, available)
if to_remove <= 0:
continue
player.tokens[color] -= to_remove
game.bank[color] += to_remove
remaining = player.total_tokens() - limit
if remaining > 0:
auto = auto_discard_tokens(player, remaining)
for color, amount in auto.items():
game.bank[color] += amount
def _check_nobles_for_player(player: PlayerState, noble: Noble) -> bool:
# this rule is slower
for color, cost in noble.requirements.items(): # noqa: SIM110
if player.discounts[color] < cost:
return False
return True
def check_nobles_for_player(
game: GameState,
strategy: Strategy,
player: PlayerState,
) -> None:
"""Award at most one noble to player if they qualify."""
if game.noble_min_requirements > max(player.discounts.values()):
return
candidates = [noble for noble in game.available_nobles if _check_nobles_for_player(player, noble)]
if not candidates:
return
chosen = candidates[0] if len(candidates) == 1 else strategy.choose_noble(game, player, candidates)
if chosen not in game.available_nobles:
return
game.available_nobles.remove(chosen)
game.get_noble_min_requirements()
player.add_noble(chosen)
def apply_take_different(game: GameState, strategy: Strategy, action: TakeDifferent) -> None:
"""Mutate game state according to action."""
player = game.current_player
colors = [color for color in action.colors if color in BASE_COLORS and game.bank[color] > 0]
if not (1 <= len(colors) <= game.config.max_token_take):
return
for color in colors:
game.bank[color] -= 1
player.tokens[color] += 1
enforce_token_limit(game, strategy, player)
def apply_take_double(game: GameState, strategy: Strategy, action: TakeDouble) -> None:
"""Mutate game state according to action."""
player = game.current_player
color = action.color
if color not in BASE_COLORS:
return
if game.bank[color] < game.config.minimum_tokens_to_buy_2:
return
game.bank[color] -= 2
player.tokens[color] += 2
enforce_token_limit(game, strategy, player)
def apply_buy_card(game: GameState, _strategy: Strategy, action: BuyCard) -> None:
"""Mutate game state according to action."""
player = game.current_player
row = game.table_by_tier.get(action.tier)
if row is None or not (0 <= action.index < len(row)):
return
card = row[action.index]
if not player.can_afford(card):
return
row.pop(action.index)
payment = player.pay_for_card(card)
for color, amount in payment.items():
game.bank[color] += amount
game.refill_table()
def apply_buy_card_reserved(game: GameState, _strategy: Strategy, action: BuyCardReserved) -> None:
"""Mutate game state according to action."""
player = game.current_player
if not (0 <= action.index < len(player.reserved)):
return
card = player.reserved[action.index]
if not player.can_afford(card):
return
player.reserved.pop(action.index)
payment = player.pay_for_card(card)
for color, amount in payment.items():
game.bank[color] += amount
def apply_reserve_card(game: GameState, strategy: Strategy, action: ReserveCard) -> None:
"""Mutate game state according to action."""
player = game.current_player
if len(player.reserved) >= game.config.reserve_limit:
return
card: Card | None = None
if action.from_deck:
deck = game.decks_by_tier.get(action.tier)
if deck:
card = deck.pop()
else:
row = game.table_by_tier.get(action.tier)
if row is None:
return
if action.index is None or not (0 <= action.index < len(row)):
return
card = row.pop(action.index)
game.refill_table()
if card is None:
return
player.reserved.append(card)
if game.bank["gold"] > 0:
game.bank["gold"] -= 1
player.tokens["gold"] += 1
enforce_token_limit(game, strategy, player)
def apply_action(game: GameState, strategy: Strategy, action: Action) -> None:
"""Mutate game state according to action."""
actions = {
TakeDifferent: apply_take_different,
TakeDouble: apply_take_double,
BuyCard: apply_buy_card,
ReserveCard: apply_reserve_card,
BuyCardReserved: apply_buy_card_reserved,
}
action_func = actions.get(type(action))
if action_func is None:
msg = f"Unknown action type: {type(action)}"
raise ValueError(msg)
action_func(game, strategy, action)
# not sure how to simplify this yet
def get_legal_actions( # noqa: C901
game: GameState,
player: PlayerState | None = None,
) -> list[Action]:
"""Enumerate all syntactically legal actions for the given player.
This enforces:
- token-taking rules
- reserve limits
- affordability for buys
"""
if player is None:
player = game.players[game.current_player_index]
actions: list[Action] = []
colors_available = [c for c in BASE_COLORS if game.bank[c] > 0]
for r in (1, 2, 3):
actions.extend(TakeDifferent(colors=list(combo)) for combo in itertools.combinations(colors_available, r))
actions.extend(
TakeDouble(color=color) for color in BASE_COLORS if game.bank[color] >= game.config.minimum_tokens_to_buy_2
)
for tier, row in game.table_by_tier.items():
for idx, card in enumerate(row):
if player.can_afford(card):
actions.append(BuyCard(tier=tier, index=idx))
for idx, card in enumerate(player.reserved):
if player.can_afford(card):
actions.append(BuyCardReserved(index=idx))
if len(player.reserved) < game.config.reserve_limit:
for tier, row in game.table_by_tier.items():
for idx, _ in enumerate(row):
actions.append(
ReserveCard(tier=tier, index=idx, from_deck=False),
)
for tier, deck in game.decks_by_tier.items():
if deck:
actions.append(
ReserveCard(tier=tier, index=None, from_deck=True),
)
return actions
def create_random_cards_tier(
tier: int,
card_count: int,
cost_choices: list[int],
point_choices: list[int],
) -> list[Card]:
"""Create a random set of cards for a given tier."""
cards: list[Card] = []
for color in BASE_COLORS:
for _ in range(card_count):
cost = dict.fromkeys(GEM_COLORS, 0)
for c in BASE_COLORS:
if c == color:
continue
cost[c] = random.choice(cost_choices)
points = random.choice(point_choices)
cards.append(Card(tier=tier, points=points, color=color, cost=cost))
return cards
def create_random_cards() -> list[Card]:
"""Generate a generic but Splendor-ish set of cards.
This is not the official deck, but structured similarly enough for play.
"""
cards: list[Card] = []
cards.extend(
create_random_cards_tier(
tier=1,
card_count=5,
cost_choices=[0, 1, 1, 2],
point_choices=[0, 0, 1],
)
)
cards.extend(
create_random_cards_tier(
tier=2,
card_count=4,
cost_choices=[2, 3, 4],
point_choices=[1, 2, 2, 3],
)
)
cards.extend(
create_random_cards_tier(
tier=3,
card_count=3,
cost_choices=[4, 5, 6],
point_choices=[3, 4, 5],
)
)
random.shuffle(cards)
return cards
def create_random_nobles() -> list[Noble]:
"""A small set of noble tiles, roughly Splendor-ish."""
nobles: list[Noble] = []
base_requirements: list[dict[GemColor, int]] = [
{"white": 3, "blue": 3, "green": 3},
{"blue": 3, "green": 3, "red": 3},
{"green": 3, "red": 3, "black": 3},
{"red": 3, "black": 3, "white": 3},
{"black": 3, "white": 3, "blue": 3},
{"white": 4, "blue": 4},
{"green": 4, "red": 4},
{"blue": 4, "black": 4},
]
for idx, req in enumerate(base_requirements, start=1):
nobles.append(
Noble(
name=f"Noble {idx}",
points=3,
requirements=dict(req.items()),
),
)
return nobles
def load_nobles(file: Path) -> list[Noble]:
"""Load nobles from a file."""
nobles = json.loads(file.read_text())
return [Noble(**noble) for noble in nobles]
def load_cards(file: Path) -> list[Card]:
"""Load cards from a file."""
cards = json.loads(file.read_text())
return [Card(**card) for card in cards]
def new_game(
strategies: Sequence[Strategy],
config: GameConfig,
) -> GameState:
"""Create a new game state from a config + list of players."""
num_players = len(strategies)
bank = get_default_starting_tokens(num_players)
decks_by_tier: dict[int, list[Card]] = {1: [], 2: [], 3: []}
for card in config.cards:
decks_by_tier.setdefault(card.tier, []).append(card)
for deck in decks_by_tier.values():
random.shuffle(deck)
table_by_tier: dict[int, list[Card]] = {1: [], 2: [], 3: []}
players = [PlayerState(strategy=strategy) for strategy in strategies]
nobles = list(config.nobles)
random.shuffle(nobles)
nobles = nobles[: num_players + 1]
game = GameState(
config=config,
players=players,
bank=bank,
decks_by_tier=decks_by_tier,
table_by_tier=table_by_tier,
available_nobles=nobles,
)
game.refill_table()
return game
def run_game(game: GameState) -> tuple[PlayerState, int]:
"""Run a full game loop until someone wins or a player returns None."""
turn_count = 0
while not game.finished:
turn_count += 1
player = game.current_player
strategy = player.strategy
action = strategy.choose_action(game, player)
if action is None:
game.finished = True
break
apply_action(game, strategy, action)
check_nobles_for_player(game, strategy, player)
winner = game.check_winner_simple()
if winner is not None:
return winner, turn_count
game.next_player()
if turn_count >= game.config.turn_limit:
break
fallback = max(game.players, key=lambda player: player.score)
return fallback, turn_count

View File

@@ -1,288 +0,0 @@
"""Bot for Splendor game."""
from __future__ import annotations
import random
from .base import (
BASE_COLORS,
Action,
BuyCard,
BuyCardReserved,
Card,
GameState,
GemColor,
PlayerState,
ReserveCard,
Strategy,
TakeDifferent,
TakeDouble,
auto_discard_tokens,
get_legal_actions,
)
def can_bot_afford(player: PlayerState, card: Card) -> bool:
"""Check if player can afford card, using discounts + gold."""
missing = 0
gold = player.tokens["gold"]
for color, cost in card.cost.items():
missing += max(0, cost - player.discounts.get(color, 0) - player.tokens.get(color, 0))
if missing > gold:
return False
return True
class RandomBot(Strategy):
"""Dumb bot that follows rules but doesn't think."""
def __init__(self, name: str) -> None:
"""Initialize the bot."""
super().__init__(name=name)
def choose_action(self, game: GameState, player: PlayerState) -> Action | None:
"""Choose an action for the current player."""
affordable: list[tuple[int, int]] = []
for tier, row in game.table_by_tier.items():
for idx, card in enumerate(row):
if can_bot_afford(player, card):
affordable.append((tier, idx))
if affordable and random.random() < 0.5:
tier, idx = random.choice(affordable)
return BuyCard(tier=tier, index=idx)
if random.random() < 0.2:
tier = random.choice([1, 2, 3])
row = game.table_by_tier.get(tier, [])
if row:
idx = random.randrange(len(row))
return ReserveCard(tier=tier, index=idx, from_deck=False)
if random.random() < 0.5:
colors_for_double = [c for c in BASE_COLORS if game.bank[c] >= 4]
if colors_for_double:
return TakeDouble(color=random.choice(colors_for_double))
colors_for_diff = [c for c in BASE_COLORS if game.bank[c] > 0]
random.shuffle(colors_for_diff)
return TakeDifferent(colors=colors_for_diff[:3])
def choose_discard(
self,
game: GameState, # noqa: ARG002
player: PlayerState,
excess: int,
) -> dict[GemColor, int]:
"""Choose how many tokens to discard."""
return auto_discard_tokens(player, excess)
def check_cards_in_tier(row: list[Card], player: PlayerState) -> list[int]:
"""Check if player can afford card, using discounts + gold."""
return [index for index, card in enumerate(row) if can_bot_afford(player, card)]
class PersonalizedBot(Strategy):
"""PersonalizedBot."""
"""Dumb bot that follows rules but doesn't think."""
def __init__(self, name: str) -> None:
"""Initialize the bot."""
super().__init__(name=name)
def choose_action(self, game: GameState, player: PlayerState) -> Action | None:
"""Choose an action for the current player."""
for tier in (1, 2, 3):
row = game.table_by_tier[tier]
if affordable := check_cards_in_tier(row, player):
index = random.choice(affordable)
return BuyCard(tier=tier, index=index)
colors_for_diff = [c for c in BASE_COLORS if game.bank[c] > 0]
random.shuffle(colors_for_diff)
return TakeDifferent(colors=colors_for_diff[:3])
def choose_discard(
self,
game: GameState, # noqa: ARG002
player: PlayerState,
excess: int,
) -> dict[GemColor, int]:
"""Choose how many tokens to discard."""
return auto_discard_tokens(player, excess)
class PersonalizedBot2(Strategy):
"""PersonalizedBot2."""
"""Dumb bot that follows rules but doesn't think."""
def __init__(self, name: str) -> None:
"""Initialize the bot."""
super().__init__(name=name)
def choose_action(self, game: GameState, player: PlayerState) -> Action | None:
"""Choose an action for the current player."""
tiers = (1, 2, 3)
for tier in tiers:
row = game.table_by_tier[tier]
if affordable := check_cards_in_tier(row, player):
index = random.choice(affordable)
return BuyCard(tier=tier, index=index)
if affordable := check_cards_in_tier(player.reserved, player):
index = random.choice(affordable)
return BuyCardReserved(index=index)
colors_for_diff = [c for c in BASE_COLORS if game.bank[c] > 0]
if len(colors_for_diff) >= 3:
random.shuffle(colors_for_diff)
return TakeDifferent(colors=colors_for_diff[:3])
for tier in tiers:
len_deck = len(game.decks_by_tier[tier])
if len_deck:
return ReserveCard(tier=tier, index=None, from_deck=True)
return TakeDifferent(colors=colors_for_diff[:3])
def choose_discard(
self,
game: GameState, # noqa: ARG002
player: PlayerState,
excess: int,
) -> dict[GemColor, int]:
"""Choose how many tokens to discard."""
return auto_discard_tokens(player, excess)
def buy_card_reserved(player: PlayerState) -> Action | None:
"""Buy a card reserved."""
if affordable := check_cards_in_tier(player.reserved, player):
index = random.choice(affordable)
return BuyCardReserved(index=index)
return None
def buy_card(game: GameState, player: PlayerState) -> Action | None:
"""Buy a card."""
for tier in (1, 2, 3):
row = game.table_by_tier[tier]
if affordable := check_cards_in_tier(row, player):
index = random.choice(affordable)
return BuyCard(tier=tier, index=index)
return None
def take_tokens(game: GameState) -> Action | None:
"""Take tokens."""
colors_for_diff = [color for color in BASE_COLORS if game.bank[color] > 0]
if len(colors_for_diff) >= 3:
random.shuffle(colors_for_diff)
return TakeDifferent(colors=colors_for_diff[: game.config.max_token_take])
return None
class PersonalizedBot3(Strategy):
"""PersonalizedBot3."""
"""Dumb bot that follows rules but doesn't think."""
def __init__(self, name: str) -> None:
"""Initialize the bot."""
super().__init__(name=name)
def choose_action(self, game: GameState, player: PlayerState) -> Action | None:
"""Choose an action for the current player."""
print(len(get_legal_actions(game, player)))
print(get_legal_actions(game, player))
if action := buy_card_reserved(player):
return action
if action := buy_card(game, player):
return action
colors_for_diff = [color for color in BASE_COLORS if game.bank[color] > 0]
if len(colors_for_diff) >= 3:
random.shuffle(colors_for_diff)
return TakeDifferent(colors=colors_for_diff[:3])
for tier in (1, 2, 3):
len_deck = len(game.decks_by_tier[tier])
if len_deck:
return ReserveCard(tier=tier, index=None, from_deck=True)
return TakeDifferent(colors=colors_for_diff[:3])
def choose_discard(
self,
game: GameState, # noqa: ARG002
player: PlayerState,
excess: int,
) -> dict[GemColor, int]:
"""Choose how many tokens to discard."""
return auto_discard_tokens(player, excess)
def estimate_value_of_card(game: GameState, player: PlayerState, color: GemColor) -> int:
"""Estimate value of a color in the player's bank."""
return game.bank[color] - player.discounts.get(color, 0)
def estimate_value_of_token(game: GameState, player: PlayerState, color: GemColor) -> int:
"""Estimate value of a color in the player's bank."""
return game.bank[color] - player.discounts.get(color, 0)
class PersonalizedBot4(Strategy):
"""PersonalizedBot4."""
def __init__(self, name: str) -> None:
"""Initialize the bot."""
super().__init__(name=name)
def filter_actions(self, actions: list[Action]) -> list[Action]:
"""Filter actions to only take different."""
return [
action
for action in actions
if (isinstance(action, TakeDifferent) and len(action.colors) == 3) or not isinstance(action, TakeDifferent)
]
def choose_action(self, game: GameState, player: PlayerState) -> Action | None:
"""Choose an action for the current player."""
legal_actions = get_legal_actions(game, player)
print(len(legal_actions))
good_actions = self.filter_actions(legal_actions)
print(len(good_actions))
print(good_actions)
print(len(get_legal_actions(game, player)))
if action := buy_card_reserved(player):
return action
if action := buy_card(game, player):
return action
colors_for_diff = [color for color in BASE_COLORS if game.bank[color] > 0]
if len(colors_for_diff) >= 3:
random.shuffle(colors_for_diff)
return TakeDifferent(colors=colors_for_diff[:3])
for tier in (1, 2, 3):
len_deck = len(game.decks_by_tier[tier])
if len_deck:
return ReserveCard(tier=tier, index=None, from_deck=True)
return TakeDifferent(colors=colors_for_diff[:3])
def choose_discard(
self,
game: GameState, # noqa: ARG002
player: PlayerState,
excess: int,
) -> dict[GemColor, int]:
"""Choose how many tokens to discard."""
return auto_discard_tokens(player, excess)

View File

@@ -1,724 +0,0 @@
"""Splendor game."""
from __future__ import annotations
import sys
from typing import TYPE_CHECKING, Any
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical
from textual.widget import Widget
from textual.widgets import Footer, Header, Input, Static
from .base import (
BASE_COLORS,
GEM_COLORS,
Action,
BuyCard,
BuyCardReserved,
Card,
GameState,
GemColor,
Noble,
PlayerState,
ReserveCard,
Strategy,
TakeDifferent,
TakeDouble,
)
if TYPE_CHECKING:
from collections.abc import Mapping
# Abbreviations used when rendering costs
COST_ABBR: dict[GemColor, str] = {
"white": "W",
"blue": "B",
"green": "G",
"red": "R",
"black": "K",
"gold": "O",
}
# Abbreviations players can type on the command line
COLOR_ABBR_TO_FULL: dict[str, GemColor] = {
"w": "white",
"b": "blue",
"g": "green",
"r": "red",
"k": "black",
"o": "gold",
}
def parse_color_token(raw: str) -> GemColor:
"""Convert user input into a GemColor.
Supports:
- full names: white, blue, green, red, black, gold
- abbreviations: w, b, g, r, k, o
"""
key = raw.lower()
# full color names first
if key in BASE_COLORS:
return key # type: ignore[return-value]
# abbreviations
if key in COLOR_ABBR_TO_FULL:
return COLOR_ABBR_TO_FULL[key]
error = f"Unknown color: {raw}"
raise ValueError(error)
def format_cost(cost: Mapping[GemColor, int]) -> str:
"""Format a cost/requirements dict as colored tokens like 'B:2, R:1'.
Uses `color_token` internally so colors are guaranteed to match your bank.
"""
parts: list[str] = []
for color in GEM_COLORS:
n = cost.get(color, 0)
if not n:
continue
# color_token gives us e.g. "[blue]blue: 3[/]"
token = color_token(color, n)
# Turn the leading color name into the abbreviation (blue: 3 → B:3)
# We only replace the first occurrence.
full = f"{color}:"
abbr = f"{COST_ABBR[color]}:"
token = token.replace(full, abbr, 1)
parts.append(token)
return ", ".join(parts) if parts else "-"
def format_card(card: Card) -> str:
"""Readable card line using dataclass fields instead of __str__."""
color_abbr = COST_ABBR[card.color]
header = f"T{card.tier} {color_abbr} P{card.points}"
cost_str = format_cost(card.cost)
return f"{header} ({cost_str})"
def format_noble(noble: Noble) -> str:
"""Readable noble line using dataclass fields instead of __str__."""
cost_str = format_cost(noble.requirements)
return f"{noble.name} +{noble.points} ({cost_str})"
def format_tokens(tokens: Mapping[GemColor, int]) -> str:
"""Colored 'color: n' list for a token dict."""
return " ".join(color_token(c, tokens.get(c, 0)) for c in GEM_COLORS)
def format_discounts(discounts: Mapping[GemColor, int]) -> str:
"""Colored discounts, skipping zeros."""
parts: list[str] = []
for c in GEM_COLORS:
n = discounts.get(c, 0)
if not n:
continue
abbr = COST_ABBR[c]
fg, bg = COLOR_STYLE[c]
parts.append(f"[{fg} on {bg}]{abbr}:{n}[/{fg} on {bg}]")
return ", ".join(parts) if parts else "-"
COLOR_STYLE: dict[GemColor, tuple[str, str]] = {
"white": ("black", "white"), # fg, bg
"blue": ("bright_white", "blue"),
"green": ("bright_white", "sea_green4"),
"red": ("white", "red3"),
"black": ("white", "grey0"),
"gold": ("black", "yellow3"),
}
def fmt_gem(color: GemColor) -> str:
"""Render gem name with fg/bg matching real token color."""
fg, bg = COLOR_STYLE[color]
return f"[{fg} on {bg}] {color} [/{fg} on {bg}]"
def fmt_number(value: int) -> str:
"""Return a Rich-markup colored 'value' string."""
return f"[bold cyan]{value}[/]"
def color_token(name: GemColor, amount: int) -> str:
"""Return a Rich-markup colored 'name: n' string."""
# Map Splendor colors -> terminal colors
color_map: Mapping[GemColor, str] = {
"white": "white",
"blue": "blue",
"green": "green",
"red": "red",
"black": "grey70", # 'black' is unreadable on dark backgrounds
"gold": "yellow",
}
style = color_map.get(name, "white")
return f"[{style}]{name}: {amount}[/]"
class Board(Widget):
"""Big board widget with the layout you sketched."""
def __init__(self, game: GameState, me: PlayerState, **kwargs: Any) -> None: # noqa: ANN401
"""Initialize the board widget."""
super().__init__(**kwargs)
self.game = game
self.me = me
def compose(self) -> ComposeResult:
"""Compose the board widget."""
# Structure:
# ┌ bank row
# ├ middle row (tiers | nobles)
# └ players row
with Vertical(id="board_root"):
yield Static(id="bank_box")
with Horizontal(id="middle_row"):
with Vertical(id="tiers_box"):
yield Static(id="tier1_box")
yield Static(id="tier2_box")
yield Static(id="tier3_box")
yield Static(id="nobles_box")
yield Static(id="players_box")
def on_mount(self) -> None:
"""Refresh the board content."""
self.refresh_content()
def refresh_content(self) -> None:
"""Refresh the board content."""
self._render_bank()
self._render_tiers()
self._render_nobles()
self._render_players()
# --- sections ----------------------------------------------------
def _render_bank(self) -> None:
bank = self.game.bank
parts: list[str] = ["[b]Bank:[/b]"]
# One line, all tokens colored
parts.append(format_tokens(bank))
self.query_one("#bank_box", Static).update("\n".join(parts))
def _render_tiers(self) -> None:
for tier in (1, 2, 3):
box = self.query_one(f"#tier{tier}_box", Static)
cards: list[Card] = self.game.table_by_tier.get(tier, [])
lines: list[str] = [f"[b]Tier {tier} cards:[/b]"]
if not cards:
lines.append(" (none)")
else:
for idx, card in enumerate(cards):
lines.append(f" [{idx}] {format_card(card)}")
box.update("\n".join(lines))
def _render_nobles(self) -> None:
nobles_box = self.query_one("#nobles_box", Static)
lines: list[str] = ["[b]Nobles[/b]"]
if not self.game.available_nobles:
lines.append(" (none)")
else:
lines.extend(" - " + format_noble(noble) for noble in self.game.available_nobles)
nobles_box.update("\n".join(lines))
def _render_players(self) -> None:
players_box = self.query_one("#players_box", Static)
lines: list[str] = ["[b]Players:[/b]", ""]
for player in self.game.players:
mark = "*" if player is self.me else " "
token_str = format_tokens(player.tokens)
discount_str = format_discounts(player.discounts)
lines.append(
f"{mark} {player.name:10} Score={player.score:2d} Discounts={discount_str}",
)
lines.append(f" Tokens: {token_str}")
if player.nobles:
noble_names = ", ".join(n.name for n in player.nobles)
lines.append(f" Nobles: {noble_names}")
# Optional: show counts of cards / reserved
if player.cards:
lines.append(f" Cards: {len(player.cards)}")
if player.reserved:
lines.append(f" Reserved: {len(player.reserved)}")
lines.append("")
players_box.update("\n".join(lines))
class ActionApp(App[None]):
"""Textual app that asks for a single action command and returns an Action."""
CSS = """
Screen {
/* 3 rows: command zone, board, footer */
layout: grid;
grid-size: 1 3;
grid-rows: auto 1fr auto;
}
/* Top area with input + instructions */
#command_zone {
grid-columns: 1;
grid-rows: 1;
padding: 1 1;
}
/* Board sits in the middle row and can grow */
#board {
grid-columns: 1;
grid-rows: 2;
padding: 0 1 1 1;
}
Footer {
grid-columns: 1;
grid-rows: 3;
}
Input {
border: round $accent;
}
/* === Board layout === */
#board_root {
/* outer frame around the whole board area */
border: heavy white;
padding: 0 1;
}
/* Bank row: full width */
#bank_box {
border: heavy white;
padding: 0 1;
}
/* Middle row: tiers (left) + nobles (right) */
#middle_row {
layout: horizontal;
}
#tiers_box {
border: heavy white;
padding: 0 1;
width: 70%;
}
#tier1_box,
#tier2_box,
#tier3_box {
border-bottom: heavy white;
padding: 0 0 1 0;
margin-bottom: 1;
}
#nobles_box {
border: heavy white;
padding: 0 1;
width: 30%;
}
/* Players row: full width at bottom */
#players_box {
border: heavy white;
padding: 0 1;
}
"""
def __init__(self, game: GameState, player: PlayerState) -> None:
"""Initialize the action app."""
super().__init__()
self.game = game
self.player = player
self.result: Action | None = None
self.message: str = ""
def compose(self) -> ComposeResult:
"""Compose the action app."""
# Row 1: input + Actions text
with Vertical(id="command_zone"):
yield Input(
placeholder="Enter command, e.g. '1 white blue red' or '1 w b r' or 'q'",
id="input_line",
)
yield Static("", id="prompt")
# Row 2: board
yield Board(self.game, self.player, id="board")
# Row 3: footer
yield Footer()
def on_mount(self) -> None:
"""Mount the action app."""
self._update_prompt()
self.query_one(Input).focus()
def _update_prompt(self) -> None:
lines: list[str] = []
lines.append("[bold underline]Actions:[/]")
lines.append(
" [bold green]1[/] <colors...> - Take up to 3 different gem colors "
"(e.g. [cyan]1 white blue red[/] or [cyan]1 w b r[/])",
)
lines.append(
f" [bold green]2[/] <color> - Take 2 of the same color (needs {fmt_number(4)} in bank, "
"e.g. [cyan]2 blue[/] or [cyan]2 b[/])",
)
lines.append(
" [bold green]3[/] <tier> <idx> - Buy a face-up card (e.g. [cyan]3 1 0[/] for tier 1, index 0)",
)
lines.append(" [bold green]4[/] <idx> - Buy a reserved card")
lines.append(" [bold green]5[/] <tier> <idx> - Reserve a face-up card")
lines.append(" [bold green]6[/] <tier> - Reserve top card of a deck")
lines.append(" [bold red]q[/] - Quit game")
if self.message:
lines.append("")
lines.append(f"[bold red]Message:[/] {self.message}")
self.query_one("#prompt", Static).update("\n".join(lines))
def _cmd_1(self, parts: list[str]) -> str | None:
"""Take up to 3 different gem colors: 1 white blue red OR 1 w b r."""
color_names = parts[1:]
if not color_names:
return "Need at least one color (full name or abbreviation)."
colors: list[GemColor] = []
for name in color_names:
color = parse_color_token(name)
if self.game.bank[color] <= 0:
return f"No tokens left for color: {color}"
colors.append(color)
self.result = TakeDifferent(colors=colors[:3])
self.exit()
return None
def _cmd_2(self, parts: list[str]) -> str | None:
"""Take two of the same color."""
if len(parts) < 2:
return "Usage: 2 <color>"
color = parse_color_token(parts[1])
if self.game.bank[color] < self.game.config.minimum_tokens_to_buy_2:
return "Bank must have at least 4 of that color."
self.result = TakeDouble(color=color)
self.exit()
return None
def _cmd_3(self, parts: list[str]) -> str | None:
"""Buy face-up card."""
if len(parts) < 3:
return "Usage: 3 <tier> <index>"
tier = int(parts[1])
idx = int(parts[2])
self.result = BuyCard(tier=tier, index=idx)
self.exit()
return None
def _cmd_4(self, parts: list[str]) -> str | None:
"""Buy reserved card."""
if len(parts) < 2:
return "Usage: 4 <reserved_index>"
idx = int(parts[1])
if not (0 <= idx < len(self.player.reserved)):
return "Reserved index out of range."
self.result = BuyCardReserved(tier=0, index=idx)
self.exit()
return None
def _cmd_5(self, parts: list[str]) -> str | None:
"""Reserve face-up card."""
if len(parts) < 3:
return "Usage: 5 <tier> <index>"
tier = int(parts[1])
idx = int(parts[2])
self.result = ReserveCard(tier=tier, index=idx, from_deck=False)
self.exit()
return None
def _cmd_6(self, parts: list[str]) -> str | None:
"""Reserve top of deck."""
if len(parts) < 2:
return "Usage: 6 <tier>"
tier = int(parts[1])
self.result = ReserveCard(tier=tier, index=None, from_deck=True)
self.exit()
return None
def _unknown_cmd(self, _parts: list[str]) -> str:
return "Unknown command."
def on_input_submitted(self, event: Input.Submitted) -> None:
"""Handle user input."""
text = (event.value or "").strip()
event.input.value = ""
if not text:
return
if text.lower() in {"q", "quit", "0"}:
self.result = None
self.exit()
return
parts = text.split()
cmds = {
"1": self._cmd_1,
"2": self._cmd_2,
"3": self._cmd_3,
"4": self._cmd_4,
"5": self._cmd_5,
"6": self._cmd_6,
}
cmd = parts[0]
error = cmds.get(cmd, self._unknown_cmd)(parts)
if error:
self.message = error
self._update_prompt()
return
class DiscardApp(App[None]):
"""Textual app to choose discards when over token limit."""
CSS = """
Screen {
layout: vertical;
}
#command_zone {
padding: 1 1;
}
#board {
padding: 0 1 1 1;
}
Input {
border: round $accent;
}
"""
def __init__(self, game: GameState, player: PlayerState) -> None:
"""Initialize the discard app."""
super().__init__()
self.game = game
self.player = player
self.discards: dict[GemColor, int] = dict.fromkeys(GEM_COLORS, 0)
self.message: str = ""
def compose(self) -> ComposeResult: # type: ignore[override]
"""Compose the discard app."""
yield Header(show_clock=False)
with Vertical(id="command_zone"):
yield Input(
placeholder="Enter color to discard, e.g. 'blue' or 'b'",
id="input_line",
)
yield Static("", id="prompt")
# Board directly under the command zone
yield Board(self.game, self.player, id="board")
yield Footer()
def on_mount(self) -> None: # type: ignore[override]
"""Mount the discard app."""
self._update_prompt()
self.query_one(Input).focus()
def _remaining_to_discard(self) -> int:
return self.player.total_tokens() - sum(self.discards.values()) - self.game.config.token_limit
def _update_prompt(self) -> None:
remaining = max(self._remaining_to_discard(), 0)
lines: list[str] = []
lines.append(
"You must discard "
f"{fmt_number(remaining)} token(s) "
f"to get down to {fmt_number(self.game.config.token_limit)}.",
)
disc_str = ", ".join(f"{fmt_gem(c)}={fmt_number(self.discards[c])}" for c in GEM_COLORS)
lines.append(f"Current planned discards: {{ {disc_str} }}")
lines.append(
"Type a color name or abbreviation (e.g. 'blue' or 'b') to discard one token.",
)
if self.message:
lines.append("")
lines.append(f"[bold red]Message:[/] {self.message}")
self.query_one("#prompt", Static).update("\n".join(lines))
def on_input_submitted(self, event: Input.Submitted) -> None: # type: ignore[override]
"""Handle user input."""
raw = (event.value or "").strip()
event.input.value = ""
if not raw:
return
try:
color = parse_color_token(raw)
except ValueError:
self.message = f"Unknown color: {raw}"
self._update_prompt()
return
available = self.player.tokens[color] - self.discards[color]
if available <= 0:
self.message = f"No more {color} tokens available to discard."
self._update_prompt()
return
self.discards[color] += 1
if self._remaining_to_discard() <= 0:
self.exit()
return
self.message = ""
self._update_prompt()
# ---------------------------------------------------------------------------
# Noble choice app
# ---------------------------------------------------------------------------
class NobleChoiceApp(App[None]):
"""Textual app to choose one noble."""
CSS = """
Screen {
layout: vertical;
}
#command_zone {
padding: 1 1;
}
#board {
padding: 0 1 1 1;
}
Input {
border: round $accent;
}
"""
def __init__(
self,
game: GameState,
player: PlayerState,
nobles: list[Noble],
) -> None:
"""Initialize the noble choice app."""
super().__init__()
self.game = game
self.player = player
self.nobles = nobles
self.result: Noble | None = None
self.message: str = ""
def compose(self) -> ComposeResult: # type: ignore[override]
"""Compose the noble choice app."""
yield Header(show_clock=False)
with Vertical(id="command_zone"):
yield Input(
placeholder="Enter noble index, e.g. '0'",
id="input_line",
)
yield Static("", id="prompt")
# Board directly under the command zone
yield Board(self.game, self.player, id="board")
yield Footer()
def on_mount(self) -> None: # type: ignore[override]
"""Mount the noble choice app."""
self._update_prompt()
self.query_one(Input).focus()
def _update_prompt(self) -> None:
lines: list[str] = []
lines.append("[bold underline]You qualify for nobles:[/]")
for i, noble in enumerate(self.nobles):
lines.append(f" [bright_cyan]{i})[/] {format_noble(noble)}")
lines.append("Enter the index of the noble you want.")
if self.message:
lines.append("")
lines.append(f"[bold red]Message:[/] {self.message}")
self.query_one("#prompt", Static).update("\n".join(lines))
def on_input_submitted(self, event: Input.Submitted) -> None: # type: ignore[override]
"""Handle user input."""
raw = (event.value or "").strip()
event.input.value = ""
if not raw:
return
try:
idx = int(raw)
except ValueError:
self.message = "Please enter a valid integer index."
self._update_prompt()
return
if not (0 <= idx < len(self.nobles)):
self.message = "Index out of range."
self._update_prompt()
return
self.result = self.nobles[idx]
self.exit()
class TuiHuman(Strategy):
"""Textual-based human player Strategy with colorful board."""
def choose_action(
self,
game: GameState,
player: PlayerState,
) -> Action | None:
"""Choose an action for the player."""
if not sys.stdout.isatty():
return None
app = ActionApp(game, player)
app.run()
return app.result
def choose_discard(
self,
game: GameState,
player: PlayerState,
excess: int, # noqa: ARG002
) -> dict[GemColor, int]:
"""Choose tokens to discard."""
if not sys.stdout.isatty():
return dict.fromkeys(GEM_COLORS, 0)
app = DiscardApp(game, player)
app.run()
return app.discards
def choose_noble(
self,
game: GameState,
player: PlayerState,
nobles: list[Noble],
) -> Noble:
"""Choose a noble for the player."""
if not sys.stdout.isatty():
return nobles[0]
app = NobleChoiceApp(game, player, nobles)
app.run()
return app.result

View File

@@ -1,19 +0,0 @@
"""Main entry point for Splendor game."""
from __future__ import annotations
from .base import new_game, run_game
from .bot import RandomBot
from .human import TuiHuman
def main() -> None:
"""Main entry point."""
human = TuiHuman()
bot = RandomBot()
game_state = new_game(["You", "Bot A"])
run_game(game_state, [human, bot])
if __name__ == "__main__":
main()

View File

@@ -1,111 +0,0 @@
"""Public state for RL/search."""
from __future__ import annotations
from dataclasses import dataclass
from .base import (
BASE_COLORS,
BASE_INDEX,
GEM_ORDER,
Card,
GameState,
Noble,
PlayerState,
)
@dataclass(frozen=True)
class ObsCard:
"""Numeric-ish card view for RL/search."""
tier: int
points: int
color_index: int
cost: list[int]
@dataclass(frozen=True)
class ObsNoble:
"""Numeric-ish noble view for RL/search."""
points: int
requirements: list[int]
@dataclass(frozen=True)
class ObsPlayer:
"""Numeric-ish player view for RL/search."""
tokens: list[int]
discounts: list[int]
score: int
cards: list[ObsCard]
reserved: list[ObsCard]
nobles: list[ObsNoble]
@dataclass(frozen=True)
class Observation:
"""Full public state for RL/search."""
current_player: int
bank: list[int]
players: list[ObsPlayer]
table_by_tier: dict[int, list[ObsCard]]
decks_remaining: dict[int, int]
available_nobles: list[ObsNoble]
def _encode_card(card: Card) -> ObsCard:
color_index = BASE_INDEX.get(card.color, -1)
cost_vec = [card.cost.get(c, 0) for c in BASE_COLORS]
return ObsCard(
tier=card.tier,
points=card.points,
color_index=color_index,
cost=cost_vec,
)
def _encode_noble(noble: Noble) -> ObsNoble:
req_vec = [noble.requirements.get(c, 0) for c in BASE_COLORS]
return ObsNoble(
points=noble.points,
requirements=req_vec,
)
def _encode_player(player: PlayerState) -> ObsPlayer:
tokens_vec = [player.tokens[c] for c in GEM_ORDER]
discounts_vec = [player.discounts[c] for c in GEM_ORDER]
cards_enc = [_encode_card(c) for c in player.cards]
reserved_enc = [_encode_card(c) for c in player.reserved]
nobles_enc = [_encode_noble(n) for n in player.nobles]
return ObsPlayer(
tokens=tokens_vec,
discounts=discounts_vec,
score=player.score,
cards=cards_enc,
reserved=reserved_enc,
nobles=nobles_enc,
)
def to_observation(game: GameState) -> Observation:
"""Create a structured observation of the full public state."""
bank_vec = [game.bank[c] for c in GEM_ORDER]
players_enc = [_encode_player(p) for p in game.players]
table_enc: dict[int, list[ObsCard]] = {
tier: [_encode_card(c) for c in row] for tier, row in game.table_by_tier.items()
}
decks_remaining = {tier: len(deck) for tier, deck in game.decks_by_tier.items()}
nobles_enc = [_encode_noble(n) for n in game.available_nobles]
return Observation(
current_player=game.current_player_index,
bank=bank_vec,
players=players_enc,
table_by_tier=table_enc,
decks_remaining=decks_remaining,
available_nobles=nobles_enc,
)

View File

@@ -1,36 +0,0 @@
"""Simulate a step in the game."""
from __future__ import annotations
import copy
from .base import Action, GameState, PlayerState, apply_action, check_nobles_for_player
from .bot import RandomBot
class SimStrategy(RandomBot):
"""Strategy used in simulate_step.
We never call choose_action here (caller chooses actions),
but we reuse discard/noble-selection logic.
"""
def choose_action(self, game: GameState, player: PlayerState) -> Action | None: # noqa: ARG002
"""Choose an action for the current player."""
msg = "SimStrategy.choose_action should not be used in simulate_step"
raise RuntimeError(msg)
def simulate_step(game: GameState, action: Action) -> GameState:
"""Return a deep-copied next state after applying action for the current player.
Useful for tree search / MCTS:
next_state = simulate_step(state, action)
"""
next_state = copy.deepcopy(game)
sim_strategy = SimStrategy()
apply_action(next_state, sim_strategy, action)
check_nobles_for_player(next_state, sim_strategy, next_state.current_player)
next_state.next_player()
return next_state

View File

@@ -1,50 +0,0 @@
"""Simulator for Splendor game."""
from __future__ import annotations
from collections import defaultdict
from pathlib import Path
from statistics import mean
from .base import GameConfig, load_cards, load_nobles, new_game, run_game
from .bot import PersonalizedBot4, RandomBot
def main() -> None:
"""Main entry point."""
turn_limit = 1000
good_games = 0
games = 1
winners: dict[str, list] = defaultdict(list)
game_data = Path(__file__).parent / "game_data"
cards = load_cards(game_data / "cards/default.json")
nobles = load_nobles(game_data / "nobles/default.json")
for _ in range(games):
bot_a = RandomBot("bot_a")
bot_b = RandomBot("bot_b")
bot_c = RandomBot("bot_c")
bot_d = PersonalizedBot4("my_bot")
config = GameConfig(
cards=cards,
nobles=nobles,
turn_limit=turn_limit,
)
players = (bot_a, bot_b, bot_c, bot_d)
game_state = new_game(players, config)
winner, turns = run_game(game_state)
if turns < turn_limit:
good_games += 1
winners[winner.strategy.name].append(turns)
print(
f"out of {games} {turn_limit} turn games with {len(players)}"
f"random bots there where {good_games} games where a bot won"
)
for name, turns in winners.items():
print(f"{name} won {len(turns)} games in {mean(turns):.2f} turns")
if __name__ == "__main__":
main()

View File

@@ -1 +0,0 @@
"""init."""

View File

@@ -1,40 +0,0 @@
"""capasitor."""
def calculate_capacitor_capacity(voltage: float, farads: float) -> float:
"""Calculate capacitor capacity."""
joules = (farads * voltage**2) // 2
return joules // 3600
def calculate_pack_capacity(cells: int, cell_voltage: float, farads: float) -> float:
"""Calculate pack capacity."""
return calculate_capacitor_capacity(cells * cell_voltage, farads / cells)
def calculate_pack_capacity2(cells: int, cell_voltage: float, farads: float, cell_cost: float) -> tuple[float, float]:
"""Calculate pack capacity."""
capacitor_capacity = calculate_capacitor_capacity(cells * cell_voltage, farads / cells)
return capacitor_capacity, cell_cost * cells
def main() -> None:
"""Main."""
watt_hours = calculate_pack_capacity(cells=10, cell_voltage=2.7, farads=500)
print(f"{watt_hours=}")
print(f"{watt_hours*16=}")
watt_hours = calculate_pack_capacity(cells=1, cell_voltage=2.7, farads=5000)
print(f"{watt_hours=}")
watt_hours, cost = calculate_pack_capacity2(
cells=10,
cell_voltage=2.7,
farads=3000,
cell_cost=11.60,
)
print(f"{watt_hours=}")
print(f"{cost=}")
if __name__ == "__main__":
main()

View File

@@ -1,25 +0,0 @@
"""thing."""
def caculat_batry_specs(
cell_amp_hour: int,
cell_voltage: float,
cells_per_pack: int,
packs: int,
) -> tuple[float, float]:
"""Caculat battry specs."""
pack_voltage = cell_voltage * cells_per_pack
pack_watt_hours = pack_voltage * cell_amp_hour
battry_capacity = pack_watt_hours * packs
return (
battry_capacity,
pack_voltage,
)
battry_capacity, pack_voltage = caculat_batry_specs(300, 3.2, 8, 2)
print(f"{battry_capacity=} {pack_voltage=}")
cost = 1700
print(f"$/kWh {cost / battry_capacity}")

View File

@@ -1,196 +0,0 @@
"""voltage_drop."""
import math
from enum import Enum
class TemperatureUnit(Enum):
"""Temperature unit."""
CELSIUS = "c"
FAHRENHEIT = "f"
KELVIN = "k"
class Temperature:
"""Temperature."""
def __init__(
self,
temperature: float,
unit: TemperatureUnit = TemperatureUnit.CELSIUS,
) -> None:
"""__init__."""
unit_modifier = {
TemperatureUnit.CELSIUS: 1,
TemperatureUnit.FAHRENHEIT: 0.5556,
TemperatureUnit.KELVIN: 1.8,
}
self.temperature = temperature * unit_modifier[unit]
def __float__(self) -> float:
"""Return the temperature in degrees Celsius."""
return self.temperature
class LengthUnit(Enum):
"""Length unit."""
METERS = "m"
FEET = "ft"
INCHES = "in"
class Length:
"""Length."""
def __init__(self, length: float, unit: LengthUnit) -> None:
"""__init__."""
self.meters = self._convert_to_meters(length, unit)
def _convert_to_meters(self, length: float, unit: LengthUnit) -> float:
thing = {
LengthUnit.METERS: 1,
LengthUnit.FEET: 0.3048,
LengthUnit.INCHES: 0.0254,
}
test = thing.get(unit)
if test:
return length * test
error = f"Unsupported unit: {unit}"
raise ValueError(error)
def __float__(self) -> float:
"""Return the length in meters."""
return self.meters
def feet(self) -> float:
"""Return the length in feet."""
return self.meters * 3.2808
class MaterialType(Enum):
"""Material type."""
COPPER = "copper"
ALUMINUM = "aluminum"
CCA = "cca"
SILVER = "silver"
GOLD = "gold"
def get_material_resistivity(
material: MaterialType,
temperature: Temperature | None = None,
) -> float:
"""Get the resistivity of a material."""
if not temperature:
temperature = Temperature(20.0)
material_info = {
MaterialType.COPPER: (1.724e-8, 0.00393),
MaterialType.ALUMINUM: (2.908e-8, 0.00403),
MaterialType.CCA: (2.577e-8, 0.00397),
MaterialType.SILVER: (1.632e-8, 0.00380),
MaterialType.GOLD: (2.503e-8, 0.00340),
}
base_resistivity, temp_coefficient = material_info[material]
return base_resistivity * (1 + temp_coefficient * float(temperature))
def calculate_awg_diameter_mm(gauge: int) -> float:
"""Calculate wire diameter in millimeters for a given AWG gauge."""
return round(0.127 * 92 ** ((36 - gauge) / 39), 3)
def calculate_wire_area_m2(gauge: int) -> float:
"""Calculate the area of a wire in square meters.
Args:
gauge (int): The AWG (American Wire Gauge) number of the wire
Returns:
float: The area of the wire in square meters
"""
return math.pi * (calculate_awg_diameter_mm(gauge) / 2000) ** 2
def calculate_resistance_per_meter(gauge: int) -> float:
"""Calculate the resistance per meter of a wire.
Args:
gauge (int): The AWG (American Wire Gauge) number of the wire
Returns:
float: The resistance per meter of the wire
"""
return get_material_resistivity(MaterialType.COPPER) / calculate_wire_area_m2(gauge)
def voltage_drop(
gauge: int,
material: MaterialType,
length: Length,
current_a: float,
) -> float:
"""Calculate the voltage drop of a wire.
Args:
gauge (int): The AWG (American Wire Gauge) number of the wire
material (MaterialType): The type of conductor material (e.g., copper, aluminum)
length (Length): The length of the wire in meters
current_a (float): The current flowing through the wire in amperes
Returns:
float: The voltage drop of the wire in volts
"""
resistivity = get_material_resistivity(material)
resistance_per_meter = resistivity / calculate_wire_area_m2(gauge)
total_resistance = resistance_per_meter * float(length) * 2 # round-trip
return total_resistance * current_a
print(
voltage_drop(
gauge=10,
material=MaterialType.CCA,
length=Length(length=20, unit=LengthUnit.FEET),
current_a=20,
)
)
def max_wire_length(
gauge: int,
material: MaterialType,
current_amps: float,
voltage_drop: float = 0.3,
temperature: Temperature | None = None,
) -> Length:
"""Calculate the maximum allowable wire length based on voltage drop criteria.
Args:
gauge (int): The AWG (American Wire Gauge) number of the wire
material (MaterialType): The type of conductor material (e.g., copper, aluminum)
current_amps (float): The current flowing through the wire in amperes
voltage_drop (float, optional): Maximum allowable voltage drop as a decimal (default 0.1 or 10%)
temperature (Temperature | None, optional): The temperature of the wire. Defaults to None.
Returns:
float: Maximum wire length in meters that maintains the specified voltage drop
"""
if not temperature:
temperature = Temperature(100.0, unit=TemperatureUnit.FAHRENHEIT)
resistivity = get_material_resistivity(material, temperature)
resistance_per_meter = resistivity / calculate_wire_area_m2(gauge)
# V = IR, solve for length where V is the allowed voltage drop
return Length(
voltage_drop / (current_amps * resistance_per_meter),
LengthUnit.METERS,
)
print(max_wire_length(gauge=10, material=MaterialType.CCA, current_amps=20).feet())
print(max_wire_length(gauge=10, material=MaterialType.CCA, current_amps=10).feet())
print(max_wire_length(gauge=10, material=MaterialType.CCA, current_amps=5).feet())

Some files were not shown because too many files have changed in this diff Show More