Skip to content

Index

Job management utilities for simulation servers.

Provides sample generation, output handling, and run management utilities adapted from mada-job-management jobkit.

FolderOutputHandler

Bases: BaseOutputHandler

Output handler for writing samples to a folder.

Attributes:

Name Type Description
settings_class

The settings class for this output handler.
supported_kwargs

The full set of supported keyword arguments for the write method.
required_kwargs

The set of required keyword arguments for the write method.

Methods:

Name Description
write

Writes samples to a folder structure.
Source code in src/mada_tools/simulation/simutils/samples/output/folder_output_handler.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class FolderOutputHandler(BaseOutputHandler):
    """
    Output handler for writing samples to a folder.

    Attributes:
        settings_class: The settings class for this output handler.
        supported_kwargs: The full set of supported keyword arguments for the `write` method.
        required_kwargs: The set of required keyword arguments for the `write` method.

    Methods:
        write: Writes samples to a folder structure.
    """

    def _get_settings_class(self) -> FolderOutputSettings:
        """
        Get the settings class for this output handler.

        Returns:
            The settings class for this output handler.
        """
        return FolderOutputSettings

    def _get_supported_kwargs(self) -> Set[str]:
        """
        Get the set of supported keyword arguments for the `write` method.

        Returns:
            A set of supported keyword argument names for output configuration.
        """
        return {"output_dir", "param_file"}

    def _get_required_kwargs(self) -> Set[str]:
        """
        Get the set of required keyword arguments for the `write` method.

        Returns:
            A set of required keyword argument names for output configuration.
        """
        return {}

    def write(
        self,
        samples: np.ndarray,
        parameter_names: List[str] | None = None,
        **kwargs: Dict[str, Any],
    ) -> SampleOutputResult:
        """
        Creates a folder structure for each sample, writes the parameters to individual files,
        and dumps all `RunInstances` to a JSON file.

        Args:
            samples: A list of samples, where each sample is a list of floats.
            parameter_names: List of parameter names.
            **kwargs: Additional keyword arguments for output configuration.

        Returns:
            A `SampleOutputResult` object containing the output path, output type, and run instances.
        """
        self.check_if_samples_are_empty(samples)

        output_settings = self.build_settings_from_kwargs(kwargs)

        print(f"Creating folder structure in '{output_settings.output_dir}'...")

        # Ensure the base output directory exists
        if not os.path.exists(output_settings.output_dir):
            os.makedirs(output_settings.output_dir)
            print(f"Created base output directory: {output_settings.output_dir}")
        else:
            print(f"Base output directory already exists: {output_settings.output_dir}")

        # Create subdirectories and RunInstance objects for each sample
        padding = len(str(len(samples)))  # Determine padding for folder names (e.g., run001)
        run_instances = []  # List to store all RunInstance objects

        if parameter_names is not None:
            # Build DataFrame
            run_ids = [str(i).zfill(padding) for i in range(len(samples))]
            df = pd.DataFrame(samples, columns=parameter_names)
            df.insert(0, "run#", run_ids)

            # Save CSV
            csv_path = os.path.join(output_settings.output_dir, "run_parameters.csv")
            df.to_csv(csv_path, index=False)
            print(f"Saved run parameter summary CSV: {csv_path}\n")

        for i, sample in enumerate(samples):
            # Create a subdirectory for the sample
            sample_dir = os.path.join(output_settings.output_dir, f"run{str(i).zfill(padding)}")
            os.makedirs(sample_dir, exist_ok=True)

            # Write the sample parameters to the specified file
            param_file_path = os.path.join(sample_dir, output_settings.param_file)
            with open(param_file_path, "w") as param_file:
                if parameter_names is not None:
                    param_file.write("\n".join(f"{name}: {value}" for name, value in zip(parameter_names, sample)))
                else:
                    param_file.write("\n".join(map(str, sample)))

            # Create a RunInstance object
            run_id = str(i).zfill(padding)
            run_instance = RunInstance(run_location=sample_dir, id=run_id)
            run_instances.append(run_instance)

            print(f"Created folder: {sample_dir}")
            print(f"\tWritten parameters to: {param_file_path}")
            print(f"\tRunInstance created: {run_instance}")

        # Dump all RunInstances to a JSON file in case they need to be loaded later
        run_instances_json_path = os.path.join(output_settings.output_dir, "run_instances.json")
        with open(run_instances_json_path, "w") as json_file:
            json.dump([run.to_dict() for run in run_instances], json_file, indent=4)

        print(f"RunInstances dumped to JSON file: {run_instances_json_path}")
        print("Folder structure created successfully.")

        return SampleOutputResult(
            output_path=output_settings.output_dir,
            output_type="folders",
            run_instances=run_instances,
        )

