### Full Stream Processing Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/06_stream.md

A comprehensive example demonstrating stream creation, response queue setup, parameter configuration, frame generation, and simulated processing loop.

```python
from aiko_services.main import (
    Stream, StreamState, StreamEvent, Frame, DEFAULT_STREAM_ID
)
import queue
import time

def process_stream_example():
    # Create stream
    stream = Stream(stream_id="data_stream", graph_path="PE_1")
    
    # Setup response queue
    response_queue = queue.Queue()
    stream.queue_response = response_queue
    stream.topic_response = "AIKO/responses"
    
    # Add parameters
    stream.parameters["batch_size"] = 10
    stream.parameters["timeout_seconds"] = 30
    
    # Generate frames
    for i in range(5):
        frame = Frame()
        frame.swag["index"] = i
        frame.swag["data"] = [i, i+1, i+2]
        frame.metrics["created_at"] = time.time()
        stream.frames[i] = frame
    
    # Simulate processing
    while stream.state != StreamState.STOP:
        if len(stream.frames) == 0:
            stream.set_state(StreamState.STOP)
            break
        
        # Process next frame
        frame_id = stream.frame_id
        frame = stream.frames.pop(frame_id, None)
        
        if frame:
            print(f"Processing: {frame.swag}")
            frame.metrics["processed_at"] = time.time()
            stream.frame_id += 1
        else:
            stream.set_state(StreamState.NO_FRAME)
    
    print(f"Stream {stream.stream_id} complete")
    print(f"Total frames processed: {stream.frame_id}")

process_stream_example()
```

--------------------------------

### Start mosquitto, aiko_registrar, and aiko_dashboard on Linux/Mac

Source: https://github.com/geekscape/aiko_services/blob/master/ReadMe.md

This script starts the mosquitto MQTT server, aiko_registrar, and aiko_dashboard. It assumes the default AIKO_MQTT_HOST is set to localhost. Ensure mosquitto is installed and running.

```bash
./scripts/system_start.sh  # default AIKO_MQTT_HOST=localhost
```

--------------------------------

### Install Mosquitto MQTT Broker

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Install the Mosquitto MQTT broker on macOS or Ubuntu/Debian. This is a one-time setup.

```bash
# Install mosquitto (one-time)
brew install mosquitto    # macOS
sudo apt install mosquitto # Ubuntu/Debian
```

--------------------------------

### Navigate to HyperSpace Example Directory

Source: https://github.com/geekscape/aiko_services/blob/master/src/aiko_services/examples/hyperspace/ReadMe.md

Change the current working directory to the HyperSpace example directory to begin.

```bash
$ cd src/aiko_services/examples/hyperspace
```

--------------------------------

### Install Aiko Services from PyPI

Source: https://github.com/geekscape/aiko_services/blob/master/ReadMe.md

Use this command to install the Aiko Services package from the Python Package Index. This is recommended for users who want to quickly use existing examples and tools.

```bash
pip install aiko_services
```

--------------------------------

### Start Mosquitto MQTT Broker

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Run this command to start the Mosquitto MQTT broker service.

```bash
mosquitto
```

--------------------------------

### Run AlohaHonua Actor Locally

Source: https://github.com/geekscape/aiko_services/blob/master/src/aiko_services/examples/aloha_honua/ReadMe.md

After starting the core services, navigate to the aloha_honua example directory and run the Python script. This will start the AlohaHonua Actor and connect it to the MQTT server.

```bash
# Use another terminal session, also starting at the top-level of the repository
cd examples/aloha_honua
./aloha_honua_0.py
    MQTT topic: aiko/nomad.local/123/1/in
```

--------------------------------

### Start Aiko Services and Core Services

Source: https://github.com/geekscape/aiko_services/blob/master/src/aiko_services/examples/aloha_honua/ReadMe.md

This command starts the local MQTT server (mosquitto), Aiko Registrar, and Aiko Dashboard. Ensure your current working directory is the top-level of the Aiko Services repository before running.

```bash
# Current working directory should be the top-level of the Aiko Services repository
# The first command starts mosquitto, Aiko Registrar and Aiko Dashboard
./scripts/system_start.sh
    Starting: /usr/sbin/mosquitto
    Starting: aiko_registrar
```

--------------------------------

### ServiceFilter Examples

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

Examples demonstrating how to create ServiceFilter instances for various discovery scenarios, including finding all services, specific services on localhost, and services with specific tags.

```python
# Find all services
filter_all = ServiceFilter("*", "*", "*", "*", "*", "*")

# Find pipelines on localhost
filter = ServiceFilter(
    topic_paths="AIKO/localhost/+/+",
    protocol="*/pipeline:*",
    owner="*"
)

# Find production actors with GPU
filter = ServiceFilter(
    protocol="*/actor:*",
    tags=["env=prod", "gpu=true"]
)
```

--------------------------------

### Complete Actor Example: Calculator

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/04_actor.md

A full implementation of an actor, demonstrating interface definition, implementation, and direct/delayed message invocation. This example shows how to set up and run a service actor.

