Basic Application Example

This example shows a simple control application that monitors a single node using ai_nn_controller.

Complete Code

"""
basic_app.py - A simple monitoring application

This app monitors node 3 (an amplifier) and prints its measurements.
"""

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="BasicMonitor")
class BasicMonitor(AicApp):
    """
    Simple application that monitors an amplifier.
    """

    # Application identifier
    aic_app_id = 100

    # How often to process measurements (seconds)
    control_loop_update_time = 2

    # Measurements to subscribe to
    read_measurements = {
        3: [  # Node 3 (Amplifier)
            "amp1_target_gain",
            "amp1_gain_tilt",
            "amp1_target_power",
            "amp1_control_mode"
        ]
    }

    # No control functions - this is monitoring only
    control_functions = {}

    # Note: cell_ids and send_commands are auto-generated by @aic_app decorator

    @classmethod
    def process(cls, measurements):
        """
        Process incoming measurements.

        This method is called every control_loop_update_time seconds.
        """
        print("\n" + "=" * 50)
        print("[BasicMonitor] Processing measurements")
        print("=" * 50)

        if not measurements:
            print("  No measurements received yet")
            return

        # Get measurements from node 3
        node_3_data = measurements.get(3, [])

        if not node_3_data:
            print("  No data from node 3")
            return

        # Get the latest measurement (last in list)
        latest = node_3_data[-1]

        # Print all received metrics
        print(f"  Received {len(node_3_data)} measurement(s)")
        print(f"  Latest values:")
        print(f"    Target Gain:  {latest.get('amp1_target_gain', 'N/A')} dB")
        print(f"    Gain Tilt:    {latest.get('amp1_gain_tilt', 'N/A')} dB")
        print(f"    Target Power: {latest.get('amp1_target_power', 'N/A')} dBm")
        print(f"    Control Mode: {latest.get('amp1_control_mode', 'N/A')}")


if __name__ == "__main__":
    print("Starting BasicMonitor application...")
    print("API will be available at http://localhost:8000")
    print("Swagger docs at http://localhost:8000/docs")
    print()

    # Start the controller with API
    AicController(
        with_api=True,
        api_port=8000,
        verbose=True
    ).run()

Running the Example

1. Make sure the infrastructure is running:

   docker compose up -d redis aic_register node_msg_broker amp1_node
   

2. Run the application:

   python basic_app.py --verbose
   

3. Start the app via API:

   curl -X PUT http://localhost:8000/apps/BasicMonitor/state \
     -H "Content-Type: application/json" \
     -d '{"state": "running"}'
   

4. View measurements:

   curl http://localhost:8000/apps/BasicMonitor/measurements
   

Expected Output

Starting BasicMonitor application...
API will be available at http://localhost:8000
Swagger docs at http://localhost:8000/docs

==================================================
[BasicMonitor] Processing measurements
==================================================
  Received 3 measurement(s)
  Latest values:
    Target Gain:  20.5 dB
    Gain Tilt:    1.2 dB
    Target Power: -10.0 dBm
    Control Mode: auto

Key Concepts Demonstrated

1. @aic_app decorator: Registers the app with the framework

2. read_measurements: Declares what data to receive from which nodes

3. cell_ids: Auto-generated from read_measurements and control_functions keys

4. control_loop_update_time: Controls processing frequency

5. process() method: Receives measurements as a dictionary

Next Steps

• Add control functions to send commands

• Monitor multiple nodes

• Implement control logic based on measurements