write(samples, parameter_names=None, **kwargs)

Creates a folder structure for each sample, writes the parameters to individual files, and dumps all RunInstances to a JSON file.

Parameters:

Name Type Description Default
samples ndarray

A list of samples, where each sample is a list of floats.

 required
parameter_names List[str] | None

List of parameter names.

 None
**kwargs Dict[str, Any]

Additional keyword arguments for output configuration.

 {}

Returns:

Type Description
SampleOutputResult

A SampleOutputResult object containing the output path, output type, and run instances.
Source code in src/mada_tools/simulation/simutils/samples/output/folder_output_handler.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def write(
    self,
    samples: np.ndarray,
    parameter_names: List[str] | None = None,
    **kwargs: Dict[str, Any],
) -> SampleOutputResult:
    """
    Creates a folder structure for each sample, writes the parameters to individual files,
    and dumps all `RunInstances` to a JSON file.

    Args:
        samples: A list of samples, where each sample is a list of floats.
        parameter_names: List of parameter names.
        **kwargs: Additional keyword arguments for output configuration.

    Returns:
        A `SampleOutputResult` object containing the output path, output type, and run instances.
    """
    self.check_if_samples_are_empty(samples)

    output_settings = self.build_settings_from_kwargs(kwargs)

    print(f"Creating folder structure in '{output_settings.output_dir}'...")

    # Ensure the base output directory exists
    if not os.path.exists(output_settings.output_dir):
        os.makedirs(output_settings.output_dir)
        print(f"Created base output directory: {output_settings.output_dir}")
    else:
        print(f"Base output directory already exists: {output_settings.output_dir}")

    # Create subdirectories and RunInstance objects for each sample
    padding = len(str(len(samples)))  # Determine padding for folder names (e.g., run001)
    run_instances = []  # List to store all RunInstance objects

    if parameter_names is not None:
        # Build DataFrame
        run_ids = [str(i).zfill(padding) for i in range(len(samples))]
        df = pd.DataFrame(samples, columns=parameter_names)
        df.insert(0, "run#", run_ids)

        # Save CSV
        csv_path = os.path.join(output_settings.output_dir, "run_parameters.csv")
        df.to_csv(csv_path, index=False)
        print(f"Saved run parameter summary CSV: {csv_path}\n")

    for i, sample in enumerate(samples):
        # Create a subdirectory for the sample
        sample_dir = os.path.join(output_settings.output_dir, f"run{str(i).zfill(padding)}")
        os.makedirs(sample_dir, exist_ok=True)

        # Write the sample parameters to the specified file
        param_file_path = os.path.join(sample_dir, output_settings.param_file)
        with open(param_file_path, "w") as param_file:
            if parameter_names is not None:
                param_file.write("\n".join(f"{name}: {value}" for name, value in zip(parameter_names, sample)))
            else:
                param_file.write("\n".join(map(str, sample)))

        # Create a RunInstance object
        run_id = str(i).zfill(padding)
        run_instance = RunInstance(run_location=sample_dir, id=run_id)
        run_instances.append(run_instance)

        print(f"Created folder: {sample_dir}")
        print(f"\tWritten parameters to: {param_file_path}")
        print(f"\tRunInstance created: {run_instance}")

    # Dump all RunInstances to a JSON file in case they need to be loaded later
    run_instances_json_path = os.path.join(output_settings.output_dir, "run_instances.json")
    with open(run_instances_json_path, "w") as json_file:
        json.dump([run.to_dict() for run in run_instances], json_file, indent=4)

    print(f"RunInstances dumped to JSON file: {run_instances_json_path}")
    print("Folder structure created successfully.")

    return SampleOutputResult(
        output_path=output_settings.output_dir,
        output_type="folders",
        run_instances=run_instances,
    )

