Nodes and Lifecycle

pub trait Node: Send {
    // Required
    fn tick(&mut self);

    // Name (defaults to struct type name, e.g. `MotorController`)
    fn name(&self) -> &str { /* derived from type name */ }

    // Optional lifecycle
    fn init(&mut self) -> Result<()> { Ok(()) }
    // SAFETY: implement shutdown() for any node that controls actuators or holds hardware resources
    fn shutdown(&mut self) -> Result<()> { Ok(()) }
    fn on_error(&mut self, error: &str) { /* logs error */ }

    // Metadata (auto-generated by node! macro — rarely implemented manually)
    fn publishers(&self) -> Vec<TopicMetadata> { Vec::new() }
    fn subscribers(&self) -> Vec<TopicMetadata> { Vec::new() }

    // SAFETY: used by safety monitor when Miss::SafeMode triggers — override for actuator nodes
    fn is_safe_state(&self) -> bool { true }
    fn enter_safe_state(&mut self) { /* no-op */ }
}

fn tick(&mut self) {
    // Your node logic here
}

fn name(&self) -> &str {
    "MyNode"
}

fn init(&mut self) -> Result<()>

fn init(&mut self) -> Result<()> {
    self.sensor = Sensor::open("/dev/i2c-1")
        .horus_context("opening IMU sensor")?;
    hlog!(info, "Sensor initialized");
    Ok(())
}

// SAFETY: always implement shutdown() for nodes controlling motors, servos, or other actuators
fn shutdown(&mut self) -> Result<()> {
    hlog!(info, "Node shutting down");
    // Clean up resources, close connections, etc.
    Ok(())
}

// SAFETY: override for actuator nodes — scheduler calls this during Miss::SafeMode
fn is_safe_state(&self) -> bool {
    self.velocity == 0.0 && self.motor_disabled
}

// SAFETY: stop all actuators and disable outputs when entering safe state
fn enter_safe_state(&mut self) {
    self.velocity = 0.0;
    self.disable_motor();
    hlog!(warn, "Entered safe state");
}

fn on_error(&mut self, error: &str)

fn on_error(&mut self, error: &str) {
    hlog!(error, "Node error: {}", error);
    self.consecutive_errors += 1;
    if self.consecutive_errors > 5 {
        self.enter_safe_state();
    }
}

fn publishers(&self) -> Vec<TopicMetadata>
fn subscribers(&self) -> Vec<TopicMetadata>

fn publishers(&self) -> Vec<TopicMetadata> {
    vec![TopicMetadata {
        topic_name: "cmd_vel".to_string(),
        type_name: std::any::type_name::<CmdVel>().to_string(),
    }]
}

use horus::prelude::*;

struct LifecycleDemo {
    counter: u32,
}

impl Node for LifecycleDemo {
    fn name(&self) -> &str {
        "LifecycleDemo"
    }

    fn init(&mut self) -> Result<()> {
        // NOTE: called ONCE when node starts — open resources, allocate buffers here
        hlog!(info, "Initializing resources");
        self.counter = 0;
        Ok(())
    }

    fn tick(&mut self) {
        // NOTE: called REPEATEDLY in main loop (~60 FPS default)
        self.counter += 1;

        hlog!(debug, "Tick #{}", self.counter);
    }

    // SAFETY: called ONCE during graceful shutdown — release resources, stop actuators here
    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "Shutting down after {} ticks", self.counter);
        Ok(())
    }
}

hlog!(info, "Robot ready");

hlog!(warn, "Battery low");

hlog!(error, "Sensor disconnected");

hlog!(debug, "Position: ({}, {})", x, y);

fn tick(&mut self) {
    self.velocity_pub.send(1.5);

    // IMPORTANT: call recv() every tick to avoid stale data accumulation
    if let Some(scan) = self.lidar_sub.recv() {
        self.process(scan);
    }
}

pub struct NodeMetrics {
    pub name: String,
    pub order: u32,
    pub total_ticks: u64,
    pub successful_ticks: u64,
    pub failed_ticks: u64,
    pub avg_tick_duration_ms: f64,
    pub max_tick_duration_ms: f64,
    pub min_tick_duration_ms: f64,
    pub last_tick_duration_ms: f64,
    pub messages_sent: u64,
    pub messages_received: u64,
    pub errors_count: u64,
    pub warnings_count: u64,
    pub uptime_seconds: f64,
}

fn init(&mut self) -> Result<()> {
    hlog!(info, "Node initializing");
    Ok(())
}

fn tick(&mut self) {
    // Track state internally if needed
    self.tick_count += 1;
}