```python
from abc import abstractmethod
from aiko_services.main import (
    Actor, ActorImpl, actor_args, compose_instance,
    aiko, process_create, SERVICE_PROTOCOL_AIKO, event
)

# Define actor interface
class Calculator(Actor):
    Interface.default("Calculator", "__main__.CalculatorImpl")
    
    @abstractmethod
    def add(self, a: int, b: int) -> None:
        pass
    
    @abstractmethod
    def multiply(self, a: int, b: int) -> None:
        pass

# Implement actor
class CalculatorImpl(Calculator):
    def __init__(self, context):
        context.call_init(self, "Actor", context)
        self.last_result = None
    
    def add(self, a: int, b: int):
        self.last_result = a + b
        print(f"{a} + {b} = {self.last_result}")
    
    def multiply(self, a: int, b: int):
        self.last_result = a * b
        print(f"{a} * {b} = {self.last_result}")

# Setup and run
protocol = f"{SERVICE_PROTOCOL_AIKO}/calculator:0"
init_args = actor_args("calculator", protocol=protocol)
actor = compose_instance(CalculatorImpl, init_args)

# Use actor directly (synchronous)
actor.add(5, 3)
actor.multiply(4, 7)

# Post delayed message
actor._post_message(
    actor._actor_mailbox_name(actor.ActorTopic.IN),
    "add",
    (10, 20),
    delay=1.0  # Execute after 1 second
)

# Run
aiko.process.run()
```

--------------------------------

### Initialize Service and Add to Process

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Example of initializing a custom service and registering it with the aiko process.

```python
class MyService(Service):
    def initialize(self):
        print("Service initialized")
        aiko.process.add_service(self)

service = MyService(name="my_service")
service.initialize()
```

--------------------------------

### ServiceFields Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

An example of creating a ServiceFields instance with sample metadata.

```python
fields = ServiceFields(
    topic_path="AIKO/server/12345/1",
    name="image_processor",
    protocol=ServiceProtocol(...),
    transport="mqtt",
    owner="alice",
    tags=["env=production", "gpu=true"]
)
```

--------------------------------

### Actor Run Method Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/04_actor.md

Shows how to instantiate and start an actor's event loop. The 'run' method is blocking and continues until 'terminate' is called.

```python
actor = MyActorImpl(context)
actor.run()  # Runs until terminate() called
```

--------------------------------

### Create and Run a Simple Service

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Example of creating a custom service, composing its instance, and registering it with the Aiko process.

```python
from aiko_services.main import Service, service_args, compose_instance

class HelloService(Service):
    def __init__(self, context):
        context.call_init(self, "Service", context)

    def greet(self, name: str):
        print(f"Hello, {name}!")

# Create instance
protocol = f"{SERVICE_PROTOCOL_AIKO}/hello:0"
init_args = service_args("hello_service", protocol)
service = compose_instance(HelloService, init_args)

# Register with process
aiko.process.add_service(service)
aiko.process.run()
```

--------------------------------

### PipelineDefinition Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

An example of how to structure a PipelineDefinition dictionary. This shows typical values for version, name, runtime, graph, and parameters.

```python
definition = {
    "version": 0,
    "name": "video_processing",
    "runtime": "python",
    "graph": ["(VideoReader Decoder Filter Encoder Writer)"],
    "parameters": {
        "timeout_ms": 1000,
        "max_queue": 10
    },
    "elements": [...]  # PipelineElementDefinition list
}
```

--------------------------------

### Prepare Aiko Services for PyPI Release

Source: https://github.com/geekscape/aiko_services/blob/master/ReadMe.md

After installing from GitHub, install Hatch and use it to manage dependencies and build the package for release to PyPI. This is for package maintainers.

```bash
pip install -U hatch  # Install latest Hatch build and package manager
hatch shell           # Run shell using Hatch to manage dependencies
# hatch test          # Run local tests (to be completed)
hatch build           # Publish Aiko Services package to PyPI
```

--------------------------------

### Complete Actor Usage Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Demonstrates how to define, implement, and compose an actor using the Aiko Services framework. This example shows actor definition with an interface, implementation with a constructor and methods, and instantiation using a context object.

```python
from abc import abstractmethod
from aiko_services.main import \
    Actor, ActorImpl, actor_args, compose_instance, \
    aiko, process_create, SERVICE_PROTOCOL_AIKO

# Define actor with interface
class Counter(Actor):
    Interface.default("Counter", "__main__.CounterImpl")
    
    @abstractmethod
    def increment(self, amount: int = 1) → None:
        pass

# Implement actor
class CounterImpl(Counter):
    def __init__(self, context):
        context.call_init(self, "Actor", context)
        self.count = 0
    
    def increment(self, amount: int = 1):
        self.count += amount
        print(f"Count: {self.count}")

# Create using context
protocol = f"{SERVICE_PROTOCOL_AIKO}/counter:0"
context = actor_args(
    name="counter_1",
    protocol=protocol,
    tags=["test=true"],
    parameters={"initial": "0"}
)

actor = compose_instance(CounterImpl, context)

# Use
actor.increment(5)
actor.increment(3)

aiko.process.run()
```

--------------------------------

### Frame Usage Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/06_stream.md

Demonstrates how to instantiate a Frame object and populate its swag and metrics fields.

```python
import time

frame = Frame()
frame.swag["image"] = image_data
frame.swag["timestamp"] = time.time()
frame.metrics["processing_time_ms"] = 42.5
```

--------------------------------

### Complete Event Loop Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/03_event.md

This example demonstrates setting up and running the Aiko Services event loop with timer, mailbox, and queue handlers. It includes a self-shutdown mechanism after a specified duration.

```python
from aiko_services.main import event, process_create, aiko
import time

# Setup
aiko.process = process_create()

# Timer: runs every 5 seconds
def check_status():
    print(f"Status check at {time.time()}")

event.add_timer_handler(check_status, 5.0, immediate=True)

# Mailbox: high-priority messages
def process_command(cmd, timestamp):
    print(f"Command: {cmd} at {timestamp}")

event.add_mailbox_handler(process_command, "command")

# Queue: general items
def handle_item(item, item_type):
    print(f"Item ({item_type}): {item}")
    return False  # Continue to other handlers

event.add_queue_handler(handle_item, ["message", "sensor"])

# Send items
event.mailbox_put("command", {"action": "start"})
event.queue_put({"temp": 25.5}, "sensor")

# Shutdown after 20 seconds
def auto_shutdown():
    print("Timeout - shutting down")
    event.terminate()

event.add_timer_handler(auto_shutdown, 20.0)

# Run
event.loop()
```