LHSampleGenerator

Bases: BaseSampleGenerator

Latin Hypercube Sample generator component.

Attributes:

Name Type Description
settings_class

The settings class for this output handler.
supported_kwargs

The full set of supported keyword arguments for the write method.
required_kwargs

The set of required keyword arguments for the write method.

Methods:

Name Description
find_missing_required_kwargs

Validates that all required keyword arguments are present.
build_settings_from_kwargs

Validates the keyword arguments and builds the settings object.
generate

Generate a list of samples from the provided data.
Source code in src/mada_tools/simulation/simutils/samples/generation/lhs_sample_generator.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class LHSampleGenerator(BaseSampleGenerator):
    """
    Latin Hypercube Sample generator component.

    Attributes:
        settings_class: The settings class for this output handler.
        supported_kwargs: The full set of supported keyword arguments for the `write` method.
        required_kwargs: The set of required keyword arguments for the `write` method.

    Methods:
        find_missing_required_kwargs: Validates that all required keyword arguments are present.
        build_settings_from_kwargs: Validates the keyword arguments and builds the settings object.
        generate: Generate a list of samples from the provided data.
    """

    def _get_settings_class(self) -> LHSampleSettings:
        """
        Get the settings class for this sample generator.

        Returns:
            The settings class for this sample generator.
        """
        return LHSampleSettings

    def _get_supported_kwargs(self):
        """
        Get the full set of supported keyword arguments for the `generate` method.

        Returns:
            A set of supported keyword argument names.
        """
        return {"dims", "n_samples", "lower_bounds", "upper_bounds"}

    def _get_required_kwargs(self) -> Set[str]:
        """
        Get the set of required keyword arguments for the `generate` method.

        Returns:
            A set of required keyword argument names for sample generation.
        """
        return {"dims", "n_samples", "lower_bounds", "upper_bounds"}

    def generate(self, **kwargs: Dict[str, int | List[float]]) -> np.ndarray:
        """
        Generate a list of samples from the provided data.

        This method will convert the keyword arguments into a `LHSampleSettings` object
        for validation and logging purposes. It will then use these settings to generate
        samples using the Latin Hypercube Sampling method.

        Args:
            **kwargs: Keyword arguments for sample generation. For LHS sampling, this must include:
                - dims: int, number of dimensions
                - n_samples: int, number of samples to generate
                - lower_bounds: List[float], lower bounds for each dimension
                - upper_bounds: List[float], upper bounds for each dimension

        Returns:
            A numpy array of generated samples.
        """
        print("Generating samples...")

        sample_settings = self.build_settings_from_kwargs(kwargs)

        lower = np.asarray(sample_settings.lower_bounds, dtype=float)
        upper = np.asarray(sample_settings.upper_bounds, dtype=float)

        varying_mask = lower < upper
        fixed_mask = lower == upper

        result = np.empty((sample_settings.n_samples, sample_settings.dims), dtype=float)

        if np.any(varying_mask):
            lhs = qmc.LatinHypercube(d=int(np.sum(varying_mask)))
            samples = lhs.random(n=sample_settings.n_samples)

            scaled = qmc.scale(
                samples,
                l_bounds=lower[varying_mask],
                u_bounds=upper[varying_mask],
            )

            result[:, varying_mask] = scaled

        if np.any(fixed_mask):
            result[:, fixed_mask] = lower[fixed_mask]

        print("Samples generated successfully!\n")
        return result

generate(**kwargs)

Generate a list of samples from the provided data.

This method will convert the keyword arguments into a LHSampleSettings object for validation and logging purposes. It will then use these settings to generate samples using the Latin Hypercube Sampling method.

Parameters:

Name Type Description Default
**kwargs Dict[str, int | List[float]]

Keyword arguments for sample generation. For LHS sampling, this must include: - dims: int, number of dimensions - n_samples: int, number of samples to generate - lower_bounds: List[float], lower bounds for each dimension - upper_bounds: List[float], upper bounds for each dimension

 {}

Returns:

Type Description
ndarray