// Recommended priority constants
const CRITICAL: u32    = 0;   // Highest priority
const HIGH: u32        = 10;
const NORMAL: u32      = 50;  // Default
const LOW: u32         = 80;
const BACKGROUND: u32  = 100; // Lowest priority

use horus::prelude::*;

let mut scheduler = Scheduler::new();

// Safety monitor runs FIRST every tick (order 0)
scheduler.add(safety_node).order(0).build()?;

// Controller runs second (order 10)
scheduler.add(control_node).order(10).build()?;

// Sensors run third (order 50)
scheduler.add(sensor_node).order(50).build()?;

// Logging runs LAST (order 100)
scheduler.add(logger_node).order(100).build()?;

// Fine-grained priorities for complex systems
scheduler.add(emergency_stop).order(0).build()?;      // Highest
scheduler.add(motor_control).order(15).build()?;      // Between HIGH and NORMAL
scheduler.add(vision_processing).order(55).build()?;  // Slightly lower than normal
scheduler.add(telemetry).order(90).build()?;          // Between LOW and BACKGROUND

use horus::prelude::*;

struct SensorNode {
    data_pub: Topic<f32>,
}

impl SensorNode {
    fn new() -> Result<Self> {
        Ok(Self {
            data_pub: Topic::new("sensor_data")?,
        })
    }
}

impl Node for SensorNode {
    fn name(&self) -> &str {
        "SensorNode"
    }

    fn tick(&mut self) {
        let data = 42.0; // Read sensor
        self.data_pub.send(data);
    }

    // SAFETY: no actuators controlled — shutdown is optional for pure publisher nodes
    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "SensorNode stopped");
        Ok(())
    }
}

use horus::prelude::*;

node! {
    SensorNode {
        pub {
            sensor_data: f32 -> "sensor_data",
        }

        tick {
            let data = 42.0;
            self.sensor_data.send(data);
        }
    }
}

struct Publisher {
    data_pub: Topic<f32>,
}

impl Node for Publisher {
    fn tick(&mut self) {
        let data = self.generate_data();
        self.data_pub.send(data);
    }

    // SAFETY: implement shutdown() if this publisher controls actuators
    fn shutdown(&mut self) -> Result<()> { Ok(()) }
}

struct Subscriber {
    data_sub: Topic<f32>,
}

impl Node for Subscriber {
    fn tick(&mut self) {
        // IMPORTANT: call recv() every tick to drain the buffer and avoid stale data
        if let Some(data) = self.data_sub.recv() {
            self.process(data);
        }
    }

    fn shutdown(&mut self) -> Result<()> { Ok(()) }
}

// PATTERN: Pipeline — subscribe, transform, republish
struct Filter {
    input_sub: Topic<f32>,
    output_pub: Topic<f32>,
}

impl Node for Filter {
    fn tick(&mut self) {
        // IMPORTANT: call recv() every tick to drain the buffer and avoid stale data
        if let Some(input) = self.input_sub.recv() {
            let output = input * 2.0;
            self.output_pub.send(output);
        }
    }

    fn shutdown(&mut self) -> Result<()> { Ok(()) }
}

// PATTERN: Aggregator — combine multiple inputs into one output
// NOTE: if either input has no data, BOTH recv() calls still execute but output is skipped.
// This means data from the other input is consumed and lost. For independent draining,
// use the Multi-Topic Synchronization pattern below instead.
struct Aggregator {
    input_a: Topic<f32>,
    input_b: Topic<f32>,
    output_pub: Topic<f32>,
}

impl Node for Aggregator {
    fn tick(&mut self) {
        // IMPORTANT: both recv() calls run every tick — data is consumed even if the other is None
        if let (Some(a), Some(b)) = (self.input_a.recv(), self.input_b.recv()) {
            let result = a + b;
            self.output_pub.send(result);
        }
    }

    fn shutdown(&mut self) -> Result<()> { Ok(()) }
}

// IMPORTANT: keep tick() under 1ms — the scheduler monitors tick duration
fn tick(&mut self) {
    let result = self.compute_quickly();
    self.pub.send(result);
}

// WARNING: blocking I/O in tick() causes deadline misses — use .async_io() or .compute() instead
fn tick(&mut self) {
    let data = std::fs::read_to_string("file.txt").unwrap(); // Blocks!
    // ...
}

