CovidContagionNetworkVisual

This example builds upon covid_contagion_network by adding a Visualizer. It demonstrates how to use MelodieStudio to interactively run the simulation and view real-time animations of the network topology and data charts.

Visualizer: Project Structure

The structure is similar to the network example but adds a visualizer component:

examples/covid_contagion_network_visual
├── core/
│   ├── agent.py
│   ├── environment.py
│   ├── model.py
│   ├── scenario.py
│   ├── data_collector.py
│   ├── visualizer.py       # Defines charts and network styling
│   └── ...
├── data/
│   └── ...
├── run_studio.py           # Launches MelodieStudio
└── run_simulator.py        # Runs headless simulation

Visualizer: Key Changes

1. Visualizer Class: A new class CovidVisualizer inherits from Melodie.Visualizer. It defines:

   • Charts: A line chart tracking the number of Susceptible, Infected, and Recovered agents.

   • Network View: A visual representation of the network where agents appear as colored nodes (Blue=Susceptible, Red=Infected, Gray=Recovered) connected by edges.

2. MelodieStudio: Instead of running a batch simulation immediately, the run_studio.py script starts a local web server (MelodieStudio). You can control the simulation (Start, Pause, Reset) from the browser.

Visualizer: Running the Model

To run the model with the visualizer, execute the main script:

python examples/covid_contagion_network_visual/run_studio.py

Then, open your browser and navigate to http://localhost:8089. This is the web gateway for MelodieStudio. The backend simulation service will automatically run on 127.0.0.1:8765.

• In the web interface, go to the Simulator page.

• The interactive parameter controls will appear on the left. You can adjust them and click Reset to apply the new values.

• Click Start to begin the simulation.

• The network animation will be displayed in the center, and the line chart showing health state trends will be on the right.

Visualizer: Customization Guide

Here’s how you can customize the visualizer components:

1. Adding Data Charts (e.g., Line Chart)

• In core/visualizer.py, use self.plot_charts.add_line_chart("unique_chart_name") to create a new chart.

• Chain it with .set_data_source({...}) to bind data series. The values of the dictionary should be functions that return the numerical data for each time step (e.g., self.model.environment.num_susceptible).

• You can add multiple charts (line, pie, bar) as long as each has a unique name.

2. Configuring the Network View

• Use self.add_network(...) to link the visualizer to your model’s Network object.

• The var_style dictionary is key: it maps the integer values returned by var_getter to a specific color and a label for the legend.

• The legend for the network is automatically generated based on these labels.

• The network layout is automatically computed using NetworkX’s spring layout algorithm.

3. Adding an Interactive Parameter Panel

• Use self.params_manager.add_param(...) to register interactive parameters. The example adds three FloatParam instances for:

  • initial_infected_percentage

  • infection_prob

  • recovery_prob

• Each parameter requires a getter to read its current value from the Scenario and a setter to write the new value from the web UI back to the Scenario.

• After changing a parameter in the UI, click the Reset button to restart the simulation with the new settings.

Visualizer: Code

This section shows the key code additions for the visualizer. Core logic files are the same as the covid_contagion_network example.

Visualizer Definition

This file is the core addition. It maps model data to visual elements. Defined in core/visualizer.py.

from Melodie import Visualizer
from .model import CovidModel
from MelodieInfra.lowcode.params import FloatParam


class CovidVisualizer(Visualizer):
    model: CovidModel

    def setup(self):
        # Add interactive parameters to the left control panel
        self.params_manager.add_param(
            FloatParam(
                name="initial_infected_percentage",
                value_range=(0.0, 1.0),
                step=0.01,
                getter=lambda scenario: scenario.initial_infected_percentage,
                setter=lambda scenario, val: setattr(
                    scenario, "initial_infected_percentage", val
                ),
                label="Initial infected %",
                description="Share of agents initially infected (0-1).",
            )
        )
        self.params_manager.add_param(
            FloatParam(
                name="infection_prob",
                value_range=(0.0, 1.0),
                step=0.01,
                getter=lambda scenario: scenario.infection_prob,
                setter=lambda scenario, val: setattr(
                    scenario, "infection_prob", val
                ),
                label="Infection prob",
                description="Probability an infected agent infects a neighbor each step.",
            )
        )
        self.params_manager.add_param(
            FloatParam(
                name="recovery_prob",
                value_range=(0.0, 1.0),
                step=0.01,
                getter=lambda scenario: scenario.recovery_prob,
                setter=lambda scenario, val: setattr(
                    scenario, "recovery_prob", val
                ),
                label="Recovery prob",
                description="Probability an infected agent recovers each step.",
            )
        )

        # Configure the chart to show population health states over time
        self.plot_charts.add_line_chart("health_state_trend") \
            .set_data_source({
                "Susceptible": lambda: self.model.environment.num_susceptible,
                "Infected": lambda: self.model.environment.num_infected,
                "Recovered": lambda: self.model.environment.num_recovered
            })

        # Configure the network visualization
        self.add_network(
            name="covid_network",
            network_getter=lambda: self.model.network,
            var_getter=lambda agent: agent.health_state,
            var_style={
                0: {"label": "Susceptible", "color": "#0000FF"},
                1: {"label": "Infected", "color": "#FF0000"},
                2: {"label": "Recovered", "color": "#808080"}
            },
        )

