homelab/stack
1
0
Fork 0
Code Issues 262 Pull Requests 2 Actions Packages Projects 1 Releases Wiki Activity

feat(pfs): lineage for every payable code — one inverted pass over fr_anchors, stack pfs lineage --all-payable (refs #691 #698)

Browse Source
This commit is contained in:
kert
2026-09-10 10:03:59 -04:00
parent d62480c3e1
commit 78da985b3a
4 changed files with 328 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
src/cli/pfs.py
View File

@@ -42,7 +42,7 @@ from pfs.cpt_load import ingest as cpt_ingest
from pfs.extract import extract_code
from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
from pfs.guidance import build as build_guidance
from pfs.lineage import lineage
from pfs.lineage import lineage, lineage_all
from pfs.reaction import series as reaction_series
log = logging.getLogger(__name__)
@@ -205,20 +205,83 @@ def _print_events(events: list[Any]) -> None:
typer.echo(f"{e.year} {e.kind:<18}{arrow:<16}{anchor}{flag}")
def _print_lineage_all(by_code: dict[str, list[Any]]) -> None:
per_source: dict[str, int] = {}
with_events = 0
unanchored = 0
for events in by_code.values():
if events:
with_events += 1
for e in events:
per_source[e.source] = per_source.get(e.source, 0) + 1
if e.source == "rvu" and not e.anchored:
unanchored += 1
detail = ", ".join(f"{k}={v}" for k, v in sorted(per_source.items())) or "none"
typer.echo(
f"lineage: {len(by_code)} codes targeted, {with_events} with >=1 event, "
f"events per source: {detail}, unanchored rvu events: {unanchored}"
)
@app.command("lineage")
def lineage_cmd(
code: str = typer.Option(..., "--code"),
code: str = typer.Option("", "--code"),
all_payable: bool = typer.Option(
False,
"--all-payable",
help="Every A/R/T code in the newest RVU year, one inverted pass over fr_anchors.",
),
limit: int = typer.Option(
0,
"--limit",
help="Only the first N target codes (sorted) — for smoke runs with --all-payable.",
),
write: bool = typer.Option(
False, "--write", help="Persist to pfs.code_event and republish."
),
) -> None:
"""Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked."""
"""Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked.
--all-payable computes lineage for every payable code in ONE pass over
fr_anchors instead of one query per code.
"""
if not code and not all_payable:
raise typer.BadParameter("pass --code or --all-payable")
store = _store()
if all_payable:
# Ruling A13: resolve targets and build every code's lineage
# against a plain DuckDB read (I4) — never the write lock a
# notebook may be holding (#508-#514). The batch below only ever
# calls write_events and publishes.
con = _read()
try:
targets = _codes_for(con, [], [], True)
if limit:
targets = targets[:limit]
by_code = lineage_all(con, store, targets)
finally:
con.close()
if write:
with _batch() as con:
ensure_tables(con)
for c in targets:
write_events(con, c, by_code.get(c, []))
_publish()
_print_lineage_all(by_code)
return
if write:
# Same A13 discipline for the single-code path: build the events
# against a plain read, then open the batch only to write them.
con = _read()
try:
events = lineage(con, store, code.upper())
finally:
con.close()
_print_events(events)
with _batch() as con:
ensure_tables(con)
events = lineage(con, store, code.upper())
_print_events(events)
write_events(con, code.upper(), events)
_publish()
return

139
src/pfs/lineage.py
View File

@@ -23,7 +23,7 @@ from __future__ import annotations
import dataclasses
import re
from typing import Any
from typing import Any, Sequence
from pfs.codetables import EventRow, cpt_years, is_missing_table_error
from pfs.descriptors import rule_year_of
@@ -194,28 +194,39 @@ def _expand_ranges(text: str) -> str:
return text
def fr_events(store: Any, code: str) -> list[EventRow]:
def _events_for(
code: str, rows: Sequence[Any], *, expanded: bool = False
) -> list[EventRow]:
"""The event-verb logic for *code* over already-fetched *rows*
(``item_key, p_id, page, text, title, date_published`` — a sqlite
``Row`` mapping or a plain dict). ``expanded=True`` means ``text`` is
already range-expanded and FR-citation-stripped (``fr_events_bucketed``
computes that once per row, shared across every code it hits) — pass
``expanded=False`` (the default) for raw ``fr_anchors.text``.
Presence and "other codes nearby" both go through ``find_codes`` (word
boundaries, FR-citation-safe) rather than a bare substring check, so
this agrees exactly with the ``find_codes(text) & targets`` hit test
``fr_events_bucketed`` uses to decide which bucket a row lands in."""
code = code.upper()
con = store._con()
rows = con.execute(
"SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
"FROM fr_anchors a JOIN items i ON i.key = a.item_key WHERE a.text LIKE ? OR a.text LIKE ?",
(f"%{code}%", f"%{code[:2]}___-{code[:2]}___%"),
).fetchall()
patterns = _patterns(code)
seen: set[tuple[int, str, str, int]] = set()
out: list[EventRow] = []
for r in rows:
text = _expand_ranges(r["text"])
# An FR page citation's number ("91 FR 99490") is not a mention of
# the code — strip it before both the presence check and every
# event pattern below (M1: a "crosswalk ... 91 FR 99490" sentence
# used to read as a crosswalk event for 99490).
text = FR_CITE_RE.sub(" ", text)
if code not in text.upper():
if expanded:
text = r["text"]
else:
# An FR page citation's number ("91 FR 99490") is not a
# mention of the code — strip it before both the presence
# check and every event pattern below (M1: a "crosswalk ...
# 91 FR 99490" sentence used to read as a crosswalk event
# for 99490).
text = FR_CITE_RE.sub(" ", _expand_ranges(r["text"]))
codes_here = find_codes(text)
if code not in codes_here:
continue
year = rule_year_of(r["title"], r["date_published"])
others = tuple(c for c in find_codes(text) if c != code)
others = tuple(c for c in codes_here if c != code)
for kind, pat, direction in patterns:
m = pat.search(text)
if not m:
@@ -265,6 +276,53 @@ def fr_events(store: Any, code: str) -> list[EventRow]:
return sorted(out, key=lambda e: (e.year, FR_KINDS.index(e.kind), e.p_id))
def fr_events(store: Any, code: str) -> list[EventRow]:
code = code.upper()
con = store._con()
rows = con.execute(
"SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
"FROM fr_anchors a JOIN items i ON i.key = a.item_key WHERE a.text LIKE ? OR a.text LIKE ?",
(f"%{code}%", f"%{code[:2]}___-{code[:2]}___%"),
).fetchall()
return _events_for(code, rows)
def fr_events_bucketed(store: Any, codes: Sequence[str]) -> dict[str, list[EventRow]]:
"""``_events_for`` for every one of *codes* in ONE streaming pass over
``fr_anchors`` instead of one SQL query (two LIKE prefilters) per
code — the whole point for an 8.7k-code universe over 193k
paragraphs. Each row's text is range-expanded and FR-citation-stripped
exactly once, then dropped into every target code's bucket that
``find_codes`` finds it naming.
``fr_events_bucketed(store, [c])[c] == fr_events(store, c)`` for any
single code ``c``: ``fr_events``'s LIKE prefilter is a superset of its
own presence check, and both presence checks now agree (``_events_for``
docstring), so the two paths select exactly the same rows for *c*."""
targets = {c.upper() for c in codes}
buckets: dict[str, list[dict[str, Any]]] = {c: [] for c in targets}
con = store._con()
for r in con.execute(
"SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
"FROM fr_anchors a JOIN items i ON i.key = a.item_key ORDER BY a.item_key, a.p_id"
):
text = FR_CITE_RE.sub(" ", _expand_ranges(r["text"]))
hits = targets & set(find_codes(text))
if not hits:
continue
row = {
"item_key": r["item_key"],
"p_id": r["p_id"],
"page": r["page"],
"text": text,
"title": r["title"],
"date_published": r["date_published"],
}
for c in hits:
buckets[c].append(row)
return {c: _events_for(c, rows, expanded=True) for c, rows in buckets.items()}
def _cpt_ev(code: str, year: int, kind: str, item_key: str, note: str) -> EventRow:
return EventRow(code, year, kind, "", "", item_key, 0, 0, "cpt", True, note)
@@ -318,14 +376,51 @@ def cpt_events(con: Any, code: str) -> list[EventRow]:
return []
def lineage(con: Any, store: Any, code: str) -> list[EventRow]:
fr = fr_events(store, code)
cpt = cpt_events(con, code)
def _merge(
code: str,
rvu: list[EventRow],
fr: list[EventRow],
cpt: list[EventRow],
) -> list[EventRow]:
"""Merge one code's rvu/fr/cpt events, marking each rvu event
``anchored`` when an fr or cpt event for the same code lies within
±1 rule year — shared by ``lineage`` (per-code) and ``lineage_all``
(bucketed) so both anchor identically."""
anchor_years = {e.year for e in fr} | {e.year for e in cpt}
rvu = [
anchored_rvu = [
dataclasses.replace(
ev, anchored=any(abs(ev.year - y) <= 1 for y in anchor_years)
)
for ev in rvu_events(con, code)
for ev in rvu
]
return sorted([*rvu, *fr, *cpt], key=lambda e: (e.year, e.source != "fr", e.kind))
return sorted(
[*anchored_rvu, *fr, *cpt], key=lambda e: (e.year, e.source != "fr", e.kind)
)
def lineage(con: Any, store: Any, code: str) -> list[EventRow]:
fr = fr_events(store, code)
cpt = cpt_events(con, code)
rvu = rvu_events(con, code)
return _merge(code, rvu, fr, cpt)
def lineage_all(
con: Any, store: Any, codes: Sequence[str]
) -> dict[str, list[EventRow]]:
"""``lineage`` for every one of *codes*, computing fr events with ONE
pass over ``fr_anchors`` (``fr_events_bucketed``) instead of one per
code; rvu/cpt events stay per-code (cheap replica reads) and are
merged with the same ±1-year anchoring ``lineage`` uses, via
``_merge``."""
codes_u = [c.upper() for c in codes]
fr_by_code = fr_events_bucketed(store, codes_u)
return {
code: _merge(
code,
rvu_events(con, code),
fr_by_code.get(code, []),
cpt_events(con, code),
)
for code in codes_u
}

71
tests/cli/test_pfs_cli.py
View File

@@ -270,6 +270,77 @@ class TestLineage:
assert read_calls == [True]
assert con.published == []
def test_requires_code_or_all_payable(self, con):
res = runner.invoke(app, ["pfs", "lineage"])
assert res.exit_code != 0
def test_all_payable_limit_write_calls_writer_once_per_code_after_build(
self, con, monkeypatch
):
# Extra A/R/T, newest-year codes so --limit has something to
# trim: sorted, the newest-year A/R/T targets are
# 99490 < A0001 < B0002 ('9' sorts before letters); 99999 is
# status B (excluded) and G2058 is a stale year (excluded).
con.execute(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
["A0001", None, "extra a", "A", 1.0, 2026],
)
con.execute(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
["B0002", None, "extra b", "R", 1.0, 2026],
)
order: list[tuple] = []
def fake_lineage_all(con_, store_, codes):
order.append(("lineage_all", tuple(codes)))
return {c: [] for c in codes}
monkeypatch.setattr(pfs_cli, "lineage_all", fake_lineage_all)
written: list[str] = []
def fake_write_events(con_, code, rows):
order.append(("write", code))
written.append(code)
return 0
monkeypatch.setattr(pfs_cli, "write_events", fake_write_events)
real_batch = pfs_cli._batch
def recording_batch():
order.append(("batch",))
return real_batch()
monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
res = runner.invoke(
app, ["pfs", "lineage", "--all-payable", "--limit", "2", "--write"]
)
assert res.exit_code == 0, res.output
assert order[0] == ("lineage_all", ("99490", "A0001"))
assert order[1] == ("batch",)
assert order[2:] == [("write", "99490"), ("write", "A0001")]
assert written == ["99490", "A0001"]
assert con.published == [True]
def test_all_payable_without_write_never_opens_the_batch(self, con, monkeypatch):
monkeypatch.setattr(
pfs_cli, "lineage_all", lambda c, s, codes: {c: [] for c in codes}
)
def fail_batch():
raise AssertionError(
"lineage --all-payable without --write must not open a RW "
"duckdb_batch connection"
)
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "lineage", "--all-payable"])
assert res.exit_code == 0, res.output
assert con.published == []
assert "lineage: 1 codes targeted" in res.output
class TestFamilies:
def test_derives_from_tables_and_writes(self, con):

