Skip to content

fitter

Master fitting and map making script. You typically want to run the witcher command instead of this.

deep_merge(a, b)

Based on https://gist.github.com/angstwad/bf22d1822c38a92ec0a9?permalink_comment_id=3517209

Source code in witch/fitter.py
208
209
210
211
212
213
214
215
216
217
218
219
def deep_merge(a: dict, b: dict) -> dict:
    """
    Based on https://gist.github.com/angstwad/bf22d1822c38a92ec0a9?permalink_comment_id=3517209
    """
    result = deepcopy(a)
    for bk, bv in b.items():
        av = result.get(bk)
        if isinstance(av, dict) and isinstance(bv, dict):
            result[bk] = deep_merge(av, bv)
        else:
            result[bk] = deepcopy(bv)
    return result

load_config(start_cfg, cfg_path)

We want to load a config and if it has the key "base", load that as well and merge them. We only want to take things from base that are not in the original config so we merge the original into the newly loaded one.

Source code in witch/fitter.py
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def load_config(start_cfg, cfg_path):
    """
    We want to load a config and if it has the key "base",
    load that as well and merge them.
    We only want to take things from base that are not in the original config
    so we merge the original into the newly loaded one.
    """
    with open(cfg_path) as file:
        new_cfg = yaml.safe_load(file)
    cfg = deep_merge(new_cfg, start_cfg)
    if "base" in new_cfg:
        base_path = new_cfg["base"]
        if not os.path.isabs(base_path):
            base_path = os.path.join(os.path.dirname(cfg_path), base_path)
        return load_config(cfg, base_path)
    return cfg

print_once(*args)

Helper function to print only once when running with MPI. Only the rank 0 process will print.

Parameters:

Name Type Description Default
*args Unpack[tuple[Any, ...]]

Arguments to pass to print.

()
Source code in witch/fitter.py
25
26
27
28
29
30
31
32
33
34
35
36
37
def print_once(*args: Unpack[tuple[Any, ...]]):
    """
    Helper function to print only once when running with MPI.
    Only the rank 0 process will print.

    Parameters
    ----------
    *args : Unpack[tuple[Any, ...]]
        Arguments to pass to print.
    """
    if minkasi.myrank == 0:
        print(*args)
        sys.stdout.flush()