fn init(&mut self) -> Result<()> {
    hlog!(info, "Initializing MyMotorNode");

    // 1. Open hardware connections
    self.serial_port = serialport::new("/dev/ttyUSB0", 115200)
        .open()
        .map_err(|e| Error::node("MyMotorNode", format!("Failed to open serial: {}", e)))?;

    // IMPORTANT: pre-allocate buffers here — allocation in tick() causes jitter
    self.command_buffer = vec![0u8; 256];

    // 3. Initialize hardware state
    self.send_init_sequence()?;

    // SAFETY: start with actuators in a known safe state
    self.velocity = 0.0;
    self.is_armed = false;

    hlog!(info, "MyMotorNode initialized successfully");
    Ok(())
}

// SAFETY: shutdown() is called once on SIGINT/SIGTERM — stop all actuators before releasing resources
fn shutdown(&mut self) -> Result<()> {
    hlog!(info, "MyMotorNode shutting down");

    // CRITICAL: stop all actuators FIRST — before closing connections or releasing resources
    self.velocity = 0.0;
    self.send_stop_command();

    // SAFETY: disable hardware outputs to prevent runaway on power cycle
    self.disable_motor_driver();

    // 3. Close hardware connections
    if let Some(port) = self.serial_port.take() {
        drop(port);  // Closes the port
    }

    // 4. Save any important state
    self.save_position_to_file()?;

    hlog!(info, "MyMotorNode shutdown complete");
    Ok(())
}

use horus::prelude::*;

struct MyMotorController {
    // Hardware
    serial_port: Option<Box<dyn serialport::SerialPort>>,

    // Communication
    cmd_sub: Topic<MotorCommand>,
    status_pub: Topic<MotorStatus>,

    // State
    velocity: f64,
    position: i32,
    is_enabled: bool,
}

impl MyMotorController {
    fn new() -> Result<Self> {
        Ok(Self {
            serial_port: None,
            cmd_sub: Topic::new("motor.cmd")?,
            status_pub: Topic::new("motor.status")?,
            velocity: 0.0,
            position: 0,
            is_enabled: false,
        })
    }

    fn send_velocity(&mut self, vel: f64) {
        if let Some(ref mut port) = self.serial_port {
            let cmd = format!("V{}\n", vel);
            let _ = port.write(cmd.as_bytes());
        }
    }
}

impl Node for MyMotorController {
    fn name(&self) -> &str { "MyMotorController" }

    fn init(&mut self) -> Result<()> {
        hlog!(info, "Opening serial connection to motor controller");

        // Open hardware connection
        self.serial_port = Some(
            serialport::new("/dev/ttyUSB0", 115200)
                .timeout(std::time::Duration::from_millis(100))
                .open()
                .map_err(|e| Error::node("MyMotorController", format!("Serial open failed: {}", e)))?
        );

        // Initialize motor to stopped state
        self.send_velocity(0.0);
        self.is_enabled = true;

        hlog!(info, "Motor controller ready");
        Ok(())
    }

    fn tick(&mut self) {
        // IMPORTANT: call recv() every tick to drain the command buffer
        if let Some(cmd) = self.cmd_sub.recv() {
            self.velocity = cmd.velocity;
            self.send_velocity(self.velocity);
        }

        // Publish status
        let status = MotorStatus {
            velocity: self.velocity,
            position: self.position,
        };
        self.status_pub.send(status);
    }

    // SAFETY: must stop motors before releasing serial port
    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "Stopping motor for safe shutdown");

        // CRITICAL: stop motor first — before closing serial port
        self.velocity = 0.0;
        self.send_velocity(0.0);

        // Close serial port
        self.serial_port = None;
        self.is_enabled = false;

        hlog!(info, "Motor stopped safely");
        Ok(())
    }
}

fn init(&mut self) -> Result<()> {
    if !self.sensor.is_available() {
        return Err(Error::node("MyNode", "Sensor not found"));
    }
    Ok(())
}

fn tick(&mut self) {
    hlog!(info, "Processing data");
    hlog!(debug, "Detailed debug info: {:?}", self.state);
}

// IMPORTANT: store state in the node struct — the scheduler owns your node's lifetime
struct MyNode {
    counter: u32,  // Instance state
}

// WARNING: static mut is unsound and breaks multi-node isolation
static mut COUNTER: u32 = 0;  // Unsafe global state

fn init(&mut self) -> Result<()> {
    self.device = Device::open().map_err(|e| {
        Error::node("MyNode", format!("Failed to open device: {}", e))
    })?;

    hlog!(info, "Device opened successfully");
    Ok(())
}

fn tick(&mut self) {
    // IMPORTANT: call recv() every tick — None is normal (no new data this tick)
    match self.data_sub.recv() {
        Some(data) => self.process(data),
        None => {
            // No data available - this is normal
        }
    }
}

