From ef4e2f2ac63cbe280259c8d050aa7b2eb0aa69bf Mon Sep 17 00:00:00 2001 From: Retief Lubbe Date: Fri, 24 Nov 2023 22:23:13 +0100 Subject: [PATCH] Removed class attributes and added class documentation instead. Related to #55 --- grainlearning/dynamic_systems.py | 228 +++++++++++++++---------------- 1 file changed, 108 insertions(+), 120 deletions(-) diff --git a/grainlearning/dynamic_systems.py b/grainlearning/dynamic_systems.py index a579121..d0996cb 100644 --- a/grainlearning/dynamic_systems.py +++ b/grainlearning/dynamic_systems.py @@ -2,11 +2,12 @@ This module contains various classes of state-space systems that describe the time evolution of the system's hidden state based on partial observations from the real world. """ -from typing import Type, List, Callable import os from datetime import datetime -from math import floor, log from glob import glob +from math import floor, log +from typing import Callable, List, Type + import numpy as np from grainlearning.tools import get_keys_and_data, write_to_table @@ -85,13 +86,16 @@ class DynamicSystem: The simulation data is a numpy array of shape (num_samples, num_obs, num_steps). - :param obs_data: observation or reference data + :param obs_data: Observation or reference data :param num_samples: Sample size of the ensemble of model evaluations + :param num_steps: Number of steps or sequence size in the dataset + :param num_obs: Number of observations in the dataset + :param num_ctrl: Number of control data (identical between simulation and observation) :param param_min: List of parameter lower bounds :param param_max: List of parameter Upper bounds - :param callback: Callback function, defaults to None - :param curr_iter: current iteration ID, defaults to 0 - :param ctrl_data: control data (e.g, time), defaults to None, optional + :param callback: Callback function to run simulations, defaults to None + :param curr_iter: Current iteration ID, defaults to 0 + :param ctrl_data: Control data (e.g, time), defaults to None, optional :param obs_names: Column names of the observation data, defaults to None, optional :param ctrl_name: Column name of the control data, defaults to None, optional :param inv_obs_weight: Inverse of the observation weight, defaults to None, optional @@ -101,91 +105,12 @@ class DynamicSystem: :param sigma_max: Maximum uncertainty, defaults to 1.0e6, optional :param sigma_tol: Tolerance of the estimated uncertainty, defaults to 1.0e-3, optional :param sim_name: Name of the simulation, defaults to 'sim', optional + :param sigma_min: Minimum uncertainty, defaults to 1.0e-6, optional + :param _inv_normalized_sigma: Calculated normalized sigma to weigh the covariance matrix + :param estimated_params: Estimated parameter as the first moment of the distribution (:math:`x_\mu = \sum_i w_i * x_i`), defaults to None, optional + :param estimated_params_cv: Estimated parameter coefficient of variation as the second moment of the distribution (:math:`x_\sigma = \sqrt{\sum_i w_i * (x_i - x_\mu)^2} / x_\mu`), defaults to None, optional """ - ##### Parameters ##### - - #: Parameter data of shape (num_samples, num_params) - param_data: np.ndarray - - #: Parameter data of previous iteration - param_data_prev: np.ndarray - - #: Number of unknown parameters - num_params: int - - #: Lower bound of the parameters - param_min: List - - #: Upper bound of the parameters - param_max: List - - #: Names of the parameters - param_names: List[str] - - ##### Observations ##### - - #: Observation (or reference) data of shape (num_obs, num_steps) - obs_data: np.ndarray - - #: Observation keys - obs_names: List[str] - - #: Inverse of the observation weight - inv_obs_weight: List[float] - - #: Number of steps or sequence size in the dataset - num_steps: int - - #: Number of observations in the dataset - num_obs: int - - #: Control dataset (num_ctrl, num_steps) - ctrl_data: np.ndarray - - #: Observation control (e.g., time) - ctrl_name: str - - #: Number of control data (identical between simulation and observation) - num_ctrl: int - - ##### Simulations ##### - - #: Name of the simulation (e.g., sim) - sim_name: str = 'sim' - - #: Simulation data of shape (num_samples, num_obs, num_steps) - sim_data: np.ndarray - - #: Number of samples (usually specified by user) - num_samples: int - - #: Callback function to run the forward predictions - callback: Callable - - #: Current iteration ID - curr_iter: int - - ##### Uncertainty ##### - - #: Minimum value of the uncertainty - sigma_min: float = 1.0e-6 - - #: Maximum value of the uncertainty - sigma_max: float = 1.0e6 - - #: Sigma tolerance - sigma_tol: float = 1.0e-3 - - #: Calculated normalized sigma to weigh the covariance matrix - _inv_normalized_sigma: np.array - - #: Estimated parameter as the first moment of the distribution (:math:`x_\mu = \sum_i w_i * x_i`) - estimated_params: np.ndarray = None - - #: Parameter uncertainty as the second moment of the distribution (:math:`x_\sigma = \sum_i w_i \cdot (x_i - x_\mu)^2 / x_\mu`) - estimated_params_cv: np.ndarray = None - def __init__( self, obs_data: np.ndarray, @@ -196,7 +121,7 @@ def __init__( obs_names: List[str] = None, ctrl_name: str = None, inv_obs_weight: List[float] = None, - sim_name: str = None, + sim_name: str = 'sim', sim_data: np.ndarray = None, callback: Callable = None, curr_iter: int = 0, @@ -205,14 +130,49 @@ def __init__( sigma_max: float = 1.0e6, sigma_tol: float = 1.0e-3 ): - """Initialize the dynamic system class""" + """Initialize the dynamic system class + + Parameters + ---------- + obs_data : np.ndarray + Observation (or reference) data of shape (num_obs, num_steps) + num_samples : int + Number of samples (usually specified by user) + param_min : List[float] + Lower bound of the parameters + param_max : List[float] + Upper bound of the parameters + ctrl_data : np.ndarray, optional + Control dataset (num_ctrl, num_steps), by default None + obs_names : List[str], optional + Observation keys, by default None + ctrl_name : str, optional + Observation control (e.g., time), by default None + inv_obs_weight : List[float], optional + Inverse of the observation weight, by default None + sim_name : str, optional + Name of the simulation, by default None + sim_data : np.ndarray, optional + Simulation data of shape (num_samples, num_obs, num_steps), by default None + callback : Callable, optional + Callback function to run simulations, by default None + curr_iter : int, optional + Current iteration ID, by default 0 + param_data : np.ndarray, optional + Parameter data of shape (num_samples, num_params), by default None + param_names : List[str], optional + Names of the parameters, by default None + sigma_max : float, optional + Maximum value of the uncertainty, by default 1.0e6 + sigma_tol : float, optional + Tolerance of the estimated uncertainty, by default 1.0e-3 + """ #### Observations #### self.obs_data = np.array( obs_data, ndmin=2 ) # ensure data is of shape (num_obs,num_step). - if ctrl_data is not None: - self.ctrl_data = ctrl_data + self.ctrl_data = ctrl_data self.obs_names = obs_names @@ -247,19 +207,27 @@ def __init__( if param_min: self.num_params = len(param_min) + else: + self.num_params = 0 self.param_data = param_data self.param_names = param_names #### Uncertainty #### + + self.sigma_min = 1.0e-6 self.sigma_max = sigma_max self.sigma_tol = sigma_tol + self._inv_normalized_sigma = None self.compute_inv_normalized_sigma() + self.estimated_params = None + self.estimated_params_cv = None + @classmethod def from_dict(cls: Type["DynamicSystem"], obj: dict): """ Initialize the class using a dictionary style @@ -283,7 +251,7 @@ def from_dict(cls: Type["DynamicSystem"], obj: dict): obs_names=obj.get("obs_names", None), ctrl_name=obj.get("ctrl_name", None), inv_obs_weight=obj.get("inv_obs_weight", None), - sim_name=obj.get("sim_name", None), + sim_name=obj.get("sim_name", 'sim'), sim_data=obj.get("sim_data", None), callback=obj.get("callback", None), param_data=obj.get("param_data", None), @@ -416,7 +384,7 @@ class IODynamicSystem(DynamicSystem): obs_data_file = 'obs_data.txt', obs_names = ['y_obs'], ctrl_name = 'y_ctrl', - sim_name = 'linear', + sim_name = 'linear',_description_ sim_data_file_ext = '.txt', callback = run_sim ) @@ -439,29 +407,9 @@ class IODynamicSystem(DynamicSystem): :param inv_obs_weight: Inverse of the observation weight, defaults to None, optional :param param_data: Parameter data, defaults to None, optional :param sim_data: Simulation data, defaults to None, optional + :param sim_data_files: List pf simulation data files (num_samples), defaults to None, optional """ - ##### Parameters ##### - - #: Name of the parameter data file - param_data_file: str - - ##### Observations ##### - - #: Name of the observation data file - obs_data_file: str - - ##### Simulations ##### - - #: Simulation data files (num_samples) - sim_data_files: List[str] - - #: Name of the directory where simulation data is stored - sim_data_dir: str = './sim_data' - - #: Extension of simulation data files - sim_data_file_ext: str = '.npy' - def __init__( self, sim_name: str, @@ -483,9 +431,47 @@ def __init__( param_data: np.ndarray = None, param_names: List[str] = None, ): - """Initialize the IO dynamic system class""" - - #### Calling base constructor #### + """Initialize the I/O dynamic system class + + Parameters + ---------- + sim_name : str + Name of the simulation + sim_data_dir : str + Name of the directory where simulation data is stored + sim_data_file_ext : str + Extension of simulation data files (e.g., .txt, .npy) + obs_data_file : str + Name of the observation data file + obs_names : List[str] + Observation names + ctrl_name : str + Control name + num_samples : int + Number of samples + param_min : List[float] + List of parameter lower bounds + param_max : List[float] + List of parameter upper bounds + obs_data : np.ndarray, optional + Observation (or reference) data of shape (num_obs, num_steps), by default None + ctrl_data : np.ndarray, optional + _description_, by default None + inv_obs_weight : List[float], optional + _description_, by default None + sim_data : np.ndarray, optional + _description_, by default None + callback : Callable, optional + _description_, by default None + curr_iter : int, optional + _description_, by default 0 + param_data_file : str, optional + Parameter data file, by default '' + param_data : np.ndarray, optional + _description_, by default None + param_names : List[str], optional + _description_, by default None + """ super().__init__( obs_data, @@ -535,6 +521,8 @@ def __init__( self.compute_inv_normalized_sigma() + self.sim_data_files = [] + @classmethod def from_dict(cls: Type["IODynamicSystem"], obj: dict): """ Initialize the class using a dictionary style @@ -710,4 +698,4 @@ def backup_sim_data(self): path = self.sim_data_dir.rstrip('/') backup_dir = f'{path}_backup_{formatted_time}' os.makedirs(backup_dir, exist_ok=True) - os.rename(self.sim_data_dir, backup_dir) + os.rename(self.sim_data_dir, backup_dir) \ No newline at end of file