Skip to content

utils

Utility functions for job management.

This module provides helper functions to interact with job-related files and data structures.

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
58
59
60
61
62
63
64
65
66
67
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_info = json.load(json_file)
            if isinstance(run_info, dict):
                run_instances_data = run_info.get("runs", [])
            else:
                run_instances_data = run_info
            if not isinstance(run_instances_data, list):
                print("ERROR: Run instances JSON must be a list or an object with a 'runs' list.")
                return []
            run_instances = []
            for data in run_instances_data:
                if not isinstance(data, dict):
                    print("ERROR: Run instance entries must be JSON objects.")
                    return []
                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 []