--------------------------------

### Frame Usage Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

Demonstrates how to instantiate and populate a Frame object with data, performance metrics, and processing status.

```python
frame = Frame()

# Data passed between elements
frame.swag["image"] = image_array
frame.swag["timestamp"] = time.time()

# Performance tracking
frame.metrics["read_time_ms"] = 5.3
frame.metrics["process_time_ms"] = 12.7

# Remote processing
frame.paused_pe_name = "RemoteInference"
```

--------------------------------

### Complete Pipeline Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/08_pipeline.md

Demonstrates the creation and execution of a simple pipeline. Includes defining custom elements, setting up the pipeline graph and elements, and processing data through the pipeline.

```python
from aiko_services.main import (
    PipelineElement, PipelineElementImpl, PipelineImpl,
    pipeline_element_args, compose_instance,
    Stream, Frame, StreamEvent,
    aiko, process_create
)
import numpy as np

# 1. Define element: Read data source
class DataReader(PipelineElement):
    Interface.default("DataReader", "__main__.DataReaderImpl")

class DataReaderImpl(PipelineElementImpl):
    def __init__(self, context):
        super().__init__(context)
        self.data = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
        self.index = 0
    
    def process_frame(self, stream, frame):
        if self.index >= len(self.data):
            return StreamEvent.STOP
        
        frame.swag["input"] = self.data[self.index]
        self.index += 1
        return StreamEvent.OKAY

# 2. Define element: Process data
class DataProcessor(PipelineElement):
    Interface.default("DataProcessor", "__main__.DataProcessorImpl")

class DataProcessorImpl(PipelineElementImpl):
    def process_frame(self, stream, frame):
        input_data = frame.swag.get("input")
        if input_data:
            output = [x * 2 for x in input_data]
            frame.swag["output"] = output
        return StreamEvent.OKAY

# 3. Create pipeline definition
definition = {
    "version": 0,
    "name": "simple_pipeline",
    "runtime": "python",
    "graph": ["(DataReader DataProcessor)"],
    "elements": [
        {
            "name": "DataReader",
            "input": {},
            "output": {"input": "list"},
            "deploy": {"type": "local", "module": "__main__"}
        },
        {
            "name": "DataProcessor",
            "input": {"input": "list"},
            "output": {"output": "list"},
            "deploy": {"type": "local", "module": "__main__"}
        }
    ]
}

# 4. Create and run pipeline
pipeline_context = pipeline_args("simple_pipeline")
pipeline = PipelineImpl(pipeline_context, definition)

# Process data
stream = Stream(stream_id="test_stream")
for i in range(3):
    frame = Frame()
    pipeline.create_frame(stream, frame)
    pipeline.process_frame(stream, frame)
    print(f"Result: {frame.swag.get('output')}")

aiko.process.run()
```

--------------------------------

### PipelineElementDefinition Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

An example of a PipelineElementDefinition dictionary, illustrating input/output ports, parameters, and deployment configuration for an image filter element.

```python
element_def = {
    "name": "ImageFilter",
    "input": {
        "image": "numpy.ndarray",
        "threshold": "float"
    },
    "output": {
        "filtered": "numpy.ndarray"
    },
    "parameters": {
        "kernel_size": 3,
        "iterations": 2
    },
    "deploy": {
        "type": "local",
        "module": "aiko_services.elements.media.image_io"
    }
}
```

--------------------------------

### Discover Services and Start Proxies

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/README.md

Performs service discovery using a filter and defines a handler for found services. The `on_found` function starts the discovered service proxy.

```python
from aiko_services.main import do_discovery, ServiceFilter

def on_found(details, proxy):
    proxy.start()  # Call remote method

filter = ServiceFilter(name="processor")
do_discovery(MyInterface, filter, discovery_add_handler=on_found)
```

--------------------------------

### Service Topic Path Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/07_discovery.md

Services are discovered at their MQTT topic path. When creating a proxy, use the full topic path.

```text
AIKO/hostname/pid/sid/in
```

--------------------------------

### Install Aiko Services from GitHub for Development

Source: https://github.com/geekscape/aiko_services/blob/master/ReadMe.md

Clone the Aiko Services repository from GitHub and set up a virtual environment for development. This method is recommended when contributing to or extending Aiko Services.

```bash
git clone https://github.com/geekscape/aiko_services.git
cd aiko_services
python3 -m venv venv      # Once only
source venv/bin/activate  # Each terminal session
pip install -U pip        # Install latest pip
pip install -e .          # Install Aiko Services for development
```

--------------------------------

### Create and Run a Custom Service

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/01_process.md

Example demonstrating how to initialize the Aiko process, define a custom service, register it, and run the main event loop. This is a common pattern for service development within the Aiko framework.

```python
from aiko_services.main import aiko, process_create, Service

class MyService(Service):
    def __init__(self):
        self.name = "my_service"
        self.protocol = "github.com/geekscape/aiko_services/protocol/myservice:0"
        self.transport = "mqtt"

# Initialize process
aiko.process = process_create()
aiko.process.initialize()

# Create and register service
service = MyService()
service_id = aiko.process.add_service(service)

# Run main event loop
aiko.process.run()
```

