|
| 1 | +"""Engine-owned projection into the AHBG presentation snapshot contract. |
| 2 | +
|
| 3 | +This module converts already-resolved engine state into |
| 4 | +``ahbg.presentation.snapshot`` data. It does not validate or decide mechanics; |
| 5 | +movement legality has already been settled by the engine before a ``move`` |
| 6 | +event exists. |
| 7 | +""" |
| 8 | + |
| 9 | +from __future__ import annotations |
| 10 | + |
| 11 | +from typing import Any |
| 12 | + |
| 13 | +from ahbg.presentation.snapshot import ( |
| 14 | + KIND as PRESENTATION_KIND, |
| 15 | + STANDING as PRESENTATION_STANDING, |
| 16 | + validate_snapshot, |
| 17 | +) |
| 18 | + |
| 19 | +from .errors import ReplayMismatch, ValidationError |
| 20 | +from .events import KIND_MOVE, KIND_PLANE_INIT, EventLog |
| 21 | +from .movement import spec_from_event_data |
| 22 | +from .persistence import replay |
| 23 | +from .plane import Plane |
| 24 | + |
| 25 | + |
| 26 | +def motion_traces_from_log(log: EventLog, turn: int | None = None) -> list[dict[str, str]]: |
| 27 | + """Return presentation traces for canonical ``move`` events. |
| 28 | +
|
| 29 | + If ``turn`` is supplied, only move events from that engine turn are |
| 30 | + returned. The trace fields name already-presented unit and tile ids; they |
| 31 | + do not re-check adjacency or War conditions. |
| 32 | + """ |
| 33 | + log.verify() |
| 34 | + traces: list[dict[str, str]] = [] |
| 35 | + for event in log.events: |
| 36 | + if event.kind != KIND_MOVE: |
| 37 | + continue |
| 38 | + if turn is not None and event.turn != turn: |
| 39 | + continue |
| 40 | + spec = spec_from_event_data(event.data) |
| 41 | + traces.append( |
| 42 | + { |
| 43 | + "unit": spec.unit_id, |
| 44 | + "from": spec.from_tile_id, |
| 45 | + "to": spec.to_tile_id, |
| 46 | + } |
| 47 | + ) |
| 48 | + return traces |
| 49 | + |
| 50 | + |
| 51 | +def feed_from_log(log: EventLog) -> list[dict[str, Any]]: |
| 52 | + """Build a compact human feed from engine provenance events.""" |
| 53 | + log.verify() |
| 54 | + feed: list[dict[str, Any]] = [] |
| 55 | + for event in log.events: |
| 56 | + if event.kind == KIND_PLANE_INIT: |
| 57 | + units = event.data.get("plane", {}).get("units", []) |
| 58 | + if units: |
| 59 | + placements = ", ".join( |
| 60 | + f"{unit.get('label') or unit.get('unit_id')} at {unit.get('tile_id')}" |
| 61 | + for unit in units |
| 62 | + ) |
| 63 | + feed.append( |
| 64 | + {"turn": event.turn, "text": f"plane loaded; {placements}"} |
| 65 | + ) |
| 66 | + else: |
| 67 | + feed.append({"turn": event.turn, "text": "plane loaded"}) |
| 68 | + elif event.kind == KIND_MOVE: |
| 69 | + spec = spec_from_event_data(event.data) |
| 70 | + feed.append( |
| 71 | + { |
| 72 | + "turn": event.turn, |
| 73 | + "text": ( |
| 74 | + f"{spec.unit_id} move " |
| 75 | + f"{spec.from_tile_id} to {spec.to_tile_id}" |
| 76 | + ), |
| 77 | + } |
| 78 | + ) |
| 79 | + return feed |
| 80 | + |
| 81 | + |
| 82 | +def snapshot_from_plane( |
| 83 | + plane: Plane, |
| 84 | + log: EventLog | None = None, |
| 85 | + *, |
| 86 | + plane_id: str = "plane-0", |
| 87 | + selected_tile_id: str | None = None, |
| 88 | +) -> dict[str, Any]: |
| 89 | + """Project an engine plane into the presentation snapshot format. |
| 90 | +
|
| 91 | + When a log is supplied, it must replay exactly to ``plane`` before any |
| 92 | + presentation data is emitted. Default motion traces are the moves from the |
| 93 | + last completed turn, matching the current visual transition into the |
| 94 | + presented plane state. |
| 95 | + """ |
| 96 | + plane.validate() |
| 97 | + if not isinstance(plane_id, str) or not plane_id: |
| 98 | + raise ValidationError("presentation plane_id must be exact non-empty text") |
| 99 | + if log is not None: |
| 100 | + replayed = replay(log) |
| 101 | + if replayed.canonical_dict() != plane.canonical_dict(): |
| 102 | + raise ReplayMismatch("presentation snapshot source log does not replay to plane") |
| 103 | + |
| 104 | + tiles = [ |
| 105 | + {"id": tile.tile_id, "q": tile.q, "r": tile.r, "label": tile.tile_id} |
| 106 | + for tile in sorted(plane.tiles.values(), key=lambda item: item.tile_id) |
| 107 | + ] |
| 108 | + units = [ |
| 109 | + { |
| 110 | + "id": unit.unit_id, |
| 111 | + "tile": unit.tile_id, |
| 112 | + "label": unit.label or unit.unit_id, |
| 113 | + } |
| 114 | + for unit in sorted(plane.units.values(), key=lambda item: item.unit_id) |
| 115 | + ] |
| 116 | + tile_ids = {tile["id"] for tile in tiles} |
| 117 | + if selected_tile_id is None and units: |
| 118 | + selected_tile_id = units[0]["tile"] |
| 119 | + if selected_tile_id is not None and selected_tile_id not in tile_ids: |
| 120 | + raise ValidationError("selected_tile_id must name a plane tile") |
| 121 | + |
| 122 | + payload: dict[str, Any] = { |
| 123 | + "kind": PRESENTATION_KIND, |
| 124 | + "standing": PRESENTATION_STANDING, |
| 125 | + "plane_id": plane_id, |
| 126 | + "turn": plane.turn, |
| 127 | + "tiles": tiles, |
| 128 | + "units": units, |
| 129 | + "selected_tile": selected_tile_id, |
| 130 | + "feed": feed_from_log(log) if log is not None else [], |
| 131 | + } |
| 132 | + if log is not None: |
| 133 | + payload["motions"] = motion_traces_from_log( |
| 134 | + log, |
| 135 | + turn=plane.turn - 1 if plane.turn > 0 else None, |
| 136 | + ) |
| 137 | + return dict(validate_snapshot(payload)) |
0 commit comments