jesse-ux
b85c4a686a
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.
119 lines
6.2 KiB
Python
119 lines
6.2 KiB
Python
"""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())
|