A numpy array of generated samples.
Source code in src/mada_tools/simulation/simutils/samples/generation/lhs_sample_generator.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def generate(self, **kwargs: Dict[str, int | List[float]]) -> np.ndarray:
    """
    Generate a list of samples from the provided data.

    This method will convert the keyword arguments into a `LHSampleSettings` object
    for validation and logging purposes. It will then use these settings to generate
    samples using the Latin Hypercube Sampling method.

    Args:
        **kwargs: Keyword arguments for sample generation. For LHS sampling, this must include:
            - dims: int, number of dimensions
            - n_samples: int, number of samples to generate
            - lower_bounds: List[float], lower bounds for each dimension
            - upper_bounds: List[float], upper bounds for each dimension

    Returns:
        A numpy array of generated samples.
    """
    print("Generating samples...")

    sample_settings = self.build_settings_from_kwargs(kwargs)

    lower = np.asarray(sample_settings.lower_bounds, dtype=float)
    upper = np.asarray(sample_settings.upper_bounds, dtype=float)

    varying_mask = lower < upper
    fixed_mask = lower == upper

    result = np.empty((sample_settings.n_samples, sample_settings.dims), dtype=float)

    if np.any(varying_mask):
        lhs = qmc.LatinHypercube(d=int(np.sum(varying_mask)))
        samples = lhs.random(n=sample_settings.n_samples)

        scaled = qmc.scale(
            samples,
            l_bounds=lower[varying_mask],
            u_bounds=upper[varying_mask],
        )

        result[:, varying_mask] = scaled

    if np.any(fixed_mask):
        result[:, fixed_mask] = lower[fixed_mask]

    print("Samples generated successfully!\n")
    return result

RunInstance dataclass

Represents a single instance of a computational run, storing relevant data and metadata associated with the run.

Attributes:

Name Type Description
run_location str

The directory where the run will be executed.
id str

The identifier for a run instance.
command str

The executable command to run (optional).
args List[str]

The arguments to pass to the command (optional).

Methods:

Name Description
to_dict

Converts the RunInstance object into a JSON-serializable dictionary.
get_command_with_args

Returns the command and arguments for execution.
Source code in src/mada_tools/simulation/simutils/models.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@dataclass
class RunInstance:
    """
    Represents a single instance of a computational run, storing relevant data
    and metadata associated with the run.

    Attributes:
        run_location (str): The directory where the run will be executed.
        id (str): The identifier for a run instance.
        command (str): The executable command to run (optional).
        args (List[str]): The arguments to pass to the command (optional).

    Methods:
        to_dict: Converts the RunInstance object into a JSON-serializable dictionary.
        get_command_with_args: Returns the command and arguments for execution.
    """

    run_location: str
    id: str
    command: str = None
    args: List[str] = None

    def to_dict(self) -> Dict:
        """
        Converts the RunInstance object into a JSON-serializable dictionary.

        Returns:
            A dictionary representation of the RunInstance object.
        """
        result = {"id": self.id, "run_location": self.run_location}

        # Include command and args if they are set
        if self.command:
            result["command"] = self.command
        if self.args:
            result["args"] = self.args

        return result

    @classmethod
    def from_dict(cls, data: Dict) -> "RunInstance":
        """
        Creates a RunInstance object from a dictionary.

        Args:
            data: Dictionary containing run instance data

        Returns:
            RunInstance object with data from the dictionary
        """
        return cls(
            run_location=data.get("run_location"),
            id=data.get("id"),
            command=data.get("command"),
            args=data.get("args"),
        )

    def __str__(self) -> str:
        """
        Returns a human-readable string representation of the RunInstance object.

        Returns:
            str: A string describing the run instance.
        """
        result = f"RunInstance '{self.id}':\n\tLocation: {self.run_location}"
        if self.command:
            result += f"\n\tCommand: {self.command}"
        if self.args:
            result += f"\n\tArgs: {self.args}"
        return result

    def __repr__(self) -> str:
        """
        Returns a detailed string representation of the RunInstance object for debugging.

        Returns:
            str: A detailed string representation of the object.
        """
        return f"RunInstance(id={self.id}, run_location={self.run_location}, command={self.command}, args={self.args})"

    def get_command_with_args(self) -> Tuple[str, List[str]]:
        """
        Returns the command and arguments for execution.

        Returns:
            Tuple[str, List[str]]: The command and its arguments.
        """
        if self.command and self.args:
            return self.command, self.args
        else:
            raise ValueError("Command and args not set for this RunInstance")

    def get_run_command(self) -> str:
        """
        Constructs the shell command to execute this run instance.

        Returns:
            str: The shell command to execute this run.
        """
        if self.command and self.args:
            return f"{self.command} {' '.join(self.args)}"
        else:
            raise ValueError("No command specified for this RunInstance")

