Event scoring was measured at chance level on the 77 Rodden-AA holdout cases (BUG-1091). The new flow finds the dividing line nearest the reported time, describes how the flipped divisional charts read on either side, asks up to five self-assessment questions, and saves the user's own choice (provenance user_choice) with the domains to read in both versions. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DbBXozbkyFNqbu8pVEtZ7W
157 lines
6.8 KiB
Python
157 lines
6.8 KiB
Python
"""Chart choice engine: dividing lines, readings, and the agreed decision rules."""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
sys.path.insert(0, str(ROOT / "scripts"))
|
||
|
||
from scripts.chart_choice.boundaries import ( # noqa: E402
|
||
ChartChoiceInputError,
|
||
ChartRequest,
|
||
find_lines,
|
||
nearest_line,
|
||
representative_minute,
|
||
segment_around,
|
||
signature,
|
||
)
|
||
from scripts.chart_choice.decision import decide, reported_is_neutral # noqa: E402
|
||
from scripts.chart_choice.service import build_packet, decide_packet # noqa: E402
|
||
|
||
# Fictional reference input (not a user record): mid-latitude north China, early morning.
|
||
BODY = {
|
||
"lat": 36.42, "lon": 114.21, "tz": 8,
|
||
"start_at": "1997-08-08T04:45", "end_at": "1997-08-08T05:15", "reported_at": "1997-08-08T05:00",
|
||
}
|
||
REQ = ChartRequest("1997-08-08", 36.42, 114.21, 8.0)
|
||
|
||
|
||
def test_lines_are_exact_and_merge_charts_that_flip_together():
|
||
lines = find_lines(REQ, datetime(1997, 8, 8, 4, 45), datetime(1997, 8, 8, 5, 15))
|
||
fifteen = next(line for line in lines if line.degree == 15.0)
|
||
assert set(fifteen.flips) == {"D2", "D4", "D10", "D12", "D24"}
|
||
assert fifteen.at.strftime("%H:%M") == "04:59"
|
||
assert any("D9" in line.flips for line in lines), "D9 lines must not be lost to float rounding"
|
||
for line in lines:
|
||
before = signature((line.degree - 1e-4) + 90) # Cancer: sign index 3 → 90°
|
||
after = signature((line.degree + 1e-4) + 90)
|
||
assert {k for k in before if before[k] != after[k]} == set(line.flips)
|
||
|
||
|
||
def test_nearest_line_and_segments():
|
||
start, end = datetime(1997, 8, 8, 4, 45), datetime(1997, 8, 8, 5, 15)
|
||
lines = find_lines(REQ, start, end)
|
||
reported = datetime(1997, 8, 8, 5, 0)
|
||
line = nearest_line(lines, reported)
|
||
assert line.degree == 15.0
|
||
before, after = segment_around(lines, line, start, end)
|
||
assert before[1] == after[0] == line.at
|
||
assert representative_minute(after, reported) == reported
|
||
saved_before = representative_minute(before, reported)
|
||
assert before[0] < saved_before < before[1]
|
||
assert (before[1] - saved_before).total_seconds() >= 60
|
||
|
||
|
||
def test_packet_lists_only_charts_that_flip_and_unchanged_ones():
|
||
packet = build_packet(BODY)
|
||
assert packet["status"] == "choose"
|
||
assert packet["line"]["charts"] == ["D10", "D4", "D2", "D12", "D24"]
|
||
assert [u["chart"] for u in packet["unchanged"]] == ["D1", "D9"]
|
||
assert packet["reported_side"] == "after"
|
||
assert packet["reported_neutral"] is True
|
||
assert 1 <= len(packet["questions"]) <= 5
|
||
for question in packet["questions"]:
|
||
texts = [o["text"] for o in question["options"]]
|
||
assert texts[0] != texts[1] and texts[2] == "说不清"
|
||
|
||
|
||
def test_decision_majority_when_reported_time_is_neutral():
|
||
answers = {"career": "before", "home": "after", "money": "before", "parents": "before", "study": "after"}
|
||
result = decide_packet({**BODY, "answers": answers})["decision"]
|
||
assert result["chosen_side"] == "before"
|
||
assert result["rule"] == "majority"
|
||
assert result["both_versions_domains"] == ["home", "study"]
|
||
assert result["saved_at"] == "1997-08-08T04:56"
|
||
|
||
|
||
def test_decision_rules():
|
||
assert decide({"a": "before", "b": "after"}, reported_side="after", neutral=True)["rule"] == "tie_keeps_reported"
|
||
kept = decide({"a": "before", "b": "before", "c": "before", "d": "after"}, reported_side="after", neutral=False)
|
||
assert kept["chosen_side"] == "after" and kept["rule"] == "reported_kept"
|
||
flipped = decide({"a": "before", "b": "before", "c": "before", "d": "before", "e": "after"},
|
||
reported_side="after", neutral=False)
|
||
assert flipped["chosen_side"] == "before" and flipped["rule"] == "strong_override"
|
||
assert flipped["both_versions_domains"] == ["e"]
|
||
unsure = decide({"a": "unsure"}, reported_side="before", neutral=True)
|
||
assert unsure["chosen_side"] == "before" and unsure["both_versions_domains"] == []
|
||
|
||
|
||
def test_reported_neutral_rules():
|
||
line = datetime(1997, 8, 8, 5, 20, 30)
|
||
assert reported_is_neutral(datetime(1997, 8, 8, 5, 30), line, reported_known=True) # rounded
|
||
assert reported_is_neutral(datetime(1997, 8, 8, 5, 22), line, reported_known=True) # near line
|
||
assert not reported_is_neutral(datetime(1997, 8, 8, 5, 37), line, reported_known=True)
|
||
assert reported_is_neutral(datetime(1997, 8, 8, 5, 37), line, reported_known=False)
|
||
|
||
|
||
def test_override_and_read_both():
|
||
result = decide_packet({**BODY, "override_side": "before", "read_both_all": True})["decision"]
|
||
assert result["chosen_side"] == "before" and result["rule"] == "user_override"
|
||
assert set(result["both_versions_domains"]) == {"career", "home", "money", "parents", "study"}
|
||
|
||
|
||
def test_answers_for_unasked_domains_are_ignored():
|
||
result = decide_packet({**BODY, "answers": {"relationship": "before", "self": "before"}})
|
||
assert result["answers"] == {}
|
||
assert result["decision"]["chosen_side"] == "after"
|
||
|
||
|
||
def test_wide_window_asks_to_narrow_first():
|
||
packet = build_packet({**BODY, "start_at": "1997-08-08T03:00", "end_at": "1997-08-08T08:00"})
|
||
assert packet["status"] == "needs_narrowing"
|
||
options = packet["narrowing"]["options"]
|
||
assert options[0]["start_at"] == "1997-08-08T03:00" and options[1]["end_at"] == "1997-08-08T08:00"
|
||
assert options[0]["end_at"] == options[1]["start_at"]
|
||
assert packet["narrowing"]["anchor"] == "sunrise" # sunrise falls inside 03:00–08:00 here
|
||
|
||
|
||
def test_no_line_means_certain():
|
||
packet = build_packet({**BODY, "start_at": "1997-08-08T04:55", "end_at": "1997-08-08T04:58", "reported_at": "1997-08-08T04:56"})
|
||
assert packet["status"] == "certain" and packet["saved_at"] == "1997-08-08T04:56"
|
||
|
||
|
||
def test_reported_outside_window_falls_back_to_middle():
|
||
packet = build_packet({**BODY, "reported_at": "1997-08-08T06:00"})
|
||
assert packet["reported_known"] is False and packet["reported_at"] == "1997-08-08T05:00"
|
||
|
||
|
||
@pytest.mark.parametrize("patch", [
|
||
{"start_at": "1997-08-08T05:15", "end_at": "1997-08-08T04:45"},
|
||
{"lat": 120},
|
||
{"start_at": None},
|
||
{"start_at": "5pm"},
|
||
{"end_at": "1997-08-10T05:15"},
|
||
{"node_mode": "osculating"},
|
||
])
|
||
def test_bad_input(patch):
|
||
with pytest.raises(ChartChoiceInputError):
|
||
build_packet({**BODY, **patch})
|
||
|
||
|
||
def test_decide_requires_a_choice():
|
||
with pytest.raises(ChartChoiceInputError):
|
||
decide_packet({**BODY, "start_at": "1997-08-08T04:55", "end_at": "1997-08-08T04:58", "reported_at": "1997-08-08T04:56"})
|
||
|
||
|
||
def test_window_across_midnight():
|
||
packet = build_packet({**BODY, "start_at": "1997-08-07T23:40", "end_at": "1997-08-08T00:20",
|
||
"reported_at": "1997-08-08T00:00"})
|
||
assert packet["status"] in ("choose", "certain")
|
||
if packet["status"] == "choose":
|
||
assert packet["sides"]["before"]["from"] >= "1997-08-07T23:40"
|