72
tests/pfs/test_lineage.py
View File

@@ -12,7 +12,9 @@ from pfs.lineage import (
_expand_ranges,
cpt_events,
fr_events,
fr_events_bucketed,
lineage,
lineage_all,
rvu_events,
rvu_year_span,
)
@@ -363,3 +365,73 @@ class TestCrossCheck:
def test_sorted_by_year_then_source(self, con, store):
ev = lineage(con, store, "99439")
assert [e.year for e in ev] == sorted(e.year for e in ev)
class TestFrEventsBucketed:
# The four target codes from `store`'s three rule items: a plain
# mention (99490), a replaced_by/crosswalk pair (G2058 -> 99439) and
# a range-form deletion (99441, from "99441-99443").
_CODES = ("99490", "G2058", "99439", "99441")
def test_matches_fr_events_per_code(self, store):
bucketed = fr_events_bucketed(store, self._CODES)
for c in self._CODES:
assert bucketed[c] == fr_events(store, c), c
def test_range_form_middle_member_matches(self, store):
# 90868 is a middle member of a "90867-90869" range and appears
# nowhere else in the paragraph literally — both paths must find
# it via range expansion, not a literal LIKE hit.
bucketed = fr_events_bucketed(store, ["90868"])
assert bucketed["90868"] == fr_events(store, "90868")
assert any(e.kind == "deleted" for e in bucketed["90868"])
def test_fr_citation_page_number_is_not_a_mention(self, store):
# M1, bucketed path: a page citation ("91 FR 99490") must not
# make 99490 register a hit at all, not just skip the crosswalk
# pattern — matching fr_events's own guard.
store.con.execute(
"INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
(
"DE2VH9PD",
1500,
67999,
1500,
"We note the crosswalk methodology discussed at 91 FR 99490 for related codes.",
),
)
bucketed = fr_events_bucketed(store, ["99490"])
assert bucketed["99490"] == fr_events(store, "99490")
assert not any(e.kind == "crosswalk" for e in bucketed["99490"])
def test_unmentioned_code_gets_empty_bucket(self, store):
assert fr_events_bucketed(store, ["00000"]) == {"00000": []}
class TestLineageAll:
def test_matches_lineage_per_code(self, con, store):
codes = ["99487", "G2058", "99439"]
by_code = lineage_all(con, store, codes)
for c in codes:
assert by_code[c] == lineage(con, store, c), c
def test_rvu_event_anchored_by_cpt_changed_within_one_year(self, con, store):
_cpt_section(con, 2024, "GQGTPGYV")
_cpt_reference(con, 2024, "GQGTPGYV", "99487", [2016])
by_code = lineage_all(con, store, ["99487"])
by = {(e.kind, e.year): e for e in by_code["99487"]}
assert by[("status_change", 2017)].anchored is True
assert ("cpt_changed", 2016) in by
def test_rvu_events_anchored_by_nearby_fr_event(self, con, store):
by_code = lineage_all(con, store, ["G2058", "99487"])
by_kind = {e.kind: e for e in by_code["G2058"]}
assert by_kind["appeared"].year == 2020 and by_kind["appeared"].anchored is True
assert (
by_kind["disappeared"].year == 2021
and by_kind["disappeared"].anchored is True
)
assert all(not e.anchored for e in by_code["99487"] if e.source == "rvu")
def test_unknown_code_yields_empty_list(self, con, store):
assert lineage_all(con, store, ["99999"]) == {"99999": []}