sift_py.yaml.calculated_channels

 1from pathlib import Path
 2from typing import Any, Dict, List, cast
 3
 4import yaml
 5
 6from sift_py.calculated_channels.config import CalculatedChannelConfig
 7from sift_py.ingestion.config.yaml.error import YamlConfigError
 8from sift_py.yaml.utils import _handle_subdir
 9
10
11def load_calculated_channels(paths: List[Path]) -> List[CalculatedChannelConfig]:
12    """
13    Takes in a list of paths to YAML files which contains calculated channel configs and processes them into a list of
14    `CalculatedChannelConfig` objects. For more information on report templates see
15    `sift_py.report_templates.config.CalculatedChannelConfig`.
16    """
17    calculated_channel_configs: List[CalculatedChannelConfig] = []
18
19    def update_calculated_channels(path: Path):
20        calculated_channel_configs.extend(_read_calculated_channels_yaml(path))
21
22    for path in paths:
23        if path.is_dir():
24            _handle_subdir(path, update_calculated_channels)
25        elif path.is_file():
26            update_calculated_channels(path)
27    return calculated_channel_configs
28
29
30def _read_calculated_channels_yaml(path: Path) -> List[CalculatedChannelConfig]:
31    calculated_channel_configs = []
32    with open(path, "r") as f:
33        channel_config_yaml = cast(Dict[str, Any], yaml.safe_load(f.read()))
34
35        calculated_channel_list = channel_config_yaml.get("calculated_channels", [])
36        for calc_channel in calculated_channel_list:
37            if not isinstance(calc_channel, dict):
38                raise YamlConfigError(
39                    f"Expected 'calculated_channels' to be a list of dictionaries in yaml: '{path}'"
40                )
41            for channel_ref in calc_channel.get("channel_references", []):
42                parsed_channel_refs = []
43                if not isinstance(channel_ref, dict):
44                    raise YamlConfigError(
45                        f"Expected 'channel_references' to be a list of dictionaries in yaml: '{path}'"
46                    )
47                if "channel_reference" not in channel_ref:
48                    for k, v in channel_ref.items():
49                        parsed_channel_refs.append(dict(channel_reference=k, channel_identifier=v))
50                else:
51                    parsed_channel_refs.append(channel_ref)
52                calc_channel["channel_references"] = parsed_channel_refs
53
54        if not isinstance(calculated_channel_list, list):
55            raise YamlConfigError(f"Expected 'calculated_channels' to be a list in yaml: '{path}'")
56
57        for calc_channel in calculated_channel_list:
58            try:
59                calc_channel_cfg = CalculatedChannelConfig(**calc_channel)
60                calculated_channel_configs.append(calc_channel_cfg)
61            except Exception as e:
62                raise YamlConfigError(f"Error parsing calculated channel '{calc_channel}'") from e
63
64        return calculated_channel_configs