pamoja.profile

141class AlertKind(str, enum.Enum):
142    """Which threshold a reading crossed."""
143
144    #: A controlled reading drifted outside its safe band.
145    OUT_OF_RANGE = "OutOfRange"
146    #: A falling level will reach empty within a few more samples.
147    RUNNING_OUT = "RunningOut"
148    #: A reading is changing faster than its safe rate.
149    CHANGING_FAST = "ChangingFast"
150    #: A reading that is not a finite number, such as the NaN a failed probe produces.
151    #: It changes nothing: a setpoint's output holds, and a level or a surge carries on
152    #: from the last good reading.
153    INVALID_READING = "InvalidReading"
154    #: A condition a policy of the program's own raised, named by ``code``.
155    CUSTOM = "Custom"

125class ControlKind(str, enum.Enum):
126    """Which control policy a profile applies to each reading."""
127
128    #: Hold a reading near a setpoint by switching an output on and off.
129    SETPOINT = "Setpoint"
130    #: Watch a falling level and warn before it reaches empty.
131    LEVEL = "Level"
132    #: Warn when a reading changes faster than a limit.
133    SURGE = "Surge"
134    #: Report readings only, with no output and no alerts.
135    MONITOR = "Monitor"
136    #: A kind the library does not ship, named by the manifest in ``custom_kind`` and
137    #: decided by the program's own code from ``params``.
138    CUSTOM = "Custom"

38@dataclass(frozen=True)
39class CustomAlert:
40    """A condition a policy of the program's own raises, named by a code it chose."""
41
42    #: The condition's name, such as ``FrostRisk``.
43    code: str
44    #: The measurement behind it.
45    value: float
46    #: Always ``Custom``, as the built-in alerts name their own kind.
47    kind: str = "Custom"

50@dataclass(frozen=True)
51class Decision:
52    """What a policy of the program's own decided about one reading."""
53
54    #: The setting the output should take, or ``None`` to leave it alone.
55    actuator: Optional[bool] = None
56    #: A condition the reading raised, or ``None``.
57    alert: Optional[CustomAlert] = None

21class Link(Protocol):
22    """A link a node publishes its readings over: a ``LoopbackTransport``, an
23    ``MqttClient``, a ``CoapClient``, a ``Ladder``, or anything else with the same
24    ``send``. Connect it first."""
25
26    def send(self, topic: str, payload: Payload) -> Awaitable[Any]:
27        """Publish one message to a topic."""

