Multi-Node Monitoring Example

This example demonstrates monitoring multiple network nodes and implementing basic control logic using ai_nn_controller. The framework supports any type of network node — optical, wireless, RAN, core network, and more.

Complete Code

"""
multi_node_app.py - Multi-node monitoring with control

Monitors multiple network nodes, sends corrections when needed.
This example uses optical network nodes, but the same pattern applies
to any network domain (wireless, RAN, core, etc.).
"""

import time
import commands  # Register commands

from ai_nn_controller.decorators.aic_app import aic_app
from ai_nn_controller.AicApp import AicApp
from ai_nn_controller.AicController import AicController

@aic_app(name="MultiNodeController")
class MultiNodeController(AicApp):
    """
    Controls multiple amplifiers and ROADMs in a network segment.
    """

    aic_app_id = 101
    control_loop_update_time = 3

    # Monitor multiple nodes
    read_measurements = {
        3: [  # Amp1
            "amp1_target_gain",
            "amp1_gain_tilt",
        ],
        5: [  # Amp2
            "amp2_target_gain",
            "amp2_gain_tilt",
        ],
        8: [  # ROADM3
            "roadm3_preamp_gain",
            "roadm3_booster_gain",
        ],
    }

    # Control functions
    control_functions = {
        3: ["SET_GAIN"],
        5: ["SET_GAIN"],
        8: ["SET_GAIN", "SET_VOA"],
    }

    # cell_ids and send_commands are auto-generated by @aic_app
    # cell_ids will be: [3, 5, 8]

    # Configuration
    TARGET_GAIN = 20.0
    TOLERANCE = 2.0

    # State tracking
    last_correction = {}

    @classmethod
    def process(cls, measurements):
        """Monitor nodes and apply corrections."""
        if not measurements:
            print("[MultiNode] Waiting for measurements...")
            return

        current_time = time.time()

        # Process each node
        for node_id, data_list in measurements.items():
            if not data_list:
                continue

            latest = data_list[-1]
            cls._process_node(node_id, latest, current_time)

    @classmethod
    def _process_node(cls, node_id, data, current_time):
        """Process measurements from a single node."""
        # Get the appropriate gain metric
        gain_key = cls._get_gain_key(node_id)
        if gain_key not in data:
            return

        current_gain = data[gain_key]
        gain_error = current_gain - cls.TARGET_GAIN

        print(f"[Node {node_id}] {gain_key}: {current_gain:.2f} dB (error: {gain_error:+.2f})")

        # Check if correction needed
        if abs(gain_error) > cls.TOLERANCE:
            # Rate limit corrections (once per 10 seconds per node)
            last_time = cls.last_correction.get(node_id, 0)
            if current_time - last_time > 10:
                cls._send_correction(node_id, current_gain)
                cls.last_correction[node_id] = current_time

    @classmethod
    def _get_gain_key(cls, node_id):
        """Get the gain metric key for a node."""
        mapping = {
            3: "amp1_target_gain",
            5: "amp2_target_gain",
            8: "roadm3_preamp_gain",
        }
        return mapping.get(node_id, "unknown")

    @classmethod
    def _send_correction(cls, node_id, current_gain):
        """Send a gain correction command."""
        print(f"  [CORRECTION] Node {node_id}: {current_gain:.2f} -> {cls.TARGET_GAIN:.2f}")

        # Determine amp_type based on node
        amp_type = "preamp" if node_id == 8 else "line"

        cls.add_command((
            "SET_GAIN",
            {
                "node_id": node_id,
                "value": {
                    "amp_type": amp_type,
                    "target_gain": cls.TARGET_GAIN
                }
            }
        ))


if __name__ == "__main__":
    AicController(with_api=True, verbose=True).run()

Commands Module

Create commands.py in the same directory:

"""commands.py - Command definitions for multi-node controller."""

import json
from ai_nn_controller.registry import register_command

def set_gain_handler(node_id: int, value: dict) -> str:
    return json.dumps({
        "command": "SET_GAIN",
        "amp_type": value.get("amp_type", "line"),
        "target_gain": value.get("target_gain", 0)
    })

SET_GAIN_SCHEMA = {
    "description": "Set amplifier target gain",
    "properties": {
        "node_id": {"type": "integer"},
        "amp_type": {"type": "string", "enum": ["line", "preamp", "booster"]},
        "target_gain": {"type": "number"}
    },
    "required": ["node_id", "target_gain"]
}

def set_voa_handler(node_id: int, value: dict) -> str:
    return json.dumps({
        "command": "SET_VOA",
        "attenuation": value.get("attenuation", 0)
    })

SET_VOA_SCHEMA = {
    "description": "Set variable optical attenuator",
    "properties": {
        "node_id": {"type": "integer"},
        "attenuation": {"type": "number"}
    },
    "required": ["node_id", "attenuation"]
}

register_command("SET_GAIN", set_gain_handler, SET_GAIN_SCHEMA)
register_command("SET_VOA", set_voa_handler, SET_VOA_SCHEMA)

Running

# Start infrastructure
docker compose up -d

# Run the application
python multi_node_app.py --verbose

# Start via API
curl -X PUT http://localhost:8000/apps/MultiNodeController/state \
  -H "Content-Type: application/json" \
  -d '{"state": "running"}'

Output Example

[Node 3] amp1_target_gain: 23.50 dB (error: +3.50)
  [CORRECTION] Node 3: 23.50 -> 20.00
[Node 5] amp2_target_gain: 19.80 dB (error: -0.20)
[Node 8] roadm3_preamp_gain: 25.00 dB (error: +5.00)
  [CORRECTION] Node 8: 25.00 -> 20.00

Key Patterns

1. Multiple node subscriptions: Monitor several nodes simultaneously

2. State tracking: Use class attributes to track correction history

3. Rate limiting: Prevent command flooding with time-based throttling

4. Helper methods: Organize code with private methods

5. Node-specific logic: Handle different nodes differently