problems with to_dict() functions working with mypy.

simulation-system/libs/csle-common/src/csle_common/controllers/container_controller.py

@@ -517,22 +517,24 @@ def stop_docker_stats_thread(execution: Union[None, EmulationExecution], physica
         :return: None
         """
         if not ManagementSystemController.is_statsmanager_running():
-            ManagementSystemController.start_docker_stats_manager(
-                logger=logger,
-                port=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_port,
-                log_file=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_log_file,
-                log_dir=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_log_dir,
-                max_workers=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_max_workers
-            )
+            if execution is not None:
+                ManagementSystemController.start_docker_stats_manager(
+                    logger=logger,
+                    port=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_port,
+                    log_file=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_log_file,
+                    log_dir=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_log_dir,
+                    max_workers=execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_max_workers
+                )
             time.sleep(5)
-        ip = physical_server_ip
-        with grpc.insecure_channel(
-                f'{ip}:'
-                f'{execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_port}',
-                options=constants.GRPC_SERVERS.GRPC_OPTIONS) as channel:
-            stub = csle_collector.docker_stats_manager.docker_stats_manager_pb2_grpc.DockerStatsManagerStub(channel)
-            csle_collector.docker_stats_manager.query_docker_stats_manager.stop_docker_stats_monitor(
-                stub=stub, emulation=execution.emulation_name, execution_first_ip_octet=execution.ip_first_octet)
+        if execution is not None:
+            ip = physical_server_ip
+            with grpc.insecure_channel(
+                    f'{ip}:'
+                    f'{execution.emulation_env_config.docker_stats_manager_config.docker_stats_manager_port}',
+                    options=constants.GRPC_SERVERS.GRPC_OPTIONS) as channel:
+                stub = csle_collector.docker_stats_manager.docker_stats_manager_pb2_grpc.DockerStatsManagerStub(channel)
+                csle_collector.docker_stats_manager.query_docker_stats_manager.stop_docker_stats_monitor(
+                    stub=stub, emulation=execution.emulation_name, execution_first_ip_octet=execution.ip_first_octet)
 
     @staticmethod
     def get_docker_stats_manager_status(docker_stats_manager_config: DockerStatsManagerConfig) \

simulation-system/libs/csle-common/src/csle_common/dao/emulation_config/emulation_statistics_windowed.py

@@ -1,4 +1,4 @@
-from typing import Dict, Any
+from typing import Dict, Any, List
 from csle_common.dao.emulation_config.emulation_env_state import EmulationEnvState
 from csle_common.dao.system_identification.emulation_statistics import EmulationStatistics
 from csle_common.dao.emulation_action.attacker.emulation_attacker_action import EmulationAttackerAction
@@ -21,8 +21,8 @@ def __init__(self, window_size: int, emulation_name: str, descr: str):
         :param descr: the description
         """
         self.window_size = window_size
-        self.initial_states = []
-        self.state_transitions = []
+        self.initial_states: List[Any] = []
+        self.state_transitions: List[Any] = []
         self.emulation_name = emulation_name
         self.descr = descr
         self.emulation_statistics = EmulationStatistics(emulation_name=self.emulation_name, descr=self.descr)
@@ -93,7 +93,7 @@ def from_dict(d: Dict[str, Any]) -> "EmulationStatisticsWindowed":
         obj.statistics_id = d["statistics_id"]
         return obj
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self):# -> Dict[str, Any]:
         """
         Converts the object to a dict representation
 

simulation-system/libs/csle-common/src/csle_common/dao/emulation_config/out/csle_common/dao/emulation_config/emulation_statistics_windowed.pyi

@@ -0,0 +1,26 @@
+from _typeshed import Incomplete
+from csle_base.json_serializable import JSONSerializable
+from csle_common.dao.emulation_action.attacker.emulation_attacker_action import EmulationAttackerAction as EmulationAttackerAction
+from csle_common.dao.emulation_action.defender.emulation_defender_action import EmulationDefenderAction as EmulationDefenderAction
+from csle_common.dao.emulation_config.emulation_env_state import EmulationEnvState as EmulationEnvState
+from csle_common.dao.system_identification.emulation_statistics import EmulationStatistics as EmulationStatistics
+from csle_common.metastore.metastore_facade import MetastoreFacade as MetastoreFacade
+from typing import Any, Dict
+
+class EmulationStatisticsWindowed(JSONSerializable):
+    window_size: Incomplete
+    initial_states: Incomplete
+    state_transitions: Incomplete
+    emulation_name: Incomplete
+    descr: Incomplete
+    emulation_statistics: Incomplete
+    statistics_id: Incomplete
+    def __init__(self, window_size: int, emulation_name: str, descr: str) -> None: ...
+    def add_initial_state(self, s: EmulationEnvState) -> None: ...
+    def add_state_transition(self, s: EmulationEnvState, s_prime: EmulationEnvState, a1: EmulationDefenderAction, a2: EmulationAttackerAction) -> None: ...
+    def update_emulation_statistics(self) -> None: ...
+    @staticmethod
+    def from_dict(d: Dict[str, Any]) -> EmulationStatisticsWindowed: ...
+    def to_dict(self) -> Dict[str, Any]: ...
+    @staticmethod
+    def from_json_file(json_file_path: str) -> EmulationStatisticsWindowed: ...