112class Node:
113    """A profile assembled around the parts that make it run.
114
115    Each :meth:`tick` reads, lets the profile's policy decide, switches the output when the
116    policy calls for it, and publishes the reading to the profile's topic. :meth:`run`
117    repeats that at the cadence the profile's power schedule sets for the battery's charge.
118    """
119
120    def __init__(
121        self,
122        profile: Profile,
123        *,
124        read: Callable[[], Union[float, Awaitable[float]]],
125        link: Link,
126        drive: Optional[Callable[[bool], Any]] = None,
127        policy: Union[Policy, PolicyRegistry, None] = None,
128        encode: Optional[Callable[[float], Payload]] = None,
129    ) -> None:
130        """Assemble a node.
131
132        :param profile: The profile the node runs.
133        :param read: Takes one reading, in the unit the profile reads.
134        :param link: The link each reading is published over, connected before the node
135            runs.
136        :param drive: Switches the output a profile drives; required for a setpoint
137            profile.
138        :param policy: What decides each reading: the profile's own controller unless
139            given, a policy of the program's own, or a registry that resolves the
140            profile's control kind.
141        :param encode: Writes a reading as the payload published; the number as JSON text
142            unless given.
143        :raises PamojaError: When the profile drives an output and no ``drive`` is given,
144            or when its control kind is custom and no policy or registry decides it.
145        """
146        self.profile = profile
147        self._read = read
148        self._link = link
149        self._drive = drive
150        self._encode = encode or json.dumps
151        if isinstance(policy, PolicyRegistry):
152            self._policy = policy.resolve(profile)
153        else:
154            self._policy = policy if policy is not None else profile.controller()
155        if profile.control.kind == "Setpoint" and drive is None:
156            raise PamojaError(
157                f"the profile `{profile.name}` switches an output, so the node needs `drive`"
158            )
159        self._plan = profile.power_plan()
160        self._mode: Optional[str] = None
161
162    @property
163    def power_mode(self) -> Optional[str]:
164        """The power mode the last :meth:`schedule` chose, or ``None`` before the first."""
165        return self._mode
166
167    async def tick(self) -> Tick:
168        """Run one read, decide, act, and publish cycle.
169
170        :returns: The reading and what the policy decided about it.
171        :raises Exception: What the reading, the output, or the link raises.
172        """
173        reading = await _settle(self._read())
174        reaction = self._policy.evaluate(reading)
175        if reaction.actuator is not None and self._drive is not None:
176            await _settle(self._drive(reaction.actuator))
177        await self._link.send(self.profile.topic, self._encode(reading))
178        return Tick(reading, reaction)
179
180    def schedule(self, charge: float, charging: bool = False) -> tuple[str, float]:
181        """Say what power mode the battery's charge puts the node in and how long to wait
182        before the next tick.
183
184        The node remembers the mode it chose, so a charge hovering at a threshold keeps the
185        slower cadence until it clears the schedule's hysteresis.
186
187        :param charge: The state of charge, from 0 to 1.
188        :param charging: Whether the panel is delivering charge.
189        :returns: The mode, and the wait in seconds.
190        """
191        if self._mode is None:
192            mode = self._plan.mode_while_charging(charge, charging)
193        else:
194            mode = self._plan.next_mode_while_charging(self._mode, charge, charging)
195        self._mode = mode
196        return mode, self._plan.interval_for_us(mode) / 1_000_000
197
198    async def run(
199        self,
200        *,
201        battery: Optional[Callable[[], Any]] = None,
202        on_tick: Optional[Callable[[Tick], Any]] = None,
203        on_error: Optional[Callable[[BaseException], Any]] = None,
204        ticks: Optional[int] = None,
205        wait: Optional[Callable[[float], Awaitable[None]]] = None,
206    ) -> None:
207        """Tick, then wait the interval the battery's charge calls for, until cancelled.
208
209        Cancel the task running it to stop; the tick under way is abandoned where it
210        stands.
211
212        :param battery: Reads the battery before each wait, as a charge from 0 to 1 or a
213            ``(charge, charging)`` pair; a node without one samples at the active cadence.
214        :param on_tick: Hears each tick, such as to log it or to act on an alert.
215        :param on_error: Hears a tick that failed. The loop carries on at the same cadence;
216            without one, the failure ends the loop and is raised.
217        :param ticks: How many ticks to run before returning; the loop runs until cancelled
218            unless given.
219        :param wait: Waits between ticks, given seconds; :func:`asyncio.sleep` unless given,
220            which a test replaces to run at once.
221        """
222        pause = wait or asyncio.sleep
223        count = 0
224        while ticks is None or count < ticks:
225            try:
226                tick = await self.tick()
227                if on_tick is not None:
228                    await _settle(on_tick(tick))
229            except Exception as error:
230                if on_error is None:
231                    raise
232                await _settle(on_error(error))
233            charge = 1.0 if battery is None else await _settle(battery())
234            charging = False
235            if isinstance(charge, tuple):
236                charge, charging = charge
237            _, seconds = self.schedule(charge, charging)
238            count += 1
239            if ticks is None or count < ticks:
240                await pause(seconds)