--------------------------------

### Setting up and Running the Event Loop

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/03_event.md

Configure and start the main event loop, which processes timer events, mailbox messages, queue items, and flatout handlers. The loop blocks execution.

```python
from aiko_services.main import event

event.add_timer_handler(my_timer, 1.0)
event.add_flatout_handler(my_flatout)
event.add_queue_handler(my_queue_handler)

event.loop()  # Blocks until terminate() called
```

--------------------------------

### Run Aiko Application

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Run your Aiko application after starting the MQTT broker. Ensure mosquitto is running before executing the Python application.

```bash
# Start MQTT broker first
mosquitto &

# Run your application
python app.py
```

--------------------------------

### Get Service Parameters

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Retrieves the configuration parameters for a service. Ensure the context object is initialized.

```python
params = context.get_parameters()
# Returns: {"key1": "value1", ...}
```

--------------------------------

### ActorTopic Usage Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/04_actor.md

Demonstrates how to construct a topic path using ActorTopic.IN for subscribing to application input messages.

```python
from aiko_services.main import ActorTopic

topic_path = f"{actor.topic_path}/{ActorTopic.IN}"
aiko.message.subscribe(topic_path)
```

--------------------------------

### Create and Run a Process

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Initializes and starts the main Aiko process. The `run()` method blocks execution until the process is terminated.

```python
aiko.process = process_create()
aiko.process.initialize()
aiko.process.run()  # Blocks until terminate()
```

--------------------------------

### Event Timer Handler and Loop

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Set up a timer to trigger a callback function at a specified interval and start the event loop.

```python
from aiko_services.main import event
event.add_timer_handler(callback, 1.0)
event.loop()
```

--------------------------------

### Complete Aiko Service Discovery Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/07_discovery.md

This Python script demonstrates a full cycle of service discovery and interaction within the Aiko framework. It defines a 'Printer' actor, implements it, composes an instance, and then uses `do_discovery` to find and communicate with the printer service.

```python
from aiko_services.main import (
    aiko, process_create, Actor, ActorImpl,
    actor_args, compose_instance,
    do_discovery, ServiceFilter,
    SERVICE_PROTOCOL_AIKO
)

# Define actor interface
class Printer(Actor):
    """Interface for a remote printer service"""
    Interface.default("Printer", "__main__.PrinterImpl")
    
    def print_document(self, text: str) → None:
        pass
    
    def get_status(self) → dict:
        pass

# Implement printer actor
class PrinterImpl(ActorImpl):
    def __init__(self, context):
        super().__init__(context)
        self.queue = []
    
    def print_document(self, text: str):
        self.queue.append(text)
        print(f"Printing: {text}")
    
    def get_status(self) → dict:
        return {"queue_size": len(self.queue)}

# Setup printer service
protocol = f"{SERVICE_PROTOCOL_AIKO}/printer:0"
context = actor_args(
    name="office_printer",
    protocol=protocol,
    tags=["location=floor2"]
)
printer = compose_instance(PrinterImpl, context)

# Client: discover and use printer
def discover_printers():
    def on_printer_found(service_details, proxy):
        topic, name, proto, transport, owner, tags = service_details
        print(f"Found printer: {name} at {topic}")
        
        # Use printer via proxy
        proxy.print_document("Hello from discovery!")
        status = proxy.get_status()
        print(f"Printer status: {status}")
    
    def on_printer_lost(service_details):
        print(f"Lost printer: {service_details[1]}")
    
    # Discover all printer services
    filter = ServiceFilter.with_topic_path(
        protocol=f"{SERVICE_PROTOCOL_AIKO}/printer:*"
    )
    
    do_discovery(
        Printer,
        filter,
        discovery_add_handler=on_printer_found,
        discovery_remove_handler=on_printer_lost
    )

# Main program
if __name__ == "__main__":
    aiko.process = process_create()
    aiko.process.initialize()
    
    # Start printer service
    printer.initialize()
    
    # Start discovery in separate task
    discover_printers()
    
    # Run event loop
    aiko.process.run()

```

--------------------------------

### Get Service Tags as String

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Retrieves the service's tags formatted as a space-separated string.

```python
service = Service(tags=["env=prod", "version=1"])
tags_str = service.get_tags_string()
# Returns: "env=prod version=1"
```

--------------------------------

### Basic Aiko Service Application

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/README.md

A minimal example demonstrating the creation and execution of a basic actor within the Aiko Services Framework. This includes setting up the process, defining an actor, and running the application.

```python
from aiko_services.main import aiko, process_create, Actor, ActorImpl, actor_args, compose_instance

# Create process
aiko.process = process_create()

# Create actor
class MyActor(Actor):
    Interface.default("MyActor", "__main__.MyActorImpl")

class MyActorImpl(ActorImpl):
    def __init__(self, context):
        super().__init__(context)
    
    def process(self, data):
        print(f"Processing: {data}")

context = actor_args("my_actor")
actor = compose_instance(MyActorImpl, context)

# Run
aiko.process.run()
```

--------------------------------

### Implement PipelineElement Frame Creation

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/08_pipeline.md

Example implementation for creating and initializing a new frame within a PipelineElement. It populates frame data and adds it to the stream.

```python
def create_frame(self, stream, frame_data):
    frame = Frame()
    frame.swag["data"] = frame_data
    frame.metrics["created_at"] = time.time()
    stream.frames[stream.frame_id] = frame
    return StreamEvent.OKAY
```

--------------------------------

### Register Timer Handler for Resource Monitoring

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/03_event.md

Example of registering a timer handler to periodically check system resources like CPU and memory usage.

