Diagnostics Messages
System health and safety monitoring types.
# simplified
from horus import Heartbeat, DiagnosticStatus, EmergencyStop, SafetyStatus
Heartbeat
Node alive signal — publish periodically to indicate health.
# simplified
hb = horus.Heartbeat(node_name="motor_ctrl", node_id=1, timestamp_ns=horus.timestamp_ns())
| Field | Type | Default | Description |
|---|
node_name | str | "" | Name of the sending node |
node_id | int | 0 | Numeric node identifier |
sequence | int | — | Heartbeat sequence number (getter only) |
alive | bool | — | Whether the node is alive (getter only) |
uptime | float | — | Node uptime in seconds (getter only) |
timestamp_ns | int | 0 | Timestamp in nanoseconds |
| Method | Returns | Description |
|---|
update(uptime) | None | Update heartbeat with current uptime, increments sequence |
DiagnosticStatus
Diagnostic report from a subsystem.
# simplified
status = horus.DiagnosticStatus(component="battery", level=0, message="OK")
# level: 0=OK, 1=WARN, 2=ERROR, 3=FATAL
| Field | Type | Default | Description |
|---|
level | int | 0 | 0=OK, 1=WARN, 2=ERROR, 3=FATAL |
code | int | 0 | Diagnostic error code |
message | str | "" | Human-readable status message |
component | str | "" | Component name |
timestamp_ns | int | 0 | Timestamp in nanoseconds |
Static Methods:
| Method | Returns | Description |
|---|
DiagnosticStatus.ok(message) | DiagnosticStatus | Create an OK status |
DiagnosticStatus.warn(code, message) | DiagnosticStatus | Create a warning |
DiagnosticStatus.error(code, message) | DiagnosticStatus | Create an error |
DiagnosticStatus.fatal(code, message) | DiagnosticStatus | Create a fatal error |
Methods:
| Method | Returns | Description |
|---|
with_component(component) | DiagnosticStatus | Return a copy with component name set |
message_str() | str | Get message as a string |
component_str() | str | Get component as a string |
EmergencyStop
E-stop signal — safety-critical.
# simplified
estop = horus.EmergencyStop(engaged=True, reason="obstacle detected", timestamp_ns=horus.timestamp_ns())
| Field | Type | Default | Description |
|---|
engaged | bool | True | Whether the E-stop is engaged |
reason | str | "" | Reason for engagement |
auto_reset | bool | — | Whether auto-reset is enabled (getter only) |
timestamp_ns | int | 0 | Timestamp in nanoseconds |
Static Methods:
| Method | Returns | Description |
|---|
EmergencyStop.engage(reason) | EmergencyStop | Create an engaged E-stop with reason |
EmergencyStop.release() | EmergencyStop | Create a released E-stop |
Methods:
| Method | Returns | Description |
|---|
with_source(source) | EmergencyStop | Return a copy with source name set |
reason_str() | str | Get reason as a string |
SafetyStatus
Overall safety monitoring state. Created with no parameters — read state via getters.
# simplified
safety = horus.SafetyStatus()
# Read-only getters
safety.enabled # bool
safety.estop_engaged # bool
safety.watchdog_ok # bool
safety.limits_ok # bool
safety.comms_ok # bool
safety.mode # int
safety.fault_code # int
safety.timestamp_ns # int
| Getter | Type | Description |
|---|
enabled | bool | Safety system enabled |
estop_engaged | bool | Emergency stop is engaged |
watchdog_ok | bool | Watchdog is healthy |
limits_ok | bool | All limits within bounds |
comms_ok | bool | Communications healthy |
mode | int | Current safety mode |
fault_code | int | Active fault code |
timestamp_ns | int | Timestamp in nanoseconds |
| Method | Returns | Description |
|---|
is_safe() | bool | True if all safety checks pass |
set_fault(code) | None | Set a fault code |
clear_faults() | None | Clear all fault codes |
NodeHeartbeat
Per-node health heartbeat with state and health level.
# simplified
nhb = horus.NodeHeartbeat(state=1, health=0)
# state and health are u8 integer codes
| Field | Type | Default | Description |
|---|
state | int | 0 | Node state (u8) |
health | int | 0 | Node health level (u8) |
tick_count | int | — | Total tick count (getter only) |
target_rate_hz | int | — | Target tick rate (getter only) |
actual_rate_hz | int | — | Actual measured tick rate (getter only) |
error_count | int | — | Cumulative error count (getter only) |
last_tick_timestamp | int | — | Last tick timestamp in ns (getter only) |
heartbeat_timestamp | int | — | Heartbeat timestamp in ns (getter only) |
| Method | Returns | Description |
|---|
is_fresh(max_age_secs) | bool | True if heartbeat is within max_age_secs of now |
update_timestamp() | None | Update the heartbeat timestamp to now |
ResourceUsage
System resource monitoring.
# simplified
usage = horus.ResourceUsage(
cpu_percent=45.0,
memory_bytes=268435456,
memory_percent=12.5,
timestamp_ns=horus.timestamp_ns(),
)
| Field | Type | Default | Description |
|---|
cpu_percent | float | 0.0 | CPU usage percentage |
memory_bytes | int | 0 | Memory usage in bytes |
memory_percent | float | 0.0 | Memory usage percentage |
disk_bytes | int | 0 | Disk usage in bytes |
disk_percent | float | 0.0 | Disk usage percentage |
network_tx_bytes | int | 0 | Network bytes transmitted |
network_rx_bytes | int | 0 | Network bytes received |
temperature | float | 0.0 | System temperature (°C) |
thread_count | int | 0 | Active thread count |
timestamp_ns | int | 0 | Timestamp in nanoseconds |
| Method | Returns | Description |
|---|
is_cpu_high(threshold) | bool | True if CPU usage exceeds threshold |
is_memory_high(threshold) | bool | True if memory usage exceeds threshold |
is_temperature_high(threshold) | bool | True if temperature exceeds threshold |
DiagnosticReport
Aggregated diagnostic report. Values are added via helper methods, not constructor args.
# simplified
report = horus.DiagnosticReport(component="motors", level=0)
report.add_value("temperature", 45.2)
report.add_string("firmware", "v2.1.0")
| Field | Type | Default | Description |
|---|
component | str | "" | Component being reported on |
level | int | 0 | Overall diagnostic level |
| Method | Description |
|---|
add_value(key, value) | Add a numeric diagnostic value |
add_string(key, value) | Add a string diagnostic value |
DiagnosticValue
Single key-value diagnostic entry.
# simplified
dv = horus.DiagnosticValue(key="temperature", value="45.2")
| Field | Type | Default | Description |
|---|
key | str | "" | Diagnostic key |
value | str | "" | Diagnostic value (as string) |
value_type | int | — | Value type code (getter only) |
Static Methods:
| Method | Returns | Description |
|---|
DiagnosticValue.int(key, value) | DiagnosticValue | Create an integer diagnostic value |
DiagnosticValue.float(key, value) | DiagnosticValue | Create a float diagnostic value |
DiagnosticValue.boolean(key, value) | DiagnosticValue | Create a boolean diagnostic value |
See Also