Spaces:
Running
Running
| """Deterministic Decidron processing engine. | |
| Ties the network topology, statistics, user commands, and queued sensor | |
| inputs together. On each run it matches commands against sensor inputs | |
| and, for every match: | |
| 1. records ECDS-like experience events on the involved nodes | |
| (sensor received -> decidron fired -> actuator acted), | |
| 2. emits the command output and drives the target actuator (if any), | |
| 3. applies topology ``network_ops`` (snapshotting history), | |
| 4. updates performance statistics. | |
| No LLM is involved -- v1 is purely rule-based. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import time | |
| from pathlib import Path | |
| from .models import ( | |
| ExperienceEvent, | |
| ProcessingCommand, | |
| RunResult, | |
| SensorInput, | |
| ) | |
| from .network import NetworkStore | |
| from .stats import StatsCollector | |
| _DATA_DIR = Path(__file__).resolve().parent.parent / "data" / "decidron" | |
| _SAMPLE_COMMANDS = _DATA_DIR / "sample_commands.json" | |
| def _matches(rule: str, value: str) -> bool: | |
| """Evaluate an ``operator:value`` match rule against a sensor value.""" | |
| if ":" in rule: | |
| op, _, target = rule.partition(":") | |
| op = op.strip().lower() | |
| target = target.strip() | |
| else: | |
| op, target = "contains", rule.strip() | |
| v = value.strip() | |
| def _nums(): | |
| try: | |
| return float(v), float(target) | |
| except (TypeError, ValueError): | |
| return None, None | |
| if op == "contains": | |
| return target.lower() in v.lower() | |
| if op == "equals": | |
| return v.lower() == target.lower() | |
| if op == "startswith": | |
| return v.lower().startswith(target.lower()) | |
| if op == "endswith": | |
| return v.lower().endswith(target.lower()) | |
| if op == "regex": | |
| try: | |
| return re.search(target, v) is not None | |
| except re.error: | |
| return False | |
| if op in {"gt", "lt", "gte", "lte"}: | |
| a, b = _nums() | |
| if a is None: | |
| return False | |
| return { | |
| "gt": a > b, | |
| "lt": a < b, | |
| "gte": a >= b, | |
| "lte": a <= b, | |
| }[op] | |
| # Unknown operator -> no match. | |
| return False | |
| class DecidronEngine: | |
| def __init__(self) -> None: | |
| self.network = NetworkStore() | |
| self.stats = StatsCollector() | |
| self.commands: dict[str, ProcessingCommand] = {} | |
| self.pending_inputs: list[SensorInput] = [] | |
| self.input_log: list[SensorInput] = [] | |
| self.run_results: list[RunResult] = [] | |
| self._load_sample_commands() | |
| def _load_sample_commands(self) -> None: | |
| if _SAMPLE_COMMANDS.is_file(): | |
| for c in json.loads(_SAMPLE_COMMANDS.read_text(encoding="utf-8")): | |
| cmd = ProcessingCommand(**c) | |
| self.commands[cmd.name] = cmd | |
| # -- commands -------------------------------------------------------- | |
| def list_commands(self) -> list[ProcessingCommand]: | |
| return list(self.commands.values()) | |
| def add_command(self, command: ProcessingCommand) -> ProcessingCommand: | |
| self.commands[command.name] = command | |
| return command | |
| def remove_command(self, name: str) -> bool: | |
| return self.commands.pop(name, None) is not None | |
| # -- sensor inputs --------------------------------------------------- | |
| def add_sensor_input(self, sensor_input: SensorInput) -> SensorInput: | |
| self.pending_inputs.append(sensor_input) | |
| return sensor_input | |
| # -- experience (ECDS-like) ----------------------------------------- | |
| def _record(self, node_id: str, event: ExperienceEvent) -> None: | |
| node = self.network.nodes.get(node_id) | |
| if node is not None: | |
| node.experience.append(event) | |
| def _decidrons_for_sensor(self, sensor_node_id: str | None) -> list[str]: | |
| """Decidron nodes coupled downstream of a sensor (fallback: all).""" | |
| if sensor_node_id is not None: | |
| coupled = [ | |
| nid for nid in self.network.neighbors_out(sensor_node_id) | |
| if self.network.nodes.get(nid) and self.network.nodes[nid].type == "decidron" | |
| ] | |
| if coupled: | |
| return coupled | |
| return [n.id for n in self.network.nodes_list() if n.type == "decidron"] | |
| # -- run ------------------------------------------------------------- | |
| def run(self, inputs: list[SensorInput] | None = None) -> list[RunResult]: | |
| batch = inputs if inputs is not None else list(self.pending_inputs) | |
| results: list[RunResult] = [] | |
| for si in batch: | |
| sensor_node = self.network.find_sensor_node(si.channel) | |
| if sensor_node is not None: | |
| self._record(sensor_node.id, ExperienceEvent( | |
| kind="sensor_received", | |
| detail=f"received '{si.value}' on channel '{si.channel}'", | |
| sensor=si.channel, | |
| value=si.value, | |
| )) | |
| for cmd in self.commands.values(): | |
| if cmd.sensor.lower() != si.channel.lower(): | |
| continue | |
| start = time.perf_counter() | |
| matched = _matches(cmd.match, si.value) | |
| ops_applied = 0 | |
| output = None | |
| actuator = None | |
| if matched: | |
| output = cmd.output | |
| for nid in self._decidrons_for_sensor( | |
| sensor_node.id if sensor_node else None | |
| ): | |
| self._record(nid, ExperienceEvent( | |
| kind="rule_fired", | |
| detail=f"command '{cmd.name}' fired -> {cmd.output}", | |
| command=cmd.name, | |
| sensor=si.channel, | |
| value=si.value, | |
| )) | |
| if cmd.actuator and cmd.actuator in self.network.nodes: | |
| actuator = cmd.actuator | |
| self._record(cmd.actuator, ExperienceEvent( | |
| kind="actuator_acted", | |
| detail=f"actuated by '{cmd.name}': {cmd.output}", | |
| command=cmd.name, | |
| )) | |
| if cmd.network_ops: | |
| ops_applied = self.network.apply_ops( | |
| cmd.network_ops, reason=f"command '{cmd.name}' fired" | |
| ) | |
| latency_ms = (time.perf_counter() - start) * 1000.0 | |
| self.stats.record(si.channel, cmd.name, output, matched, latency_ms) | |
| result = RunResult( | |
| command=cmd.name, | |
| sensor=si.channel, | |
| value=si.value, | |
| matched=matched, | |
| output=output, | |
| actuator=actuator, | |
| network_ops_applied=ops_applied, | |
| latency_ms=round(latency_ms, 4), | |
| ) | |
| results.append(result) | |
| self.input_log.append(si) | |
| if inputs is None: | |
| self.pending_inputs.clear() | |
| self.run_results.extend(results) | |
| return results | |
| def reset(self) -> None: | |
| self.network.reset() | |
| self.stats.reset() | |
| self.pending_inputs.clear() | |
| self.input_log.clear() | |
| self.run_results.clear() | |
| engine = DecidronEngine() | |