diff --git a/work/meshai/central/budget.py b/work/meshai/central/budget.py deleted file mode 100644 index 32f76c7..0000000 --- a/work/meshai/central/budget.py +++ /dev/null @@ -1,9 +0,0 @@ -"""Re-export shim — implementation has moved to meshai.notifications.formatters._budget. - -Every existing ``from meshai.central.budget import budget_for, fit_to_budget`` -import continues to work unchanged. This shim is the sole consumer of the -canonical implementation; update the implementation there, not here. -""" -from meshai.notifications.formatters._budget import budget_for, fit_to_budget - -__all__ = ["budget_for", "fit_to_budget"] diff --git a/work/meshai/central/satpass_handler.py b/work/meshai/central/satpass_handler.py index 6fdfea5..fd4569e 100644 --- a/work/meshai/central/satpass_handler.py +++ b/work/meshai/central/satpass_handler.py @@ -43,7 +43,7 @@ from typing import Any, Optional from zoneinfo import ZoneInfo from meshai.adapter_config import adapter_config -from meshai.central.budget import budget_for, fit_to_budget +from meshai.notifications.formatters._budget import budget_for, fit_to_budget from meshai.persistence import get_db logger = logging.getLogger(__name__) diff --git a/work/meshai/central/wfigs_handler.py b/work/meshai/central/wfigs_handler.py index 4368fb6..66f0698 100644 --- a/work/meshai/central/wfigs_handler.py +++ b/work/meshai/central/wfigs_handler.py @@ -29,7 +29,7 @@ inside that connection's autocommit mode. from __future__ import annotations from meshai.adapter_config import adapter_config -from meshai.central.budget import budget_for, fit_to_budget +from meshai.notifications.formatters._budget import budget_for, fit_to_budget import logging import time diff --git a/work/meshai/notifications/renderers/composer.py b/work/meshai/notifications/renderers/composer.py index 248b318..7adf1b7 100644 --- a/work/meshai/notifications/renderers/composer.py +++ b/work/meshai/notifications/renderers/composer.py @@ -309,7 +309,7 @@ def _resolve_budget(event: Event) -> int: src = (getattr(event, "source", "") or "").strip() if src: try: - from meshai.central.budget import budget_for + from meshai.notifications.formatters._budget import budget_for return budget_for(src) except Exception: pass diff --git a/work/meshai/notifications/renderers/work_zone.py b/work/meshai/notifications/renderers/work_zone.py deleted file mode 100644 index 1d09bf3..0000000 --- a/work/meshai/notifications/renderers/work_zone.py +++ /dev/null @@ -1,205 +0,0 @@ -"""work_zone mesh-string renderer. - -Consumes a `central_normalizer.normalize()` output dict (the data layer) -and produces a friendly mesh-broadcast string under an 80-byte UTF-8 cap. -Pure formatting; no adapter knowledge. - -Format (per Matt-approved spec): - - 🚧 @ mile [–][, mi of ]: - , [, ends ] - -Optional segments are dropped wholesale (lowest-priority first) when the -byte budget is over: - - 1. ends (lowest priority — dropped first) - 2. of distance segment - 3. - 4. - -Required: emoji + road. If even those overrun, the road is truncated -codepoint-safe with an ellipsis. Emojis count as 4 bytes each in UTF-8. -""" - -from datetime import datetime, timedelta -from typing import Optional - - -_BYTE_BUDGET = 80 - - -def _bytelen(s: str) -> int: - return len(s.encode("utf-8")) - - -def _format_end_short(ends_at: Optional[datetime], now: Optional[datetime] = None) -> Optional[str]: - """Format a future datetime as a tight mesh-friendly string. - - - within 24h -> 'today 6pm' or 'tomorrow 9am' - - within 7 days -> 'Fri 6pm' - - 7-365 days -> 'Jun 15' - - past or far future-> None (caller drops the segment) - """ - if ends_at is None: return None - if now is None: now = datetime.now() - # Normalize to naive (renderer doesn't care about tz here; both inputs - # should be Boise-local in practice since the upstream feed uses local). - if ends_at.tzinfo is not None: - ends_at = ends_at.replace(tzinfo=None) - if now.tzinfo is not None: - now = now.replace(tzinfo=None) - delta = ends_at - now - if delta.total_seconds() < 0: - return None - hour = ends_at.hour - minute = ends_at.minute - if hour == 0: - time_part = "12am" if minute == 0 else f"12:{minute:02d}am" - elif hour < 12: - time_part = f"{hour}am" if minute == 0 else f"{hour}:{minute:02d}am" - elif hour == 12: - time_part = "12pm" if minute == 0 else f"12:{minute:02d}pm" - else: - time_part = f"{hour-12}pm" if minute == 0 else f"{hour-12}:{minute:02d}pm" - - if delta < timedelta(hours=24) and ends_at.date() == now.date(): - return f"today {time_part}" - if delta < timedelta(hours=48) and ends_at.date() == (now + timedelta(days=1)).date(): - return f"tomorrow {time_part}" - if delta < timedelta(days=7): - wd = ends_at.strftime("%a") # 'Mon' .. 'Sun' - return f"{wd} {time_part}" - if delta < timedelta(days=365): - return ends_at.strftime("%b %-d") if hasattr(datetime, "now") else ends_at.strftime("%b ") + str(ends_at.day) - return None - - -def _format_direction_phrase(direction: Optional[str]) -> Optional[str]: - """Render the normalized direction as a mesh-friendly noun phrase.""" - if not direction or direction == "unknown": - return None - if direction == "both": - return "both directions" - return direction - - -def _format_mile_segment(mile_start: Optional[int], mile_end: Optional[int]) -> Optional[str]: - if mile_start is None: return None - if mile_end is not None and mile_end != mile_start: - return f"@ mile {mile_start}–{mile_end}" - return f"@ mile {mile_start}" - - -def _format_distance_segment(distance_mi: Optional[int], bearing: Optional[str], town: Optional[str]) -> Optional[str]: - if not town: return None - # Issue 2 polish: distances under 1 mi are uninformative ("0 mi S of - # McCall" reads worse than "near McCall"). Drop the distance/bearing - # pair and fall back to plain "near " form. - if distance_mi is not None and bearing and distance_mi >= 1: - return f"{distance_mi} mi {bearing} of {town}" - return f"near {town}" - - -def _truncate_road(road: str, budget: int) -> str: - """Truncate `road` to fit in `budget` UTF-8 bytes, codepoint-safe.""" - if _bytelen(road) <= budget: - return road - cut = road - while cut and _bytelen(cut + "…") > budget: - cut = cut[:-1] - return cut + "…" if cut else "…" - - -def format_work_zone_mesh(n: dict, now: Optional[datetime] = None) -> str: - """Render a normalized work_zone dict to a mesh-friendly string. - - Always returns a string; drops segments to fit the 80-byte cap. - """ - emoji = "🚧" - raw_road = n.get("road") - town = n.get("town") - - # Issue 3d: when the road is uninformative (set None by the normalizer - # for "Exit 80 Southbound On Ramp" shapes) AND we have a town, lead - # with the town as the head instead of a useless placeholder. - if raw_road: - road = raw_road - head = f"{emoji} {road}" - suppress_distance_seg = False - elif town: - # "🚧 near " or "🚧 mi of " as head. - # distance/bearing are folded INTO the head; suppress the separate - # distance segment below to avoid duplication. - road = town # used only by the last-resort road-truncation branch - head = f"{emoji} {_format_distance_segment(n.get('distance_mi'), n.get('bearing'), town)}" - suppress_distance_seg = True - else: - road = "Road event" - head = f"{emoji} {road}" - suppress_distance_seg = False - - mile_seg = _format_mile_segment(n.get("mile_start"), n.get("mile_end")) if raw_road else None - dist_seg = None if suppress_distance_seg else \ - _format_distance_segment(n.get("distance_mi"), n.get("bearing"), town) - dir_phrase = _format_direction_phrase(n.get("direction")) - sub = n.get("sub_type") - impact = n.get("impact") - if impact == "full_closure": - # Promote full-closure into the description slot so it's louder. - sub = f"all lanes closed{' (' + sub + ')' if sub else ''}" - ends_seg = _format_end_short(n.get("ends_at"), now=now) - - # Optional segments in build order; each has a drop_priority where - # HIGHER number = dropped FIRST when over budget. - Segment = tuple # (drop_priority, joiner_before, text) - segs: list[tuple[int, str, str]] = [] - if mile_seg: - segs.append((10, " ", mile_seg)) # mile segment is high-value, keep longest - if dist_seg: - segs.append((20, ", ", dist_seg)) - if dir_phrase or sub: - segs.append((30, ": ", "")) # marker for the colon transition - if dir_phrase: - segs.append((30, "", dir_phrase)) - if sub: - segs.append((40, ", " if dir_phrase else "", sub)) - if ends_seg: - segs.append((50, ", ends ", ends_seg)) - - # Iteratively assemble; drop highest-priority segments until under budget. - kept = list(range(len(segs))) - while True: - out = head - last_was_colon = False - for i in kept: - prio, joiner, text = segs[i] - if text == "" and joiner == ": ": - out += ": " - last_was_colon = True - continue - # If we're right after a ": " marker, the first content segment - # joins with no extra delimiter even if its joiner was ", ". - if last_was_colon: - out += text - last_was_colon = False - else: - out += joiner + text - if _bytelen(out) <= _BYTE_BUDGET: - return out - # Find the highest-priority kept segment with non-empty content; - # drop it (and its preceding colon marker if it was the only one - # past the colon). - droppable = [i for i in kept if segs[i][2] != ""] - if not droppable: - # All optional segments gone; truncate road. - budget_for_road = _BYTE_BUDGET - len((emoji + " ").encode("utf-8")) - return f"{emoji} {_truncate_road(road, budget_for_road)}" - worst = max(droppable, key=lambda i: segs[i][0]) - kept.remove(worst) - # If we dropped both dir_phrase and sub, remove the ": " marker too. - remaining_after_colon = any( - segs[i][2] != "" for i in kept - if any(segs[j][0] == 30 and segs[j][2] == "" for j in range(len(segs)) if j < i) - ) - if not remaining_after_colon: - kept = [i for i in kept if not (segs[i][2] == "" and segs[i][1] == ": ")] diff --git a/work/tests/test_budget_shim.py b/work/tests/test_budget_shim.py deleted file mode 100644 index 540a886..0000000 --- a/work/tests/test_budget_shim.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Phase-0: verify the budget.py re-export shim is identity-equal to the impl. - -Both import paths must resolve to the SAME function objects so any runtime -patching (e.g. in existing tests) affects both paths simultaneously. -""" - -import meshai.central.budget as _shim -import meshai.notifications.formatters._budget as _impl - - -def test_budget_for_is_same_object(): - assert _shim.budget_for is _impl.budget_for, ( - "meshai.central.budget.budget_for must be the same object as " - "meshai.notifications.formatters._budget.budget_for" - ) - - -def test_fit_to_budget_is_same_object(): - assert _shim.fit_to_budget is _impl.fit_to_budget, ( - "meshai.central.budget.fit_to_budget must be the same object as " - "meshai.notifications.formatters._budget.fit_to_budget" - ) - - -def test_shim_imports_work(): - """Smoke: the public names are importable from the legacy path.""" - from meshai.central.budget import budget_for, fit_to_budget # noqa: F401 - assert callable(budget_for) - assert callable(fit_to_budget) - - -def test_budget_for_returns_default_without_adapter_config(monkeypatch): - """budget_for falls back to 140 when the adapter key is absent.""" - val = _shim.budget_for("__no_such_adapter__") - assert val == 140 diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py index 7201955..c7aa747 100644 --- a/work/tests/test_fire_refactor.py +++ b/work/tests/test_fire_refactor.py @@ -22,7 +22,7 @@ from __future__ import annotations import pytest from meshai import central_normalizer as cn -from meshai.central.budget import budget_for +from meshai.notifications.formatters._budget import budget_for from meshai.central.wfigs_handler import ( _build_canonical, _render as _wfigs_render, diff --git a/work/tests/test_firms_refactor.py b/work/tests/test_firms_refactor.py index 231bf32..abc3d96 100644 --- a/work/tests/test_firms_refactor.py +++ b/work/tests/test_firms_refactor.py @@ -31,7 +31,7 @@ import uuid import pytest -from meshai.central.budget import budget_for +from meshai.notifications.formatters._budget import budget_for from meshai.notifications.formatters.fire import format as fire_format from meshai.notifications.formatters.firms import format as firms_format from meshai.notifications.gating.firms import decide as firms_decide diff --git a/work/tests/test_incident_refactor.py b/work/tests/test_incident_refactor.py index 811a2f0..dee0802 100644 --- a/work/tests/test_incident_refactor.py +++ b/work/tests/test_incident_refactor.py @@ -18,10 +18,13 @@ Original diffs are preserved in git history. This is a real production gap flagged for Matt: TomTom road-incident ingestion in particular has no native replacement. -Work-zone parity is unaffected — `meshai.central_normalizer` (a top-level, -non-Central-NATS module; note the name is legacy) and -`meshai.notifications.renderers.work_zone` were never part of the deleted -consumer path and remain live. +Work-zone parity: `meshai.central_normalizer` (a top-level, non-Central-NATS +module; note the name is legacy) was never part of the deleted consumer path +and remains live. `meshai.notifications.renderers.work_zone` was ALSO never +part of the deleted consumer path, but was itself dead code (zero production +callers) once formatters.incident absorbed it as `_render_work_zone()`; it +was removed in a later ripout pass. See TestWorkZoneGolden below for how its +golden-parity coverage was preserved as pinned literals. Groups ------ @@ -143,16 +146,30 @@ class TestWorkZoneGolden: """traffic_last/0002 (itd_511 work_zone) and traffic_last/0003 (wzdx) must produce byte-identical output from the new formatter. - Golden is computed via normalize() → format_work_zone_mesh() (old path). - New path: normalize() → canonical data → formatters.incident.format(). + Originally the golden was computed live via normalize() → + format_work_zone_mesh() (old renderers.work_zone path) and compared + against normalize() → canonical data → formatters.incident.format() + (new path). renderers.work_zone.py has since been deleted (dead code, + zero production callers post-Central-excision; formatters.incident's + _render_work_zone() is its byte-identical live replacement — see that + module's docstring). Mirroring the approach in test_nws_refactor.py for + the same situation: the golden strings below were captured by running + format_work_zone_mesh() against these exact fixtures immediately before + its deletion (confirmed byte-identical to the new formatter's output at + that time) and are now pinned as literals, so this test exercises the + LIVE formatters.incident.format() path only. now is pinned to captured_epoch (1783206522) for both paths so the ends-at segment is deterministic. """ + _GOLDEN = { + "0002.json": "🚧 US-91, near Chubbuck: southbound, road construction, ends Aug 17", + "0003.json": "🚧 US-95, near Wilder: southbound, ends Jul 19", + } + def _run_wz(self, fixture_name: str, adapter_expected: str): from meshai.central_normalizer import normalize - from meshai.notifications.renderers.work_zone import format_work_zone_mesh from meshai.notifications.formatters.incident import format as fmt # Find the fixture by name @@ -167,12 +184,10 @@ class TestWorkZoneGolden: envelope = fx["envelope"] now_epoch = float(fx.get("captured_epoch", time.time())) - now_dt = datetime.fromtimestamp(now_epoch) - # Old renderer golden n = normalize(envelope) assert n is not None, f"normalize() returned None for {fixture_name!r}" - golden = format_work_zone_mesh(n, now=now_dt) + golden = self._GOLDEN[fixture_name] # New formatter canonical = _n_to_canonical_workzone(n) diff --git a/work/tests/test_itd_511_work_zone.py b/work/tests/test_itd_511_work_zone.py index 59ed064..4c8cf49 100644 --- a/work/tests/test_itd_511_work_zone.py +++ b/work/tests/test_itd_511_work_zone.py @@ -2,7 +2,9 @@ Covers the cutover path: itd_511 supplies all Idaho work_zone broadcasts now (state_511_atis ID is skipped). The parser must produce the same flat dict -shape as _parse_state_511_atis so format_work_zone_mesh works unchanged. +shape as _parse_state_511_atis so the work-zone renderer (now +formatters.incident._render_work_zone(), via _n_to_canonical_workzone()) +works unchanged. """ from datetime import datetime, timezone @@ -97,18 +99,6 @@ def test_itd_511_work_zone_full_closure_impact(no_photon): assert n["impact"] == "full_closure" -def test_itd_511_work_zone_renderer_produces_wire(no_photon): - from meshai.notifications.renderers.work_zone import format_work_zone_mesh - env = _itd_work_zone_env() - n = cn.normalize(env) - wire = format_work_zone_mesh(n) - assert wire is not None - # Format matches state_511 convention: 🚧 emoji + road - assert "🚧" in wire - assert "SH-55" in wire - assert "McCall" in wire - - def test_itd_511_work_zone_end_date_formatting(no_photon): """planned_end_epoch should serialize to a datetime that the renderer can format consistently with state_511.""" diff --git a/work/tests/test_nws_refactor.py b/work/tests/test_nws_refactor.py index 8621613..c0cef70 100644 --- a/work/tests/test_nws_refactor.py +++ b/work/tests/test_nws_refactor.py @@ -38,7 +38,7 @@ from datetime import datetime import pytest -from meshai.central.budget import budget_for +from meshai.notifications.formatters._budget import budget_for from meshai.notifications.formatters.nws import format as nws_format from meshai.notifications.gating.nws import decide as nws_decide from meshai.persistence import close_thread_connection, init_db diff --git a/work/tests/test_work_zone_renderer.py b/work/tests/test_work_zone_renderer.py deleted file mode 100644 index be8c3f6..0000000 --- a/work/tests/test_work_zone_renderer.py +++ /dev/null @@ -1,207 +0,0 @@ -"""Tests for the work_zone mesh renderer.""" - -from datetime import datetime, timedelta - -import pytest - -from meshai.notifications.renderers.work_zone import format_work_zone_mesh - - -def _bytelen(s: str) -> int: - return len(s.encode("utf-8")) - - -# ---------- canonical / fully-populated case ------------------------------ - - -def test_all_fields_present_produces_canonical_format(): - n = { - "source": "state_511_atis", "road": "SH-3", "direction": "both", - "mile_start": 60, "mile_end": None, "description": "...", - "sub_type": "construction work", "impact": "partial", - "ends_at": datetime(2026, 6, 2, 17, 0), - "town": "Plummer", "distance_mi": 8, "bearing": "N", - } - out = format_work_zone_mesh(n, now=datetime(2026, 6, 1, 14, 0)) - assert out.startswith("🚧 SH-3 @ mile 60") - assert "8 mi N of Plummer" in out - assert "both directions" in out - assert "construction work" in out - assert _bytelen(out) <= 80 - - -# ---------- segment-drop progression -------------------------------------- - - -def test_no_mile_drops_at_mile_segment(): - n = {"source": "state_511_atis", "road": "W Prairie Ave", "direction": "both", - "mile_start": None, "mile_end": None, "sub_type": "paving", - "impact": "partial", "town": "Coeur d'Alene", "distance_mi": 5, "bearing": "E", - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert "@ mile" not in out - assert "W Prairie Ave" in out - assert "5 mi E of Coeur d'Alene" in out - - -def test_no_town_drops_distance_segment(): - n = {"source": "state_511_atis", "road": "SH-55", "direction": "both", - "mile_start": 17, "mile_end": 18, "sub_type": "paving", - "impact": "partial", "town": None, "distance_mi": None, "bearing": None, - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert " mi " not in out - assert " of " not in out - assert "@ mile 17–18" in out - - -def test_no_ends_drops_ends_suffix(): - n = {"source": "state_511_atis", "road": "I-86", "direction": "both", - "mile_start": 58, "mile_end": 59, "sub_type": "bridge maintenance", - "impact": "partial", "town": "Pocatello", "distance_mi": 15, "bearing": "W", - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert ", ends" not in out - - -def test_unknown_direction_drops_direction_phrase(): - n = {"source": "state_511_atis", "road": "SH-36", "direction": "unknown", - "mile_start": 17, "mile_end": 18, "sub_type": "paving", - "impact": "partial", "town": None, "distance_mi": None, "bearing": None, - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert "unknown" not in out.lower().split(":")[-1] # no 'unknown' in tail - - -def test_full_closure_promoted(): - n = {"source": "state_511_atis", "road": "I-15", "direction": "southbound", - "mile_start": None, "mile_end": None, "sub_type": "road construction", - "impact": "full_closure", "town": None, "distance_mi": None, "bearing": None, - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert "all lanes closed" in out - - -# ---------- byte budget --------------------------------------------------- - - -def test_byte_length_under_80_for_canonical(): - n = {"source": "state_511_atis", "road": "SH-3", "direction": "both", - "mile_start": 60, "mile_end": None, "sub_type": "construction work", - "impact": "partial", "town": "Plummer", "distance_mi": 8, "bearing": "N", - "ends_at": datetime(2026, 6, 2, 17, 0), "description": ""} - out = format_work_zone_mesh(n, now=datetime(2026, 6, 1, 14, 0)) - assert _bytelen(out) <= 80 - - -def test_byte_length_under_80_with_long_road_name(): - n = {"source": "state_511_atis", - "road": "SCIENCE CENTER DR / E ANDERSON ST / N THIRD WAY", - "direction": "both", - "mile_start": 100, "mile_end": 200, "sub_type": "construction work", - "impact": "partial", "town": "Idaho Falls", "distance_mi": 12, "bearing": "SE", - "ends_at": datetime(2026, 9, 16, 1, 0), "description": ""} - out = format_work_zone_mesh(n, now=datetime(2026, 6, 1, 14, 0)) - assert _bytelen(out) <= 80, f"over budget: {len(out.encode('utf-8'))} = {out!r}" - - -def test_emoji_counts_as_4_bytes(): - # 🚧 is U+1F6A7 → 4 bytes in UTF-8. - assert _bytelen("🚧") == 4 - - -def test_extreme_road_name_truncated(): - # Force the renderer into the truncate-road last-resort branch. - long_road = "VERY-LONG-ROAD-NAME-" * 10 - n = {"source": "state_511_atis", "road": long_road, "direction": None, - "mile_start": None, "mile_end": None, "sub_type": None, - "impact": "partial", "town": None, "distance_mi": None, "bearing": None, - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert _bytelen(out) <= 80 - assert out.startswith("🚧 ") - assert "…" in out or _bytelen(long_road) <= 80 # truncated with ellipsis - - -# ---------- ends_at relative-time formatting ------------------------------ - - -def test_ends_today_format(): - now = datetime(2026, 6, 1, 9, 0) - ends = datetime(2026, 6, 1, 18, 0) - n = {"source": "state_511_atis", "road": "X", "direction": None, - "mile_start": None, "mile_end": None, "sub_type": None, "impact": "partial", - "town": None, "distance_mi": None, "bearing": None, - "ends_at": ends, "description": ""} - out = format_work_zone_mesh(n, now=now) - assert "today" in out and "6pm" in out - - -def test_ends_within_week_uses_weekday(): - now = datetime(2026, 6, 1, 9, 0) # Monday - ends = datetime(2026, 6, 5, 16, 30) - n = {"source": "state_511_atis", "road": "X", "direction": None, - "mile_start": None, "mile_end": None, "sub_type": None, "impact": "partial", - "town": None, "distance_mi": None, "bearing": None, - "ends_at": ends, "description": ""} - out = format_work_zone_mesh(n, now=now) - assert "Fri" in out - assert "4:30pm" in out - - -def test_ends_past_drops_segment(): - now = datetime(2026, 6, 10, 9, 0) - ends = datetime(2026, 5, 5, 17, 0) # already past - n = {"source": "state_511_atis", "road": "X", "direction": None, - "mile_start": None, "mile_end": None, "sub_type": None, "impact": "partial", - "town": None, "distance_mi": None, "bearing": None, - "ends_at": ends, "description": ""} - out = format_work_zone_mesh(n, now=now) - assert ", ends" not in out - - -# ---------- v0.5.8 distance < 1 mi → "near X" ----------------------------- - - -def test_distance_zero_drops_to_near_only(): - n = {"source": "state_511_atis", "road": "SH-55", "direction": "both", - "mile_start": None, "mile_end": None, "sub_type": "emergency repairs", - "impact": "partial", "town": "McCall", "distance_mi": 0, "bearing": "S", - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert "near McCall" in out - assert "0 mi" not in out - assert "S of McCall" not in out - - -def test_distance_one_keeps_bearing_segment(): - n = {"source": "state_511_atis", "road": "SH-41", "direction": "southbound", - "mile_start": None, "mile_end": None, "sub_type": "utility work", - "impact": "partial", "town": "Rathdrum", "distance_mi": 1, "bearing": "S", - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - assert "1 mi S of Rathdrum" in out - - -# ---------- v0.5.8 no-road fallback (leads with town) --------------------- - - -def test_no_road_leads_with_town_distance(): - n = {"source": "state_511_atis", "road": None, "direction": "southbound", - "mile_start": None, "mile_end": None, "sub_type": "ramp work", - "impact": "partial", "town": "Stanley", "distance_mi": 3, "bearing": "NE", - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - # Head is the distance/town form, not a placeholder. - assert out.startswith("🚧 3 mi NE of Stanley") - assert "Road event" not in out - - -def test_no_road_no_town_falls_back_to_placeholder(): - n = {"source": "state_511_atis", "road": None, "direction": "both", - "mile_start": None, "mile_end": None, "sub_type": None, - "impact": "partial", "town": None, "distance_mi": None, "bearing": None, - "ends_at": None, "description": ""} - out = format_work_zone_mesh(n) - # Placeholder is acceptable when we have literally nothing. - assert out.startswith("🚧 ")