Model Structure

Same as the network example. Defined in core/model.py.

from typing import TYPE_CHECKING

from Melodie import Model

from .agent import CovidAgent
from .data_collector import CovidDataCollector
from .environment import CovidEnvironment
from .scenario import CovidScenario

if TYPE_CHECKING:
    from Melodie import AgentList


class CovidModel(Model):
    """
    Network contagion model with visualization support.
    Demonstrates the use of `Melodie.Network` for agent interaction,
    combined with the Visualizer module.
    """

    scenario: "CovidScenario"
    agents: "AgentList[CovidAgent]"
    environment: CovidEnvironment
    data_collector: CovidDataCollector

    def create(self) -> None:
        """
        Create model components: Agents, Environment, DataCollector, and Network.
        """
        self.agents = self.create_agent_list(CovidAgent)
        self.environment = self.create_environment(CovidEnvironment)
        self.data_collector = self.create_data_collector(CovidDataCollector)
        
        # Create an empty network structure
        self.network = self.create_network()

    def setup(self) -> None:
        """
        Setup model components and build the network.
        """
        # 1. Initialize agents
        self.agents.setup_agents(self.scenario.agent_num)

        # 2. Build network connections based on scenario parameters
        # This uses NetworkX generators under the hood.
        self.network.setup_agent_connections(
            agent_lists=[self.agents],
            network_type=self.scenario.network_type,
            network_params=self.scenario.get_network_params(),
        )

        # 3. Seed initial infections
        self.environment.seed_infection(self.agents)
        
        # 4. Initial population stats for visualization
        self.environment.update_population_stats(self.agents)

    def run(self) -> None:
        """
        Main simulation loop.
        """
        for t in self.iterator(self.scenario.period_num):
            self.environment.spread_on_network(self.agents)
            self.environment.recover(self.agents)
            self.environment.update_population_stats(self.agents)
            self.data_collector.collect(t)
        self.data_collector.save()

Environment Logic

Same as the network example. Defined in core/environment.py.

from Melodie import Environment
from typing import TYPE_CHECKING
import random

if TYPE_CHECKING:
    from Melodie import AgentList
    from .agent import CovidAgent
    from .scenario import CovidScenario


class CovidEnvironment(Environment):
    """
    Environment handles infection/recovery logic on the network structure.
    """

    scenario: "CovidScenario"

    def setup(self):
        """
        Initialize macro-level statistical counters.
        """
        self.num_susceptible = 0
        self.num_infected = 0
        self.num_recovered = 0

    def seed_infection(self, agents: "AgentList[CovidAgent]"):
        """
        Initial seeding of infection among agents.
        """
        for agent in agents:
            agent.seed_infection(self.scenario.initial_infected_percentage)

    def spread_on_network(self, agents: "AgentList[CovidAgent]"):
        """
        Network-based infection logic:
        1. Iterate through all infected agents.
        2. Find their neighbors in the network.
        3. If a neighbor is susceptible, infect them with probability `infection_prob`.
        """
        newly_infected = []
        for agent in agents:
            # Only infected agents can spread the virus
            if agent.health_state != 1:
                continue
            
            # Get neighbors from the network structure
            neighbors = self.model.network.get_neighbors(agent)
            
            for category, neighbor_id in neighbors:
                # Retrieve the actual agent object using category and ID
                neighbor = self.model.network.agent_categories[category].get_agent(neighbor_id)
                
                # Infect susceptible neighbors
                if neighbor.health_state == 0:
                    if random.random() < self.scenario.infection_prob:
                        newly_infected.append(neighbor)
        
        # Update states after iterating to avoid chain reactions in a single step
        for agent in newly_infected:
            agent.health_state = 1

    def recover(self, agents: "AgentList[CovidAgent]"):
        """
        Infected agents recover with a certain probability.
        """
        for agent in agents:
            if agent.health_state == 1 and random.random() < self.scenario.recovery_prob:
                agent.health_state = 2

    def update_population_stats(self, agents: "AgentList[CovidAgent]"):
        """
        Update statistical counts for the current step.
        """
        self.num_susceptible = 0
        self.num_infected = 0
        self.num_recovered = 0
        for agent in agents:
            if agent.health_state == 0:
                self.num_susceptible += 1
            elif agent.health_state == 1:
                self.num_infected += 1
            else:
                self.num_recovered += 1

Agent Behavior

Same as the network example. Defined in core/agent.py.

from Melodie import NetworkAgent
from typing import TYPE_CHECKING
import random

if TYPE_CHECKING:
    from Melodie import AgentList


class CovidAgent(NetworkAgent):
    """
    Network-based Covid agent with basic SIR states.
    
    Attributes:
        health_state (int): 0=Susceptible, 1=Infected, 2=Recovered.
        category (int): Agent category (0 for all agents in this simple model).
    """

    def setup(self):
        """
        Initialize agent properties.
        """
        self.health_state: int = 0
        self.category: int = 0

    def set_category(self):
        """
        Required by NetworkAgent to determine the category ID.
        We use a single category (0) for simplicity.
        """
        self.category = 0

    def seed_infection(self, initial_infected_percentage: float):
        """
        Randomly infect the agent based on the initial percentage.
        """
        if random.random() < initial_infected_percentage:
            self.health_state = 1