30class Policy(Protocol):
31    """What decides each reading: a profile's built-in ``Controller``, or a policy of the
32    program's own, which returns a :class:`Decision`."""
33
34    def evaluate(self, reading: float) -> Any:
35        """Decide what one reading calls for."""

 70class PolicyRegistry:
 71    """What resolves a profile's control kind to the code that decides it: a built-in kind
 72    to its ``Controller``, and a kind the library never shipped to the policy the factory
 73    registered under its name builds. One program then runs any manifest its registry
 74    covers."""
 75
 76    def __init__(self) -> None:
 77        self._factories: dict[str, Callable[[Params], Policy]] = {}
 78
 79    def register(self, kind: str, factory: Callable[[Params], Policy]) -> "PolicyRegistry":
 80        """Register the factory for a custom kind, replacing one of the same name.
 81
 82        :param kind: The kind as a manifest names it, such as ``frost_guard``.
 83        :param factory: Builds the policy from the parameters beside the kind.
 84        :returns: The registry, for chaining.
 85        """
 86        self._factories[kind] = factory
 87        return self
 88
 89    @property
 90    def kinds(self) -> list[str]:
 91        """The custom kinds registered, in name order."""
 92        return sorted(self._factories)
 93
 94    def resolve(self, profile: Profile) -> Policy:
 95        """Resolve a profile's control kind to the policy that decides it.
 96
 97        :param profile: The profile whose kind is resolved.
 98        :returns: A fresh policy, ready to evaluate readings.
 99        :raises PamojaError: When the kind is custom and no factory is registered for it,
100            naming the kind and the one it was probably meant to be.
101        """
102        control = profile.control
103        if control.kind != "Custom":
104            return profile.controller()
105        kind = control.custom_kind or ""
106        factory = self._factories.get(kind)
107        if factory is None:
108            raise PamojaError(unresolved(kind, self.kinds))
109        return factory(dict(control.params or {}))

15class ReceivingLink(Link, Protocol):
16    """A link a rule engine listens on as well as publishes over: a ``LoopbackTransport``,
17    an ``MqttClient``, a ``CoapClient``, or anything else with the same three calls.
18    Connect it first."""
19
20    def subscribe(self, topic: str) -> Awaitable[Any]:
21        """Subscribe to one topic."""
22
23    def recv(self) -> Awaitable[Any]:
24        """Wait for the next message, with its ``topic``, ``text``, and ``number``."""

80class RuleActionKind(str, enum.Enum):
81    """What a rule calls for when its condition sets or clears."""
82
83    #: Switch the output the program holds under ``actuator``.
84    DRIVE = "drive"
85    #: Publish ``payload`` to ``topic``.
86    PUBLISH = "publish"

 27class RuleEngine:
 28    """Runs a rule file off a link.
 29
 30    :meth:`listen` checks that every output a rule drives was given and subscribes to every
 31    watched topic; each :meth:`step` handles one message, switching the outputs by name and
 32    publishing over the same link; :meth:`run` repeats it until cancelled.
 33    """
 34
 35    def __init__(
 36        self,
 37        rules: Union[str, RuleEvaluator],
 38        link: ReceivingLink,
 39        *,
 40        actuators: Optional[dict[str, Callable[[bool], Any]]] = None,
 41        decode: Optional[Callable[[Any], float]] = None,
 42    ) -> None:
 43        """Assemble an engine.
 44
 45        :param rules: The rule file's text, or an evaluator already loaded from it.
 46        :param link: The link readings arrive on and actions publish over.
 47        :param actuators: The outputs the rules drive, each under the name the rule file
 48            gives it.
 49        :param decode: Reads a message as a reading; its ``number`` unless given, which
 50            raises ``ValueError`` for a payload that is not a number written out.
 51        :raises PamojaError: When the rule file is refused, with the rule and the reason.
 52        """
 53        self.evaluator = RuleEvaluator.from_json(rules) if isinstance(rules, str) else rules
 54        self._link = link
 55        self._actuators = dict(actuators or {})
 56        self._decode = decode or (lambda message: message.number)
 57
 58    @property
 59    def topics(self) -> list[str]:
 60        """The topics the rules watch, each once, in name order."""
 61        return self.evaluator.topics
 62
 63    @property
 64    def actuators(self) -> list[str]:
 65        """The outputs the rules drive, each once, in name order."""
 66        return self.evaluator.actuators
 67
 68    def is_set(self, rule: str) -> Optional[bool]:
 69        """Whether a rule's condition holds.
 70
 71        :param rule: The rule's name.
 72        :returns: The state, or ``None`` for a name no rule has.
 73        """
 74        return self.evaluator.is_set(rule)
 75
 76    async def listen(self) -> None:
 77        """Check the engine was given every output a rule drives, and subscribe to every
 78        watched topic.
 79
 80        :raises PamojaError: When a rule drives an output the engine was not given,
 81            naming it.
 82        """
 83        missing = [name for name in self.actuators if name not in self._actuators]
 84        if missing:
 85            raise PamojaError(
 86                "a rule drives `"
 87                + "`, `".join(missing)
 88                + "`, which the engine was not given under `actuators`"
 89            )
 90        for topic in self.topics:
 91            await self._link.subscribe(topic)
 92
 93    async def step(self, timeout: Optional[float] = None) -> Optional[list[RuleFired]]:
 94        """Wait for one message and run every rule that watches its topic.
 95
 96        :param timeout: How long to wait, in seconds; for good unless given.
 97        :returns: What fired, which is empty when the message set or cleared nothing, or
 98            ``None`` when no message arrived in time or the link has ended.
 99        :raises PamojaError: When a reading on a watched topic is not a finite number.
100        :raises Exception: What decoding or an action raises.
101        """
102        try:
103            message = await asyncio.wait_for(self._link.recv(), timeout)
104        except asyncio.TimeoutError:
105            return None
106        if message is None:
107            return None
108        if not self.evaluator.watches(message.topic):
109            return []
110        fired = self.evaluator.evaluate(message.topic, self._decode(message))
111        for one in fired:
112            for action in one.actions:
113                if action.kind == "drive":
114                    await _settle(self._actuators[action.actuator](bool(action.on)))
115                else:
116                    await self._link.send(action.topic, action.payload)
117        return fired
118
119    async def run(
120        self,
121        *,
122        on_fired: Optional[Callable[[list[RuleFired]], Any]] = None,
123        on_error: Optional[Callable[[BaseException], Any]] = None,
124        messages: Optional[int] = None,
125    ) -> None:
126        """Handle messages until cancelled.
127
128        :param on_fired: Hears each message that set or cleared a rule, with what fired.
129        :param on_error: Hears a message that could not be judged or an action that failed.
130            The loop carries on with the next message; without one, the failure ends the
131            loop and is raised.
132        :param messages: How many messages to handle before returning; the loop runs until
133            cancelled or the link ends unless given.
134        """
135        handled = 0
136        while messages is None or handled < messages:
137            try:
138                fired = await self.step()
139                if fired is None:
140                    return
141                handled += 1
142                if fired and on_fired is not None:
143                    await _settle(on_fired(fired))
144            except Exception as error:
145                handled += 1
146                if on_error is None:
147                    raise
148                await _settle(on_error(error))

