1"""Config parsing and window-shape normalisation helpers.
2
3These helpers translate a device's raw ``config`` dict into the flat list
4of parameter codes we expect to see in sensor data, and rename the
5dust-channel columns into the ``z1..z4`` scheme that the threshold table
6uses.
7
8No I/O happens here — everything is pure transformation on dicts and
9column-oriented numpy windows.
10"""
11
12from __future__ import annotations
13
14import logging
15from typing import Any, Dict, List, Optional, Tuple
16
17from OzWrapper.OzAnomaly.data_provider import Window, is_empty
18from OzWrapper.OzAnomaly.thresholds import PARA_CHECK_IGNORE_LIST, PROCESS_CONFIG_IGNORE_LIST
19
20logger = logging.getLogger(__name__)
21
22
[docs]
23def get_ogs_list(
24 config: Dict[str, Any], device_type: Optional[str]
25) -> List[Dict[str, Any]]:
26 """Return the OGS sensor list for skip-check lookups.
27
28 DUSTROID devices don't carry gas cartridges, so their config does not
29 hold OGS entries with the ``lb`` label that the skip lists key off. We
30 return an empty list for that device type so the skip helpers simply
31 fall through without raising. For every other device type we return the
32 raw ``config['ogs']`` list (or an empty list if absent).
33 """
34 if device_type and device_type.lower() == "dustroid":
35 return []
36 return config.get("ogs", []) or []
37
38
[docs]
39def process_config(config: Dict[str, Any]) -> List[str]:
40 """Flatten a device config into a sorted list of expected parameter codes.
41
42 Walks the nested config dict, collects every enabled sensor's `sc` value
43 plus the gas keys under `soft.config`, and filters out metadata keys
44 (`lat`, `lon`, etc.). For every `p1..p4` present we also emit `p11..p41`
45 as expected data channels, then drop the `p1..p4` stubs.
46 """
47
48 def extract_sc_values(data: Any) -> List[str]:
49 sc_values: List[str] = []
50
51 def extract_recursively(node: Any) -> None:
52 if isinstance(node, list):
53 for item in node:
54 extract_recursively(item)
55 elif isinstance(node, dict):
56 if "en" in node and node["en"] == 0:
57 return
58 if "sc" in node:
59 sc_values.append(node["sc"])
60 for value in node.values():
61 extract_recursively(value)
62
63 extract_recursively(data)
64 return sc_values
65
66 def get_gases(data: Dict[str, Any]) -> List[str]:
67 keys_to_drop = ["p1", "p2", "p3", "p4"]
68 all_keys: List[str] = []
69
70 if "soft" in data and "config" in data["soft"]:
71 soft_conf = data["soft"]["config"]
72 for gas in soft_conf:
73 include_gas = False
74 for ogs_sensor in data.get("ogs", []):
75 if ogs_sensor.get("en", 0) == 0:
76 continue
77 for param in ogs_sensor.get("parameters", []):
78 sc = param.get("sc")
79 if not sc:
80 continue
81 if sc == gas or sc.startswith(gas):
82 include_gas = True
83 break
84 if include_gas:
85 break
86 if include_gas:
87 all_keys.append(gas)
88
89 if "p1" in all_keys:
90 all_keys.append("p11")
91 if "p2" in all_keys:
92 all_keys.append("p21")
93 if "p3" in all_keys:
94 all_keys.append("p31")
95 if "p4" in all_keys:
96 all_keys.append("p41")
97
98 for key in keys_to_drop:
99 if key in all_keys:
100 all_keys.remove(key)
101
102 return all_keys
103
104 combined_list = extract_sc_values(config)
105 combined_list.extend(get_gases(config))
106 combined_list = [k for k in combined_list if k not in PROCESS_CONFIG_IGNORE_LIST]
107 return sorted(combined_list)
108
109
[docs]
110def para_check(window: Optional[Window], config: Dict[str, Any]) -> Dict[str, int]:
111 """Flag parameters expected by config that never appear in the window.
112
113 For each parameter returned by :func:`process_config` that is not a key
114 in ``window`` (and not in the ignore list), emit a ``{param}_e6: 1``
115 entry.
116 """
117 if is_empty(window):
118 return {}
119
120 para_list = process_config(config)
121 window_keys = list(window.keys())
122 return {
123 f"{para}_e6": 1
124 for para in para_list
125 if para not in window_keys and para not in PARA_CHECK_IGNORE_LIST
126 }
127
128
[docs]
129def should_drop(element: str, dust_list: List[str]) -> bool:
130 """Return True if `element` has any sibling raw-dust channel (`pN1..pN4`).
131
132 Used to pick the raw dust columns over the aggregated ones when both
133 are present.
134 """
135 num = element[1:]
136 return (
137 f"p{num}1" in dust_list
138 or f"p{num}2" in dust_list
139 or f"p{num}3" in dust_list
140 or f"p{num}4" in dust_list
141 )
142
143
[docs]
144def filter_and_rename_columns(
145 window: Optional[Window],
146) -> Tuple[Optional[Window], Dict[str, str]]:
147 """Rename dust columns to ``z1..z4`` for threshold lookups.
148
149 Maps either ``p1..p4`` or ``p11..p41`` to ``z1..z4`` depending on which
150 set is the raw-value set in the window. The rename mapping is returned
151 so the caller can restore original names on the output keys.
152 """
153 if is_empty(window):
154 return window, {}
155
156 keys = list(window.keys())
157 dust_list = [
158 col for col in keys
159 if col.startswith("p") and col.endswith(("1", "2", "3", "4"))
160 ]
161 # Sort before comparing — `keys` is derived from a dict built off a
162 # `set` upstream, so insertion order is arbitrary. Without the sort,
163 # the equality check below silently falls through to the empty
164 # rename mapping and dust columns are never renamed to z1..z4.
165 dust_list_filtered = sorted(col for col in dust_list if not should_drop(col, dust_list))
166
167 if dust_list_filtered == ["p1", "p2", "p3", "p4"]:
168 rename_mapping = {f"p{i}": f"z{i}" for i in range(1, 5)}
169 elif dust_list_filtered == ["p11", "p21", "p31", "p41"]:
170 rename_mapping = {f"p{i}1": f"z{i}" for i in range(1, 5)}
171 else:
172 rename_mapping = {}
173
174 if not rename_mapping:
175 return window, {}
176
177 new_window: Window = {
178 rename_mapping.get(k, k): v for k, v in window.items()
179 }
180 return new_window, rename_mapping
181
182
[docs]
183def replace_keys_with_original(
184 output_dict: Dict[str, Any], rename_dict: Dict[str, str]
185) -> Dict[str, Any]:
186 """Undo the ``p -> z`` column rename on output field keys.
187
188 When :func:`filter_and_rename_columns` renamed ``p11 -> z1``, any
189 output key that starts with ``z1`` is restored to ``p11`` so the
190 caller sees the original parameter naming.
191 """
192 new_output_dict: Dict[str, Any] = {}
193 for key, value in output_dict.items():
194 found_replacement = False
195 for coded_key_rename, original_key in rename_dict.items():
196 if key.startswith(original_key):
197 new_key = key.replace(original_key, coded_key_rename)
198 new_output_dict[new_key] = value
199 found_replacement = True
200 break
201 if not found_replacement:
202 new_output_dict[key] = value
203 return new_output_dict
204
205
[docs]
206def select_columns(window: Optional[Window]) -> List[str]:
207 """Pick the subset of window columns the per-parameter checks care about.
208
209 Returns gases (``g*1``, ``g*5``, ``o*1``, ``v*1``, ``v*5``), dust
210 (``z1..z4``), and a small fixed list of environmental columns
211 (``light, leq, t1, t2, temp, hum, pr``).
212 """
213 if is_empty(window):
214 return []
215
216 keys = list(window.keys())
217 gases_list = [
218 col for col in keys
219 if col.startswith(("g", "o", "v")) and col.endswith(("1", "5"))
220 ]
221 dust_list = [
222 col for col in keys
223 if col.startswith("z") and col.endswith(("1", "2", "3", "4"))
224 ]
225 other_params = ["light", "leq", "t1", "t2", "temp", "hum", "pr"]
226 matching_columns = [col for col in keys if col in other_params]
227 return gases_list + dust_list + matching_columns