112 lines
3.6 KiB
Python
112 lines
3.6 KiB
Python
"""Nearest divisional-chart change around a recorded minute.
|
|
|
|
Reads ``window_scan`` transitions only. Does not score and does not change
|
|
the scan payload.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
# Same domain map as event_probes.LAYER_DOMAIN. Topics are the opening words.
|
|
CHART_LAYERS = ("d9", "d10", "d4", "d5", "d24", "d7", "d12", "d2", "d11", "d30", "d1")
|
|
LAYER_DOMAIN = {
|
|
"d9": "relationship",
|
|
"d10": "career",
|
|
"d4": "relocation",
|
|
"d5": "education",
|
|
"d24": "education",
|
|
"d7": "family",
|
|
"d12": "family",
|
|
"d2": "finance",
|
|
"d11": "finance",
|
|
"d30": "health_pressure",
|
|
}
|
|
LAYER_TOPIC = {
|
|
"d9": "婚恋",
|
|
"d10": "工作",
|
|
"d4": "住处",
|
|
"d5": "学业",
|
|
"d24": "学业",
|
|
"d7": "子女",
|
|
"d12": "父母",
|
|
"d2": "钱",
|
|
"d11": "钱",
|
|
"d30": "身体",
|
|
}
|
|
|
|
|
|
def _clock(value: Any) -> int | None:
|
|
raw = str(value or "")[:5]
|
|
if len(raw) != 5 or raw[2] != ":" or not (raw[:2].isdigit() and raw[3:].isdigit()):
|
|
return None
|
|
hour, minute = int(raw[:2]), int(raw[3:])
|
|
if hour > 23 or minute > 59:
|
|
return None
|
|
return hour * 60 + minute
|
|
|
|
|
|
def _distance(left: int, right: int) -> int:
|
|
span = abs(left - right)
|
|
return min(span, 1440 - span)
|
|
|
|
|
|
def nearest_chart_boundary(scan: dict[str, Any] | None, recorded_time: str | None) -> dict[str, Any]:
|
|
"""Closest real sign change to ``recorded_time``.
|
|
|
|
Display-only layers and segment-start rows that do not change sign are
|
|
ignored. An equal distance keeps the earlier clock.
|
|
"""
|
|
recorded = _clock(recorded_time)
|
|
rows = scan.get("transitions") if isinstance(scan, dict) else None
|
|
if recorded is None or not isinstance(rows, list):
|
|
return {"found": False, "reason": "unavailable", "layers": [], "topics": [], "domains": []}
|
|
changes: list[tuple[int, str, str, str]] = []
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
layer = str(row.get("layer") or "")
|
|
if layer not in CHART_LAYERS:
|
|
continue
|
|
before, after = row.get("from_sign"), row.get("to_sign")
|
|
if not isinstance(before, str) or not isinstance(after, str) or before == after:
|
|
continue
|
|
at = _clock(row.get("at"))
|
|
if at is None:
|
|
continue
|
|
changes.append((at, layer, before, after))
|
|
if not changes:
|
|
return {"found": False, "reason": "no_transition", "layers": [], "topics": [], "domains": []}
|
|
best_at = min(changes, key=lambda item: (_distance(item[0], recorded), item[0]))[0]
|
|
order = {layer: index for index, layer in enumerate(CHART_LAYERS)}
|
|
layers: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for at, layer, before, after in sorted(changes, key=lambda item: order[item[1]]):
|
|
if at != best_at or layer in seen:
|
|
continue
|
|
seen.add(layer)
|
|
layers.append({
|
|
"layer": layer,
|
|
"from_sign": before,
|
|
"to_sign": after,
|
|
"topic": LAYER_TOPIC.get(layer),
|
|
"domain": LAYER_DOMAIN.get(layer),
|
|
})
|
|
topics: list[str] = []
|
|
domains: list[str] = []
|
|
for row in layers:
|
|
topic, domain = row["topic"], row["domain"]
|
|
if isinstance(topic, str) and topic not in topics:
|
|
topics.append(topic)
|
|
if isinstance(domain, str) and domain not in domains:
|
|
domains.append(domain)
|
|
return {
|
|
"found": True,
|
|
"reason": None,
|
|
"at": f"{best_at // 60:02d}:{best_at % 60:02d}",
|
|
"distance_minutes": _distance(best_at, recorded),
|
|
"layers": layers,
|
|
"topics": topics,
|
|
"domains": domains,
|
|
}
|