diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d1c5857..b156eb54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,26 @@ # Changelog +## 0.9.0 +- Add an optional dashboard function to the mas utility, giving a live overview over all MAS module results. Modules optionally implement the visualization themselves. + - The Dashboard can be launched after a run, blocking main, not blocking main, or experimentally also live during the run. + - Usage: + ```python + mas = LocalMASAgency(...) + mas.show_results_dashboard(live=True) # show live + mas.run(until=None) + # mas.show_results_dashboard(block_main=False) # or show after the run + # ... other plotting or analysis code + ``` +- Add optional results logging for communicators, with available granularity: none, basic, detail. Communication results are also available in mas-dashboard. + ```json + { + "type": "local", + "communication_log_level": "detail" + } + ``` +- Add subcription based multiprocessing. +- Fixed a bug, where the AgentLogger would load an old run when no filename was specified, but overwrite_log was true + ## 0.8.9 - Improve error message to include agent and module id in validation errors - Allow agent configs to specify modules as a dict instead of list, with dict keys corresponding to module_id. Old list still supported as normal. diff --git a/README.md b/README.md index a7cfd146..6b42d599 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,8 @@ AgentLib has a number of optional dependencies, ranging from additional features - **plot**: Installs matplotlib, allows to plot the result of examples. - **orjson**: Faster json library, improves performance when using network communicators. - **fuzz**: Improves error messages when providing wrong configurations. + - **interactive**: Supports result visualization in dashboards using plotly dash. + **clonemap**: Support the execution of agents and their communication through [clonemap](https://github.com/sogno-platform/clonemap). As clonemapy is not available through PYPI, please install it from source, or through the AgentLib's ``requirements.txt`` . diff --git a/agentlib/__init__.py b/agentlib/__init__.py index 51b0e566..0f39bf55 100644 --- a/agentlib/__init__.py +++ b/agentlib/__init__.py @@ -12,7 +12,7 @@ LocalCloneMAPAgency, ) -__version__ = "0.8.9" +__version__ = "0.9.0" __all__ = [ "core", diff --git a/agentlib/core/data_broker.py b/agentlib/core/data_broker.py index 2713958f..2cde1738 100644 --- a/agentlib/core/data_broker.py +++ b/agentlib/core/data_broker.py @@ -450,7 +450,7 @@ def __init__( self.thread = threading.Thread( target=self._callback_thread, daemon=True, name="DataBroker" ) - self._module_queues: dict[Union[str, None], queue.Queue] = {} + self._module_queues: Dict[Union[str, None], queue.Queue] = {} env.process(self._start_executing_callbacks(env)) diff --git a/agentlib/core/logging_.py b/agentlib/core/logging_.py index 73c5b34a..16ecd35e 100644 --- a/agentlib/core/logging_.py +++ b/agentlib/core/logging_.py @@ -35,7 +35,7 @@ def makeRecord( _time = self.env.pretty_time() if _until is environment.UNTIL_UNSET: # Add "INIT" prefix to clearly indicate initialization phase - record.env_time = f"" + record.env_time = "" elif _until is None: record.env_time = _time else: diff --git a/agentlib/core/module.py b/agentlib/core/module.py index 5b3edfc3..8844c62a 100644 --- a/agentlib/core/module.py +++ b/agentlib/core/module.py @@ -16,6 +16,7 @@ Optional, get_type_hints, Type, + Tuple, ) import pydantic @@ -32,7 +33,7 @@ AttrsToPydanticAdaptor, ) from agentlib.core.environment import CustomSimpyEnvironment -from agentlib.core.errors import ConfigurationError +from agentlib.core.errors import ConfigurationError, OptionalDependencyError from agentlib.utils.fuzzy_matching import fuzzy_match, RAPIDFUZZ_IS_INSTALLED from agentlib.utils.validators import ( include_defaults_in_root, @@ -44,6 +45,7 @@ if TYPE_CHECKING: # this avoids circular import from agentlib.core import Agent + from dash import html # For type hinting logger = logging.getLogger(__name__) @@ -218,7 +220,7 @@ def merge_variables( for field_name, field in cls.model_fields.items(): # If field is missing in values, validation of field was not # successful. Continue and pydantic will later raise the ValidationError - if field_name not in pre_validated_instance.model_fields: + if field_name not in cls.model_fields: continue pre_merged_attr = pre_validated_instance.__getattribute__(field_name) @@ -639,6 +641,45 @@ def _copy_list_to_dict(ls: List[AgentVariable]): # pylint: disable=invalid-name return {var.name: var for var in ls.copy()} + @classmethod + def visualize_results( + cls, results_data: Any, module_id: str, agent_id: str + ) -> Optional["html.Div"]: + """ + Generate a visualization for the module's results. + This method should be overridden by subclasses to provide + custom visualizations. + + Args: + results_data: The data returned by the module's get_results() method. + module_id: The ID of the module instance. + agent_id: The ID of the agent this module belongs to. + + Returns: + A Dash HTML component (e.g., dash.html.Div) containing the visualization, + or None if not implemented or Dash is unavailable. + """ + try: + # Import here to avoid circular dependency issues at module load time + # and to only require Dash when this method is actually called. + from dash import html as dash_html + except ImportError: + raise OptionalDependencyError( + used_object=f"{cls.__name__}.visualize_results", + dependency_install="agentlib[interactive]", + dependency_name="interactive", + ) + # Base implementation returns None, indicating no visualization is provided. + # Subclasses should override this method. + logger.debug( + "Visualization not implemented for module type %s (module_id: '%s', agent_id: '%s'). " + "Returning None.", + cls.__name__, + module_id, + agent_id, + ) + return None + def get_results(self): """ Returns results of this modules run. @@ -650,6 +691,45 @@ def get_results(self): Some form of results data, often in the form of a pandas DataFrame. """ + def get_results_incremental( + self, update_token: Optional[Any] = None + ) -> Tuple[Any, Optional[Any]]: + """ + Fetches results suitable for incremental updates in a live dashboard. + + This method is intended to be overridden by subclasses that support + incremental data fetching for live visualization. The `update_token` + allows the module to return only new or changed data since the last + call. + + Args: + update_token: An optional token from the previous call. If None, + the module should typically return its full current data. + The nature of the token is module-specific. + + Returns: + A tuple (results_chunk, next_update_token): + - results_chunk: The data to be sent to the dashboard (e.g., + a pandas DataFrame, a list of new entries). + - next_update_token: A token for the dashboard to use in the + next call to this method. Can be None if no + further incremental updates are supported by + this specific call or if the module doesn't + support incremental updates beyond the initial fetch. + """ + if update_token is None: + # Default behavior for the first call or for modules not fully + # implementing incremental logic: return all results from the + # existing get_results() method. The None token indicates that + # this is the complete data for now, or that no further + # *incremental* steps are defined by this default implementation. + return self.get_results(), None + # If an update_token is provided but this base method is called + # (i.e., not overridden with more specific incremental logic), + # it implies no further *new* data based on this token according + # to the default implementation. + return None, None + def cleanup_results(self): """ Deletes all files this module created. diff --git a/agentlib/models/fmu_model.py b/agentlib/models/fmu_model.py index e1420aa1..44dd23b0 100644 --- a/agentlib/models/fmu_model.py +++ b/agentlib/models/fmu_model.py @@ -126,7 +126,7 @@ def initialize(self, **kwargs): t_start (float): Start time of simulation t_stop (float): Stop time of simulation """ - logger.info("Initializing model...") + logger.debug("Initializing model...") # Handle Logging of the FMU itself: try: callbacks = fmpy.fmi2.fmi2CallbackFunctions() @@ -156,7 +156,7 @@ def initialize(self, **kwargs): ) self.system.enterInitializationMode() self.system.exitInitializationMode() - logger.info("Model: %s initialized", self.name) + logger.debug("Model: %s initialized", self.name) def _fmu_logger(self, component, instanceName, status, category, message): """Print the FMU's log messages to the command line (works for both FMI 1.0 and 2.0)""" diff --git a/agentlib/modules/communicator/__init__.py b/agentlib/modules/communicator/__init__.py index 77d65f5e..4c848acb 100644 --- a/agentlib/modules/communicator/__init__.py +++ b/agentlib/modules/communicator/__init__.py @@ -24,4 +24,8 @@ import_path="agentlib.modules.communicator.local_multiprocessing", class_name="MultiProcessingBroadcastClient", ), + "multiprocessing_subscription": ModuleImport( + import_path="agentlib.modules.communicator.local_multiprocessing", + class_name="MultiProcessingSubscriptionClient", + ), } diff --git a/agentlib/modules/communicator/communication_logging_handling.py b/agentlib/modules/communicator/communication_logging_handling.py new file mode 100644 index 00000000..4f045d8d --- /dev/null +++ b/agentlib/modules/communicator/communication_logging_handling.py @@ -0,0 +1,502 @@ +""" +Communication logging and result handling functionality +""" + +import collections +import json +import logging +import os +from pathlib import Path +from typing import Union, List, Any, Optional, Tuple, Callable + +import pandas as pd + +from agentlib.core.errors import OptionalDependencyError + +logger = logging.getLogger(__name__) + + +class CommunicationLogger: + """Handles all communication logging and result processing""" + + def __init__(self, config, agent_id: str, module_id: str, env): + self.config = config + self.agent_id = agent_id + self.module_id = module_id + self.env = env + self.logger = logging.getLogger(f"{__name__}.{agent_id}.{module_id}") + + # Initialize attributes that might be used + self._sent_alias_counts: Optional[collections.Counter] = None + self._received_source_alias_counts: Optional[collections.Counter] = None + self._communication_log_filename: Optional[str] = None + self._communication_log_batch: Optional[List[dict]] = None + + # Set up logging methods based on level + self._setup_logging_strategy() + + def _setup_logging_strategy(self): + """Set up the appropriate logging methods based on config level""" + log_level = self.config.communication_log_level + + if log_level == "none": + self.log_sent_message = self._log_sent_none + self.log_received_message = self._log_received_none + elif log_level == "basic": + self._init_basic_logging() + self.log_sent_message = self._log_sent_basic + self.log_received_message = self._log_received_basic + elif log_level == "detail": + self._init_detail_logging() + self.log_sent_message = self._log_sent_detail + self.log_received_message = self._log_received_detail + + def _init_basic_logging(self): + """Initialize basic logging with counters""" + self._sent_alias_counts = collections.Counter() + self._received_source_alias_counts = collections.Counter() + + def _init_detail_logging(self): + """Initialize detail logging with file handling""" + self._communication_log_batch = [] + + if self.config.communication_log_file is None: + logs_dir = Path("communicator_logs") + logs_dir.mkdir(parents=True, exist_ok=True) + self._communication_log_filename = str( + logs_dir / f"{self.agent_id}_{self.module_id}.jsonl" + ) + else: + self._communication_log_filename = self.config.communication_log_file + # Ensure parent directory exists if a custom path is given + Path(self._communication_log_filename).parent.mkdir( + parents=True, exist_ok=True + ) + + if ( + self._communication_log_filename is not None + and Path(self._communication_log_filename).exists() + and self.config.communication_log_overwrite + ): + os.remove(self._communication_log_filename) + self.logger.info( + f"Overwriting existing communication log: {self._communication_log_filename}" + ) + + # For non-realtime, schedule periodic flushing + if not self.env.config.rt and self.config.communication_log_t_sample > 0: + self.env.process(self._log_detail_process()) + elif self.config.communication_log_t_sample <= 0 and not self.env.config.rt: + self.logger.info( + "communication_log_t_sample <= 0, detail logs will only be written on terminate." + ) + + # No-op logging methods for "none" level + def _log_sent_none(self, alias: str): + """No-op logging for sent messages""" + pass + + def _log_received_none(self, alias: str, source_agent_id: Optional[str]): + """No-op logging for received messages""" + pass + + # Basic logging methods + def _log_sent_basic(self, alias: str): + """Log sent message for basic level""" + self._sent_alias_counts[alias] += 1 + + def _log_received_basic(self, alias: str, source_agent_id: Optional[str]): + """Log received message for basic level""" + self._received_source_alias_counts[(source_agent_id, alias)] += 1 + + # Detail logging methods + def _log_sent_detail(self, alias: str): + """Log sent message for detail level""" + log_entry = { + "timestamp": self.env.time, + "direction": "sent", + "own_agent_id": self.agent_id, + "remote_agent_id": None, # Cannot be known for sent messages generally + "alias": alias, + } + self._communication_log_batch.append(log_entry) + + def _log_received_detail(self, alias: str, source_agent_id: Optional[str]): + """Log received message for detail level""" + log_entry = { + "timestamp": self.env.time, + "direction": "received", + "own_agent_id": self.agent_id, + "remote_agent_id": source_agent_id, + "alias": alias, + } + self._communication_log_batch.append(log_entry) + + # These methods will be assigned during init + def log_sent_message(self, alias: str): + """Log a sent message - implementation assigned during init""" + pass + + def log_received_message(self, alias: str, source_agent_id: Optional[str]): + """Log a received message - implementation assigned during init""" + pass + + def _log_detail_process(self): + """SimPy process to periodically flush detail logs.""" + while True: + yield self.env.timeout(self.config.communication_log_t_sample) + self._flush_detail_log() + + def _flush_detail_log(self): + """Writes the current batch of detail logs to file.""" + if ( + not self._communication_log_batch + or self._communication_log_filename is None + ): + return + + with open(self._communication_log_filename, "a", encoding="utf-8") as f: + for entry in self._communication_log_batch: + json.dump(entry, f) + f.write("\n") + self.logger.debug( + f"Flushed {len(self._communication_log_batch)} entries to {self._communication_log_filename}" + ) + self._communication_log_batch = [] # Clear batch + + def terminate(self): + """Handle termination cleanup""" + if self.config.communication_log_level == "detail": + self.logger.debug( + f"Terminating communicator {self.module_id}, flushing any remaining detail logs." + ) + self._flush_detail_log() + + def get_results(self) -> Optional[Union[dict, pd.DataFrame]]: + """Returns logged communication data based on the log level.""" + log_level = self.config.communication_log_level + if log_level == "none": + return None + elif log_level == "basic": + return { + "sent_counts": dict(self._sent_alias_counts or {}), + "received_counts": { + str(k): v + for k, v in (self._received_source_alias_counts or {}).items() + }, + } + elif log_level == "detail": + self._flush_detail_log() # Ensure all data is written before reading + log_entries = [] + with open(self._communication_log_filename, "r", encoding="utf-8") as f: + for line in f: + log_entries.append(json.loads(line)) + if not log_entries: # Handle empty log file + return pd.DataFrame() + df = pd.DataFrame(log_entries) + return df + return None + + def get_results_incremental( + self, update_token: Optional[Any] = None + ) -> Tuple[Optional[Union[dict, pd.DataFrame]], Optional[Any]]: + """Returns logged communication data incrementally.""" + log_level = self.config.communication_log_level + + if log_level == "none": + return None, None + elif log_level == "basic": + current_counts = { + "sent_counts": dict(self._sent_alias_counts or {}), + "received_counts": { + str(k): v + for k, v in (self._received_source_alias_counts or {}).items() + }, + } + if update_token is None or update_token != current_counts: + return current_counts, current_counts + return None, update_token + elif log_level == "detail": + self._flush_detail_log() + + if update_token is None: # Initial call + df = self.get_results() + if df is None or df.empty: + return pd.DataFrame(), 0 + with open(self._communication_log_filename, "r", encoding="utf-8") as f: + lines_in_file = sum(1 for _ in f) + return df, lines_in_file + else: # Incremental call + df_chunk, lines_read = self._load_detail_log_incremental( + filename=self._communication_log_filename, + start_line_index=update_token, + ) + if df_chunk is None or df_chunk.empty: + return None, update_token + return df_chunk, update_token + lines_read + + @classmethod + def _load_detail_log_incremental( + cls, filename: str, start_line_index: int + ) -> Tuple[Optional[pd.DataFrame], int]: + """Loads detail log file from a specific line index.""" + log_entries = [] + lines_read_count = 0 + + with open(filename, "r", encoding="utf-8") as f: + for i, line in enumerate(f): + if i < start_line_index: + continue + log_entries.append(json.loads(line)) + lines_read_count += 1 + + if not log_entries: + return None, 0 + + df = pd.DataFrame(log_entries) + return df, lines_read_count + + def cleanup_results(self): + """Deletes the communication log file if 'detail' logging was active.""" + if ( + self.config.communication_log_level == "detail" + and self._communication_log_filename + ): + try: + if Path(self._communication_log_filename).exists(): + os.remove(self._communication_log_filename) + self.logger.info( + f"Cleaned up communication log file: {self._communication_log_filename}" + ) + except OSError as e: + self.logger.error( + f"Error cleaning up communication log file {self._communication_log_filename}: {e}" + ) + + @classmethod + def visualize_results( + cls, + results_data: Optional[Union[dict, pd.DataFrame]], + module_id: str, + agent_id: str, + ) -> "Optional[html.Div]": + try: + from dash import dcc, html + import plotly.graph_objects as go + import dash_bootstrap_components as dbc + except ImportError: + raise OptionalDependencyError( + used_object="Dashboard", + dependency_install="dash", + dependency_name="interactive", + ) + if results_data is None: + return html.Div( + f"No communication data to visualize for {module_id} (Agent: {agent_id}). Log level might be 'none'." + ) + + main_container_children = [ + html.H4( + f"Communicator Activity: {module_id} (Agent: {agent_id})", + className="mb-3", + ) + ] + + if isinstance(results_data, dict): # Basic logging + log_type_info = html.P("Log Level: basic (Counters)") + sent_counts = results_data.get("sent_counts", {}) + received_counts_str_keys = results_data.get("received_counts", {}) + + received_counts_display = { + k: v for k, v in received_counts_str_keys.items() + } + + if sent_counts: + sent_fig = go.Figure( + data=[ + go.Bar(x=list(sent_counts.keys()), y=list(sent_counts.values())) + ] + ) + sent_fig.update_layout( + title_text="Sent Message Counts by Alias", + xaxis_title="Alias", + yaxis_title="Count", + height=300, + margin=dict(l=40, r=20, t=40, b=30), + ) + main_container_children.append(dcc.Graph(figure=sent_fig)) + else: + main_container_children.append( + html.P( + "No messages sent or 'basic' logging did not capture sent messages." + ) + ) + + if received_counts_display: + received_fig = go.Figure( + data=[ + go.Bar( + x=list(received_counts_display.keys()), + y=list(received_counts_display.values()), + ) + ] + ) + received_fig.update_layout( + title_text="Received Message Counts by (Source Agent, Alias)", + xaxis_title="(Source Agent, Alias)", + yaxis_title="Count", + height=300, + margin=dict(l=40, r=20, t=40, b=30), + ) + main_container_children.append(dcc.Graph(figure=received_fig)) + else: + main_container_children.append( + html.P( + "No messages received or 'basic' logging did not capture received messages." + ) + ) + + elif isinstance(results_data, pd.DataFrame): + log_type_info = html.P("Log Level: detail (Timeline & Aggregates)") + if results_data.empty: + main_container_children.append(html.P("Detail log is empty.")) + else: + # Sent Messages Summary (from DataFrame) + sent_df = results_data[results_data["direction"] == "sent"] + if not sent_df.empty: + sent_counts_df = ( + sent_df.groupby("alias").size().reset_index(name="count") + ) + sent_fig_df = go.Figure( + data=[ + go.Bar(x=sent_counts_df["alias"], y=sent_counts_df["count"]) + ] + ) + sent_fig_df.update_layout( + title_text="Sent Message Counts by Alias (from Detail Log)", + xaxis_title="Alias", + yaxis_title="Count", + height=300, + margin=dict(l=40, r=20, t=40, b=30), + ) + main_container_children.append(dcc.Graph(figure=sent_fig_df)) + else: + main_container_children.append( + html.P("No 'sent' messages in detail log.") + ) + + # Received Messages Summary (from DataFrame) + received_df = results_data[results_data["direction"] == "received"] + if not received_df.empty: + # Create a combined key for grouping, e.g., "agent_id | alias" + # Ensure remote_agent_id is not None before concatenating + received_df_copy = ( + received_df.copy() + ) # Avoid SettingWithCopyWarning + received_df_copy["source_alias_key"] = ( + received_df_copy["remote_agent_id"].fillna("Unknown") + + " | " + + received_df_copy["alias"] + ) + received_counts_df = ( + received_df_copy.groupby("source_alias_key") + .size() + .reset_index(name="count") + ) + received_counts_df = received_counts_df.sort_values( + by="source_alias_key" + ) + + received_fig_df = go.Figure( + data=[ + go.Bar( + x=received_counts_df["source_alias_key"], + y=received_counts_df["count"], + ) + ] + ) + received_fig_df.update_layout( + title_text="Received Message Counts by (Source Agent | Alias) (from Detail Log)", + xaxis_title="(Source Agent | Alias)", + yaxis_title="Count", + height=300, + margin=dict(l=40, r=20, t=40, b=30), + ) + main_container_children.append(dcc.Graph(figure=received_fig_df)) + else: + main_container_children.append( + html.P("No 'received' messages in detail log.") + ) + + # Timelines (at the bottom) + timeline_children = [] + if not sent_df.empty: + sent_timeline_fig = go.Figure( + data=[ + go.Scatter( + x=sent_df["timestamp"], + y=sent_df["alias"], + mode="markers", + name="Sent", + ) + ] + ) + sent_timeline_fig.update_layout( + title_text="Sent Messages Timeline", + xaxis_title="Time", + yaxis_title="Alias", + height=350, + margin=dict(l=40, r=20, t=40, b=30), + ) + timeline_children.append( + dbc.Col(dcc.Graph(figure=sent_timeline_fig), md=6) + ) + + if not received_df.empty: + received_timeline_fig = go.Figure() + # Group by remote_agent_id, handling potential None values + for r_agent_id, group in received_df.fillna( + {"remote_agent_id": "Unknown"} + ).groupby("remote_agent_id"): + received_timeline_fig.add_trace( + go.Scatter( + x=group["timestamp"], + y=group["alias"], + mode="markers", + name=f"From {r_agent_id}", + hovertext=[ + f"From: {ra} Alias: {al}" + for ra, al in zip( + group["remote_agent_id"], group["alias"] + ) + ], + ) + ) + received_timeline_fig.update_layout( + title_text="Received Messages Timeline", + xaxis_title="Time", + yaxis_title="Alias", + height=350, + margin=dict(l=40, r=20, t=40, b=30), + ) + timeline_children.append( + dbc.Col(dcc.Graph(figure=received_timeline_fig), md=6) + ) + + if timeline_children: + main_container_children.append(html.Hr()) + main_container_children.append( + html.H5("Message Timelines (Detail Log)", className="mt-4 mb-3") + ) + main_container_children.append(dbc.Row(timeline_children)) + + else: + return html.Div( + f"Unknown results data type for {module_id} (Agent: {agent_id}): {type(results_data)}" + ) + + main_container_children.insert( + 1, log_type_info + ) # Insert log type info after H4 + + return html.Div(main_container_children, style={"padding": "10px"}) diff --git a/agentlib/modules/communicator/communicator.py b/agentlib/modules/communicator/communicator.py index d17ba881..98ca98fe 100644 --- a/agentlib/modules/communicator/communicator.py +++ b/agentlib/modules/communicator/communicator.py @@ -4,9 +4,10 @@ import abc import json +import logging import queue import threading -from typing import Union, List, TypedDict, Any +from typing import Union, List, TypedDict, Any, Optional, Literal, Tuple import pandas as pd from pydantic import Field, field_validator @@ -16,6 +17,11 @@ from agentlib.core.errors import OptionalDependencyError from agentlib.utils.broker import Broker from agentlib.utils.validators import convert_to_list +from agentlib.modules.communicator.communication_logging_handling import ( + CommunicationLogger, +) + +logger = logging.getLogger(__name__) # Added logger initialization class CommunicationDict(TypedDict): @@ -33,6 +39,27 @@ class CommunicatorConfig(BaseModuleConfig): description="If true, the faster orjson library will be used for serialization " "deserialization. Requires the optional dependency.", ) + communication_log_level: Literal["none", "basic", "detail"] = Field( + default="none", + title="Communication Log Level", + description="Level of communication logging: 'none', 'basic', 'detail'.", + ) + communication_log_file: Optional[str] = Field( + default=None, + title="Communication Log File", + description="Filename for 'detail' logging. Defaults to 'communicator_logs/{agent_id}_{module_id}.jsonl'.", + ) + communication_log_overwrite: bool = Field( + default=True, + title="Overwrite Communication Log", + description="If true, existing log file will be overwritten at the start.", + ) + communication_log_t_sample: Union[float, int] = Field( + default=300, + title="Communication Log Sampling Time", + description="Interval in seconds for batch writing 'detail' logs to file. Only for non-realtime.", + ge=0, + ) class SubscriptionCommunicatorConfig(CommunicatorConfig): @@ -50,10 +77,19 @@ class Communicator(BaseModule): """ config: CommunicatorConfig + parse_json = True def __init__(self, *, config: dict, agent: Agent): super().__init__(config=config, agent=agent) + # Initialize communication logger + self._communication_logger = CommunicationLogger( + config=self.config, + agent_id=self.agent.id, + module_id=self.id, + env=self.env, + ) + if self.config.use_orjson: try: import orjson @@ -91,6 +127,10 @@ def _send_only_shared_variables(self, variable: AgentVariable): payload = self.short_dict(variable) self.logger.debug("Sending variable %s=%s", variable.alias, variable.value) + + # Delegate logging to communication logger + self._communication_logger.log_sent_message(payload["alias"]) + self._send(payload=payload) def _variable_can_be_send(self, variable): @@ -105,13 +145,13 @@ def _send(self, payload: CommunicationDict): "This method needs to be implemented " "individually for each communicator" ) - def short_dict(self, variable: AgentVariable, parse_json: bool = True) -> CommunicationDict: + def short_dict(self, variable: AgentVariable) -> CommunicationDict: """Creates a short dict serialization of the Variable. Only contains attributes of the AgentVariable, that are relevant for other modules or agents. For performance and privacy reasons, this function should be called for communicators.""" - if isinstance(variable.value, pd.Series) and parse_json: + if isinstance(variable.value, pd.Series) and self.parse_json: value = variable.value.to_json() else: value = variable.value @@ -123,6 +163,27 @@ def short_dict(self, variable: AgentVariable, parse_json: bool = True) -> Commun source=self.agent.id, ) + def _handle_received_variable( + self, variable: AgentVariable, remote_agent_id: Optional[str] = None + ): + """ + Centralized handler for received variables that manages logging and forwarding. + """ + source_agent_id = remote_agent_id or variable.source.agent_id + + # Delegate logging to communication logger + self._communication_logger.log_received_message(variable.alias, source_agent_id) + + # Forward to data broker + self.agent.data_broker.send_variable(variable) + + self.logger.debug( + "Received and processed variable %s=%s from source %s", + variable.alias, + variable.value, + source_agent_id, + ) + def to_json(self, payload: CommunicationDict) -> Union[bytes, str]: """Transforms the payload into json serialized form. Dynamically uses orjson if it is installed, and the builtin json otherwise. @@ -133,6 +194,33 @@ def to_json(self, payload: CommunicationDict) -> Union[bytes, str]: # implemented on init pass + def terminate(self): + self._communication_logger.terminate() + super().terminate() + + def get_results(self) -> Optional[Union[dict, pd.DataFrame]]: + """Returns logged communication data based on the log level.""" + return self._communication_logger.get_results() + + def get_results_incremental( + self, update_token: Optional[Any] = None + ) -> Tuple[Optional[Union[dict, pd.DataFrame]], Optional[Any]]: + """Returns logged communication data incrementally.""" + return self._communication_logger.get_results_incremental(update_token) + + def cleanup_results(self): + """Deletes the communication log file if 'detail' logging was active.""" + self._communication_logger.cleanup_results() + + @classmethod + def visualize_results( + cls, + results_data: Optional[Union[dict, pd.DataFrame]], + module_id: str, + agent_id: str, + ) -> "Optional[html.Div]": + return CommunicationLogger.visualize_results(results_data, module_id, agent_id) + class LocalCommunicatorConfig(CommunicatorConfig): parse_json: bool = Field( @@ -141,10 +229,7 @@ class LocalCommunicatorConfig(CommunicatorConfig): "which use MQTT or similar.", default=False, ) - queue_size: int = Field( - title="Size of the queue", - default=10000 - ) + queue_size: int = Field(title="Size of the queue", default=10000) class LocalCommunicator(Communicator): @@ -169,6 +254,7 @@ def __init__(self, config: dict, agent: Agent): super().__init__(config=config, agent=agent) self.broker = self.setup_broker() + self.parse_json = self.config.parse_json self._msg_q_in = queue.Queue(self.config.queue_size) self.broker.register_client(client=self) @@ -188,18 +274,9 @@ def setup_broker(self): """Function to set up the broker object. Needs to return a valid broker option.""" raise NotImplementedError( - "This method needs to be implemented " "individually for each communicator" + "This method needs to be implemented individually for each communicator" ) - def _send_only_shared_variables(self, variable: AgentVariable): - """Send only variables with field ``shared=True``""" - if not self._variable_can_be_send(variable): - return - - payload = self.short_dict(variable, parse_json=self.config.parse_json) - self.logger.debug("Sending variable %s=%s", variable.alias, variable.value) - self._send(payload=payload) - def _process(self): """Waits for new messages, sends them to the broker.""" yield self.env.event() @@ -218,38 +295,33 @@ def _send_simpy(self, ignored): """Sends new messages to the broker when receiving them, adhering to the simpy event queue. To be appended to a simpy event callback.""" variable = self._msg_q_in.get_nowait() - self.agent.data_broker.send_variable(variable) + self._handle_received_variable(variable) - def _receive(self, msg_obj): + def _receive(self, msg_obj): # msg_obj is raw from broker """Receive a given message and put it in the queue and set the corresponding simpy event.""" - if self.config.parse_json: - variable = AgentVariable.from_json(msg_obj) - else: - variable = msg_obj - self._msg_q_in.put(variable, block=False) + variable_to_queue = ( + AgentVariable.from_json(msg_obj) if self.config.parse_json else msg_obj + ) + self._msg_q_in.put(variable_to_queue, block=False) self._received_variable.callbacks.append(self._send_simpy) self._received_variable.succeed() self._received_variable = self.env.event() - def _receive_realtime(self, msg_obj): + def _receive_realtime(self, msg_obj): # msg_obj is raw from broker """Receive a given message and put it in the queue. No event setting is required for realtime.""" - if self.config.parse_json: - variable = AgentVariable.from_json(msg_obj) - else: - variable = msg_obj - self._msg_q_in.put(variable) + variable_to_queue = ( + AgentVariable.from_json(msg_obj) if self.config.parse_json else msg_obj + ) + self._msg_q_in.put(variable_to_queue) - def _message_handler(self): + def _message_handler(self): # Realtime message handler """Reads messages that were put in the message queue.""" while True: variable = self._msg_q_in.get() - self.agent.data_broker.send_variable(variable) + self._handle_received_variable(variable) def terminate(self): - # Terminating is important when running multiple - # simulations/environments, otherwise the broker will keep spamming all - # agents from the previous simulation, potentially filling their queues. self.broker.delete_client(self) super().terminate() diff --git a/agentlib/modules/communicator/local_multiprocessing.py b/agentlib/modules/communicator/local_multiprocessing.py index 370a4a41..daaa9f8f 100644 --- a/agentlib/modules/communicator/local_multiprocessing.py +++ b/agentlib/modules/communicator/local_multiprocessing.py @@ -1,21 +1,25 @@ -import time import multiprocessing import threading - -from pydantic import Field +import time +from abc import abstractmethod from ipaddress import IPv4Address +from pydantic import Field, BaseModel + from agentlib.core import Agent from agentlib.core.datamodels import AgentVariable from agentlib.modules.communicator.communicator import ( Communicator, CommunicationDict, CommunicatorConfig, + SubscriptionCommunicatorConfig, ) from agentlib.utils import multi_processing_broker -class MultiProcessingBroadcastClientConfig(CommunicatorConfig): +class MultiProcessingConfigMixin(BaseModel): + """Mixin containing common multiprocessing configuration fields""" + ipv4: IPv4Address = Field( default="127.0.0.1", description="IP Address for the communication server. Defaults to localhost.", @@ -30,13 +34,29 @@ class MultiProcessingBroadcastClientConfig(CommunicatorConfig): ) -class MultiProcessingBroadcastClient(Communicator): +class MultiProcessingBroadcastClientConfig( + CommunicatorConfig, MultiProcessingConfigMixin +): + """Config for the Multiprocessing Broadcast Communicator""" + + pass + + +class MultiProcessingSubscriptionClientConfig( + SubscriptionCommunicatorConfig, MultiProcessingConfigMixin +): + """Config for the Multiprocessing Subscription Communicator""" + + pass + + +class MultiProcessingCommunicatorBase(Communicator): """ - This communicator implements the communication between agents via a - central process broker. + Base class for multiprocessing communicators. + Implements the common communication logic between agents via a central process broker. """ - config: MultiProcessingBroadcastClientConfig + config: MultiProcessingConfigMixin def __init__(self, config: dict, agent: Agent): super().__init__(config=config, agent=agent) @@ -57,10 +77,9 @@ def __init__(self, config: dict, agent: Agent): signup_queue = manager.get_queue() client_read, broker_write = multiprocessing.Pipe(duplex=False) broker_read, client_write = multiprocessing.Pipe(duplex=False) - signup = multi_processing_broker.MPClient( - agent_id=self.agent.id, read=broker_read, write=broker_write - ) + # Create the appropriate client (to be implemented by subclasses) + signup = self._create_client(broker_read, broker_write) signup_queue.put(signup) self._client_write = client_write @@ -71,6 +90,14 @@ def __init__(self, config: dict, agent: Agent): # ensure the broker has set up the connection and sends its ack self._client_read.recv() + @abstractmethod + def _create_client(self, broker_read, broker_write): + """ + Create the appropriate client type (broadcast or subscription). + To be implemented by subclasses. + """ + pass + def process(self): """Only start the loop once the env is running.""" _thread = threading.Thread( @@ -88,14 +115,11 @@ def _message_handler(self): except EOFError: break variable = AgentVariable.from_json(msg) - self.logger.debug(f"Received variable {variable.name}.") - self.agent.data_broker.send_variable(variable) + self.logger.debug(f"Received variable {variable.alias}.") + self._handle_received_variable(variable) def terminate(self): """Closes all of the connections.""" - # Terminating is important when running multiple - # simulations/environments, otherwise the broker will keep spamming all - # agents from the previous simulation, potentially filling their queues. self._client_write.close() self._client_read.close() self._broker_write.close() @@ -109,3 +133,36 @@ def _send(self, payload: CommunicationDict): agent_id=agent_id, payload=self.to_json(payload) ) self._client_write.send(msg) + + +class MultiProcessingBroadcastClient(MultiProcessingCommunicatorBase): + """ + This communicator implements broadcast communication between agents via a + central process broker. + """ + + config: MultiProcessingBroadcastClientConfig + + def _create_client(self, broker_read, broker_write): + """Creates a broadcast client.""" + return multi_processing_broker.MPClient( + agent_id=self.agent.id, read=broker_read, write=broker_write + ) + + +class MultiProcessingSubscriptionClient(MultiProcessingCommunicatorBase): + """ + This communicator implements subscription-based communication between agents + via a central process broker. + """ + + config: MultiProcessingSubscriptionClientConfig + + def _create_client(self, broker_read, broker_write): + """Creates a subscription client with subscription information.""" + return multi_processing_broker.MPSubscriptionClient( + agent_id=self.agent.id, + read=broker_read, + write=broker_write, + subscriptions=tuple(self.config.subscriptions), + ) diff --git a/agentlib/modules/communicator/mqtt.py b/agentlib/modules/communicator/mqtt.py index d063772d..cbac5922 100644 --- a/agentlib/modules/communicator/mqtt.py +++ b/agentlib/modules/communicator/mqtt.py @@ -5,15 +5,14 @@ from pydantic import AnyUrl, Field, ValidationError, field_validator +from agentlib.core import Agent +from agentlib.core.datamodels import AgentVariable +from agentlib.core.errors import InitializationError, OptionalDependencyError from agentlib.modules.communicator.communicator import ( Communicator, SubscriptionCommunicatorConfig, ) -from agentlib.core import Agent -from agentlib.core.datamodels import AgentVariable -from agentlib.core.errors import InitializationError from agentlib.utils.validators import convert_to_list -from agentlib.core.errors import OptionalDependencyError try: from paho.mqtt.client import ( @@ -189,7 +188,9 @@ def disconnect(self, reasoncode=None, properties=None): """Trigger the disconnect""" self._mqttc.disconnect(reasoncode=reasoncode, properties=properties) - def _disconnect_callback(self, client, userdata, disconnect_flags, reasonCode, properties): + def _disconnect_callback( + self, client, userdata, disconnect_flag, reasonCode, properties + ): """Stop the loop as a result of the disconnect""" self.logger.warning( "Disconnected with result code: %s | userdata: %s | properties: %s", @@ -206,12 +207,11 @@ def _message_callback(self, client, userdata, msg): """ agent_inp = AgentVariable.from_json(msg.payload) self.logger.debug( - "Received variable %s = %s from source %s", + "Received MQTT message for variable %s from source %s", agent_inp.alias, - agent_inp.value, agent_inp.source, ) - self.agent.data_broker.send_variable(agent_inp) + self._handle_received_variable(agent_inp) def _subscribe_callback(self, client, userdata, mid, reasonCodes, properties): """Log if the subscription was successful""" diff --git a/agentlib/modules/simulation/simulator.py b/agentlib/modules/simulation/simulator.py index dae366e1..19c9807c 100644 --- a/agentlib/modules/simulation/simulator.py +++ b/agentlib/modules/simulation/simulator.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from math import inf from pathlib import Path -from typing import Union, Dict, List, Optional +from typing import Union, Dict, List, Optional, Tuple import numpy as np import pandas as pd @@ -187,7 +187,7 @@ def check_nonexisting_csv(cls, result_filename, info: FieldValidationInfo): if not result_filename.endswith(".csv"): raise TypeError( f"Given result_filename ends with " - f'{result_filename.split(".")[-1]} ' + f"{result_filename.split('.')[-1]} " f"but should be a .csv file" ) if os.path.isfile(result_filename): @@ -251,7 +251,7 @@ def check_model(cls, model, info: FieldValidationInfo): states = info.data.get("states") if "type" not in model: raise KeyError( - "Given model config does not " "contain key 'type' (type of the model)." + "Given model config does not contain key 'type' (type of the model)." ) _type = model.pop("type") if isinstance(_type, dict): @@ -340,7 +340,7 @@ def model(self, model: Model): t_start=self.config.t_start + self.env.config.offset, t_stop=self.config.t_stop, ) - self.logger.info("Model successfully loaded model: %s", self.model.name) + self.logger.debug("Model successfully loaded model: %s", self.model.name) def run(self, until=None): """ @@ -369,7 +369,7 @@ def _register_input_callbacks(self): ): for var in ag_vars: if var.name in model_var_names: - self.logger.info( + self.logger.debug( "Registered callback for model %s %s ", _type, var.name ) self.agent.data_broker.register_callback( @@ -497,8 +497,141 @@ def get_results(self) -> Optional[pd.DataFrame]: def cleanup_results(self): if not self.config.save_results or not self.config.result_filename: return - os.remove(self.config.result_filename) + if self.config.result_filename and Path(self.config.result_filename).exists(): + os.remove(self.config.result_filename) + + def get_results_incremental( + self, update_token: Optional[int] = None + ) -> Tuple[Optional[pd.DataFrame], Optional[int]]: + """Fetches simulation results incrementally.""" + if not self.config.save_results: + return None, None + + if self.config.result_filename: + # Results are in CSV file + file_path = Path(self.config.result_filename) + + if not file_path.exists(): + return None, 0 + + if update_token is None: # Initial call + df = read_simulator_results(str(file_path)) + # Apply same processing as get_results() + df = df.droplevel(level=2, axis=1).droplevel(level=0, axis=1) + return df, len(df) + else: # Incremental call + try: + # Skip header (3 lines) + previous data rows + df_chunk = pd.read_csv( + file_path, + header=[0, 1, 2], + index_col=0, + skiprows=range(1, update_token + 1), + ) + if df_chunk.empty: + return None, update_token + # Apply same processing as get_results() + df_chunk = df_chunk.droplevel(level=2, axis=1).droplevel( + level=0, axis=1 + ) + return df_chunk, update_token + len(df_chunk) + except pd.errors.EmptyDataError: + return None, update_token + else: + # Results are in memory + current_total_rows = ( + len(self._result.index) - 1 + ) # Exclude incomplete last row + if current_total_rows <= 0: + return None, 0 + + if update_token is None: # Initial call + df = self._result.df() + df = df.droplevel(level=2, axis=1).droplevel(level=0, axis=1) + return df, len(df) + else: # Incremental call + if update_token >= current_total_rows: + return None, update_token + + new_data_slice = self._result.data[update_token:current_total_rows] + new_index_slice = self._result.index[update_token:current_total_rows] + + if not new_data_slice: + return None, update_token + + df_chunk = pd.DataFrame( + new_data_slice, index=new_index_slice, columns=self._result.columns + ) + df_chunk = df_chunk.droplevel(level=2, axis=1).droplevel( + level=0, axis=1 + ) + return df_chunk, current_total_rows + + @classmethod + def visualize_results( + cls, results_data: pd.DataFrame, module_id: str, agent_id: str + ) -> "Optional[html.Div]": + try: + from dash import dcc, html + import plotly.graph_objs as go + import dash_bootstrap_components as dbc + except ImportError: + raise OptionalDependencyError( + used_object=f"{cls.__name__}.visualize_results", + dependency_install="agentlib[interactive]", + dependency_name="interactive", + ) + + if results_data is None or results_data.empty: + return None + if not isinstance(results_data, pd.DataFrame): + raise ValueError( + f"Expected pandas DataFrame for Simulator results for '{module_id}', got {type(results_data)}." + ) + + rows = [] + current_row_children = [] + for i, column in enumerate(results_data.columns): + fig = go.Figure() + fig.add_trace( + go.Scatter( + x=results_data.index, + y=results_data[column], + mode="lines", + name=str(column), + ) + ) + fig.update_layout( + title=str(column), + xaxis_title="Time", + yaxis_title="Value", + margin=dict(l=40, r=20, t=40, b=30), # Compact margins + height=250, # Reduced height for compactness + ) + # Add plot to a column, aim for 2 plots per row on medium screens + current_row_children.append(dbc.Col(dcc.Graph(figure=fig), md=6)) + + # If two plots are in the current row, or it's the last plot + if len(current_row_children) == 2 or i == len(results_data.columns) - 1: + rows.append(dbc.Row(current_row_children, className="mb-3")) + current_row_children = [] + + if not rows: + raise ValueError( + f"No plottable data generated for Simulator '{module_id}'." + ) + + return html.Div( + children=[ + html.H4( + f"Simulator Results: {module_id} (Agent: {agent_id})", + className="mb-3", # Add some margin below the title + ) + ] + + rows, # Add the list of dbc.Row components + style={"padding": "10px"}, + ) def _update_results(self): """ diff --git a/agentlib/modules/utils/agent_logger.py b/agentlib/modules/utils/agent_logger.py index d9cb8028..f9bfd1bd 100644 --- a/agentlib/modules/utils/agent_logger.py +++ b/agentlib/modules/utils/agent_logger.py @@ -5,17 +5,20 @@ import json import logging import os -import time from ast import literal_eval from pathlib import Path -from typing import Union, Optional +from typing import Union, Optional, Tuple, TYPE_CHECKING import pandas as pd -from pydantic import field_validator, Field -from pydantic_core.core_schema import FieldValidationInfo +from pydantic import Field from agentlib import AgentVariable from agentlib.core import BaseModule, Agent, BaseModuleConfig +from agentlib.core.errors import OptionalDependencyError + +if TYPE_CHECKING: + from dash import html + logger = logging.getLogger(__name__) @@ -47,34 +50,13 @@ class AgentLoggerConfig(BaseModuleConfig): filename: Optional[str] = Field( title="filename", default=None, - description="The filename where the log is stored. If None, will use 'agent_logs/{agent_id}_log.json'", + description="The filename where the log is stored. If None, will use 'agent_logs/{agent_id}_log.jsonl'", + ) + merge_sources: bool = Field( + title="Merge Sources", + default=True, + description="When loading the results file, automatically merges variables by sources, leaving only alias in the columns.", ) - - @field_validator("filename") - @classmethod - def check_existence_of_file(cls, filename, info: FieldValidationInfo): - """Checks whether the file already exists.""" - # pylint: disable=no-self-argument,no-self-use - - # Skip check for None, as it will be replaced in __init__ - if filename is None: - return filename - - file_path = Path(filename) - if file_path.exists(): - # remove result file, so a new one can be created - if info.data["overwrite_log"]: - file_path.unlink() - return filename - raise FileExistsError( - f"Given filename at {filename} " - f"already exists. We won't overwrite it automatically. " - f"You can use the key word 'overwrite_log' to " - f"activate automatic overwrite." - ) - # Create path in case it does not exist - file_path.parent.mkdir(parents=True, exist_ok=True) - return filename class AgentLogger(BaseModule): @@ -89,19 +71,7 @@ def __init__(self, *, config: dict, agent: Agent): super().__init__(config=config, agent=agent) # If filename is None, create a custom one using the agent ID - if self.config.filename is None: - # Use agent ID to create a default filename - logs_dir = Path("agent_logs") - logs_dir.mkdir(exist_ok=True) - self._filename = str(logs_dir / f"{self.agent.id}.jsonl") - - # Handle file exists case based on overwrite_log setting - if Path(self._filename).exists() and not self.config.overwrite_log: - # Generate a unique filename by appending a timestamp - timestamp = int(time.time()) - self._filename = str(logs_dir / f"{self.agent.id}_{timestamp}.jsonl") - else: - self._filename = self.config.filename + self._filename = self._setup_log_file() self._variables_to_log = {} if not self.env.config.rt and self.config.t_sample < 60: @@ -112,6 +82,54 @@ def __init__(self, *, config: dict, agent: Agent): self.config.t_sample, ) + def _setup_log_file(self) -> str: + """Centralized file setup logic""" + # Determine the target filename + if self.config.filename is None: + logs_dir = Path("agent_logs") + target_file = logs_dir / f"{self.agent.id}.jsonl" + else: + target_file = Path(self.config.filename) + + # Handle file existence and overwrite logic + return self._handle_file_existence(target_file) + + def _handle_file_existence(self, file_path: Path) -> str: + """Consistent file existence and overwrite handling""" + # Create parent directories if they don't exist + file_path.parent.mkdir(parents=True, exist_ok=True) + + # Handle existing file + if file_path.exists(): + if self.config.overwrite_log: + file_path.unlink() + self.logger.info(f"Overwrote existing log file: {file_path}") + elif self.config.filename is None: + # Auto-increment for default filenames to support quick iteration + file_path = self._get_incremented_filename(file_path) + self.logger.info(f"Using auto-incremented filename: {file_path}") + else: + # Custom filename with overwrite disabled - maintain strict behavior + raise FileExistsError( + f"Log file '{file_path}' already exists. " + f"Set 'overwrite_log=True' to enable automatic overwrite." + ) + + return str(file_path) + + def _get_incremented_filename(self, base_path: Path) -> Path: + """Generate an auto-incremented filename""" + counter = 1 + stem = base_path.stem + suffix = base_path.suffix + parent = base_path.parent + + while True: + new_path = parent / f"{stem}_{counter}{suffix}" + if not new_path.exists(): + return new_path + counter += 1 + @property def filename(self): """Return the filename where to log.""" @@ -212,8 +230,12 @@ def _load_agent_variable(var): def get_results(self) -> pd.DataFrame: """Load the own filename""" + # Ensure current in-memory logs are flushed before reading for static results + self._log() return self.load_from_file( - filename=self.filename, values_only=self.config.values_only + filename=self.filename, + values_only=self.config.values_only, + merge_sources=self.config.merge_sources, ) def cleanup_results(self): @@ -231,3 +253,180 @@ def terminate(self): # when terminating, we log one last time, since otherwise the data since the # last log interval is lost self._log() + + def get_results_incremental( + self, update_token: Optional[int] = None + ) -> Tuple[Optional[pd.DataFrame], Optional[int]]: + """Fetches results incrementally for live dashboard.""" + self._log() # Ensure current logs are written + + if update_token is None: # Initial call + df = self.load_from_file( + filename=self.filename, + values_only=self.config.values_only, + merge_sources=self.config.merge_sources, + ) + try: + with open(self.filename, "r") as f: + lines_in_file = sum(1 for _ in f) + except FileNotFoundError: + lines_in_file = 0 + return df, lines_in_file + else: # Incremental call + df_chunk, lines_read = self.load_from_file_incremental( + filename=self.filename, + start_line_index=update_token, + values_only=self.config.values_only, + merge_sources=self.config.merge_sources, + ) + if df_chunk is None or df_chunk.empty: + return None, update_token + return df_chunk, update_token + lines_read + + @classmethod + def load_from_file_incremental( + cls, + filename: str, + start_line_index: int, + values_only: bool = True, + merge_sources: bool = True, + ) -> Tuple[Optional[pd.DataFrame], int]: + """Loads log file from a specific line index.""" + chunks = [] + lines_read_count = 0 + + try: + with open(filename, "r") as file: + for i, data_line in enumerate(file): + if i < start_line_index: + continue + chunks.append(json.loads(data_line)) + lines_read_count += 1 + except FileNotFoundError: + return None, 0 + + if not any(chunks): + return None, 0 + + full_dict = collections.ChainMap(*chunks) + df = pd.DataFrame.from_dict(full_dict, orient="index") + df.index = df.index.astype(float) + columns = (literal_eval(column) for column in df.columns) + df.columns = pd.MultiIndex.from_tuples(columns) + + if not values_only: + + def _load_agent_variable(var): + try: + return AgentVariable.validate_data(var) + except TypeError: + pass + + df = df.applymap(_load_agent_variable) + + if merge_sources: + df = df.droplevel(1, axis=1) + df = df.loc[:, ~df.columns.duplicated(keep="first")] + + return df.sort_index(), lines_read_count + + @classmethod + def visualize_results( + cls, results_data: pd.DataFrame, module_id: str, agent_id: str + ) -> "Optional[html.Div]": + try: + from dash import dcc, html + import plotly.graph_objs as go + import dash_bootstrap_components as dbc # Added for responsive layout + except ImportError: + raise OptionalDependencyError( + used_object=f"{cls.__name__}.visualize_results", + dependency_install="agentlib[interactive]", + dependency_name="interactive", + ) + + if results_data is None or results_data.empty: + raise ValueError( + f"No results data for AgentLogger '{module_id}' in agent '{agent_id}'." + ) + return None + + rows = [] + current_row_children = [] + for i, col_name in enumerate(results_data.columns): + series = results_data[col_name].dropna() + if series.empty: + continue + + try: + numeric_series = pd.to_numeric(series, errors="coerce") + if numeric_series.isnull().all() and not series.isnull().all(): + is_numeric = False + else: + is_numeric = True + series_to_plot = numeric_series + except (ValueError, TypeError): + is_numeric = False + + if not is_numeric: + series_to_plot = series + + fig = go.Figure() + y_axis_title = "Value" + + if is_numeric: + fig.add_trace( + go.Scatter( + x=series_to_plot.index, + y=series_to_plot, + mode="lines+markers", + name=str(col_name), + ) + ) + else: + fig.add_trace( + go.Scatter( + x=series_to_plot.index, + y=[1] * len(series_to_plot), + mode="markers", + name=str(col_name), + hovertext=[str(v) for v in series_to_plot.values], + hoverinfo="x+text", + ) + ) + y_axis_title = "Occurrence" + + title_str = str(col_name) + if isinstance(col_name, tuple) and len(col_name) > 0: + title_str = f"{col_name[0]}" + if len(col_name) > 1 and col_name[1]: + title_str += f" (Source: {col_name[1]})" + + fig.update_layout( + title=title_str, + xaxis_title="Time", + yaxis_title=y_axis_title, + margin=dict(l=40, r=20, t=40, b=30), + height=250, + ) + current_row_children.append(dbc.Col(dcc.Graph(figure=fig), md=6)) + + if len(current_row_children) == 2 or i == len(results_data.columns) - 1: + rows.append(dbc.Row(current_row_children, className="mb-3")) + current_row_children = [] + + if not rows: + raise ValueError( + f"No plottable data generated for AgentLogger '{module_id}'." + ) + + return html.Div( + children=[ + html.H4( + f"AgentLogger Results: {module_id} (Agent: {agent_id})", + className="mb-3", + ) + ] + + rows, + style={"padding": "10px"}, + ) diff --git a/agentlib/utils/multi_agent_system.py b/agentlib/utils/multi_agent_system.py index 2e96bd46..bc3f2f4a 100644 --- a/agentlib/utils/multi_agent_system.py +++ b/agentlib/utils/multi_agent_system.py @@ -104,13 +104,11 @@ def add_agent_logger(config: AgentConfig, sampling=60) -> AgentConfig: sampling= """ # Add Logger config - filename = f"variable_logs//Agent_{config.id}_Logger.log" cfg = { "module_id": "AgentLogger", "type": "AgentLogger", "t_sample": sampling, "values_only": True, - "filename": filename, "overwrite_log": True, "clean_up": False, } @@ -179,6 +177,42 @@ def terminate_agents(self): for agent in self._agents.values(): agent.terminate() + def show_results_dashboard( + self, cleanup_results: bool = False, block_main=True, live=False + ): + """ + Launches an interactive dashboard to visualize the results of the MAS. + + Args: + cleanup_results (bool): If True, result files might be cleaned up + by modules after being read for the dashboard. + Defaults to False to preserve results. + """ + logger.info("Launching MAS Results Dashboard...") + try: + from agentlib.utils.plotting.mas_dashboard import launch_mas_dashboard + except ImportError: + logger.error( + "Failed to import launch_mas_dashboard. Ensure Dash and its dependencies are installed: " + "pip install agentlib[interactive]" + ) + return + + results = self.get_results(cleanup=cleanup_results) + if not results: + logger.warning("No results found to display in the dashboard.") + return None # Return None if no dashboard is launched + + try: + # launch_mas_dashboard now returns the process + dashboard_process = launch_mas_dashboard( + mas=self, mas_results=results, block_main=block_main, live_update=live + ) + return dashboard_process + except Exception as e: + logger.error(f"Error launching MAS dashboard: {e}", exc_info=True) + return None # Return None on error + def setup_agent(self, id: str) -> Agent: """Setup the agent matching the given id""" # pylint: disable=redefined-builtin diff --git a/agentlib/utils/multi_processing_broker.py b/agentlib/utils/multi_processing_broker.py index fb91c8ad..4ceb3967 100644 --- a/agentlib/utils/multi_processing_broker.py +++ b/agentlib/utils/multi_processing_broker.py @@ -4,16 +4,16 @@ """ import json +import logging import multiprocessing -from ipaddress import IPv4Address -from multiprocessing.managers import SyncManager import threading import time from collections import namedtuple +from ipaddress import IPv4Address +from multiprocessing.managers import SyncManager +from pathlib import Path from typing import Union -import logging -from pathlib import Path from pydantic import BaseModel, Field, FilePath from .broker import Broker @@ -23,6 +23,9 @@ MPClient = namedtuple("MPClient", ["agent_id", "read", "write"]) Message = namedtuple("Message", ["agent_id", "payload"]) +MPSubscriptionClient = namedtuple( + "MPSubscriptionClient", ["agent_id", "read", "write", "subscriptions"] +) class BrokerManager(SyncManager): @@ -67,6 +70,10 @@ def __init__(self, config: ConfigTypes = None): self.config = MultiProcessingBrokerConfig() else: self.config = config + + logger.info( + f"Starting Multiprocessing Broker on {(self.config.ipv4, self.config.port)}" + ) server = multiprocessing.Process( target=self._server, name="Broker_Server", args=(self.config,), daemon=True ) @@ -185,3 +192,71 @@ def send(self, source, message): client.write.send(message) except BrokenPipeError: pass + + +class MultiProcessingSubscriptionBroker(MultiProcessingBroker): + """ + Subscription-based multiprocessing broker that only sends messages to + clients subscribed to the sender's agent_id. + """ + + def _signup_handler(self): + """Connects to the manager queue and processes the signup requests. Starts a + child thread listening to messages from each client.""" + from multiprocessing.managers import BaseManager + + class QueueManager(BaseManager): + pass + + QueueManager.register("get_queue") + m = QueueManager( + address=(self.config.ipv4, self.config.port), authkey=self.config.authkey + ) + started_wait = time.time() + while True: + try: + m.connect() + break + except ConnectionRefusedError: + time.sleep(0.01) + if time.time() - started_wait > 10: + raise RuntimeError("Could not connect to server.") + + signup_queue = m.get_queue() + + while True: + try: + client = signup_queue.get() + except ConnectionResetError: + logger.info("Multiprocessing Subscription Broker disconnected.") + break + + with self.lock: + self._clients.add(client) + + # send the client an ack its messages are now being received + client.write.send(1) + threading.Thread( + target=self._client_loop, + args=(client,), + name=f"MPSubBroker_{client.agent_id}", + daemon=True, + ).start() + + def send(self, source, message): + """ + Send the given message to subscribed clients only. + Args: + source: Source agent_id to match against subscriptions + message: The message to send + """ + # lock is required so the clients loop does not change size during + # iteration if clients are added or removed + with self.lock: + for client in list(self._clients): + # Check if this client is subscribed to the source agent + if source in client.subscriptions: + try: + client.write.send(message) + except BrokenPipeError: + pass diff --git a/agentlib/utils/plotting/mas_dashboard.py b/agentlib/utils/plotting/mas_dashboard.py new file mode 100644 index 00000000..f87da798 --- /dev/null +++ b/agentlib/utils/plotting/mas_dashboard.py @@ -0,0 +1,494 @@ +""" +Dashboard for visualizing results from a Multi-Agent System (MAS). +Allows selecting an agent and then a module within that agent to display +its specific results visualization. +Supports both static (post-run) and live (during run) modes. +Runs in a separate process to avoid blocking the main thread. +""" + +import importlib +import logging +import multiprocessing +import queue # For queue.Empty +import socket +import threading +import time +import webbrowser +from typing import Dict, Optional, Tuple +from agentlib.core.errors import OptionalDependencyError + +try: + import dash + from dash import dcc, html + from dash.dependencies import Input, Output, State + import dash_bootstrap_components as dbc +except ImportError: + raise OptionalDependencyError( + used_object=f"MAS Dashboard", + dependency_install="agentlib[interactive]", + dependency_name="interactive", + ) + +from agentlib.utils.multi_agent_system import LocalMASAgency + +logger = logging.getLogger(__name__) + +# Sentinel value to signal the IPC listener thread to stop +_STOP_IPC_LISTENER = object() + + +def _ipc_listener_loop( + mas_instance: LocalMASAgency, + request_q: multiprocessing.Queue, + response_q: multiprocessing.Queue, + stop_event: threading.Event, +): + """Listens for data requests from the Dash app and calls the appropriate module.""" + logger.info("IPC listener thread started for MAS dashboard.") + while not stop_event.is_set(): + try: + # Timeout to allow checking stop_event periodically + agent_id, module_id, update_token = request_q.get(timeout=0.1) + if agent_id is _STOP_IPC_LISTENER: + logger.info("IPC listener received stop signal.") + break + + logger.debug( + f"IPC listener: Request for {agent_id}/{module_id}, token: {update_token}" + ) + data_chunk, next_token = None, None + agent = mas_instance.get_agent(agent_id) + module = agent.get_module(module_id) + if hasattr(module, "get_results_incremental"): + data_chunk, next_token = module.get_results_incremental( + update_token=update_token + ) + else: + if update_token is None: # Only call get_results for initial load + data_chunk = module.get_results() + # next_token remains None, implying no further incremental updates from this fallback + response_q.put((data_chunk, next_token)) + except queue.Empty: + continue # Timeout, check stop_event again + time.sleep(0.1) # Avoid busy-looping + logger.info("IPC listener thread stopped for MAS dashboard.") + + +def create_dash_app( + agent_module_info: Dict, + live_update: bool, + static_results: Optional[Dict] = None, + ipc_queues: Optional[Tuple[multiprocessing.Queue, multiprocessing.Queue]] = None, + update_interval_sec: Optional[float] = None, +): + app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP]) + app.title = "MAS Results Dashboard" + + agent_ids = list(agent_module_info.keys()) + + # In-memory cache for DataFrames within the Dash app process + # This dict is accessible to callbacks defined within create_dash_app + app_data_cache = {} + + layout_components = [ + html.H1("Multi-Agent System Results Dashboard", className="my-4"), + dbc.Row( + [ + dbc.Col( + [ + html.Label("Select Agent:"), + dcc.Dropdown( + id="agent-selector", + options=[ + {"label": agent_id, "value": agent_id} + for agent_id in agent_ids + ], + value=agent_ids[0] if agent_ids else None, + clearable=False, + ), + ], + width=6, + ), + dbc.Col( + [ + html.Label("Select Module:"), + dcc.Dropdown( + id="module-selector", + options=[], # Populated by callback + clearable=False, + ), + ], + width=6, + ), + ], + className="mb-4", + ), + dbc.Row(dbc.Col(html.Div(id="module-visualization-content"))), + ] + + if live_update: + layout_components.extend( + [ + # dcc.Store(id='data-cache', data={}), # Removed: DataFrames not JSON serializable for Store + dcc.Store( + id="token-cache", data={} + ), # Stores { "agent/module": next_token } + dcc.Store( + id="data-update-trigger", data=0 + ), # Dummy store to trigger visualization updates + dcc.Interval( + id="live-update-interval", + interval=( + int(update_interval_sec * 1000) if update_interval_sec else 5000 + ), + n_intervals=0, + ), + ] + ) + + app.layout = dbc.Container(layout_components, fluid=True) + + @app.callback( + [Output("module-selector", "options"), Output("module-selector", "value")], + [Input("agent-selector", "value")], + ) + def update_module_dropdown(selected_agent_id): + if not selected_agent_id: + return [], None + + modules_for_agent = agent_module_info.get(selected_agent_id, []) + module_options = [] + first_module_id = None + + for module_info_dict in modules_for_agent: + module_id = module_info_dict["id"] + if live_update: + module_options.append({"label": module_id, "value": module_id}) + if first_module_id is None: + first_module_id = module_id + elif static_results: + agent_modules_data = static_results.get(selected_agent_id, {}) + if ( + module_id in agent_modules_data + ): # Check if results exist for this module + module_options.append({"label": module_id, "value": module_id}) + if first_module_id is None: + first_module_id = module_id + return module_options, first_module_id + + # Callback for fetching data in live mode + if live_update: + + @app.callback( + [Output("token-cache", "data"), Output("data-update-trigger", "data")], + [ + Input("live-update-interval", "n_intervals"), + Input("agent-selector", "value"), + Input("module-selector", "value"), + ], + [State("token-cache", "data"), State("data-update-trigger", "data")], + ) + def fetch_live_data( + n_intervals, + selected_agent_id, + selected_module_id, + current_token_cache, + current_trigger_val, + ): + ctx = dash.callback_context + triggered_prop_id = ( + ctx.triggered[0]["prop_id"] if ctx.triggered else "initial_load" + ) + + if not selected_agent_id or not selected_module_id or not ipc_queues: + return ( + current_token_cache, + dash.no_update, + ) + + request_q, response_q = ipc_queues + module_key = f"{selected_agent_id}/{selected_module_id}" + + # Check if agent/module selection changed - if so, clear cache for this module + if any( + trigger["prop_id"] in ["agent-selector.value", "module-selector.value"] + for trigger in ctx.triggered + if trigger["prop_id"] != "" + ): + app_data_cache[module_key] = None # Clear cache for new selection + update_token_to_send = None # Start fresh + else: + update_token_to_send = current_token_cache.get(module_key) + + logger.debug( + f"Dash app: Requesting data for {module_key} with token: {update_token_to_send}" + ) + request_q.put((selected_agent_id, selected_module_id, update_token_to_send)) + data_chunk, next_token = response_q.get( + timeout=(update_interval_sec * 0.9) if update_interval_sec else 4.5 + ) + logger.debug( + f"Dash app: Received data for {module_key}. Next token: {next_token}" + ) + + # Accumulate data in memory + if data_chunk is not None: + if ( + app_data_cache.get(module_key) is None + or update_token_to_send is None + ): + # First load or fresh start + app_data_cache[module_key] = data_chunk + else: + # Append new data to existing dataframe + existing_df = app_data_cache[module_key] + if hasattr(data_chunk, "index") and hasattr(existing_df, "index"): + # Both are DataFrames - concatenate + import pandas as pd + + app_data_cache[module_key] = pd.concat( + [existing_df, data_chunk] + ) + else: + # Handle other data types as needed + app_data_cache[module_key] = data_chunk + + new_token_cache = current_token_cache.copy() + new_token_cache[module_key] = next_token + + # Increment trigger to signal data update for visualization callback + new_trigger_val = ( + (current_trigger_val + 1) if current_trigger_val is not None else 0 + ) + return new_token_cache, new_trigger_val + + # Callback for displaying module visualization + @app.callback( + Output("module-visualization-content", "children"), + [ + Input("agent-selector", "value"), + Input("module-selector", "value"), + ( + Input("data-update-trigger", "data") + if live_update + else Input("module-selector", "value") + ), + ], + # The dummy Input for static mode (using module-selector again) is just to make Dash happy + # with the conditional Input list. The value of this dummy input isn't used in static mode for this logic. + ) + def display_module_visualization( + selected_agent_id, selected_module_id, data_update_trigger_val_or_dummy + ): + if not selected_agent_id or not selected_module_id: + return html.P("Select an agent and a module to view visualization.") + + current_results_data = None + if live_update: + module_key = f"{selected_agent_id}/{selected_module_id}" + # Read from the in-memory Python dictionary + current_results_data = app_data_cache.get(module_key) + + if current_results_data is None: + ctx = dash.callback_context + # Check if triggered by agent/module selection, if so, data is being fetched. + if ctx.triggered and any( + trigger["prop_id"] + in ["agent-selector.value", "module-selector.value"] + for trigger in ctx.triggered + ): + return html.P(f"Fetching live data for {selected_module_id}...") + return html.P( + f"No live data currently available for module '{selected_module_id}'." + ) + else: + current_results_data = static_results.get(selected_agent_id, {}).get( + selected_module_id, "__no_key__" + ) + + agent_modules_meta = agent_module_info.get(selected_agent_id, []) + module_info_dict = next( + (m for m in agent_modules_meta if m["id"] == selected_module_id), None + ) + class_path = module_info_dict["class_path"] + module_path_str, class_name_str = class_path.rsplit(".", 1) + imported_module_obj = importlib.import_module(module_path_str) + ModuleType = getattr(imported_module_obj, class_name_str) + + visualization_layout = ModuleType.visualize_results( + results_data=current_results_data, + module_id=selected_module_id, + agent_id=selected_agent_id, + ) + if visualization_layout is None: + return html.P( + f"Visualization not implemented or not available for module " + f"'{selected_module_id}' (type: {ModuleType.__name__}) in agent '{selected_agent_id}'." + ) + return visualization_layout + + return app + + +def _run_dash_server( + app_factory_func, # This is create_dash_app + # Args for create_dash_app: + agent_module_info_arg: Dict, + live_update_arg: bool, + static_results_arg: Optional[Dict], + ipc_queues_arg: Optional[Tuple[multiprocessing.Queue, multiprocessing.Queue]], + update_interval_sec_arg: Optional[float], + # Args for app.run_server: + port: int, + host: str, +): + """ + Helper function to create and run the Dash app in a separate process. + """ + app = app_factory_func( + agent_module_info=agent_module_info_arg, + live_update=live_update_arg, + static_results=static_results_arg, + ipc_queues=ipc_queues_arg, + update_interval_sec=update_interval_sec_arg, + ) + logger.info(f"Dash app server starting on http://{host}:{port}") + app.run(debug=False, port=port, host=host, use_reloader=False) + + +def get_free_port(): + """Get an available port by trying to bind to a socket.""" + port = 8050 + while True: + port = 8050 + while True: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + is_free = s.connect_ex(("localhost", port)) != 0 + if is_free: + return port + else: + port += 1 + if port > 9000: # Avoid infinite loop in rare cases + raise OSError("Could not find a free port between 8050 and 9000.") + + +def launch_mas_dashboard( + mas: LocalMASAgency, + mas_results: Optional[dict] = None, + block_main: bool = False, + live_update: bool = False, + update_interval_sec: float = 5.0, +) -> multiprocessing.Process | None: + """ + Launches the MAS results dashboard. + + Args: + mas: The LocalMASAgency instance. Required for agent_module_info, + and for live_update mode. + mas_results: A dictionary of results for static mode. + Expected: {agent_id: {module_id: results_data}}. + Ignored if live_update is True. + block_main: If True, runs Dash server in the main process, blocking it. + If False, runs in a separate process. + live_update: If True, dashboard attempts to fetch live data from modules. + update_interval_sec: Interval for polling data in live_update mode. + Returns: + multiprocessing.Process: The process running the Dash dashboard (if not block_main). + None otherwise or if an error occurs. + """ + if not live_update and mas_results is None: + raise ValueError("mas_results must be provided if not in live_update mode.") + if live_update: + block_main = False + + agent_module_info_for_app = {} + if hasattr(mas, "_agents"): + for agent_id, agent_instance in mas._agents.items(): + agent_module_info_for_app[agent_id] = [] + if hasattr(agent_instance, "modules"): + for module_instance in agent_instance.modules: + module_class = module_instance.__class__ + class_path = f"{module_class.__module__}.{module_class.__name__}" + agent_module_info_for_app[agent_id].append( + {"id": module_instance.id, "class_path": class_path} + ) + if not agent_module_info_for_app: + logger.warning( + "No agents or modules found in the MAS instance. Dashboard might be empty." + ) + + port = get_free_port() + host = "0.0.0.0" + + factory_args_for_create_dash_app = { + "agent_module_info": agent_module_info_for_app, + "live_update": live_update, + "static_results": None if live_update else mas_results, + "ipc_queues": None, + "update_interval_sec": update_interval_sec if live_update else None, + } + + ipc_listener_thread_obj = None # To store the thread object + stop_ipc_event_obj = None # To store the event object + + if live_update: + request_q = multiprocessing.Queue() + response_q = multiprocessing.Queue() + factory_args_for_create_dash_app["ipc_queues"] = (request_q, response_q) + + stop_ipc_event_obj = threading.Event() + ipc_listener_thread_obj = threading.Thread( + target=_ipc_listener_loop, + args=(mas, request_q, response_q, stop_ipc_event_obj), + daemon=True, + ) + ipc_listener_thread_obj.start() + + webbrowser.open_new_tab(f"http://localhost:{port}") + + process_args_for_run_server = ( + create_dash_app, # app_factory_func + factory_args_for_create_dash_app["agent_module_info"], + factory_args_for_create_dash_app["live_update"], + factory_args_for_create_dash_app["static_results"], + factory_args_for_create_dash_app["ipc_queues"], + factory_args_for_create_dash_app["update_interval_sec"], + port, + host, + ) + + dashboard_process = None + if not block_main: + ctx = multiprocessing.get_context("spawn") + dashboard_process = ctx.Process( + target=_run_dash_server, args=process_args_for_run_server + ) + dashboard_process.daemon = False # So main program can wait for it if needed + dashboard_process.start() + logger.info( + f"MAS Dashboard is launching in a separate process on http://localhost:{port}" + ) + # The caller might need to manage stop_ipc_event_obj and ipc_listener_thread_obj + # if they want to explicitly stop the IPC listener when the dashboard_process is terminated. + # However, as a daemon thread, it will exit with the main process. + return dashboard_process + else: + # Run in main process + _run_dash_server(*process_args_for_run_server) + if live_update and stop_ipc_event_obj and ipc_listener_thread_obj: + logger.info( + "Main process dashboard finished. Attempting to stop IPC listener." + ) + stop_ipc_event_obj.set() + # Send a stop signal through the queue as well to break out of q.get() + if factory_args_for_create_dash_app["ipc_queues"]: + try: + factory_args_for_create_dash_app["ipc_queues"][0].put( + (_STOP_IPC_LISTENER, None, None), timeout=0.1 + ) + except Exception: # pragma: no cover + pass + ipc_listener_thread_obj.join(timeout=1.0) + if ipc_listener_thread_obj.is_alive(): + logger.warning("IPC listener thread did not stop in time.") + return None diff --git a/examples/multi-agent-systems/pingpong/pingpong_multiprocessing_subscription.py b/examples/multi-agent-systems/pingpong/pingpong_multiprocessing_subscription.py new file mode 100644 index 00000000..4c621fa8 --- /dev/null +++ b/examples/multi-agent-systems/pingpong/pingpong_multiprocessing_subscription.py @@ -0,0 +1,60 @@ +""" +Example of a local multi-agent-system consisting of two agents playing ping +pong. To set up a multi-agent-system by hand, the following steps are required: + 1. Specify the environment configuration + 2. Specify the config of all agents + 3. Create the environment and the agents + 4. Run the simulation by calling the run() method of the environment. +""" + +import logging + +import pingpong_module +from agentlib.core import Environment, Agent +from agentlib.utils.multi_processing_broker import MultiProcessingSubscriptionBroker + +logging.basicConfig(level="DEBUG") +logger = logging.getLogger(__name__) + + +env_config = {"rt": True, "factor": 1} +agent_config1 = { + "id": "FirstAgent", + "modules": [ + { + "module_id": "Ag1Com", + "type": "multiprocessing_subscription", + "port": 50001, + "subscriptions": ["SecondAgent"], + }, + { + "module_id": "Ping", + "type": {"file": pingpong_module.__file__, "class_name": "PingPong"}, + "start": True, + "initial_wait": 2, + }, + ], +} +agent_config2 = { + "id": "SecondAgent", + "modules": [ + { + "module_id": "Ag2Com", + "type": "multiprocessing_subscription", + "port": 50001, + "subscriptions": ["FirstAgent"], + }, + { + "module_id": "Pong", + "type": {"file": pingpong_module.__file__, "class_name": "PingPong"}, + }, + ], +} + + +if __name__ == "__main__": + broker = MultiProcessingSubscriptionBroker(config={"port": 50001}) + env = Environment(config=env_config) + agent1 = Agent(config=agent_config1, env=env) + agent2 = Agent(config=agent_config2, env=env) + env.run(until=None) diff --git a/examples/multi-agent-systems/room_mas/configs/PIDAgent.json b/examples/multi-agent-systems/room_mas/configs/PIDAgent.json index 212de599..9430ffd8 100644 --- a/examples/multi-agent-systems/room_mas/configs/PIDAgent.json +++ b/examples/multi-agent-systems/room_mas/configs/PIDAgent.json @@ -25,6 +25,7 @@ "module_id": "ComLocal", "type": "local", "subscriptions": ["Room1Agent"], + "communication_log_level": "basic", "parse_json": true } ] diff --git a/examples/multi-agent-systems/room_mas/configs/Room1.json b/examples/multi-agent-systems/room_mas/configs/Room1.json index 7b29ce6a..62c05741 100644 --- a/examples/multi-agent-systems/room_mas/configs/Room1.json +++ b/examples/multi-agent-systems/room_mas/configs/Room1.json @@ -21,7 +21,8 @@ { "name": "T_oda", "value": 273.15 - }], + } + ], "outputs": [ { "name": "T_air" @@ -31,8 +32,12 @@ { "module_id": "ComLocal", "type": "local", - "subscriptions": ["PIDAgent", "TRYSensorAgent"], - "parse_json": true + "subscriptions": [ + "PIDAgent", + "TRYSensorAgent" + ], + "parse_json": true, + "communication_log_level": "detail" } ] } \ No newline at end of file diff --git a/examples/multi-agent-systems/room_mas/room_mas.py b/examples/multi-agent-systems/room_mas/room_mas.py index dc44306d..e76ab989 100644 --- a/examples/multi-agent-systems/room_mas/room_mas.py +++ b/examples/multi-agent-systems/room_mas/room_mas.py @@ -8,7 +8,13 @@ logger = logging.getLogger(__name__) -def run_example(until, with_plots=True, log_level=logging.INFO, use_direct_callback_databroker: bool = False): +def run_example( + until, + with_plots=True, + log_level=logging.INFO, + use_direct_callback_databroker: bool = False, + with_dashboard: Union[bool, Literal["live", "after"]] = "after", +): """ Runs a multi-agent system (MAS) simulation with heating control agents. @@ -26,7 +32,7 @@ def run_example(until, with_plots=True, log_level=logging.INFO, use_direct_callb Logging verbosity level. use_direct_callback_databroker : bool, default=False Whether to use direct callback execution in the databroker instead of queue-based callbacks. - + with_dashboard: bool, whether to use dashboard or not """ # Start by setting the log-level logging.basicConfig(level=log_level) @@ -49,9 +55,12 @@ def run_example(until, with_plots=True, log_level=logging.INFO, use_direct_callb use_direct_callback_databroker=use_direct_callback_databroker ) # Simulate + if with_dashboard == "live": + mas.show_results_dashboard(live=True) mas.run(until=until) - # Load results: - results = mas.get_results(cleanup=True) + if with_dashboard: + mas.show_results_dashboard(cleanup_results=False, block_main=False, live=False) + results = mas.get_results(cleanup=False) if not with_plots: return results @@ -88,7 +97,7 @@ def run_example(until, with_plots=True, log_level=logging.INFO, use_direct_callb axes[2, 1].plot(df_se["T_oda"], color="green", linestyle="-.", label="SensorAgent") # Legend, titles etc: axes[0, 0].set_ylabel("$T_{Room}$ / K") - axes[1, 0].set_ylabel("$\dot{Q}_{Room\,,in}$ / W") + axes[1, 0].set_ylabel(r"$\dot{Q}_{Room\,,in}$ / W") axes[2, 0].set_ylabel("$T_{oda}$ / K") axes[2, 0].set_xlabel("Time / s") axes[2, 1].set_xlabel("Time / s") @@ -116,8 +125,9 @@ def run_example(until, with_plots=True, log_level=logging.INFO, use_direct_callb ) plt.show() + return results if __name__ == "__main__": - run_example(until=86400 / 10, with_plots=True) + run_example(until=86400 / 10, with_plots=True, with_dashboard=True) diff --git a/examples/simulation/csv_data_source.py b/examples/simulation/csv_data_source.py index 75375061..01b9aa01 100644 --- a/examples/simulation/csv_data_source.py +++ b/examples/simulation/csv_data_source.py @@ -10,7 +10,7 @@ def create_sample_data(): date_today = datetime.now() - time = pd.date_range(date_today, date_today + timedelta(minutes=10), freq="30S") + time = pd.date_range(date_today, date_today + timedelta(minutes=10), freq="30s") data1 = np.sin(np.linspace(0, 2 * np.pi, len(time))) * 10 + 20 data2 = np.cos(np.linspace(0, 2 * np.pi, len(time))) * 5 + 15 df = pd.DataFrame({"timestamp": time, "temperature": data1, "humidity": data2}) @@ -19,7 +19,9 @@ def create_sample_data(): return df -def run_example(log_level=logging.INFO, with_plots: bool = True, extrapolation: str = "constant"): +def run_example( + log_level=logging.INFO, with_plots: bool = True, extrapolation: str = "constant" +): # Set the log-level logging.basicConfig(level=log_level) diff --git a/tests/test_communicator.py b/tests/test_communicator.py index a2848966..17bcfd23 100644 --- a/tests/test_communicator.py +++ b/tests/test_communicator.py @@ -1,4 +1,7 @@ import unittest +import tempfile +import os +from pathlib import Path import pandas as pd @@ -58,13 +61,145 @@ def test_pd_series(self): pd.testing.assert_series_equal(variable.value, variable2.value) # communicator without json parsing - payload = comm_no_parse.short_dict( - variable, parse_json=comm_no_parse.config.parse_json - ) + payload = comm_no_parse.short_dict(variable) payload["name"] = payload["alias"] variable2 = AgentVariable(**payload) pd.testing.assert_series_equal(variable.value, variable2.value) + def test_logging_none(self): + """Test that 'none' log level produces no results""" + config = {**self.test_config, "communication_log_level": "none"} + comm = LocalClient(config=config, agent=self.agent_send) + + # Trigger some sends and receives + variable = AgentVariable(**default_data) + comm._send_only_shared_variables(variable) + comm._handle_received_variable(variable, remote_agent_id="RemoteAgent") + + # Should return None + results = comm.get_results() + self.assertIsNone(results) + + def test_logging_basic(self): + """Test that 'basic' log level produces count dictionaries""" + config = {**self.test_config, "communication_log_level": "basic"} + comm = LocalClient(config=config, agent=self.agent_send) + + # Send multiple variables with different aliases + var1 = AgentVariable(**{**default_data, "name": "var1", "alias": "alias1"}) + var2 = AgentVariable(**{**default_data, "name": "var2", "alias": "alias2"}) + var3 = AgentVariable(**{**default_data, "name": "var1", "alias": "alias1"}) + + comm._send_only_shared_variables(var1) + comm._send_only_shared_variables(var2) + comm._send_only_shared_variables(var3) # alias1 sent twice + + # Receive variables from different sources + comm._handle_received_variable(var1, remote_agent_id="Agent1") + comm._handle_received_variable(var2, remote_agent_id="Agent2") + comm._handle_received_variable(var1, remote_agent_id="Agent1") # Agent1/alias1 twice + + results = comm.get_results() + + # Check structure + self.assertIsInstance(results, dict) + self.assertIn("sent_counts", results) + self.assertIn("received_counts", results) + + # Check sent counts + self.assertEqual(results["sent_counts"]["alias1"], 2) + self.assertEqual(results["sent_counts"]["alias2"], 1) + + # Check received counts (keys are strings of tuples) + received = results["received_counts"] + self.assertEqual(received["('Agent1', 'alias1')"], 2) + self.assertEqual(received["('Agent2', 'alias2')"], 1) + + def test_logging_detail(self): + """Test that 'detail' log level produces a DataFrame with timeline data""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "test_comm.jsonl") + config = { + **self.test_config, + "communication_log_level": "detail", + "communication_log_file": log_file, + } + comm = LocalClient(config=config, agent=self.agent_send) + + # Send and receive some variables + var1 = AgentVariable(**{**default_data, "name": "var1", "alias": "temp_sensor"}) + var2 = AgentVariable(**{**default_data, "name": "var2", "alias": "pressure_sensor"}) + + comm._send_only_shared_variables(var1) + comm._send_only_shared_variables(var2) + comm._handle_received_variable(var1, remote_agent_id="SensorAgent1") + comm._handle_received_variable(var2, remote_agent_id="SensorAgent2") + + # Get results + results = comm.get_results() + + # Check it's a DataFrame + self.assertIsInstance(results, pd.DataFrame) + + # Check structure + self.assertEqual(len(results), 4) # 2 sent + 2 received + self.assertIn("timestamp", results.columns) + self.assertIn("direction", results.columns) + self.assertIn("alias", results.columns) + self.assertIn("own_agent_id", results.columns) + self.assertIn("remote_agent_id", results.columns) + + # Check sent messages + sent_df = results[results["direction"] == "sent"] + self.assertEqual(len(sent_df), 2) + self.assertIn("temp_sensor", sent_df["alias"].values) + self.assertIn("pressure_sensor", sent_df["alias"].values) + + # Check received messages + received_df = results[results["direction"] == "received"] + self.assertEqual(len(received_df), 2) + self.assertIn("SensorAgent1", received_df["remote_agent_id"].values) + self.assertIn("SensorAgent2", received_df["remote_agent_id"].values) + + # Cleanup + comm.cleanup_results() + self.assertFalse(Path(log_file).exists()) + + def test_logging_detail_incremental(self): + """Test incremental result retrieval for detail logging""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "test_comm_incremental.jsonl") + config = { + **self.test_config, + "communication_log_level": "detail", + "communication_log_file": log_file, + } + comm = LocalClient(config=config, agent=self.agent_send) + + # Send first batch + var1 = AgentVariable(**{**default_data, "name": "var1", "alias": "alias1"}) + comm._send_only_shared_variables(var1) + + # Get initial results + results1, token1 = comm.get_results_incremental(update_token=None) + self.assertIsInstance(results1, pd.DataFrame) + self.assertEqual(len(results1), 1) + self.assertIsNotNone(token1) + + # Send second batch + var2 = AgentVariable(**{**default_data, "name": "var2", "alias": "alias2"}) + comm._send_only_shared_variables(var2) + comm._handle_received_variable(var1, remote_agent_id="Agent1") + + # Get incremental results + results2, token2 = comm.get_results_incremental(update_token=token1) + self.assertIsInstance(results2, pd.DataFrame) + self.assertEqual(len(results2), 2) # 1 sent + 1 received + self.assertEqual(token2, token1 + 2) + + # Cleanup + comm.cleanup_results() + def test_pd_series_no_json(self): """Tests whether pandas series are sent correctly""" data = {**default_data, "value": pd.Series({0: 1, 10: 2}), "type": "pd.Series"} @@ -72,9 +207,9 @@ def test_pd_series_no_json(self): _config = self.test_config.copy() _config["parse_json"] = False comm = LocalClient(config=_config, agent=self.agent_send) - payload = comm.short_dict(variable, parse_json=comm.config.parse_json) + payload = comm.short_dict(variable) pd.testing.assert_series_equal(variable.value, payload["value"]) if __name__ == "__main__": - unittest.main() + unittest.main() \ No newline at end of file diff --git a/tests/test_examples.py b/tests/test_examples.py index 00a18c99..10353c45 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -1,6 +1,8 @@ """Module with functions to test the if examples are executable.""" +import shutil +import tempfile import unittest import os import subprocess @@ -17,14 +19,18 @@ class TestExamples(unittest.TestCase): """Test all examples inside the agentlib""" def setUp(self) -> None: - self.timeout = 10 # Seconds which the script is allowed to run + self.timeout = 10 self.main_cwd = os.getcwd() + # Create a unique temp directory for each test + self.temp_dir = tempfile.mkdtemp(prefix="agentlib_test_") + os.chdir(self.temp_dir) def tearDown(self) -> None: broker = LocalBroadcastBroker() broker.delete_all_clients() - # Change back cwd: os.chdir(self.main_cwd) + # Clean up the temporary directory + shutil.rmtree(self.temp_dir, ignore_errors=True) def _run_example(self, example, timeout=None): if timeout is None: @@ -65,7 +71,7 @@ def test_room_mas(self): until=8640, with_plots=False, log_level=logging.DEBUG, - use_direct_callback_databroker=False + use_direct_callback_databroker=False, ) def test_room_mas_direct_callback(self): @@ -80,7 +86,7 @@ def test_room_mas_direct_callback(self): until=8640, with_plots=False, log_level=logging.DEBUG, - use_direct_callback_databroker=True + use_direct_callback_databroker=True, ) def test_pingpong(self): @@ -116,21 +122,21 @@ def test_csv_data_source(self): func_name="run_example", with_plots=False, log_level=logging.FATAL, - extrapolation="constant" + extrapolation="constant", ) self._run_example_with_return( file="simulation//csv_data_source.py", func_name="run_example", with_plots=False, log_level=logging.FATAL, - extrapolation="repeat" + extrapolation="repeat", ) self._run_example_with_return( file="simulation//csv_data_source.py", func_name="run_example", with_plots=False, log_level=logging.FATAL, - extrapolation="backwards" + extrapolation="backwards", ) def test_scipy_model(self): diff --git a/tests/test_n_pingpong.py b/tests/test_n_pingpong.py index e69085a5..91205913 100644 --- a/tests/test_n_pingpong.py +++ b/tests/test_n_pingpong.py @@ -3,17 +3,21 @@ """ import itertools -import unittest import time -from pydantic import Field +import unittest + import numpy as np +from pydantic import Field + from agentlib.core import BaseModule, Agent, BaseModuleConfig from agentlib.core.datamodels import AgentVariable from agentlib.utils import MultiProcessingBroker, LocalBroadcastBroker, LocalBroker from agentlib.utils.multi_agent_system import LocalMASAgency +from agentlib.utils.multi_processing_broker import MultiProcessingSubscriptionBroker # pylint: disable=missing-module-docstring,missing-class-docstring PORT = 39920 +SUBSCRIPTION_PORT = 39921 # Different port for subscription broker class NPingPongConfig(BaseModuleConfig): @@ -64,7 +68,7 @@ def register_callbacks(self): if self.id == f"Pong_{self.n}": alias = f"Ping_{self.n}" elif self.id == f"Ping_{self.n}": - alias = f"Pong_{self.n-1}" + alias = f"Pong_{self.n - 1}" self.agent.data_broker.register_callback( alias=alias, source=None, callback=self._callback ) @@ -120,6 +124,14 @@ def test_multiprocessing_broadcast(self): ) as mas: mas.run(until=1) + def test_multiprocessing_subscription(self): + """Test the NPingPong system using multiprocessing subscription as communicator""" + broker = MultiProcessingSubscriptionBroker(config={"port": SUBSCRIPTION_PORT}) + with self._create_2n_pingpong_agents( + com_type="multiprocessing_subscription" + ) as mas: + mas.run(until=1) + @unittest.skip("MQTT refuses connection in ci") def test_mqtt(self): """Test the NPingPong system using local_broadcast as communicator""" @@ -142,24 +154,28 @@ def _create_2n_pingpong_agents( "type": com_type, } + # Add subscriptions for subscription-based communicators if com_type not in ["local_broadcast", "multiprocessing_broadcast"]: _com_ping["subscriptions"] = [f"AgPong_{_n - 1 if _n > 1 else self.n}"] _com_pong["subscriptions"] = [f"AgPing_{_n}"] pong_module = {"module_id": f"Pong_{_n}", "type": _pingpong_module, "n": _n} ping_module = {"module_id": f"Ping_{_n}", "type": _pingpong_module, "n": _n} + if com_type == "mqtt": _com_ping.update({"url": "mqtt://test.mosquitto.org"}) _com_pong.update({"url": "mqtt://test.mosquitto.org"}) - if com_type == "multiprocessing_broadcast": + elif com_type == "multiprocessing_broadcast": _com_ping.update({"port": PORT}) _com_pong.update({"port": PORT}) ping_module["initial_wait"] = 3 + elif com_type == "multiprocessing_subscription": + _com_ping.update({"port": SUBSCRIPTION_PORT}) + _com_pong.update({"port": SUBSCRIPTION_PORT}) + ping_module["initial_wait"] = 3 elif com_type in ["local_broadcast", "local"]: _com_ping.update({"parse_json": parse_json}) _com_pong.update({"parse_json": parse_json}) - elif com_type in ["multiprocessing_broadcast"]: - pass # No subs needed else: raise TypeError(f"Com_type '{com_type}' not supported")