Spaces:
Running
Running
| """Pydantic schemas for the Decidron simulator. | |
| These model the patent's core objects in a deliberately simplified, | |
| deterministic form: | |
| - ``DecidronNode`` -> a coupled MLU node (decidron) or a sensor/actuator. | |
| - ``Edge`` -> a coupling between nodes. | |
| - ``ExperienceEvent`` -> a simplified ECDS entry recorded on a node. | |
| - ``ProcessingCommand`` -> a user-defined IF (sensor matches) THEN | |
| (emit output / drive actuator / modify network) rule. | |
| """ | |
| from __future__ import annotations | |
| import time | |
| from typing import Literal, Optional | |
| from pydantic import BaseModel, ConfigDict, Field | |
| NodeType = Literal["decidron", "sensor", "actuator"] | |
| NetworkOpType = Literal["add_edge", "remove_edge", "add_node"] | |
| ExperienceKind = Literal["sensor_received", "rule_fired", "actuator_acted"] | |
| class ExperienceEvent(BaseModel): | |
| """A simplified ECDS (Experience Chain Data Structure) entry. | |
| The patent's signature behavior is that each unit records its | |
| experience; in v1 we capture the salient facts of each run. | |
| """ | |
| kind: ExperienceKind | |
| detail: str | |
| command: Optional[str] = None | |
| sensor: Optional[str] = None | |
| value: Optional[str] = None | |
| timestamp: float = Field(default_factory=time.time) | |
| class DecidronNode(BaseModel): | |
| id: str | |
| label: str | |
| type: NodeType = "decidron" | |
| experience: list[ExperienceEvent] = Field(default_factory=list) | |
| class Edge(BaseModel): | |
| """A directed coupling between two nodes (``from`` -> ``to``).""" | |
| model_config = ConfigDict(populate_by_name=True) | |
| source: str = Field(alias="from") | |
| target: str = Field(alias="to") | |
| label: Optional[str] = None | |
| class NetworkOp(BaseModel): | |
| """A topology modification applied when a command fires.""" | |
| model_config = ConfigDict(populate_by_name=True) | |
| op: NetworkOpType | |
| source: Optional[str] = Field(default=None, alias="from") | |
| target: Optional[str] = Field(default=None, alias="to") | |
| node: Optional[DecidronNode] = None | |
| class SensorInput(BaseModel): | |
| channel: str | |
| value: str | |
| timestamp: float = Field(default_factory=time.time) | |
| class ProcessingCommand(BaseModel): | |
| """A user-defined processing rule. | |
| ``match`` is an ``operator:value`` string, e.g. ``contains:85``, | |
| ``equals:on``, ``gt:80``. A bare value is treated as ``contains``. | |
| """ | |
| name: str | |
| sensor: str | |
| match: str | |
| output: str | |
| actuator: Optional[str] = None | |
| network_ops: list[NetworkOp] = Field(default_factory=list) | |
| class NetworkSnapshot(BaseModel): | |
| version: int | |
| reason: str | |
| nodes: list[DecidronNode] | |
| edges: list[Edge] | |
| timestamp: float = Field(default_factory=time.time) | |
| class RunResult(BaseModel): | |
| command: str | |
| sensor: str | |
| value: str | |
| matched: bool | |
| output: Optional[str] = None | |
| actuator: Optional[str] = None | |
| network_ops_applied: int = 0 | |
| latency_ms: float = 0.0 | |
| timestamp: float = Field(default_factory=time.time) | |