matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

chore: delete dead renderers/work_zone.py

Browse source 
format_work_zone_mesh() had exactly one production caller, the
central/consumer.py NATS bridge deleted in the prior Central-excision
pass. Its live replacement, formatters.incident._render_work_zone()
(registered for category "work_zone" in formatters/__init__.py), is an
already-shipped byte-identical replica per that module's own docstring.
All remaining references to renderers.work_zone were prose/comments
describing the replica relationship, not imports.

Test fallout:
- tests/test_work_zone_renderer.py tested only the dead renderer in
  isolation (17 cases). Deleted — the live path has its own coverage
  (test_adapter_wzdx.py's formatter-integration tests, plus
  TestCrossSourceIdentity::test_work_zone_category_uses_wz_renderer and
  TestWorkZoneGolden in test_incident_refactor.py).
- tests/test_itd_511_work_zone.py::test_itd_511_work_zone_renderer_produces_wire
  only smoke-tested the dead renderer's wire output for itd_511 data;
  redundant with TestWorkZoneGolden's byte-identical fixture coverage
  for the same adapter. Deleted.
- tests/test_incident_refactor.py::TestWorkZoneGolden compared the live
  formatters.incident.format() output against a golden computed by
  calling the dead renderer live on two real fixtures. Mirroring the
  precedent already in test_nws_refactor.py for this exact situation
  (golden generator deleted out from under a parity test), the two
  golden strings were captured by running format_work_zone_mesh()
  against these fixtures immediately before deletion and are now
  pinned as literals — same coverage, no live dependency on the dead
  module.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-17 21:03:35 +00:00
commit d98608db22
4 changed files with 28 additions and 435 deletions
Download patch file Download diff file Expand all files Collapse all files

205
work/meshai/notifications/renderers/work_zone.py
View file

@ -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):
🚧 <road> @ mile <start>[–<end>][, <dist> mi <bearing> of <town>]:
<direction phrase>, <sub_type>[, ends <when>]
Optional segments are dropped wholesale (lowest-priority first) when the
byte budget is over:
1. ends <when> (lowest priority — dropped first)
2. <bearing> of <town> distance segment
3. <sub_type>
4. <direction phrase>
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 <town>" 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 <town>" or "🚧 <dist> mi <bearing> of <town>" 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] == ": ")]

35
work/tests/test_incident_refactor.py
View file

@ -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)

16
work/tests/test_itd_511_work_zone.py
View file

@ -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."""

207
work/tests/test_work_zone_renderer.py
View file

@ -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("🚧 ")