```python
def check_resources():
    import psutil
    cpu = psutil.cpu_percent()
    memory = psutil.virtual_memory().percent
    print(f"CPU: {cpu}%, Memory: {memory}%")

event.add_timer_handler(check_resources, 10.0)
```

--------------------------------

### Complete Data Processing Service Example

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/INTEGRATION_GUIDE.md

This Python script demonstrates a full integration of an Aiko data processing service. It includes defining the service interface, implementing the service logic, creating a client interface, and setting up the main program to register, discover, and manage the service lifecycle.

```python
from abc import abstractmethod
from aiko_services.main import \
    aiko, process_create, Actor, ActorImpl, \
    actor_args, compose_instance, SERVICE_PROTOCOL_AIKO, \
    event, do_discovery, ServiceFilter, Stream, Frame, StreamEvent

# ============================================================================ 
# 1. DEFINE SERVICE INTERFACE
# ============================================================================ 

class DataProcessor(Actor):
    Interface.default("DataProcessor", "__main__.DataProcessorImpl")
    
    @abstractmethod
    def process_batch(self, batch_id: str) 	-> None:
        pass

# ============================================================================ 
# 2. IMPLEMENT SERVICE
# ============================================================================ 

class DataProcessorImpl(ActorImpl):
    def __init__(self, context):
        super().__init__(context)
        self.results = {}
    
    def process_batch(self, batch_id: str):
        # Simulate processing
        self.results[batch_id] = f"Processed {batch_id}"
        print(f"Batch {batch_id} complete")
    
    def get_results(self, batch_id: str):
        return self.results.get(batch_id, "Not found")

# ============================================================================ 
# 3. DEFINE CLIENT INTERFACE
# ============================================================================ 

class ProcessorInterface:
    def process_batch(self, batch_id: str) 	-> None:
        pass
    
    def get_results(self, batch_id: str) 	-> str:
        pass

# ============================================================================ 
# 4. MAIN PROGRAM
# ============================================================================ 

def main():
    # Create and initialize process
    aiko.process = process_create()
    aiko.process.initialize()
    
    # Create processor service
    protocol = f"{SERVICE_PROTOCOL_AIKO}/processor:0"
    context = actor_args("processor", protocol=protocol)
    processor = compose_instance(DataProcessorImpl, context)
    
    # Register service with process
    processor.initialize()
    
    # Add timer: generate work every 3 seconds
    counter = [0]
    def generate_work():
        counter[0] += 1
        batch_id = f"batch_{counter[0]}"
        processor.process_batch(batch_id)
    
    event.add_timer_handler(generate_work, 3.0, immediate=True)
    
    # Add timer: auto-shutdown after 30 seconds
    def auto_shutdown():
        print("Shutting down...")
        event.terminate()
    
    event.add_timer_handler(auto_shutdown, 30.0)
    
    # Optional: discover and call another processor
    def on_peer_found(details, proxy):
        print(f"Found peer processor at: {details[0]}")
        # Could call: proxy.process_batch("external_batch")
    
    filter = ServiceFilter.with_topic_path(
        name="processor",
        protocol=f"{SERVICE_PROTOCOL_AIKO}/processor:*"
    )
    do_discovery(ProcessorInterface, filter, discovery_add_handler=on_peer_found)
    
    # Run event loop (blocking)
    aiko.process.run()

if __name__ == "__main__":
    main()

```

--------------------------------

### Get Process and Service Topic Paths

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Extracts both the process and service-level MQTT topic paths from a full topic path string.

```python
process_path, service_path = ServiceTopicPath.topic_paths("AIKO/localhost/1234/1")
# Returns: ("AIKO/localhost/1234", "AIKO/localhost/1234/1")
```

--------------------------------

### Actor Set Log Level Method Examples

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/04_actor.md

Demonstrates dynamically changing the logging level of an actor during runtime. Accepts standard log level strings.

```python
actor.set_log_level("DEBUG")
actor.set_log_level("INFO")
```

--------------------------------

### Example Pipeline Definition JSON

Source: https://github.com/geekscape/aiko_services/wiki/PipelineDefinition

This JSON defines a simple pipeline named 'p_example' using Python runtime. It includes a graph structure, parameters, and definitions for two pipeline elements: PE_RandomIntegers and PE_Add.

```json
{
  "version": 0,
  "name":    "p_example",
  "runtime": "python",

  "graph": ["(PE_RandomIntegers PE_Add)"],

  "parameters": { "constant": 1},

  "elements": [
    { "name":   "PE_RandomIntegers",
      "input":  [{ "name": "random", "type": "int" }],
      "output": [{ "name": "random", "type": "int" }],
      "deploy": { "local": { "module": "aiko_services.examples.pipeline.elements" } }
    },
    { "name":   "PE_Add",
      "input":  [{ "name": "i", "type": "int" }],
      "output": [{ "name": "i", "type": "int" }],
      "deploy": { "local": { "module": "aiko_services.examples.pipeline.elements" } }
    }
  ]
}
```

--------------------------------

### initialize

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/01_process.md

Initializes the process, optionally requiring an MQTT connection. This method sets up the necessary components for the process to run.

```APIDOC
## initialize(mqtt_connection_required: bool = True)

### Description
Initialize the process with MQTT connection and event handlers.

### Method
`initialize`

### Parameters
#### Path Parameters
- **mqtt_connection_required** (bool) - Optional - Raise SystemExit if MQTT connection fails. Defaults to True.

### Raises
`SystemExit` if MQTT connection required but fails.
```

--------------------------------

