58 make a standard dashboard for mas results

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.

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`` .
 

agentlib/__init__.py

@@ -12,7 +12,7 @@
     LocalCloneMAPAgency,
 )
 
-__version__ = "0.8.9"
+__version__ = "0.9.0"
 
 __all__ = [
     "core",

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))
 

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"<INIT>"
+            record.env_time = "<INIT>"
         elif _until is None:
             record.env_time = _time
         else:

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.

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)"""

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",
+    ),
 }