(add): stats monitoring schema

Author: kenya-sk

stats-monitoring/monitoring/generate_animation.py

@@ -11,22 +11,23 @@
 from monitoring.frame import load_one_hour_density, set_frame
 from monitoring.logger import logger
 from monitoring.position_distribution import load_current_coordinate, set_histogram
+from monitoring.schema import StatsData
 from monitoring.stats_metrics import load_statistics, set_stats_metrics
 
 
 def update(
     frame_num: int,
     cfg: MonitoringConfig,
     axs: List,
-    stats_dict: Dict,
+    stats_data: StatsData,
 ) -> None:
     """Update function for movie frame
 
     Args:
         frame_num (int): current frame number
         cfg (MonitoringConfig): config for monitoring environment
         axs (List): list of matplotlib axis
-        stats_dict (Dict): dictionary containing each stats array
+        stats_data (StatsData): instance of each statistics data
     """
     # clear all axis
     for ax in axs:
@@ -50,7 +51,7 @@ def update(
     set_frame(cfg, frame_num, detected_coordinate_df, one_hour_density_df, axs[0])
 
     # set statistics metrics
-    set_stats_metrics(cfg, frame_num, stats_dict, axs[3], axs[4])
+    set_stats_metrics(cfg, frame_num, stats_data, axs[3], axs[4])
 
 
 def generate_animation(cfg: MonitoringConfig, fig: Figure, axs: List) -> None:
@@ -62,15 +63,15 @@ def generate_animation(cfg: MonitoringConfig, fig: Figure, axs: List) -> None:
         axs (List): list of matplotlib axis
     """
     logger.info("[START] Load Statistics Data ...")
-    stats_dict = load_statistics(cfg)
+    stats_data = load_statistics(cfg)
     logger.info("------------ [DONE] ------------")
 
     anim = FuncAnimation(
         fig,
         update,
         frames=trange(cfg.animation.frame_number),
         interval=cfg.animation.interval,
-        fargs=(cfg, axs, stats_dict),
+        fargs=(cfg, axs, stats_data),
     )
     save_movie_path = str(DATA_DIR / cfg.path.save_movie_path)
     anim.save(save_movie_path, writer=cfg.animation.format)

stats-monitoring/monitoring/schema.py

@@ -0,0 +1,29 @@
+from typing import Annotated, Any
+
+import numpy as np
+from pydantic import BaseModel, PlainSerializer, PlainValidator
+
+
+def validate(v: Any) -> np.ndarray:
+    if isinstance(v, np.ndarray):
+        return v
+    else:
+        raise TypeError(f"Expected numpy array, got {type(v)}")
+
+
+def serialize(v: np.ndarray) -> list[list[float]]:
+    return v.tolist()
+
+
+DataArray = Annotated[
+    np.ndarray,
+    PlainValidator(validate),
+    PlainSerializer(serialize),
+]
+
+
+class StatsData(BaseModel):
+    mean: DataArray
+    past_mean: DataArray
+    acceleration: DataArray
+    past_acceleration: DataArray

stats-monitoring/monitoring/stats_metrics.py

@@ -1,6 +1,5 @@
 import os
 from pathlib import Path
-from typing import Dict
 
 import numpy as np
 from matplotlib.axes import Axes
@@ -13,18 +12,19 @@
     LABEL_IDX_MAX,
     TIME_LABELS,
 )
-from monitoring.exceptions import PathNotExistError, StatsKeyNotExistError
+from monitoring.exceptions import PathNotExistError
 from monitoring.logger import logger
+from monitoring.schema import StatsData
 
 
-def load_statistics(cfg: MonitoringConfig) -> Dict[str, np.ndarray]:
+def load_statistics(cfg: MonitoringConfig) -> StatsData:
     """Load the statistics used in the monitoring environment
 
     Args:
         cfg (MonitoringConfig): config for monitoring environment
 
     Returns:
-        dict: dictionary containing each stats array
+        StatsData: instance of each statistics data
     """
 
     def load_value(path: Path) -> np.ndarray:
@@ -43,38 +43,20 @@ def load_value(path: Path) -> np.ndarray:
             logger.error(message)
             raise PathNotExistError(message)
 
-    stats_dict = {
-        "mean": load_value(DATA_DIR / cfg.path.mean_speed_path),
-        "past_mean": load_value(DATA_DIR / cfg.path.past_mean_speed_path),
-        "acceleration": load_value(DATA_DIR / cfg.path.acceleration_count_path),
-        "past_acceleration": load_value(DATA_DIR / cfg.path.past_acceleration_count_path),
-    }
+    stats_data = StatsData(
+        mean=load_value(DATA_DIR / cfg.path.mean_speed_path),
+        past_mean=load_value(DATA_DIR / cfg.path.past_mean_speed_path),
+        acceleration=load_value(DATA_DIR / cfg.path.acceleration_count_path),
+        past_acceleration=load_value(DATA_DIR / cfg.path.past_acceleration_count_path),
+    )
 
-    return stats_dict
-
-
-def load_array(stats_dict: Dict[str, np.ndarray], key: str) -> np.ndarray:
-    """Load array from statistics dictionary
-
-    Args:
-        stats_dict (Dict): dictionary containing each stats array
-        key (str): dictionary key
-
-    Returns:
-        np.ndarray: target stats array
-    """
-    if key in stats_dict.keys():
-        return stats_dict[key]
-    else:
-        message = f'key="{key}" is not exist in stats_dict.'
-        logger.error(message)
-        raise StatsKeyNotExistError(message)
+    return stats_data
 
 
 def set_stats_metrics(
     cfg: MonitoringConfig,
     frame_num: int,
-    stats_dict: Dict[str, np.ndarray],
+    stats_data: StatsData,
     mean_ax: Axes,
     acc_ax: Axes,
 ) -> None:
@@ -83,11 +65,11 @@ def set_stats_metrics(
     Args:
         cfg (MonitoringConfig): config for monitoring environment
         frame_num (int): current frame number
-        stats_dict (Dict): dictionary containing each stats array
+        stats_data (StatsData): instance of each statistics data
         mean_ax (Axes): matplotlib figure axis of mean speed
         acc_ax (Axes): matplotlib figure axis of acceleration count
     """
-    mean_arr = load_array(stats_dict, "mean")
+    mean_arr = stats_data.mean
     x = [i for i in range(len(mean_arr))]
 
     # plot mean speed
@@ -99,7 +81,7 @@ def set_stats_metrics(
     mean_ax.axvline(frame_num, 0, 100, color="black", linestyle="dashed")
 
     # plot cumulate acceleration count
-    acc_arr = load_array(stats_dict, "acceleration")
+    acc_arr = stats_data.acceleration
     acc_ax.plot(x[: frame_num + 1], acc_arr[: frame_num + 1])
     acc_ax.set_xlim(0, LABEL_IDX_MAX)
     acc_ax.set_ylim(0, cfg.statistics.acceleration_max)

stats-monitoring/mypy.ini

@@ -1,5 +1,5 @@
 [mypy]
-python_version = 3.8
+python_version = 3.9.16
 follow_imports = skip
 ignore_missing_imports = True
 disallow_untyped_defs = True

stats-monitoring/pyproject.toml

@@ -17,7 +17,7 @@ tqdm = "^4.62.3"
 seaborn = "^0.11.2"
 pandas = "^1.3.4"
 matplotlib = "^3.5.0"
-pydantic = "^2.3.0"
+pydantic = "2.8.2"
 pathlib = "^1.0.1"
 pyyaml = "^6.0.1"
 types-pyyaml = "^6.0.12.11"