__repr__()

Returns a detailed string representation of the RunInstance object for debugging.

Returns:

Name Type Description
str str

A detailed string representation of the object.
Source code in src/mada_tools/simulation/simutils/models.py 
85
86
87
88
89
90
91
92
def __repr__(self) -> str:
    """
    Returns a detailed string representation of the RunInstance object for debugging.

    Returns:
        str: A detailed string representation of the object.
    """
    return f"RunInstance(id={self.id}, run_location={self.run_location}, command={self.command}, args={self.args})"

__str__()

Returns a human-readable string representation of the RunInstance object.

Returns:

Name Type Description
str str

A string describing the run instance.
Source code in src/mada_tools/simulation/simutils/models.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def __str__(self) -> str:
    """
    Returns a human-readable string representation of the RunInstance object.

    Returns:
        str: A string describing the run instance.
    """
    result = f"RunInstance '{self.id}':\n\tLocation: {self.run_location}"
    if self.command:
        result += f"\n\tCommand: {self.command}"
    if self.args:
        result += f"\n\tArgs: {self.args}"
    return result

from_dict(data) classmethod

Creates a RunInstance object from a dictionary.

Parameters:

Name Type Description Default
data Dict

Dictionary containing run instance data

 required

Returns:

Type Description
RunInstance

RunInstance object with data from the dictionary
Source code in src/mada_tools/simulation/simutils/models.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@classmethod
def from_dict(cls, data: Dict) -> "RunInstance":
    """
    Creates a RunInstance object from a dictionary.

    Args:
        data: Dictionary containing run instance data

    Returns:
        RunInstance object with data from the dictionary
    """
    return cls(
        run_location=data.get("run_location"),
        id=data.get("id"),
        command=data.get("command"),
        args=data.get("args"),
    )

get_command_with_args()

Returns the command and arguments for execution.

Returns:

Type Description
Tuple[str, List[str]]

Tuple[str, List[str]]: The command and its arguments.
Source code in src/mada_tools/simulation/simutils/models.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_command_with_args(self) -> Tuple[str, List[str]]:
    """
    Returns the command and arguments for execution.

    Returns:
        Tuple[str, List[str]]: The command and its arguments.
    """
    if self.command and self.args:
        return self.command, self.args
    else:
        raise ValueError("Command and args not set for this RunInstance")

get_run_command()

Constructs the shell command to execute this run instance.

Returns:

Name Type Description
str str

The shell command to execute this run.
Source code in src/mada_tools/simulation/simutils/models.py 
106
107
108
109
110
111
112
113
114
115
116
def get_run_command(self) -> str:
    """
    Constructs the shell command to execute this run instance.

    Returns:
        str: The shell command to execute this run.
    """
    if self.command and self.args:
        return f"{self.command} {' '.join(self.args)}"
    else:
        raise ValueError("No command specified for this RunInstance")

to_dict()

Converts the RunInstance object into a JSON-serializable dictionary.

Returns:

Type Description
Dict

A dictionary representation of the RunInstance object.
Source code in src/mada_tools/simulation/simutils/models.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def to_dict(self) -> Dict:
    """
    Converts the RunInstance object into a JSON-serializable dictionary.

    Returns:
        A dictionary representation of the RunInstance object.
    """
    result = {"id": self.id, "run_location": self.run_location}

    # Include command and args if they are set
    if self.command:
        result["command"] = self.command
    if self.args:
        result["args"] = self.args

    return result

SampleOutputResult dataclass

Represents the result of sample output generation, storing the output path, output type, and optionally the run instances (if necessary).

Attributes:

Name Type Description
output_path str

The path where the samples are written. For "csv" output, this is the path to the CSV file. For "folder" output, this is the path to the directory.
output_type str

The type of output generated (e.g., "csv", "folder").
run_instances List[RunInstance]

