Monitoring Module

SPDX-FileCopyrightText: 2024 DESY and the Constellation authors SPDX-License-Identifier: EUPL-1.2

class core.monitoring.MonitoringSender(mon_port: int | None = None, **kwds: Any)

Bases: BaseSatelliteFrame

Sender mixin class for Constellation Monitoring Distribution Protocol.

Any method of inheriting classes that has the @schedule_metric decorator, will be regularly polled for new values and a corresponding Metric be sent on the monitoring port.

reentry() → None

Orderly shut down monitoring communication infrastructure.

register_metric(name: str, unit: str, description: str) → None

Register a manually triggered metric

register_scheduled_metric(name: str, unit: str, description: str, interval: float, value_cb: Callable[[], Any], allowed_states: list[SatelliteState] = [SatelliteState.INIT, SatelliteState.ORBIT, SatelliteState.RUN, SatelliteState.SAFE, SatelliteState.launching, SatelliteState.landing, SatelliteState.starting, SatelliteState.stopping, SatelliteState.interrupting, SatelliteState.DEAD]) → None

Register a scheduled metric

reset_metrics() → None

Reset all metric

This removes all metrics and resets it to the ones added via the @schedule_metric decorator.

should_log(levelname: str, topic: str | None = None) → bool

Checks if a log level or topic has any subscribers

should_stat(metric_name) → bool

Checks if a metric has any subscribers

stat(metric_name: str, value: Any) → None

Manually emit a registered metric

unregister_metric(metric_name: str) → None

Removes a previously registered metric

class core.monitoring.ScheduledMetric(name: str, unit: str, description: str, interval: float, value_cb: collections.abc.Callable[[], Any], allowed_states: list[core.protocol.cscp1.SatelliteState] | None)

Bases: Metric

allowed_states: list[SatelliteState] | None

interval: float

value_cb: Callable[[], Any]

core.monitoring.schedule_metric(unit: str, interval: float, allowed_states: list[SatelliteState] | None = [SatelliteState.INIT, SatelliteState.ORBIT, SatelliteState.RUN, SatelliteState.SAFE, SatelliteState.launching, SatelliteState.landing, SatelliteState.starting, SatelliteState.stopping, SatelliteState.interrupting, SatelliteState.DEAD]) → Callable[[Callable[[P], T]], Callable[[P], T]]

Schedule a metric for a function with given interval in seconds