Image prediction

The image prediction example shows a remote application with trainer, predictor, and end-user services. It combines service implementations, custom metrics, and a degradation policy in emulation mode.

Use it when you want to understand:

• how to attach concrete service implementations to an application,

• how to configure a remote simulation,

• how to collect custom metrics from remote service behaviour.

The full code lives in the examples/image_prediction directory.

Run it from the repository root with:

uv run image-prediction

Application

Application code

from .services import (
    EndService,
    PredictorService,
    TrainerService,
)

from eclypse.graph import Application
from eclypse.utils.constants import MAX_LATENCY

image_app = Application("ImagePrediction", include_default_assets=True)

image_app.add_service(
    EndService("EndService"),
    cpu=1,
    gpu=0,
    ram=0.5,
    storage=0.5,
    availability=0.9,
    processing_time=5,
)

image_app.add_service(
    PredictorService("PredictorService"),
    cpu=1,
    gpu=1,
    ram=16.0,
    storage=64.0,
    availability=0.9,
)

image_app.add_service(
    TrainerService("TrainerService"),
    cpu=16,
    gpu=4,
    ram=16.0,
    storage=2.0,
    availability=0.9,
)


image_app.add_edge(
    "EndService",
    "PredictorService",
    latency=MAX_LATENCY,
    bandwidth=20.0,
    symmetric=True,
)

image_app.add_edge(
    "PredictorService",
    "TrainerService",
    latency=MAX_LATENCY,
    bandwidth=100.0,
    symmetric=True,
)

Simulation

Simulation code

from .application import image_app as app
from .metrics import get_metrics
from .services.utils import (
    BASE_PATH,
    STEP_EVERY_MS,
    STEPS,
)
from .update_policy import DegradePolicy

from eclypse.builders.infrastructure import get_star
from eclypse.placement.strategies import RandomStrategy
from eclypse.simulation import (
    Simulation,
    SimulationConfig,
)


def main() -> None:
    """Run the Image Prediction example."""
    seed = 2

    sim_config = SimulationConfig(
        seed=seed,
        max_steps=STEPS,
        step_every_ms=STEP_EVERY_MS,
        include_default_metrics=False,
        events=get_metrics(),
        log_to_file=True,
        path=BASE_PATH,
        # Pass a RemoteBootstrap instance here to request specific Ray resources.
        remote=True,
    )

    sim = Simulation(
        get_star(
            infrastructure_id="IPInfr",
            n_clients=5,
            seed=seed,
            update_policies=DegradePolicy(epochs=STEPS),
            include_default_assets=True,
            resource_init="max",
            symmetric=True,
        ),
        simulation_config=sim_config,
    )
    strategy = RandomStrategy(spread=True)

    sim.register(app, strategy)
    sim.run()


if __name__ == "__main__":
    main()

Metrics

Metrics code

import os

import psutil

from eclypse.report.metrics import metric
from eclypse.report.metrics.defaults import featured_latency
from eclypse.utils.constants import DRIVING_EVENT


@metric.service(name="img_counter", remote=True)
def get_img_counter(self):
    if self.id == "EndService":
        if self.img_counter == 0:
            self.logger.warning("No images were processed")
            return 0
        return self.img_counter
    return None


@metric.service(name="accuracy", remote=True)
def get_acc(self):
    if self.id == "EndService" and self.img_counter != 0:
        acc = self.correct / self.img_counter
        self.correct = 0
        self.img_counter = 0
        return acc
    return None


@metric.service(name="model_change", remote=True)
def is_changed(self):
    if self.id == "PredictorService":
        if self.new_model_signal:
            self.new_model_signal = False
            return 1
    return None


@metric.simulation(name="cpu_usage", activates_on=[DRIVING_EVENT, "stop"])
class CPUMonitor:

    def __init__(self):
        self.process = None

    def __call__(self, event):
        if self.process is None:
            self.process = psutil.Process(os.getpid())
        return self.process.cpu_percent(interval=0.1)


@metric.service(
    name="remote_cpu_usage",
    activates_on=[DRIVING_EVENT, "stop"],
    remote=True,
)
class RemoteCPUMonitor:

    def __init__(self):
        self.process = None

    def __call__(self, event):
        if self.process is None:
            self.process = psutil.Process(os.getpid())
        return self.process.cpu_percent(interval=0.1)


@metric.simulation(
    name="memory_usage",
    activates_on=[DRIVING_EVENT, "stop"],
)
class MemoryMonitor:

    def __init__(self):
        self.process = None

    def __call__(self, event):
        if self.process is None:
            self.process = psutil.Process(os.getpid())
        memory_usage = self.process.memory_info().rss
        return memory_usage / (1024 * 1024)  # Convert to MB


@metric.service(
    name="remote_memory_usage",
    activates_on=[DRIVING_EVENT, "stop"],
    remote=True,
)
class RemoteMemoryMonitor:

    def __init__(self):
        self.process = None

    def __call__(self, event):
        if self.process is None:
            self.process = psutil.Process(os.getpid())
        memory_usage = self.process.memory_info().rss
        return memory_usage / (1024 * 1024)


def get_metrics():
    return [
        get_img_counter,
        get_acc,
        is_changed,
        featured_latency,
        CPUMonitor(),
        MemoryMonitor(),
        RemoteCPUMonitor(),
        RemoteMemoryMonitor(),
    ]

What to inspect

This example is useful when the application behaviour matters, not only the placement. Inspect the custom metrics together with the service implementations to see how emulated runtime state becomes reportable simulation data.

Related concepts:

• Emulation

• Reporting