benstull/human-experience-filter-art
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

docs: implementation plan for catalog+selection core (sub-project 1/5)

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-06-04 01:31:10 -07:00
parent 46e5d8e475
commit f15fe739ee
1 changed files with 965 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/plans/2026-06-04-catalog-and-selection-core.md
+965
View File
@@ -0,0 +1,965 @@
# Catalog & Selection Core Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build the tested Python library that defines the content catalog (load/save/validate JSONL records) and the nearest-match selection algorithm that maps a knob coordinate to a media record.
**Architecture:** A small dependency-free package `hef/` with two modules — `catalog.py` (the `Record` data model + JSONL IO + validation) and `selection.py` (coordinate distance + mode filtering + nearest-match `select`). Everything is plain stdlib Python, fully unit-tested with pytest. The actual media files and hardware are out of scope here; this is the spine that the ingest tools, review tool, and Pi player (later sub-projects) all import.
**Tech Stack:** Python 3.11+, stdlib only (`dataclasses`, `json`, `math`, `pathlib`), pytest for tests.
**Where this sits:** This is sub-project **1 of 5** from the design spec (`docs/superpowers/specs/2026-06-04-human-experience-filter-design.md`). The others get their own plans later: (2) ingest & tagging tools + review CLI, (3) Pi player runtime, (4) Arduino firmware, (5) procedural side walls. This plan must be completed first because every other plan depends on `hef.catalog` and `hef.selection`.
---
## File Structure
```
pyproject.toml project + pytest config
conftest.py empty; puts repo root on sys.path so `import hef` works
.gitignore python artifacts + .venv
hef/__init__.py empty package marker
hef/catalog.py Record model, validation, JSONL load/save/append
hef/selection.py Coordinate, Weights, distance, mode filter, select()
catalog/library.jsonl the real catalog data file (starts empty)
tests/test_smoke.py package imports
tests/test_catalog.py Record validation + dict round-trip + JSONL IO
tests/test_selection.py distance + mode filtering + select()
```
`hef/` holds the shared library (imported by the future player and tools). `catalog/` holds data only. This is a small, intentional refinement of the spec's §10 layout: the spec put "selection algorithm" under `player/`, but selection is also needed by the tagging tools, so it lives in a shared `hef/` package rather than inside the Pi-only `player/`. `player/`, `tools/`, `firmware/`, and `sidewalls/` arrive in later sub-projects.
---
### Task 1: Project scaffolding + smoke test
**Files:**
- Create: `pyproject.toml`
- Create: `conftest.py`
- Create: `.gitignore`
- Create: `hef/__init__.py`
- Create: `catalog/library.jsonl`
- Test: `tests/test_smoke.py`
- [ ] **Step 1: Create the Python environment and install pytest**
Run:
```bash
cd /Users/benstull/git/benstull.org/benstull/human-experience-filter-art
python3 -m venv .venv
source .venv/bin/activate
pip install pytest
```
Expected: pytest installs without error. Use this `.venv` (activated) for every later `pytest` command.
- [ ] **Step 2: Create `pyproject.toml`**
```toml
[project]
name = "human-experience-filter"
version = "0.1.0"
description = "Coordinate-tuned public-domain media installation"
requires-python = ">=3.11"
[tool.pytest.ini_options]
testpaths = ["tests"]
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
packages = ["hef"]
```
- [ ] **Step 3: Create `conftest.py` (empty)**
An empty `conftest.py` at the repo root makes pytest insert the repo root into `sys.path`, so `import hef` resolves without `pip install -e .`.
```python
```
(The file is intentionally empty.)
- [ ] **Step 4: Create `.gitignore`**
```gitignore
__pycache__/
*.pyc
.pytest_cache/
.venv/
```
- [ ] **Step 5: Create `hef/__init__.py` (empty) and the empty catalog file**
`hef/__init__.py`:
```python
```
(Empty package marker.)
Create the empty data file:
```bash
mkdir -p hef catalog
touch hef/__init__.py catalog/library.jsonl
```
- [ ] **Step 6: Write the smoke test**
`tests/test_smoke.py`:
```python
def test_package_imports():
import hef
import hef.catalog
import hef.selection
assert hef is not None
```
- [ ] **Step 7: Run the smoke test (expect it to fail — modules don't exist yet)**
Run: `python -m pytest tests/test_smoke.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'hef.catalog'`.
- [ ] **Step 8: Create empty module stubs to make the smoke test pass**
`hef/catalog.py`:
```python
"""Catalog data model, validation, and JSONL IO."""
```
`hef/selection.py`:
```python
"""Nearest-match selection of a catalog record for a knob coordinate."""
```
- [ ] **Step 9: Run the smoke test (expect pass)**
Run: `python -m pytest tests/test_smoke.py -v`
Expected: PASS (1 passed).
- [ ] **Step 10: Commit**
```bash
git add pyproject.toml conftest.py .gitignore hef/ catalog/library.jsonl tests/test_smoke.py
git commit -m "chore: scaffold catalog+selection core package"
```
---
### Task 2: Record model + validation
**Files:**
- Modify: `hef/catalog.py`
- Test: `tests/test_catalog.py`
- [ ] **Step 1: Write the failing tests for `Record` + `validate`**
`tests/test_catalog.py`:
```python
import pytest
from hef.catalog import Record, validate, CatalogError
def make_record(**overrides):
base = dict(
id="nasa-apollo-earthrise",
title="Earthrise",
source_url="https://archive.org/details/earthrise",
source_archive="internet_archive",
license="public_domain",
mode="video",
left=1,
right=4,
dark=1,
light=3,
duration_s=600,
file_path="/media/earthrise.mp4",
)
base.update(overrides)
return Record(**base)
def test_valid_record_passes():
validate(make_record()) # must not raise
def test_defaults_are_set():
record = make_record()
assert record.review_status == "proposed"
assert record.attribution == ""
assert record.reviewed_at is None
def test_invalid_mode_raises():
with pytest.raises(CatalogError):
validate(make_record(mode="hologram"))
def test_invalid_license_raises():
with pytest.raises(CatalogError):
validate(make_record(license="all_rights_reserved"))
def test_coordinate_out_of_range_raises():
with pytest.raises(CatalogError):
validate(make_record(left=5))
with pytest.raises(CatalogError):
validate(make_record(dark=-1))
def test_boolean_coordinate_rejected():
with pytest.raises(CatalogError):
validate(make_record(right=True))
def test_cc_by_requires_attribution():
with pytest.raises(CatalogError):
validate(make_record(license="cc_by", attribution=""))
validate(make_record(license="cc_by", attribution="Jane Doe, CC BY 4.0"))
def test_empty_id_raises():
with pytest.raises(CatalogError):
validate(make_record(id=""))
def test_negative_duration_raises():
with pytest.raises(CatalogError):
validate(make_record(duration_s=-1))
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_catalog.py -v`
Expected: FAIL with `ImportError: cannot import name 'Record' from 'hef.catalog'`.
- [ ] **Step 3: Implement `Record`, constants, and `validate`**
Replace the contents of `hef/catalog.py` with:
```python
"""Catalog data model, validation, and JSONL IO."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
MODES = frozenset({"audio", "video", "av"})
LICENSES = frozenset({"public_domain", "cc0", "cc_by", "cc_by_nc"})
REVIEW_STATUSES = frozenset({"proposed", "approved"})
COORD_FIELDS = ("left", "right", "dark", "light")
COORD_MIN = 0
COORD_MAX = 4
ATTRIBUTION_LICENSES = frozenset({"cc_by", "cc_by_nc"})
class CatalogError(ValueError):
"""Raised when a catalog record is structurally invalid."""
@dataclass
class Record:
id: str
title: str
source_url: str
source_archive: str
license: str
mode: str
left: int
right: int
dark: int
light: int
duration_s: int
file_path: str
review_status: str = "proposed"
attribution: str = ""
resolution: str = ""
dominant_color: str = ""
rationale: str = ""
reviewed_at: Optional[str] = None
notes: str = ""
def validate(record: Record) -> None:
"""Raise CatalogError if the record is structurally invalid."""
if not record.id:
raise CatalogError("record id must be non-empty")
if record.mode not in MODES:
raise CatalogError(
f"invalid mode {record.mode!r}; expected one of {sorted(MODES)}"
)
if record.license not in LICENSES:
raise CatalogError(
f"invalid license {record.license!r}; expected one of {sorted(LICENSES)}"
)
if record.review_status not in REVIEW_STATUSES:
raise CatalogError(
f"invalid review_status {record.review_status!r}; "
f"expected one of {sorted(REVIEW_STATUSES)}"
)
for axis in COORD_FIELDS:
value = getattr(record, axis)
if isinstance(value, bool) or not isinstance(value, int):
raise CatalogError(f"coordinate {axis} must be an int, got {value!r}")
if not (COORD_MIN <= value <= COORD_MAX):
raise CatalogError(
f"coordinate {axis}={value} out of range {COORD_MIN}..{COORD_MAX}"
)
if record.duration_s < 0:
raise CatalogError("duration_s must be non-negative")
if record.license in ATTRIBUTION_LICENSES and not record.attribution:
raise CatalogError(f"license {record.license} requires attribution")
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_catalog.py -v`
Expected: PASS (all tests in `test_catalog.py` pass).
- [ ] **Step 5: Commit**
```bash
git add hef/catalog.py tests/test_catalog.py
git commit -m "feat: catalog Record model and validation"
```
---
### Task 3: Record <-> dict conversion
**Files:**
- Modify: `hef/catalog.py`
- Test: `tests/test_catalog.py`
- [ ] **Step 1: Add failing tests for dict conversion**
Append to `tests/test_catalog.py`:
```python
from hef.catalog import record_to_dict, record_from_dict
def test_dict_round_trip():
record = make_record()
restored = record_from_dict(record_to_dict(record))
assert restored == record
def test_record_from_dict_rejects_unknown_field():
data = record_to_dict(make_record())
data["bogus"] = 1
with pytest.raises(CatalogError):
record_from_dict(data)
def test_record_from_dict_rejects_missing_required_field():
data = record_to_dict(make_record())
del data["title"]
with pytest.raises(CatalogError):
record_from_dict(data)
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_catalog.py -k "dict" -v`
Expected: FAIL with `ImportError: cannot import name 'record_to_dict'`.
- [ ] **Step 3: Implement the conversion functions**
Add to the imports at the top of `hef/catalog.py`:
```python
from dataclasses import dataclass, asdict, fields
```
(Replace the existing `from dataclasses import dataclass` line.)
Append to `hef/catalog.py`:
```python
def record_to_dict(record: Record) -> dict:
"""Convert a Record to a plain dict suitable for JSON serialization."""
return asdict(record)
def record_from_dict(data: dict) -> Record:
"""Build a Record from a dict, rejecting unknown or missing fields."""
known = {f.name for f in fields(Record)}
unknown = set(data) - known
if unknown:
raise CatalogError(f"unknown fields: {sorted(unknown)}")
try:
return Record(**data)
except TypeError as exc:
raise CatalogError(str(exc)) from exc
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_catalog.py -v`
Expected: PASS (all `test_catalog.py` tests, including the new dict tests).
- [ ] **Step 5: Commit**
```bash
git add hef/catalog.py tests/test_catalog.py
git commit -m "feat: catalog Record dict conversion"
```
---
### Task 4: JSONL load / save / append
**Files:**
- Modify: `hef/catalog.py`
- Test: `tests/test_catalog.py`
- [ ] **Step 1: Add failing tests for JSONL IO**
Append to `tests/test_catalog.py`:
```python
from hef.catalog import load_catalog, save_catalog, append_record
def test_save_then_load_round_trip(tmp_path):
path = tmp_path / "library.jsonl"
records = [make_record(id="a"), make_record(id="b", mode="audio")]
save_catalog(records, path)
loaded = load_catalog(path)
assert loaded == records
def test_load_skips_blank_lines(tmp_path):
path = tmp_path / "library.jsonl"
save_catalog([make_record(id="a")], path)
with path.open("a", encoding="utf-8") as fh:
fh.write("\n \n")
loaded = load_catalog(path)
assert len(loaded) == 1
def test_load_empty_file_returns_empty_list(tmp_path):
path = tmp_path / "library.jsonl"
path.write_text("", encoding="utf-8")
assert load_catalog(path) == []
def test_load_invalid_json_raises(tmp_path):
path = tmp_path / "library.jsonl"
path.write_text("{not json}\n", encoding="utf-8")
with pytest.raises(CatalogError):
load_catalog(path)
def test_load_validates_records(tmp_path):
path = tmp_path / "library.jsonl"
save_catalog([make_record(id="a")], path)
import json
bad = record_to_dict(make_record(id="bad"))
bad["left"] = 9
with path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(bad) + "\n")
with pytest.raises(CatalogError):
load_catalog(path)
def test_append_record(tmp_path):
path = tmp_path / "library.jsonl"
save_catalog([make_record(id="a")], path)
append_record(make_record(id="b"), path)
loaded = load_catalog(path)
assert [r.id for r in loaded] == ["a", "b"]
def test_save_rejects_invalid_record(tmp_path):
path = tmp_path / "library.jsonl"
with pytest.raises(CatalogError):
save_catalog([make_record(mode="hologram")], path)
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_catalog.py -k "load or save or append" -v`
Expected: FAIL with `ImportError: cannot import name 'load_catalog'`.
- [ ] **Step 3: Implement the JSONL IO functions**
Add these two imports to `hef/catalog.py` **below** the existing
`from dataclasses import ...` / `from typing import Optional` lines (they must come
after `from __future__ import annotations`, which has to stay the first statement):
```python
import json
from pathlib import Path
```
Append to `hef/catalog.py`:
```python
def load_catalog(path) -> list[Record]:
"""Load and validate every record from a JSONL file. Blank lines skipped."""
path = Path(path)
records: list[Record] = []
with path.open("r", encoding="utf-8") as fh:
for lineno, raw in enumerate(fh, start=1):
line = raw.strip()
if not line:
continue
try:
data = json.loads(line)
except json.JSONDecodeError as exc:
raise CatalogError(f"line {lineno}: invalid JSON: {exc}") from exc
record = record_from_dict(data)
validate(record)
records.append(record)
return records
def save_catalog(records, path) -> None:
"""Validate and write all records to a JSONL file (overwrites)."""
path = Path(path)
with path.open("w", encoding="utf-8") as fh:
for record in records:
validate(record)
fh.write(json.dumps(record_to_dict(record), ensure_ascii=False) + "\n")
def append_record(record, path) -> None:
"""Validate and append a single record to a JSONL file."""
validate(record)
path = Path(path)
with path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(record_to_dict(record), ensure_ascii=False) + "\n")
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_catalog.py -v`
Expected: PASS (all `test_catalog.py` tests).
- [ ] **Step 5: Commit**
```bash
git add hef/catalog.py tests/test_catalog.py
git commit -m "feat: catalog JSONL load/save/append"
```
---
### Task 5: Coordinate + distance
**Files:**
- Modify: `hef/selection.py`
- Test: `tests/test_selection.py`
- [ ] **Step 1: Write failing tests for distance**
`tests/test_selection.py`:
```python
import math
import pytest
from hef.catalog import Record
from hef.selection import Coordinate, Weights, distance, record_coordinate
def make_record(**overrides):
base = dict(
id="r",
title="t",
source_url="u",
source_archive="internet_archive",
license="public_domain",
mode="video",
left=0,
right=0,
dark=0,
light=0,
duration_s=600,
file_path="/media/r.mp4",
)
base.update(overrides)
return Record(**base)
def test_distance_zero_for_same_point():
c = Coordinate(2, 2, 2, 2)
assert distance(c, c) == 0.0
def test_distance_is_euclidean_by_default():
a = Coordinate(0, 0, 0, 0)
b = Coordinate(1, 1, 1, 1)
assert distance(a, b) == pytest.approx(2.0) # sqrt(1+1+1+1)
def test_weights_scale_planes_independently():
a = Coordinate(0, 0, 0, 0)
brain_only = Coordinate(1, 0, 0, 0)
mood_only = Coordinate(0, 0, 1, 0)
w = Weights(brain=4.0, mood=1.0)
assert distance(a, brain_only, w) == pytest.approx(2.0) # sqrt(4*1)
assert distance(a, mood_only, w) == pytest.approx(1.0) # sqrt(1*1)
def test_record_coordinate_extracts_axes():
record = make_record(left=1, right=2, dark=3, light=4)
assert record_coordinate(record) == Coordinate(1, 2, 3, 4)
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_selection.py -v`
Expected: FAIL with `ImportError: cannot import name 'Coordinate' from 'hef.selection'`.
- [ ] **Step 3: Implement Coordinate, Weights, distance, record_coordinate**
Replace the contents of `hef/selection.py` with:
```python
"""Nearest-match selection of a catalog record for a knob coordinate."""
from __future__ import annotations
import math
from dataclasses import dataclass
from hef.catalog import Record
SELECTOR_MODES = frozenset({"none", "audio", "video", "av"})
CONTENT_MODES = frozenset({"audio", "video", "av"})
@dataclass(frozen=True)
class Coordinate:
left: int
right: int
dark: int
light: int
@dataclass(frozen=True)
class Weights:
brain: float = 1.0
mood: float = 1.0
def distance(a: Coordinate, b: Coordinate, weights: Weights = Weights()) -> float:
"""Weighted Euclidean distance; brain plane and mood plane weighted separately."""
brain_sq = (a.left - b.left) ** 2 + (a.right - b.right) ** 2
mood_sq = (a.dark - b.dark) ** 2 + (a.light - b.light) ** 2
return math.sqrt(weights.brain * brain_sq + weights.mood * mood_sq)
def record_coordinate(record: Record) -> Coordinate:
"""The (left, right, dark, light) coordinate of a catalog record."""
return Coordinate(record.left, record.right, record.dark, record.light)
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_selection.py -v`
Expected: PASS (the distance tests pass).
- [ ] **Step 5: Commit**
```bash
git add hef/selection.py tests/test_selection.py
git commit -m "feat: selection Coordinate and weighted distance"
```
---
### Task 6: Mode filtering with A+V fallback
**Files:**
- Modify: `hef/selection.py`
- Test: `tests/test_selection.py`
- [ ] **Step 1: Add failing tests for `candidates_for_mode`**
Append to `tests/test_selection.py`:
```python
from hef.selection import candidates_for_mode
def test_candidates_filter_to_exact_mode():
records = [make_record(id="v", mode="video"), make_record(id="a", mode="audio")]
result = candidates_for_mode(records, "audio", pool_size=4)
assert [r.id for r in result] == ["a"]
def test_av_does_not_fall_back_when_enough_av():
records = [make_record(id=f"av{i}", mode="av") for i in range(4)]
records.append(make_record(id="a", mode="audio"))
result = candidates_for_mode(records, "av", pool_size=4)
assert {r.id for r in result} == {"av0", "av1", "av2", "av3"}
def test_av_falls_back_to_audio_and_video_when_thin():
records = [
make_record(id="av0", mode="av"),
make_record(id="a", mode="audio"),
make_record(id="v", mode="video"),
]
result = candidates_for_mode(records, "av", pool_size=4)
assert {r.id for r in result} == {"av0", "a", "v"}
def test_candidates_rejects_none_mode():
with pytest.raises(ValueError):
candidates_for_mode([], "none", pool_size=4)
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_selection.py -k "candidates or av" -v`
Expected: FAIL with `ImportError: cannot import name 'candidates_for_mode'`.
- [ ] **Step 3: Implement `candidates_for_mode`**
Append to `hef/selection.py`:
```python
def candidates_for_mode(records, mode: str, pool_size: int) -> list[Record]:
"""Records eligible for a content mode.
For 'av', if fewer than pool_size native 'av' records exist, fall back to
including 'audio' and 'video' records so the pool is never starved.
"""
if mode not in CONTENT_MODES:
raise ValueError(
f"candidates_for_mode expects a content mode {sorted(CONTENT_MODES)}, "
f"got {mode!r}"
)
primary = [r for r in records if r.mode == mode]
if mode == "av" and len(primary) < pool_size:
extra = [r for r in records if r.mode in {"audio", "video"}]
return primary + extra
return primary
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_selection.py -v`
Expected: PASS (mode-filtering tests pass).
- [ ] **Step 5: Commit**
```bash
git add hef/selection.py tests/test_selection.py
git commit -m "feat: selection mode filtering with av fallback"
```
---
### Task 7: The `select` function
**Files:**
- Modify: `hef/selection.py`
- Test: `tests/test_selection.py`
- [ ] **Step 1: Add failing tests for `select`**
Append to `tests/test_selection.py`:
```python
import random
from hef.selection import select
def test_none_mode_returns_none():
records = [make_record(id="v", mode="video")]
assert select(records, Coordinate(0, 0, 0, 0), "none") is None
def test_empty_pool_returns_none():
assert select([], Coordinate(0, 0, 0, 0), "video") is None
def test_select_returns_nearest_by_default():
near = make_record(id="near", mode="video", left=2, right=2, dark=2, light=2)
far = make_record(id="far", mode="video", left=0, right=0, dark=0, light=0)
result = select([far, near], Coordinate(2, 2, 2, 2), "video")
assert result.id == "near"
def test_select_is_deterministic_without_rng():
records = [
make_record(id="b", mode="video", left=1),
make_record(id="a", mode="video", left=1),
]
# Equal distance -> tie broken by id, so "a" wins deterministically.
result = select(records, Coordinate(1, 0, 0, 0), "video")
assert result.id == "a"
def test_select_with_rng_picks_within_nearest_pool():
records = [make_record(id=f"r{i}", mode="video", left=i % 5) for i in range(10)]
coord = Coordinate(2, 0, 0, 0)
rng = random.Random(0)
chosen = {select(records, coord, "video", pool_size=3, rng=rng).id for _ in range(50)}
# Only records inside the 3-nearest pool may ever be chosen.
ranked = sorted(records, key=lambda r: abs(r.left - 2))
allowed = {r.id for r in ranked[:3]}
assert chosen <= allowed
def test_approved_only_excludes_proposed():
proposed = make_record(id="p", mode="video", review_status="proposed")
approved = make_record(id="ok", mode="video", review_status="approved")
result = select([proposed, approved], Coordinate(0, 0, 0, 0), "video",
approved_only=True)
assert result.id == "ok"
def test_invalid_selector_mode_raises():
with pytest.raises(ValueError):
select([], Coordinate(0, 0, 0, 0), "telepathy")
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_selection.py -k "select or none or approved or rng" -v`
Expected: FAIL with `ImportError: cannot import name 'select'`.
- [ ] **Step 3: Implement `select`**
Append to `hef/selection.py`:
```python
from typing import Optional
def select(
records,
coord: Coordinate,
mode: str,
*,
pool_size: int = 4,
weights: Weights = Weights(),
approved_only: bool = False,
rng=None,
) -> Optional[Record]:
"""Pick the record nearest `coord` for the given selector `mode`.
- 'none' selector mode returns None (the void/rest state).
- With rng=None, returns the single nearest record (ties broken by id) for
deterministic behavior. Pass a random.Random to shuffle within the
pool_size nearest records.
- approved_only restricts to records the human has blessed.
"""
if mode not in SELECTOR_MODES:
raise ValueError(
f"invalid selector mode {mode!r}; expected one of {sorted(SELECTOR_MODES)}"
)
if mode == "none":
return None
pool = records
if approved_only:
pool = [r for r in pool if r.review_status == "approved"]
candidates = candidates_for_mode(pool, mode, pool_size)
if not candidates:
return None
ranked = sorted(
candidates,
key=lambda r: (distance(coord, record_coordinate(r), weights), r.id),
)
nearest = ranked[:pool_size]
if rng is None:
return nearest[0]
return rng.choice(nearest)
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_selection.py -v`
Expected: PASS (all selection tests).
- [ ] **Step 5: Commit**
```bash
git add hef/selection.py tests/test_selection.py
git commit -m "feat: nearest-match select function"
```
---
### Task 8: End-to-end integration test + empty catalog check
**Files:**
- Test: `tests/test_integration.py`
- [ ] **Step 1: Write the integration test**
This proves the full spine: build records, write them to JSONL, load them back, and select against a knob coordinate. It also asserts the committed empty `catalog/library.jsonl` loads cleanly to an empty list.
`tests/test_integration.py`:
```python
from pathlib import Path
from hef.catalog import Record, save_catalog, load_catalog
from hef.selection import Coordinate, select
REPO_ROOT = Path(__file__).resolve().parent.parent
def _rec(id, mode, left, right, dark, light):
return Record(
id=id,
title=id,
source_url=f"https://example.org/{id}",
source_archive="internet_archive",
license="public_domain",
mode=mode,
left=left,
right=right,
dark=dark,
light=light,
duration_s=600,
file_path=f"/media/{id}.mp4",
)
def test_catalog_round_trip_then_select(tmp_path):
library = [
_rec("lecture", "av", left=4, right=0, dark=1, light=2),
_rec("nebula", "video", left=0, right=4, dark=2, light=3),
_rec("storm", "av", left=0, right=3, dark=4, light=0),
_rec("sunrise", "av", left=1, right=3, dark=0, light=4),
]
path = tmp_path / "library.jsonl"
save_catalog(library, path)
loaded = load_catalog(path)
# A right-brain, very-light tuning should land on "sunrise".
chosen = select(loaded, Coordinate(left=1, right=3, dark=0, light=4), "av")
assert chosen.id == "sunrise"
# The void state returns nothing regardless of library.
assert select(loaded, Coordinate(0, 0, 0, 0), "none") is None
def test_committed_empty_catalog_loads():
path = REPO_ROOT / "catalog" / "library.jsonl"
assert load_catalog(path) == []
```
- [ ] **Step 2: Run the integration test**
Run: `python -m pytest tests/test_integration.py -v`
Expected: PASS (2 passed).
- [ ] **Step 3: Run the full suite**
Run: `python -m pytest -v`
Expected: PASS (every test across smoke, catalog, selection, integration).
- [ ] **Step 4: Commit**
```bash
git add tests/test_integration.py
git commit -m "test: end-to-end catalog+selection integration"
```
---
## Done criteria
- `python -m pytest -v` passes from the repo root inside the `.venv`.
- `hef.catalog` exposes `Record`, `validate`, `record_to_dict`, `record_from_dict`, `load_catalog`, `save_catalog`, `append_record`, `CatalogError`.
- `hef.selection` exposes `Coordinate`, `Weights`, `distance`, `record_coordinate`, `candidates_for_mode`, `select`, `SELECTOR_MODES`.
- `catalog/library.jsonl` exists, is empty, and loads to `[]`.
This spine is what sub-project 2 (ingest & review tools) will write records into, and what sub-project 3 (the Pi player) will call `select()` on each time a knob moves.