60@dataclass(frozen=True)
61class Tick:
62    """One reading a node took, and what its policy decided about it."""
63
64    #: The reading, in the unit the profile reads.
65    reading: float
66    #: The output setting and the alert the policy decided on.
67    reaction: Any

 89class Viz(str, enum.Enum):
 90    """The graphic a dashboard draws an element with, named by the instrument rather
 91    than the quantity. The values are the ones a manifest carries."""
 92
 93    #: A rolling sparkline of recent values.
 94    SPARK = "spark"
 95    #: A 270-degree arch gauge, for a fraction or percentage.
 96    GAUGE = "gauge"
 97    #: A half-dial with a needle, for a pressure or flow reading.
 98    DIAL = "dial"
 99    #: A horizontal bar with a safe-band tick, for a level or stock.
100    BAR = "bar"
101    #: A thermometer, for a temperature.
102    THERMOMETER = "thermometer"
103    #: A liquid-filled droplet, for humidity or moisture.
104    DROPLET = "droplet"
105    #: A segmented battery cell, for a state of charge or voltage.
106    BATTERY = "battery"
107    #: An anemometer, for wind speed.
108    WIND = "wind"
109    #: A sun whose corona grows with the reading, for illuminance.
110    SUN = "sun"
111    #: An acoustic waveform, for sound level or an acoustic event.
112    WAVE = "wave"
113    #: A labeled state chip, lit when the state reads as on.
114    SWITCH = "switch"
115    #: A pipe valve, open along the flow or closed across it.
116    VALVE = "valve"
117    #: A row of hash-chained blocks, for a tamper-evident record count.
118    CHAIN = "chain"
119    #: A neighbor-mesh topology map, for a mesh node's peers.
120    MESH = "mesh"
121    #: A plain numeric counter, for a node or network stat.
122    COUNT = "count"