Skip to content

API reference

griddler

ParameterSet

Bases: dict

A simple extension of the dict class, requiring that:

  1. all keys are strings, and
  2. all values are valid. Valid values are integers, floats, or strings, or lists (or tuples) composed of valid values.
Source code in griddler/__init__.py 
 9
10
11
12
13
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
class ParameterSet(dict):
    """A simple extension of the `dict` class, requiring that:

    1. all keys are strings, and
    2. all values are valid. Valid values are integers, floats, or strings, or
      lists (or tuples) composed of valid values.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.validate()

    def stable_hash(self, hash_length: int = 10) -> str:
        """Create a stable hash of this parameter set. Implemented as a BLAKE2
        digest of the JSON representation of the parameter set:

        Examples:
            >>> ParameterSet({"gamma": 1.0, "R0": 0.9}).stable_hash()
            '544601bc1dbb3346faff' # pragma: allowlist secret

        Args:
            hash_length: Number of characters in the hash

        Returns:
            hash
        """
        data = json.dumps(self, sort_keys=True)
        return hashlib.blake2b(data.encode(), digest_size=hash_length).hexdigest()

    def validate(self):
        for key in self.keys():
            if not isinstance(key, str):
                raise ValueError(f"parameter set key {key!r} is not a string")

        for value in self.values():
            self.validate_value(value)

    @classmethod
    def validate_value(cls, value):
        if isinstance(value, (str, int, float)):
            return True
        elif isinstance(value, (list, tuple)):
            for x in value:
                cls.validate_value(x)
        elif isinstance(value, dict):
            assert all(isinstance(k, str) for k in value.keys())
            for x in value.values():
                cls.validate_value(x)
        else:
            raise ValueError(
                f"parameter value {value!r} is of invalid type {type(value)}"
            )

stable_hash(hash_length=10)

Create a stable hash of this parameter set. Implemented as a BLAKE2 digest of the JSON representation of the parameter set:

Examples:

>>> ParameterSet({"gamma": 1.0, "R0": 0.9}).stable_hash()
'544601bc1dbb3346faff' # pragma: allowlist secret

Parameters:

Name Type Description Default
hash_length int

Number of characters in the hash

 10

Returns:

Type Description
str

hash
Source code in griddler/__init__.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def stable_hash(self, hash_length: int = 10) -> str:
    """Create a stable hash of this parameter set. Implemented as a BLAKE2
    digest of the JSON representation of the parameter set:

    Examples:
        >>> ParameterSet({"gamma": 1.0, "R0": 0.9}).stable_hash()
        '544601bc1dbb3346faff' # pragma: allowlist secret

    Args:
        hash_length: Number of characters in the hash

    Returns:
        hash
    """
    data = json.dumps(self, sort_keys=True)
    return hashlib.blake2b(data.encode(), digest_size=hash_length).hexdigest()

griddler.run_squash(func, parameter_sets, add_parameters=True, parameter_columns=None, add_hash=True, hash_column='hash')

Source code in griddler/__init__.py 
 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
def run_squash(
    func: Callable[..., pl.DataFrame],
    parameter_sets: Sequence[dict],
    add_parameters: bool = True,
    parameter_columns: Optional[Sequence[str]] = None,
    add_hash: bool = True,
    hash_column: str = "hash",
):
    outputs = [func(ps) for ps in parameter_sets]

    output_vars = set([column for output in outputs for column in output.columns])

    # the hash column name shouldn't collide with any output or parameter columns
    if add_hash:
        assert hash_column not in output_vars

    if add_parameters:
        # if parameter_vars is not specified, then include all parameters
        if parameter_columns is None:
            parameter_columns = list(
                set([key for ps in parameter_sets for key in ps.keys()])
            )

        # there should be no names in common between the output columns and the parameters we want
        # to add
        assert len(output_vars & set(parameter_columns)) == 0

        # there should be no collision with hash column
        if add_hash:
            assert hash_column not in parameter_columns

        # add parameter columns
        outputs = [
            output.with_columns(
                [_as_expr(ps[key]).alias(key) for key in parameter_columns]
            )
            for output, ps in zip(outputs, parameter_sets)
        ]

    # add hash column
    if add_hash:
        hashes = [ParameterSet(ps).stable_hash() for ps in parameter_sets]
        outputs = [
            output.with_columns(pl.lit(hash).alias(hash_column))
            for output, hash in zip(outputs, hashes)
        ]

    return pl.concat(outputs, how="vertical")

griddler.replicated(func, n_replicates_key='n_replicates', seed_key='seed', replicate_var='replicate', replicate_type=pl.Int64)

Source code in griddler/__init__.py 
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
def replicated(
    func: Callable[..., pl.DataFrame],
    n_replicates_key="n_replicates",
    seed_key="seed",
    replicate_var="replicate",
    replicate_type=pl.Int64,
):
    def replicated_func(parameter_set, *args, **kwargs):
        assert isinstance(parameter_set, dict)
        assert n_replicates_key in parameter_set
        assert seed_key in parameter_set

        # remove replicate-related keywords
        norep_parameter_set = {
            key: value
            for key, value in parameter_set.items()
            if key not in [n_replicates_key, seed_key]
        }

        # run the function
        random.seed(parameter_set[seed_key])

        outputs = [
            func(norep_parameter_set, *args, **kwargs)
            for i in range(parameter_set[n_replicates_key])
        ]

        return pl.concat(
            [
                output.with_columns(
                    pl.lit(i, dtype=replicate_type).alias(replicate_var)
                )
                for i, output in enumerate(outputs)
            ]
        )

    return replicated_func

griddler.griddle

parse(griddle)

Convert a griddle into a list of parameter sets.

Parameters:

Name Type Description Default
griddle dict

griddle

 required

Returns:

Type Description
list[dict]

list of parameter sets
Source code in griddler/griddle.py 
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
def parse(griddle: dict) -> list[dict]:
    """Convert a griddle into a list of parameter sets.

    Args:
        griddle: griddle

    Returns:
        list of parameter sets
    """
    validate(griddle)

    # start with the grid, and if there is no grid, consider the grid empty
    if "grid_parameters" in griddle:
        parameter_sets = [
            dict(zip(griddle["grid_parameters"].keys(), values))
            for values in itertools.product(*griddle["grid_parameters"].values())
        ]
    else:
        parameter_sets = [{}]

    # find where nested values will get merged, if they are present
    if "nested_parameters" in griddle:
        ps_nests = [
            _match_ps_nest(ps, griddle["nested_parameters"]) for ps in parameter_sets
        ]
    else:
        ps_nests = [None for ps in parameter_sets]

    # merge in baseline values, if present
    if "baseline_parameters" in griddle:
        parameter_sets = [ps | griddle["baseline_parameters"] for ps in parameter_sets]

    # update the nested values, if nests were present
    for ps, nest in zip(parameter_sets, ps_nests):
        if nest is not None:
            ps |= nest

    return parameter_sets

read(path)

Read a griddle file, and convert to parameter sets.

Parameters:

Name Type Description Default
path str

path to griddle

 required

Returns:

Type Description
list[dict]

list of parameter sets
Source code in griddler/griddle.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def read(path: str) -> list[dict]:
    """Read a griddle file, and convert to parameter sets.

    Args:
        path: path to griddle

    Returns:
        list of parameter sets
    """
    with open(path) as f:
        raw = yaml.safe_load(f)

    return parse(raw)

validate(griddle)

Validate that a griddle is well-formed

Parameters:

Name Type Description Default
griddle dict

dictionary

 required

Raises:

Type Description
AssertionError

if not valid
Source code in griddler/griddle.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def validate(griddle: dict):
    """Validate that a griddle is well-formed

    Args:
        griddle: dictionary

    Raises:
        AssertionError: if not valid
    """
    # griddle is a dict
    assert isinstance(griddle, dict)
    # griddle has only the known keys
    assert all(
        [
            key in ["baseline_parameters", "grid_parameters", "nested_parameters"]
            for key in griddle.keys()
        ]
    )
    # griddle has at least one of baseline or grid
    assert any(
        key in griddle.keys() for key in ["baseline_parameters", "grid_parameters"]
    )
    # can only have a nest if it has a grid
    assert (
        "nested_parameters" not in griddle.keys() or "grid_parameters" in griddle.keys()
    )