// SAFETY: never panic in shutdown — log errors and continue cleanup
fn shutdown(&mut self) -> Result<()> {
    if let Err(e) = self.device.close() {
        hlog!(warn, "Failed to close device: {}", e);
        // Continue shutdown anyway
    }
    Ok(())
}

fn tick(&mut self) {
    self.tick_count += 1;

    // Execute every 10 ticks
    if self.tick_count % 10 == 0 {
        self.slow_operation();
    }
}

// PATTERN: State machine — use enum states for complex behavior
enum RobotState {
    Idle,
    Moving,
    Stopped,
}

struct RobotController {
    state: RobotState,
    cmd_sub: Topic<CmdVel>,
    last_cmd: Option<CmdVel>,
}

impl Node for RobotController {
    fn tick(&mut self) {
        // CRITICAL: always call recv() outside the match — calling it only in one branch
        // causes the buffer to fill up in other states, leading to stale commands
        // executing immediately on state transition
        self.last_cmd = self.cmd_sub.recv();

        match self.state {
            RobotState::Idle => {
                if self.last_cmd.is_some() {
                    self.state = RobotState::Moving;
                }
            }
            RobotState::Moving => {
                // Execute movement
                if self.is_done() {
                    self.state = RobotState::Stopped;
                }
            }
            RobotState::Stopped => {
                self.state = RobotState::Idle;
            }
        }
    }

    // SAFETY: stop actuators on shutdown even if in Moving state
    fn shutdown(&mut self) -> Result<()> {
        self.state = RobotState::Stopped;
        Ok(())
    }
}

// PATTERN: Multi-Topic Synchronization — cache latest from each input, process when all available
struct Synchronizer {
    topic_a: Topic<f32>,
    topic_b: Topic<f32>,
    last_a: Option<f32>,
    last_b: Option<f32>,
}

impl Node for Synchronizer {
    fn tick(&mut self) {
        // IMPORTANT: call recv() on ALL topics every tick — never skip a recv() conditionally
        if let Some(a) = self.topic_a.recv() {
            self.last_a = Some(a);
        }
        if let Some(b) = self.topic_b.recv() {
            self.last_b = Some(b);
        }

        // NOTE: processes with stale data from slower topic until both update
        if let (Some(a), Some(b)) = (self.last_a, self.last_b) {
            self.process(a, b);
        }
    }

    fn shutdown(&mut self) -> Result<()> { Ok(()) }
}

// SAFETY: without this, motors continue at last commanded velocity after Ctrl+C
fn shutdown(&mut self) -> Result<()> {
    hlog!(info, "Stopping all motors for safe shutdown");

    // CRITICAL: send stop command to all motors FIRST
    self.emergency_stop();

    // SAFETY: disable motor drivers to prevent runaway
    self.disable_all_drivers();

    hlog!(info, "Motors stopped safely");
    Ok(())
}

import horus

velocity = [0.0, 0.0]

def motor_tick(node):
    msg = node.recv("cmd_vel")
    if msg is not None:
        velocity[0], velocity[1] = msg

def motor_shutdown(node):
    velocity[0], velocity[1] = 0.0, 0.0
    print("Motors stopped safely")

motor = horus.Node(name="MotorController", tick=motor_tick,
                   shutdown=motor_shutdown, order=10,
                   subs=["cmd_vel"])
horus.run(motor)

impl Node for MyMotorController {
    fn name(&self) -> &str { "MyMotorController" }

    fn tick(&mut self) {
        // Normal operation - motors running
        self.motor_pub.send(self.velocity);
    }

    // SAFETY: called on SIGINT/SIGTERM — stop actuators before releasing hardware
    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "MyMotorController shutting down");

        // CRITICAL: stop all motors FIRST — before closing any connections
        self.velocity = 0.0;
        self.motor_pub.send(0.0);

        // Close hardware connections
        if let Some(port) = self.serial_port.take() {
            port.close();
        }

        hlog!(info, "All motors stopped safely");
        Ok(())
    }
}

# Start your application
horus run

# In another terminal, send SIGINT
kill -SIGINT <pid>

# Or simply press Ctrl+C in the application terminal

// SAFETY: never panic in shutdown — always attempt all cleanup steps
fn shutdown(&mut self) -> Result<()> {
    // CRITICAL: always try to stop motors, even if other cleanup fails
    if let Err(e) = self.stop_motors() {
        hlog!(error, "Failed to stop motors: {}", e);
        // Continue with other cleanup anyway
    }

    // Close connections (non-critical)
    if let Err(e) = self.close_connection() {
        hlog!(warn, "Failed to close connection: {}", e);
    }

    Ok(())
}