### Get MQTT Configuration

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the complete MQTT configuration, including host, port, and other relevant settings. This provides a consolidated view of MQTT connection parameters.

```python
from aiko_services.main.utilities import get_mqtt_configuration

config = get_mqtt_configuration()
# Returns: {"host": "localhost", "port": 1883, ...}
```

--------------------------------

### Define and Implement a Simple Service

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/README.md

Defines a service interface and its implementation, then registers and runs it. Use this pattern for basic service creation.

```python
# 1. Define service interface
class Calculator(Service):
    Interface.default("Calculator", "__main__.CalculatorImpl")

# 2. Implement service
class CalculatorImpl(Service):
    def add(self, a, b):
        return a + b

# 3. Register and run
context = service_args("calc")
service = compose_instance(CalculatorImpl, context)
aiko.process.add_service(service)
aiko.process.run()
```

--------------------------------

### local_iso_now()

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Get the current time as a local ISO 8601 formatted string.

```APIDOC
## local_iso_now()

### Description
Get current time as local ISO 8601 string.

### Method
`local_iso_now()`

### Returns
- `str`: The current time in local ISO 8601 format.

### Example
```python
from aiko_services.main.utilities import local_iso_now

iso_time = local_iso_now()
```
```

--------------------------------

### datetime_now_utc_iso()

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Get the current time as a UTC ISO 8601 formatted string.

```APIDOC
## datetime_now_utc_iso()

### Description
Get current time as UTC ISO 8601 string.

### Method
`datetime_now_utc_iso()`

### Returns
- `str`: The current time in UTC ISO 8601 format (e.g., "2026-06-22T15:30:45.123456Z")

### Example
```python
from aiko_services.main.utilities import datetime_now_utc_iso

iso_time = datetime_now_utc_iso()
# Returns: "2026-06-22T15:30:45.123456Z"
```
```

--------------------------------

### Initialize Process

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/01_process.md

Initializes the process, optionally requiring an MQTT connection. Raises SystemExit if MQTT connection fails and is required.

```python
aiko.process.initialize()
```

```python
aiko.process.initialize(mqtt_connection_required=False)  # Optional MQTT
```

--------------------------------

### datetime_epoch()

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Get the current time as a Unix epoch timestamp (seconds since 1970-01-01).

```APIDOC
## datetime_epoch()

### Description
Get Unix epoch (seconds since 1970-01-01).

### Method
`datetime_epoch()`

### Returns
- `int`: The current Unix epoch timestamp in seconds.

### Example
```python
from aiko_services.main.utilities import datetime_epoch

epoch = datetime_epoch()
# Returns: 1719078645
```
```

--------------------------------

### Get Single Tag Value

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Extracts the value of a specific tag from a list of tags.

```python
tags = ["env=prod", "version=1.0", "region=us-east"]
env_value = ServiceTags.get_tag_value("env", tags)
# Returns: "prod"
```

--------------------------------

### Create and Run Process Instance

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/01_process.md

Create or retrieve the singleton process instance and then run it. This is the entry point for Aiko services.

```python
from aiko_services.main import process_create

process = process_create()
process.run()
```

--------------------------------

### Get Service Transport

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Retrieves the transport mechanism configured for a service. This specifies how the service communicates.

```python
transport = context.get_transport()
# Returns: "mqtt"
```

--------------------------------

### Check Connection Status

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

Example of how to check if a process is connected to the Registrar service using the ConnectionState enum.

```python
from aiko_services.main import aiko, ConnectionState

if aiko.process.connection.is_connected(ConnectionState.REGISTRAR):
    print("Connected to Registrar")
```

--------------------------------

### Get Username

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the username of the current user running the process. This can be helpful for logging or access control.

```python
from aiko_services.main.utilities import get_username

user = get_username()
# Returns: "alice"
```

--------------------------------

### Service Discovery Workflow

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/INTEGRATION_GUIDE.md

Illustrates the flow of service discovery, from process calls to registrar updates and proxy creation. This is useful for understanding how services find and interact with each other in a dynamic environment.

```text
┌─────────────────────────┐
│ Process A               │
│ (calls do_discovery)    │
└──────────┬──────────────┘
           │
           ▼
┌─────────────────────────────────────────┐
│ ServiceDiscovery(Process A)             │
│ Listens for service changes            │
│ Maintains services_cache               │
└──────────┬──────────────────────────────┘
           │
           ▼
┌─────────────────────────────────────────┐
│ Registrar (MQTT Network)                │
│ Central service registry                │
└──────────┬──────────────────────────────┘
           │
           ▼
┌─────────────────────────┐
│ Process B               │
│ (service matches filter)│
│ (publishes to registrar)│
└─────────────────────────┘

When Process B's service matches filter in Process A:
  1. Registrar publishes service details (add/remove)
  2. ServiceDiscovery.add_handler() called
  3. discovery_add_handler(details, proxy) invoked
  4. Proxy auto-created with methods from interface
  5. User can immediately call: proxy.method_name(args)
```

--------------------------------

### Get AIKO Namespace

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the AIKO namespace, which defaults to 'AIKO'. This is used for organizing AIKO-related topics.

```python
from aiko_services.main.utilities import get_namespace

namespace = get_namespace()
# Returns: "AIKO"
```

--------------------------------

### Create Service Instance using Context

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/INTEGRATION_GUIDE.md

Creates a service instance using the Context pattern for configuration. This pattern reduces constructor arguments and supports lazy initialization.

```python
from aiko_services.main import actor_args, compose_instance

# Create configuration via context
context = actor_args(
    name="processor",
    protocol="github.com/geekscape/aiko_services/protocol/actor:0",
    tags=["type=processor", "version=1.0"]
)

# Create instance (deferred initialization)
actor = compose_instance(MyActorImpl, context)
```

