diff --git a/src/kitty_workbench/project.py b/src/kitty_workbench/project.py new file mode 100644 index 0000000..032f899 --- /dev/null +++ b/src/kitty_workbench/project.py @@ -0,0 +1,111 @@ +"""Project directory management — create, log, read, list.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +DEFAULT_WORKBENCH_DIR = Path.home() / "Kitty-Workbench" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") + + +def project_path(name: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR) -> Path: + return workbench_dir / name + + +def project_exists(name: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR) -> bool: + return project_path(name, workbench_dir).is_dir() + + +def create_project( + name: str, title: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR +) -> Path: + """Create a project directory with log files. Idempotent — won't overwrite existing logs.""" + pdir = project_path(name, workbench_dir) + pdir.mkdir(parents=True, exist_ok=True) + (pdir / "assets").mkdir(exist_ok=True) + + md_path = pdir / "session.md" + if not md_path.exists(): + md_path.write_text(f"# {title} — Session Log\n\nStarted: {_now_iso()}\n\n") + + jsonl_path = pdir / "session.jsonl" + if not jsonl_path.exists(): + jsonl_path.write_text("") + + cost_path = pdir / "cost-log.jsonl" + if not cost_path.exists(): + cost_path.write_text("") + + return pdir + + +def append_log( + name: str, + entry: str, + data: Optional[dict] = None, + level: str = "info", + workbench_dir: Path = DEFAULT_WORKBENCH_DIR, +) -> dict: + """Append a log entry to session.md and session.jsonl. Returns the entry dict.""" + pdir = project_path(name, workbench_dir) + ts = _now_iso() + + with open(pdir / "session.md", "a") as f: + f.write(f"\n### {ts}\n{entry}\n") + + obj = {"ts": ts, "entry": entry, "level": level} + if data: + obj.update(data) + with open(pdir / "session.jsonl", "a") as f: + f.write(json.dumps(obj) + "\n") + + return obj + + +def read_log( + name: str, tail: int = 20, workbench_dir: Path = DEFAULT_WORKBENCH_DIR +) -> list[dict]: + """Read recent log entries from session.jsonl.""" + jsonl_path = project_path(name, workbench_dir) / "session.jsonl" + if not jsonl_path.exists(): + return [] + + lines = jsonl_path.read_text().strip().split("\n") + lines = [l for l in lines if l.strip()] + recent = lines[-tail:] if len(lines) > tail else lines + + entries = [] + for line in recent: + try: + entries.append(json.loads(line)) + except json.JSONDecodeError: + entries.append({"raw": line}) + return entries + + +def list_projects(workbench_dir: Path = DEFAULT_WORKBENCH_DIR) -> list[dict]: + """List all project directories.""" + if not workbench_dir.exists(): + return [] + projects = [] + for d in sorted(workbench_dir.iterdir()): + if d.is_dir() and not d.name.startswith(".") and d.name != "START.md": + projects.append({"name": d.name}) + return projects + + +def log_session_event( + name: str, event: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR, **extra +) -> None: + """Append to cost-log.jsonl.""" + pdir = project_path(name, workbench_dir) + obj = {"ts": _now_iso(), "event": event, "project": name} + obj.update(extra) + with open(pdir / "cost-log.jsonl", "a") as f: + f.write(json.dumps(obj) + "\n") diff --git a/tests/test_project.py b/tests/test_project.py new file mode 100644 index 0000000..8d7b8f8 --- /dev/null +++ b/tests/test_project.py @@ -0,0 +1,67 @@ +import json +from kitty_workbench.project import ( + create_project, project_exists, project_path, + append_log, read_log, list_projects, log_session_event, +) + + +def test_create_project(tmp_workbench): + path = create_project("io102", "Heathkit IO-102", workbench_dir=tmp_workbench) + assert path.exists() + assert (path / "session.md").exists() + assert (path / "session.jsonl").exists() + assert (path / "cost-log.jsonl").exists() + assert (path / "assets").is_dir() + md = (path / "session.md").read_text() + assert "Heathkit IO-102" in md + + +def test_create_project_idempotent(tmp_workbench): + create_project("io102", "First", workbench_dir=tmp_workbench) + append_log("io102", "test entry", workbench_dir=tmp_workbench) + create_project("io102", "Second", workbench_dir=tmp_workbench) + entries = read_log("io102", workbench_dir=tmp_workbench) + assert len(entries) == 1 + + +def test_project_exists(tmp_workbench): + assert not project_exists("io102", workbench_dir=tmp_workbench) + create_project("io102", "Test", workbench_dir=tmp_workbench) + assert project_exists("io102", workbench_dir=tmp_workbench) + + +def test_append_and_read_log(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + append_log("io102", "R412 measured 1.05M", data={"ohms": 1050000}, workbench_dir=tmp_workbench) + append_log("io102", "R413 OK", workbench_dir=tmp_workbench) + entries = read_log("io102", tail=10, workbench_dir=tmp_workbench) + assert len(entries) == 2 + assert entries[0]["entry"] == "R412 measured 1.05M" + assert entries[0]["ohms"] == 1050000 + assert entries[1]["entry"] == "R413 OK" + + +def test_read_log_tail(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + for i in range(30): + append_log("io102", f"Entry {i}", workbench_dir=tmp_workbench) + entries = read_log("io102", tail=5, workbench_dir=tmp_workbench) + assert len(entries) == 5 + assert entries[0]["entry"] == "Entry 25" + + +def test_list_projects(tmp_workbench): + assert list_projects(workbench_dir=tmp_workbench) == [] + create_project("io102", "First", workbench_dir=tmp_workbench) + create_project("psu-rebuild", "Second", workbench_dir=tmp_workbench) + projects = list_projects(workbench_dir=tmp_workbench) + assert [p["name"] for p in projects] == ["io102", "psu-rebuild"] + + +def test_log_session_event(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + log_session_event("io102", "session_start", workbench_dir=tmp_workbench) + cost_log = (tmp_workbench / "io102" / "cost-log.jsonl").read_text().strip() + entry = json.loads(cost_log.split("\n")[-1]) + assert entry["event"] == "session_start" + assert entry["project"] == "io102"