|
| 1 | +"""Small, safety-oriented control surface for one seven-prism cluster. |
| 2 | +
|
| 3 | +This module intentionally uses the public :class:`HexMazeInterface` methods |
| 4 | +only. It is independent of any GUI toolkit so both the desktop application |
| 5 | +and automated tests use the same validation and command sequence. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +import time |
| 11 | +from dataclasses import dataclass, field, replace |
| 12 | +from typing import Any |
| 13 | + |
| 14 | +from .hex_maze_interface import ( |
| 15 | + ControllerParameters, |
| 16 | + HexMazeInterface, |
| 17 | + HomeOutcome, |
| 18 | + HomeParameters, |
| 19 | + MazeException, |
| 20 | +) |
| 21 | + |
| 22 | + |
| 23 | +@dataclass(frozen=True, slots=True) |
| 24 | +class ClusterControlSettings: |
| 25 | + """Limits and fixed homing parameters for the operator application. |
| 26 | +
|
| 27 | + The maximum velocity range is deliberately more conservative than the |
| 28 | + protocol's unsigned-byte encoding. Change these settings in code only |
| 29 | + after the corresponding values have been validated on the physical rig. |
| 30 | + """ |
| 31 | + |
| 32 | + minimum_position_mm: int = 0 |
| 33 | + maximum_position_mm: int = 550 |
| 34 | + minimum_max_velocity_mm_s: int = 1 |
| 35 | + maximum_max_velocity_mm_s: int = 40 |
| 36 | + home_parameters: HomeParameters = field( |
| 37 | + default_factory=lambda: HomeParameters( |
| 38 | + travel_limit=250, |
| 39 | + max_velocity=20, |
| 40 | + run_current=50, |
| 41 | + stall_threshold=10, |
| 42 | + ) |
| 43 | + ) |
| 44 | + home_timeout_s: float = 30.0 |
| 45 | + home_poll_interval_s: float = 0.25 |
| 46 | + |
| 47 | + def __post_init__(self) -> None: |
| 48 | + if self.minimum_position_mm > self.maximum_position_mm: |
| 49 | + raise ValueError("minimum_position_mm must not exceed maximum_position_mm") |
| 50 | + if self.minimum_max_velocity_mm_s > self.maximum_max_velocity_mm_s: |
| 51 | + raise ValueError("minimum_max_velocity_mm_s must not exceed maximum_max_velocity_mm_s") |
| 52 | + if self.home_timeout_s <= 0: |
| 53 | + raise ValueError("home_timeout_s must be positive") |
| 54 | + if self.home_poll_interval_s <= 0: |
| 55 | + raise ValueError("home_poll_interval_s must be positive") |
| 56 | + |
| 57 | + |
| 58 | +DEFAULT_CLUSTER_CONTROL_SETTINGS = ClusterControlSettings() |
| 59 | + |
| 60 | + |
| 61 | +@dataclass(frozen=True, slots=True) |
| 62 | +class ClusterState: |
| 63 | + """The status shown by the single-cluster operator application.""" |
| 64 | + |
| 65 | + positions_mm: tuple[int, ...] |
| 66 | + homed: tuple[bool, ...] |
| 67 | + home_outcomes: tuple[HomeOutcome, ...] |
| 68 | + controller_parameters: ControllerParameters |
| 69 | + |
| 70 | + |
| 71 | +class ClusterControl: |
| 72 | + """Validate and execute the limited operator workflow for one cluster.""" |
| 73 | + |
| 74 | + def __init__( |
| 75 | + self, |
| 76 | + hmi: HexMazeInterface, |
| 77 | + cluster_address: int, |
| 78 | + *, |
| 79 | + settings: ClusterControlSettings = DEFAULT_CLUSTER_CONTROL_SETTINGS, |
| 80 | + sleep_fn: Any = time.sleep, |
| 81 | + monotonic_fn: Any = time.monotonic, |
| 82 | + ) -> None: |
| 83 | + HexMazeInterface._validate_cluster_address(cluster_address) |
| 84 | + self._hmi = hmi |
| 85 | + self.cluster_address = cluster_address |
| 86 | + self.settings = settings |
| 87 | + self._sleep_fn = sleep_fn |
| 88 | + self._monotonic_fn = monotonic_fn |
| 89 | + |
| 90 | + def connect(self) -> ClusterState: |
| 91 | + """Confirm that the selected cluster is reachable and read its state.""" |
| 92 | + if not self._hmi.communicating_cluster(self.cluster_address): |
| 93 | + raise MazeException(f"cluster {self.cluster_address} did not respond over Ethernet") |
| 94 | + return self.read_state() |
| 95 | + |
| 96 | + def read_state(self) -> ClusterState: |
| 97 | + """Read all status that the GUI presents to the operator.""" |
| 98 | + positions_mm = tuple(self._hmi.read_positions_cluster(self.cluster_address)) |
| 99 | + homed = tuple(bool(value) for value in self._hmi.homed_cluster(self.cluster_address)) |
| 100 | + home_outcomes = tuple(self._hmi.read_home_outcomes_cluster(self.cluster_address)) |
| 101 | + controller_parameters = self._hmi.read_controller_parameters_cluster(self.cluster_address) |
| 102 | + return ClusterState( |
| 103 | + positions_mm=positions_mm, |
| 104 | + homed=homed, |
| 105 | + home_outcomes=home_outcomes, |
| 106 | + controller_parameters=controller_parameters, |
| 107 | + ) |
| 108 | + |
| 109 | + @staticmethod |
| 110 | + def home_succeeded(state: ClusterState) -> bool: |
| 111 | + """Return whether every prism reports a successful home outcome. |
| 112 | +
|
| 113 | + Firmware can transiently report a false ``homed`` flag after a stall |
| 114 | + home even though its terminal outcome is successful. The terminal |
| 115 | + outcomes are therefore the authoritative signal for enabling motion. |
| 116 | + """ |
| 117 | + return all( |
| 118 | + outcome in (HomeOutcome.STALL, HomeOutcome.TARGET_REACHED) |
| 119 | + for outcome in state.home_outcomes |
| 120 | + ) |
| 121 | + |
| 122 | + def set_max_velocity(self, max_velocity_mm_s: int) -> ClusterState: |
| 123 | + """Set the shared cluster maximum velocity and verify its readback.""" |
| 124 | + self._validate_max_velocity(max_velocity_mm_s) |
| 125 | + current = self._hmi.read_controller_parameters_cluster(self.cluster_address) |
| 126 | + requested = replace(current, max_velocity=max_velocity_mm_s) |
| 127 | + if not self._hmi.write_controller_parameters_cluster(self.cluster_address, requested): |
| 128 | + raise MazeException("controller-parameter write failed") |
| 129 | + |
| 130 | + actual = self._hmi.read_controller_parameters_cluster(self.cluster_address) |
| 131 | + if actual.max_velocity != max_velocity_mm_s: |
| 132 | + raise MazeException( |
| 133 | + "controller maximum velocity was not accepted: " |
| 134 | + f"requested {max_velocity_mm_s}, read back {actual.max_velocity}" |
| 135 | + ) |
| 136 | + return self.read_state() |
| 137 | + |
| 138 | + def home_all(self) -> ClusterState: |
| 139 | + """Start homing all prisms and wait for a terminal outcome.""" |
| 140 | + parameters = HomeParameters(*self.settings.home_parameters.to_tuple()) |
| 141 | + if not self._hmi.home_cluster(self.cluster_address, parameters): |
| 142 | + raise MazeException("home command was not accepted") |
| 143 | + |
| 144 | + deadline = self._monotonic_fn() + self.settings.home_timeout_s |
| 145 | + last_state = self.read_state() |
| 146 | + while True: |
| 147 | + if self.home_succeeded(last_state): |
| 148 | + return last_state |
| 149 | + if any(outcome == HomeOutcome.FAILED for outcome in last_state.home_outcomes): |
| 150 | + raise MazeException(self._home_error_message(last_state)) |
| 151 | + if self._monotonic_fn() >= deadline: |
| 152 | + raise MazeException( |
| 153 | + "homing did not complete before the " |
| 154 | + f"{self.settings.home_timeout_s:g} s timeout; " |
| 155 | + f"last outcomes: {self._home_outcomes_text(last_state)}" |
| 156 | + ) |
| 157 | + self._sleep_fn(self.settings.home_poll_interval_s) |
| 158 | + last_state = self.read_state() |
| 159 | + |
| 160 | + def move_all(self, positions_mm: Any) -> None: |
| 161 | + """Send all seven target positions after validating the operator input.""" |
| 162 | + positions = self._validate_positions(positions_mm) |
| 163 | + if not self._hmi.write_targets_cluster(self.cluster_address, positions): |
| 164 | + raise MazeException("target-position write failed") |
| 165 | + |
| 166 | + def pause(self) -> None: |
| 167 | + """Immediately pause every prism in the selected cluster.""" |
| 168 | + if not self._hmi.pause_cluster(self.cluster_address): |
| 169 | + raise MazeException("pause command was not accepted") |
| 170 | + |
| 171 | + def power_off(self) -> None: |
| 172 | + """Turn off the selected cluster's prism power.""" |
| 173 | + if not self._hmi.power_off_cluster(self.cluster_address): |
| 174 | + raise MazeException("power-off command was not accepted") |
| 175 | + |
| 176 | + def _validate_max_velocity(self, value: int) -> None: |
| 177 | + if not isinstance(value, int): |
| 178 | + raise MazeException("maximum velocity must be an integer") |
| 179 | + if not ( |
| 180 | + self.settings.minimum_max_velocity_mm_s |
| 181 | + <= value |
| 182 | + <= self.settings.maximum_max_velocity_mm_s |
| 183 | + ): |
| 184 | + raise MazeException( |
| 185 | + "maximum velocity must be between " |
| 186 | + f"{self.settings.minimum_max_velocity_mm_s} and " |
| 187 | + f"{self.settings.maximum_max_velocity_mm_s} mm/s" |
| 188 | + ) |
| 189 | + |
| 190 | + def _validate_positions(self, positions_mm: Any) -> tuple[int, ...]: |
| 191 | + try: |
| 192 | + positions = tuple(positions_mm) |
| 193 | + except TypeError as exc: |
| 194 | + raise MazeException("target positions must be iterable") from exc |
| 195 | + if len(positions) != HexMazeInterface.PRISM_COUNT: |
| 196 | + raise MazeException( |
| 197 | + f"exactly {HexMazeInterface.PRISM_COUNT} target positions are required" |
| 198 | + ) |
| 199 | + for prism_index, position_mm in enumerate(positions, start=1): |
| 200 | + if not isinstance(position_mm, int): |
| 201 | + raise MazeException(f"prism {prism_index} target position must be an integer") |
| 202 | + if not ( |
| 203 | + self.settings.minimum_position_mm |
| 204 | + <= position_mm |
| 205 | + <= self.settings.maximum_position_mm |
| 206 | + ): |
| 207 | + raise MazeException( |
| 208 | + f"prism {prism_index} target must be between " |
| 209 | + f"{self.settings.minimum_position_mm} and " |
| 210 | + f"{self.settings.maximum_position_mm} mm" |
| 211 | + ) |
| 212 | + return positions |
| 213 | + |
| 214 | + @staticmethod |
| 215 | + def _home_outcomes_text(state: ClusterState) -> str: |
| 216 | + return ", ".join(outcome.name.lower() for outcome in state.home_outcomes) |
| 217 | + |
| 218 | + def _home_error_message(self, state: ClusterState) -> str: |
| 219 | + return f"one or more prisms failed to home: {self._home_outcomes_text(state)}" |
0 commit comments