--------------------------------

### Get Service by ID

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Shows how to retrieve a service from the container using its unique ID and access its properties.

```python
service = services.get(1)
if service:
    print(service.name)
```

--------------------------------

### Create and Populate a Stream

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/06_stream.md

Demonstrates how to create a Stream instance and add Frame objects with custom data to its frames.

```python
from aiko_services.main import Stream, StreamState, Frame

# Create stream with ID
stream = Stream(stream_id="my_stream")

# Add frames
frame1 = Frame()
frame1.swag["data"] = [1, 2, 3]
stream.frames[0] = frame1

frame2 = Frame()
frame2.swag["data"] = [4, 5, 6]
stream.frames[1] = frame2
```

--------------------------------

### Run a Service

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Execute your service script using Python.

```bash
python your_service.py
```

--------------------------------

### Get MQTT Port

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the MQTT broker port. It checks the AIKO_MQTT_PORT environment variable or defaults to 1883.

```python
from aiko_services.main.utilities import get_mqtt_port

port = get_mqtt_port()  # Default: 1883
```

--------------------------------

### Get MQTT Host

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the MQTT broker hostname. It checks the AIKO_MQTT_HOST environment variable or defaults to 'localhost'.

```python
from aiko_services.main.utilities import get_mqtt_host

host = get_mqtt_host()  # Default: localhost
```

--------------------------------

### ServiceImpl Base Implementation

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Default implementation for services, providing automatic MQTT subscription and message handling.

```python
class ServiceImpl(Service):
    def __init__(self, context: Context)
```

--------------------------------

### Get System Hostname

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the system's hostname. Useful for identifying the specific machine a service is running on.

```python
from aiko_services.main.utilities import get_hostname

hostname = get_hostname()
# Returns: "my-server" or similar
```

--------------------------------

### Get Service Tags

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Retrieves the metadata tags associated with a service. Tags are useful for filtering or categorizing services.

```python
tags = context.get_tags()
# Returns: ["env=prod", "version=1.0"]
```

--------------------------------

### Initialize HyperSpace

Source: https://github.com/geekscape/aiko_services/blob/master/src/aiko_services/examples/hyperspace/ReadMe.md

Initialize the HyperSpace local file-system storage. This command sets up the necessary directory structure.

```bash
$ aiko_hyperspace initialize
```

--------------------------------

### Process Initialization

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/INTEGRATION_GUIDE.md

Initialize the Aiko process. Set `mqtt_connection_required` to `False` to continue without MQTT if needed.

```python
aiko.process.initialize(mqtt_connection_required=False)
```

--------------------------------

### Get Service Protocol

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Retrieves the protocol identifier for a service. The returned value indicates the communication protocol used.

```python
protocol = context.get_protocol()
# Returns: "github.com/geekscape/aiko_services/protocol/actor:0"
```

--------------------------------

### Get All Registered Implementations

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Retrieve a dictionary of all registered implementations from the context. This is useful for iterating through available components or for debugging.

```python
impls = context.get_implementations()
for name, impl in impls.items():
    print(f"{name}: {impl}")
```

--------------------------------

### Configuration Parameter Resolution

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/INTEGRATION_GUIDE.md

Demonstrates how configuration parameters are resolved, prioritizing environment variables over defaults defined in code. Use this to understand how to set and access parameters within service elements.

```python
# Highest priority: environment variable
# THRESHOLD=0.8 python app.py

# Or: defaults in definition
element_def = {
    "parameters": {"threshold": 0.5}
}

# Accessed via:
threshold = element.get_parameter("threshold", default=0.5)
# Returns 0.8 (from env), 0.5 (from definition), or 0.5 (default)
```

--------------------------------

### Get Stream State as Dictionary

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/06_stream.md

Serializes the current stream state into a dictionary format, useful for storage or transmission.

```python
stream_dict = stream.as_dict()
# Returns:
# {
#     "stream_id": "stream_1",
#     "frame_id": 5,
#     "graph_path": "PE_1"
# }
```

--------------------------------

### Process Creation and Execution

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Create and run the main Aiko process. This is the entry point for most Aiko applications.

```python
from aiko_services.main import aiko, process_create
aiko.process = process_create()
aiko.process.run()
```

--------------------------------

### Configure Stream Responses

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/06_stream.md

Illustrates how to configure a stream to send responses via a queue or an MQTT topic. Both methods can be used concurrently.

```python
import queue

# Response via queue
response_queue = queue.Queue()
stream.queue_response = response_queue

# Response via MQTT topic
stream.topic_response = "AIKO/localhost/1234/1/response"

# Both can be used simultaneously
```

--------------------------------

### Get System Uptime

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Obtains the system's uptime in seconds. This function is useful for monitoring system availability and performance.

```python
from aiko_services.main.utilities import get_uptime

uptime = get_uptime()
# Returns: 123456.78 (seconds)
```

--------------------------------

### ServiceProtocol Constructor

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Initializes a ServiceProtocol object with a URL prefix, service name, and version.

```python
class ServiceProtocol:
    def __init__(self, url_prefix: str, name: str, version: int)
```

--------------------------------

### Get Process ID

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the current process ID (PID). This is useful for uniquely identifying running instances of a service.

```python
from aiko_services.main.utilities import get_pid

pid = get_pid()
# Returns: 12345
```

--------------------------------

### Services Container Methods

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/02_service.md

Defines the interface for the Services container, including methods for adding, getting, removing, and iterating over services.

