"""Deterministic supervisory controller. This is not an electrical protection device.""" from dataclasses import dataclass, asdict from copy import deepcopy import math @dataclass(frozen=True) class Inputs: wet_low: bool = False start: bool = False assist: bool = False feedback_a: bool = False feedback_b: bool = False available_a: bool = True available_b: bool = True manual_a: bool = False manual_b: bool = False power: bool = True reset: bool = False @dataclass class Pump: state: str = 'OFF' command: bool = False fault: str = '' started_at: float | None = None feedback_lost_at: float | None = None class Controller: def __init__(self, start_timeout=3.0, feedback_timeout=1.0, minimum_run=10.0): if not all(math.isfinite(x) and x > 0 for x in (start_timeout, feedback_timeout, minimum_run)): raise ValueError('All timers must be finite and positive') self.start_timeout = float(start_timeout) self.feedback_timeout = float(feedback_timeout) self.minimum_run = float(minimum_run) self.pumps = [Pump(), Pump()] self.lead = 0 self.active_lead = None self.automatic_cycle = False self.cycle_has_run = False self.boot_inhibit = True self.sensor_fault = False self.last_time = None self.inputs = Inputs() self.transitions = set() def clone(self): return deepcopy(self) def _state(self, index, state): p = self.pumps[index] if p.state != state: self.transitions.add((p.state, state)) p.state = state def _off(self, index): p = self.pumps[index] p.command = False p.started_at = None p.feedback_lost_at = None self._state(index, 'FAULT' if p.fault else 'OFF') def _trip(self, index, reason): self.pumps[index].fault = reason self._off(index) def apply_events(self, time, events): """Apply one timestamp atomically; duplicate input keys are rejected, not ordered.""" updates = {} for event in events: if updates.keys() & event.keys(): raise ValueError('Conflicting updates at the same timestamp') updates.update(event) snapshot = asdict(self.inputs) snapshot['reset'] = False # reset is a pulse unless explicitly supplied snapshot.update(updates) return self.step(time, Inputs(**snapshot)) def step(self, time, inputs): if not isinstance(inputs, Inputs): raise TypeError('Inputs snapshot required') if any(type(v) is not bool for v in asdict(inputs).values()): raise TypeError('Inputs must be booleans') if not math.isfinite(time) or (self.last_time is not None and time < self.last_time): raise ValueError('Time must be finite and monotonic') self.last_time = float(time) self.inputs = inputs feedback = (inputs.feedback_a, inputs.feedback_b) available = (inputs.available_a, inputs.available_b) manual = (inputs.manual_a, inputs.manual_b) valid = not ((inputs.assist and not inputs.start) or (inputs.start and not inputs.wet_low)) quiet = valid and not any((inputs.wet_low, inputs.start, inputs.assist, *feedback, *manual)) # Highest priority: loss of power. Persistent fault latches survive this transition. if not inputs.power: for i in range(2): self._off(i) self.boot_inhibit = True self.automatic_cycle = False self.cycle_has_run = False self.active_lead = None return self.output('POWER_OFF') if inputs.reset and quiet: self.boot_inhibit = False self.sensor_fault = False for i in range(2): if available[i]: self.pumps[i].fault = '' self._off(i) if not valid: self.sensor_fault = True if self.boot_inhibit or self.sensor_fault: for i in range(2): self._off(i) self.automatic_cycle = False self.cycle_has_run = False self.active_lead = None return self.output('INHIBITED' if self.boot_inhibit else 'SENSOR_FAULT') for i, p in enumerate(self.pumps): if not available[i]: self._off(i) continue if p.fault: continue if p.command: if feedback[i]: p.feedback_lost_at = None self._state(i, 'RUNNING') if self.automatic_cycle: self.cycle_has_run = True elif p.state == 'STARTING' and time - p.started_at >= self.start_timeout: self._trip(i, 'START_TIMEOUT') elif p.state == 'RUNNING': if p.feedback_lost_at is None: p.feedback_lost_at = time if time - p.feedback_lost_at >= self.feedback_timeout: self._trip(i, 'RUN_FEEDBACK_LOST') elif feedback[i]: self._trip(i, 'UNCOMMANDED_FEEDBACK') # Low-level dry indication overrides every manual/minimum-run request. if not inputs.wet_low: if self.automatic_cycle and self.cycle_has_run: self.lead = 1 - self.lead self.automatic_cycle = False self.cycle_has_run = False self.active_lead = None for i in range(2): self._off(i) return self.output('IDLE') usable = [available[i] and not self.pumps[i].fault for i in range(2)] if inputs.start and not self.automatic_cycle: self.automatic_cycle = True self.cycle_has_run = False self.active_lead = self.lead if usable[self.lead] else 1 - self.lead if self.automatic_cycle and (self.active_lead is None or not usable[self.active_lead]): self.active_lead = next((i for i in [self.lead, 1-self.lead] if usable[i]), None) wanted = [manual[0], manual[1]] if self.automatic_cycle and self.active_lead is not None: wanted[self.active_lead] = True if inputs.assist: wanted[1-self.active_lead] = True for i, p in enumerate(self.pumps): if not usable[i]: self._off(i) continue # Only a proved-running pump may finish a minimum-run hold after demand drops. hold = p.state == 'RUNNING' and p.command and time - p.started_at < self.minimum_run command = wanted[i] or hold if command and not p.command: p.command = True p.started_at = time p.feedback_lost_at = None self._state(i, 'STARTING') elif not command: self._off(i) return self.output() def output(self, override=None): commands = [p.command for p in self.pumps] mode = override or ('BOTH' if all(commands) else 'PUMP_A' if commands[0] else 'PUMP_B' if commands[1] else 'NO_CAPACITY' if self.automatic_cycle else 'IDLE') return {'time_s': self.last_time, 'mode': mode, 'command_a': commands[0], 'command_b': commands[1], 'state_a': self.pumps[0].state, 'state_b': self.pumps[1].state, 'fault_a': self.pumps[0].fault, 'fault_b': self.pumps[1].fault, 'sensor_fault': self.sensor_fault, 'restart_inhibited': self.boot_inhibit, 'next_lead': 'A' if self.lead == 0 else 'B'}