Skip to content

cosmotech.coal.cosmotech_api.runner.parameters

parameters

Parameter handling functions.

format_parameters_list(runner_data)

Format parameters from runner data as a list of dictionaries.

Args: runner_data: Runner data object

Returns: List of parameter dictionaries

Source code in cosmotech/coal/cosmotech_api/runner/parameters.py 
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
def format_parameters_list(runner_data: Any) -> List[Dict[str, Any]]:
    """
    Format parameters from runner data as a list of dictionaries.

    Args:
        runner_data: Runner data object

    Returns:
        List of parameter dictionaries
    """
    parameters = []

    if not runner_data.parameters_values:
        return parameters

    max_name_size = max(map(lambda r: len(r.parameter_id), runner_data.parameters_values))
    max_type_size = max(map(lambda r: len(r.var_type), runner_data.parameters_values))

    for parameter_data in runner_data.parameters_values:
        parameter_name = parameter_data.parameter_id
        value = parameter_data.value
        var_type = parameter_data.var_type
        is_inherited = parameter_data.is_inherited

        parameters.append(
            {
                "parameterId": parameter_name,
                "value": value,
                "varType": var_type,
                "isInherited": is_inherited,
            }
        )

        LOGGER.debug(
            T("coal.cosmotech_api.runner.parameter_debug").format(
                param_id=parameter_name,
                max_name_size=max_name_size,
                var_type=var_type,
                max_type_size=max_type_size,
                value=value,
                inherited=" inherited" if is_inherited else "",
            )
        )

    return parameters

get_runner_parameters(runner_data)

Extract parameters from runner data.

Args: runner_data: Runner data object

Returns: Dictionary mapping parameter IDs to values

Source code in cosmotech/coal/cosmotech_api/runner/parameters.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def get_runner_parameters(runner_data: Any) -> Dict[str, Any]:
    """
    Extract parameters from runner data.

    Args:
        runner_data: Runner data object

    Returns:
        Dictionary mapping parameter IDs to values
    """
    content = dict()
    for parameter in runner_data.parameters_values:
        content[parameter.parameter_id] = parameter.value
    return content

write_parameters(parameter_folder, parameters, write_csv=True, write_json=False)

Write parameters to files based on specified formats.

Args: parameter_folder: Folder to write the files to parameters: List of parameter dictionaries write_csv: Whether to write a CSV file write_json: Whether to write a JSON file

Returns: Dictionary mapping file types to file paths

Source code in cosmotech/coal/cosmotech_api/runner/parameters.py 
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
def write_parameters(
    parameter_folder: str,
    parameters: List[Dict[str, Any]],
    write_csv: bool = True,
    write_json: bool = False,
) -> Dict[str, str]:
    """
    Write parameters to files based on specified formats.

    Args:
        parameter_folder: Folder to write the files to
        parameters: List of parameter dictionaries
        write_csv: Whether to write a CSV file
        write_json: Whether to write a JSON file

    Returns:
        Dictionary mapping file types to file paths
    """
    result = {}

    if write_csv:
        result["csv"] = write_parameters_to_csv(parameter_folder, parameters)

    if write_json:
        result["json"] = write_parameters_to_json(parameter_folder, parameters)

    return result

write_parameters_to_csv(parameter_folder, parameters)

Write parameters to a CSV file.

Args: parameter_folder: Folder to write the file to parameters: List of parameter dictionaries

Returns: Path to the created file

Source code in cosmotech/coal/cosmotech_api/runner/parameters.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def write_parameters_to_csv(parameter_folder: str, parameters: List[Dict[str, Any]]) -> str:
    """
    Write parameters to a CSV file.

    Args:
        parameter_folder: Folder to write the file to
        parameters: List of parameter dictionaries

    Returns:
        Path to the created file
    """
    pathlib.Path(parameter_folder).mkdir(exist_ok=True, parents=True)
    tmp_parameter_file = os.path.join(parameter_folder, "parameters.csv")

    LOGGER.info(T("coal.cosmotech_api.runner.generating_file").format(file=tmp_parameter_file))

    with open(tmp_parameter_file, "w") as _file:
        _w = DictWriter(_file, fieldnames=["parameterId", "value", "varType", "isInherited"])
        _w.writeheader()
        _w.writerows(parameters)

    return tmp_parameter_file

write_parameters_to_json(parameter_folder, parameters)

Write parameters to a JSON file.

Args: parameter_folder: Folder to write the file to parameters: List of parameter dictionaries

Returns: Path to the created file

Source code in cosmotech/coal/cosmotech_api/runner/parameters.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def write_parameters_to_json(parameter_folder: str, parameters: List[Dict[str, Any]]) -> str:
    """
    Write parameters to a JSON file.

    Args:
        parameter_folder: Folder to write the file to
        parameters: List of parameter dictionaries

    Returns:
        Path to the created file
    """
    pathlib.Path(parameter_folder).mkdir(exist_ok=True, parents=True)
    tmp_parameter_file = os.path.join(parameter_folder, "parameters.json")

    LOGGER.info(T("coal.cosmotech_api.runner.generating_file").format(file=tmp_parameter_file))

    with open(tmp_parameter_file, "w") as _file:
        json.dump(parameters, _file, indent=2)

    return tmp_parameter_file