```python
class Services:
    def add(self, service: Service) → int
    def get(self, service_id: int) → Service | None
    def remove(self, service_id: int) → bool
    def __iter__() → Iterator[Service]
```

--------------------------------

### High-Level Service Discovery with Callbacks

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/07_discovery.md

Use do_discovery to automatically create service proxies and handle service additions/removals via callbacks. Requires defining an interface and a ServiceFilter.

```python
from aiko_services.main import do_discovery, ServiceFilter

class MyActorInterface:
    def start(self):
        pass
    def stop(self):
        pass
    def get_status(self):
        pass

def on_actor_added(service_details, proxy):
    print(f"Actor found: {service_details[1]}")
    proxy.start()  # Call method on remote actor

def on_actor_removed(service_details):
    print(f"Actor lost: {service_details[1]}")

filter = ServiceFilter(name="my_actor")
discovery, handler = do_discovery(
    MyActorInterface,
    filter,
    discovery_add_handler=on_actor_added,
    discovery_remove_handler=on_actor_removed
)
```

--------------------------------

### Context.call_init

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Initializes a component using its registered implementation within the context. It handles checking initialization status, retrieving the implementation, and invoking its constructor.

```APIDOC
## Context.call_init

### Description
Initialize a component using its registered implementation.

### Method
`call_init(caller: Any, implementation_name: str, context: Context, **kwargs) -> None`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **caller** (Any) - Object being initialized (self)
- **implementation_name** (str) - Name of implementation to load (e.g., "Service", "Actor")
- **context** (Context) - Context instance
- **kwargs** (dict) - Additional keyword arguments to pass to implementation

### Request Example
```python
class MyActor(Actor):
    def __init__(self, context):
        context.call_init(self, "Actor", context)
        # ... actor-specific initialization
```

### Response
#### Success Response (None)
None

#### Response Example
None

### Error Handling
None
```

--------------------------------

### loop

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/03_event.md

Starts and runs the event loop, which continuously processes timer events, mailbox messages, queue items, and flatout handlers.

```APIDOC
## loop(loop_when_no_handlers: bool = False) -> None

### Description
Run the event loop in a blocking manner. This function continuously executes registered handlers based on timers, mailboxes, queues, and flatout calls. It will block until `terminate()` is called or an exception occurs.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **loop_when_no_handlers** (bool) - Optional - Default: `False` - If True, the loop will continue running even if no handlers are currently registered.

### Behavior:
- Polls for timer events and executes when ready.
- Processes all queued mailbox items.
- Invokes queue handlers for items.
- Calls flatout handlers (approximately 1000 Hz max).
- Sleeps for 0.001 seconds between iterations.

### Exit Conditions:
- `terminate()` is called.
- An exception occurs within a handler.

### Request Example
```python
from aiko_services.main import event

event.add_timer_handler(my_timer, 1.0)
event.add_flatout_handler(my_flatout)
event.add_queue_handler(my_queue_handler)

event.loop()  # Blocks until terminate() called
```

### Response
None
```

--------------------------------

### Create a Simple Aiko Service

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/00_START_HERE.md

Create a basic Aiko service with a 'hello' method. This Python code defines a service and adds it to the Aiko process.

```python
from aiko_services.main import *

aiko.process = process_create()

class MyService(Service):
    def hello(self):
        print("Hello from Aiko!")

service = compose_instance(MyService, service_args("my_service"))
aiko.process.add_service(service)
aiko.process.run()
```

--------------------------------

### Validate Pipeline Connectivity

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/08_pipeline.md

Validate the input and output connections between elements in the pipeline graph. Specify the starting element name for validation.

```python
graph.validate(pipeline_definition, "ImageReader")
```

--------------------------------

### Get Process Memory Usage

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/09_utilities.md

Retrieves the current memory usage of the process in bytes. This function is useful for monitoring resource consumption.

```python
from aiko_services.main.utilities import get_memory_used

mem = get_memory_used()
# Returns: 52428800 (bytes)
```

--------------------------------

### Stream Usage Pattern

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/10_types.md

Illustrates a common pattern for initializing a Stream, setting parameters, and safely adding frames to its queue.

```python
stream = Stream(stream_id="dataset_1")
stream.parameters["timeout"] = 30
stream.topic_response = "AIKO/results/in"

# Thread-safe modification
with stream.lock:
    frame = Frame()
    stream.frames[stream.frame_id] = frame
    stream.frame_id += 1
```

--------------------------------

### Get All Registered Interface Implementations

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/05_context.md

Retrieve a dictionary of all implementations registered for an interface. This allows inspection of available implementations for a given interface.

```python
impls = Actor.get_implementations()
```

--------------------------------

### Context APIs

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/README.md

APIs for the configuration framework and factories.

```APIDOC
## Context APIs

### Description
APIs for the configuration framework and factories.

### Types
- **Context**
- **Interface**
- **ContextService**
- **ContextPipelineElement**
- **ContextPipeline**

### Functions
- **service_args()**
- **actor_args()**
- **pipeline_element_args()**
- **pipeline_args()**
```

--------------------------------

### Discover Services with Filters and Handlers

Source: https://github.com/geekscape/aiko_services/blob/master/_autodocs/README.md

Sets up service discovery to monitor for services matching specific criteria and defines a handler for service changes. Use for dynamic service management.

```python
discovery = ServiceDiscovery(aiko.process)
filter = ServiceFilter.with_topic_path(protocol="*/actor:*")

def on_service_change(command, details):
    if command == "add":
        print(f"Found: {details[1]}")

discovery.add_handler(on_service_change, filter)
```