A list of RunInstance objects if the output type is "folder".
Source code in src/mada_tools/simulation/simutils/models.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@dataclass
class SampleOutputResult:
    """
    Represents the result of sample output generation, storing the output path,
    output type, and optionally the run instances (if necessary).

    Attributes:
        output_path (str): The path where the samples are written. For "csv" output, this
            is the path to the CSV file. For "folder" output, this is the path to the directory.
        output_type (str): The type of output generated (e.g., "csv", "folder").
        run_instances (List[RunInstance]): A list of RunInstance objects if the output type is "folder".
    """

    output_path: str
    output_type: str  # e.g., "csv", "folder"
    run_instances: List[RunInstance] = None

    def __str__(self) -> str:
        """
        Returns a human-readable string representation of the SampleOutputResult object.

        Returns:
            str: A string describing the sample output result.
        """
        return (
            f"SampleOutputResult:\n"
            f"  Output path: {self.output_path}\n"
            f"  Output type: {self.output_type}\n"
            f"  Run instances: {self.run_instances}"
        )

    def __repr__(self) -> str:
        """
        Returns a detailed string representation of the SampleOutputResult object for debugging.

        Returns:
            str: A detailed string representation of the object.
        """
        return (
            f"SampleOutputResult(output_path={self.output_path}, "
            f"output_type={self.output_type}, run_instances={self.run_instances})"
        )

__repr__()

Returns a detailed string representation of the SampleOutputResult object for debugging.

Returns:

Name Type Description
str str

A detailed string representation of the object.
Source code in src/mada_tools/simulation/simutils/models.py 
150
151
152
153
154
155
156
157
158
159
160
def __repr__(self) -> str:
    """
    Returns a detailed string representation of the SampleOutputResult object for debugging.

    Returns:
        str: A detailed string representation of the object.
    """
    return (
        f"SampleOutputResult(output_path={self.output_path}, "
        f"output_type={self.output_type}, run_instances={self.run_instances})"
    )

__str__()

Returns a human-readable string representation of the SampleOutputResult object.

Returns:

Name Type Description
str str

A string describing the sample output result.
Source code in src/mada_tools/simulation/simutils/models.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def __str__(self) -> str:
    """
    Returns a human-readable string representation of the SampleOutputResult object.

    Returns:
        str: A string describing the sample output result.
    """
    return (
        f"SampleOutputResult:\n"
        f"  Output path: {self.output_path}\n"
        f"  Output type: {self.output_type}\n"
        f"  Run instances: {self.run_instances}"
    )

get_run_instances(run_instances_json_path)

Reads the run_instances.json file from the specified study directory and converts it into a list of RunInstance objects.

Parameters:

Name Type Description Default
run_instances_json_path str

The path to the run_instances.json file.

 required

Returns:

Type Description
List[RunInstance]

A list of RunInstance objects.
Source code in src/mada_tools/simulation/simutils/utils.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_run_instances(run_instances_json_path: str) -> List[RunInstance]:
    """
    Reads the run_instances.json file from the specified study directory and
    converts it into a list of RunInstance objects.

    Args:
        run_instances_json_path: The path to the run_instances.json file.

    Returns:
        A list of RunInstance objects.
    """
    print(f"Attempting to retrieve run instances from '{run_instances_json_path}'...")

    # Check if the file exists
    if not os.path.exists(run_instances_json_path):
        print(f"ERROR: The file containing run instances does not exist: '{run_instances_json_path}'.")
        print("Did you run the sample generation script yet?")
        return []

    # Read the JSON file and convert it into RunInstance objects
    with open(run_instances_json_path, "r") as json_file:
        try:
            run_instances_data = json.load(json_file)  # Load the JSON data as a list of dictionaries
            run_instances = []
            for data in run_instances_data:
                run_instance = RunInstance(run_location=data["run_location"], id=data["id"])
                # Set command and args if present
                if "command" in data:
                    run_instance.command = data["command"]
                if "args" in data:
                    run_instance.args = data["args"]
                run_instances.append(run_instance)
            print("Successfully retrieved run instances.")
            return run_instances
        except json.JSONDecodeError as e:
            print(f"ERROR: Failed to parse JSON file: {run_instances_json_path}.")
            print(f"Details: {e}")
            return []
        except KeyError as e:
            print(f"ERROR: Missing expected key in JSON data: {e}")
            return []