fix(rectification): anchor candidate windows to civil dates across midnight
Carry explicit local date intervals instead of inferring the day from clock order. Cluster width, delivery, adoption, and reports keep the actual civil date; adopted date is stored separately from the reported birth_date. Algorithm identity is scoring-9 / spec-v5. Scoring weights, confirmation thresholds, and Skill version are unchanged. Isolated Linux final-3 gates passed; four pre-existing Python failures remain. This is not a production release.
This commit is contained in:
@@ -0,0 +1,118 @@
|
||||
"""Civil-date candidate windows. Ordinals order samples; offsets measure minutes."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
_LOCAL_MINUTE = re.compile(r"\d{4}-\d{2}-\d{2}T(?:[01]\d|2[0-3]):[0-5]\d\Z")
|
||||
|
||||
|
||||
def candidate_intervals(request: dict[str, Any]) -> list[dict[str, str]]:
|
||||
raw = request.get("candidate_intervals")
|
||||
if "candidate_intervals" not in request:
|
||||
start = datetime.fromisoformat(f'{request["birth_date"]}T{request["start_time"]}')
|
||||
end = datetime.fromisoformat(f'{request["birth_date"]}T{request["end_time"]}')
|
||||
if end < start:
|
||||
end += timedelta(days=1)
|
||||
return [{"start_at": start.isoformat(timespec="minutes"), "end_at": end.isoformat(timespec="minutes")}]
|
||||
if not isinstance(raw, list) or not 1 <= len(raw) <= 2:
|
||||
raise ValueError("candidate_intervals must contain one or two intervals")
|
||||
anchor = datetime.fromisoformat(f'{request["birth_date"]}T00:00')
|
||||
cleaned, clocks = [], set()
|
||||
previous_end = None
|
||||
total = 0
|
||||
for item in raw:
|
||||
if not isinstance(item, dict) or set(item) != {"start_at", "end_at"}:
|
||||
raise ValueError("candidate_intervals must contain start_at/end_at only")
|
||||
values = [item.get(key) for key in ("start_at", "end_at")]
|
||||
if any(not isinstance(value, str) or not _LOCAL_MINUTE.fullmatch(value) for value in values):
|
||||
raise ValueError("candidate_intervals require local YYYY-MM-DDTHH:MM")
|
||||
start, end = map(datetime.fromisoformat, values)
|
||||
if end < start or (previous_end is not None and start <= previous_end):
|
||||
raise ValueError("candidate_intervals must be ordered and non-overlapping")
|
||||
if start < anchor - timedelta(days=1) or end >= anchor + timedelta(days=2):
|
||||
raise ValueError("candidate_intervals date out of bounds")
|
||||
width = int((end - start).total_seconds() // 60) + 1
|
||||
total += width
|
||||
if total > 1440 or width < 1:
|
||||
raise ValueError("candidate_range_out_of_bounds")
|
||||
for offset in range(width):
|
||||
clock = (start + timedelta(minutes=offset)).strftime("%H:%M")
|
||||
if clock in clocks:
|
||||
raise ValueError("candidate_intervals duplicate clock identity")
|
||||
clocks.add(clock)
|
||||
cleaned.append({"start_at": values[0], "end_at": values[1]})
|
||||
previous_end = end
|
||||
# Legacy clock envelope remains an inclusion boundary, never a date anchor.
|
||||
lower, upper = request["start_time"], request["end_time"]
|
||||
if any(not (lower <= clock <= upper if lower <= upper else clock >= lower or clock <= upper) for clock in clocks):
|
||||
raise ValueError("candidate_intervals outside clock window")
|
||||
return cleaned
|
||||
|
||||
|
||||
def narrow_candidate_intervals(request: dict[str, Any], start_time: str, end_time: str) -> list[dict[str, str]]:
|
||||
result = []
|
||||
for interval in candidate_intervals(request):
|
||||
start, end = (datetime.fromisoformat(interval[key]) for key in ("start_at", "end_at"))
|
||||
run = None
|
||||
for offset in range(int((end - start).total_seconds() // 60) + 1):
|
||||
at = start + timedelta(minutes=offset)
|
||||
clock = at.strftime("%H:%M")
|
||||
inside = start_time <= clock <= end_time if start_time <= end_time else clock >= start_time or clock <= end_time
|
||||
if inside:
|
||||
stamp = at.isoformat(timespec="minutes")
|
||||
run = {"start_at": run["start_at"] if run else stamp, "end_at": stamp}
|
||||
elif run:
|
||||
result.append(run)
|
||||
run = None
|
||||
if run:
|
||||
result.append(run)
|
||||
return result
|
||||
|
||||
|
||||
def enumerate_candidate_window(request: dict[str, Any]) -> list[datetime]:
|
||||
raw_step = request.get("minute_step", 1)
|
||||
if isinstance(raw_step, bool) or not isinstance(raw_step, int) or not 1 <= raw_step <= 15:
|
||||
raise ValueError("minute_step_out_of_bounds")
|
||||
moments = []
|
||||
for interval in candidate_intervals(request):
|
||||
start, end = (datetime.fromisoformat(interval[key]) for key in ("start_at", "end_at"))
|
||||
width = int((end - start).total_seconds() // 60) + 1
|
||||
if not 1 <= width <= 1440:
|
||||
raise ValueError("candidate_range_out_of_bounds")
|
||||
moments.extend(start + timedelta(minutes=offset) for offset in range(0, width, raw_step))
|
||||
return moments
|
||||
|
||||
|
||||
def candidate_positions(request: dict[str, Any], moments: list[datetime]) -> list[dict[str, Any]]:
|
||||
intervals = candidate_intervals(request)
|
||||
origin = datetime.fromisoformat(intervals[0]["start_at"])
|
||||
bounds = [(datetime.fromisoformat(row["start_at"]), datetime.fromisoformat(row["end_at"])) for row in intervals]
|
||||
return [{"time": at.strftime("%H:%M"), "candidate_date": at.date().isoformat(),
|
||||
"window_index": index, "window_offset_minutes": int((at - origin).total_seconds() // 60),
|
||||
"segment_index": next(i for i, (start, end) in enumerate(bounds) if start <= at <= end)}
|
||||
for index, at in enumerate(moments)]
|
||||
|
||||
|
||||
def intervals_from_positions(positions: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
groups: dict[int, list[dict[str, Any]]] = {}
|
||||
for row in sorted(positions, key=lambda item: item["window_index"]):
|
||||
groups.setdefault(row["segment_index"], []).append(row)
|
||||
result = []
|
||||
for segment, members in groups.items():
|
||||
first, last = members[0], members[-1]
|
||||
result.append({"segment_index": segment, "start_index": first["window_index"], "end_index": last["window_index"],
|
||||
"start_offset_minutes": first["window_offset_minutes"], "end_offset_minutes": last["window_offset_minutes"],
|
||||
"start_at": f'{first["candidate_date"]}T{first["time"]}',
|
||||
"end_at": f'{last["candidate_date"]}T{last["time"]}'})
|
||||
return result
|
||||
|
||||
|
||||
def interval_union_width(intervals: list[dict[str, Any]]) -> int:
|
||||
"""Keep historical within-segment envelope; never fill a declared segment gap."""
|
||||
groups: dict[int, list[dict[str, Any]]] = {}
|
||||
for row in sorted(intervals, key=lambda item: item["start_index"]):
|
||||
groups.setdefault(row["segment_index"], []).append(row)
|
||||
return sum(max(row["end_offset_minutes"] for row in rows) - min(row["start_offset_minutes"] for row in rows) + 1
|
||||
for rows in groups.values())
|
||||
Reference in New Issue
Block a user