|
30 | 30 | from munch import Munch |
31 | 31 |
|
32 | 32 |
|
| 33 | +def collect_power_values( |
| 34 | + data: GwEntityData, loc: Munch, tariff: str, legacy: bool = False |
| 35 | +) -> None: |
| 36 | + """Something.""" |
| 37 | + for loc.peak_select in ("nl_peak", "nl_offpeak"): |
| 38 | + loc.locator = ( |
| 39 | + f'./{loc.log_type}[type="{loc.measurement}"]/period/' |
| 40 | + f'measurement[@{tariff}="{loc.peak_select}"]' |
| 41 | + ) |
| 42 | + if legacy: |
| 43 | + loc.locator = ( |
| 44 | + f"./{loc.meas_list[0]}_{loc.log_type}/measurement" |
| 45 | + f'[@directionality="{loc.meas_list[1]}"][@{tariff}="{loc.peak_select}"]' |
| 46 | + ) |
| 47 | + |
| 48 | + loc = power_data_peak_value(loc, legacy) |
| 49 | + if not loc.found: |
| 50 | + continue |
| 51 | + |
| 52 | + data = power_data_energy_diff(loc.measurement, loc.net_string, loc.f_val, data) |
| 53 | + key = cast(SensorType, loc.key_string) |
| 54 | + data["sensors"][key] = loc.f_val |
| 55 | + |
| 56 | + |
| 57 | +def power_data_peak_value(loc: Munch, legacy: bool) -> Munch: |
| 58 | + """Helper-function for _power_data_from_location() and _power_data_from_modules().""" |
| 59 | + loc.found = True |
| 60 | + if loc.logs.find(loc.locator) is None: |
| 61 | + loc = check_alternative_location(loc, legacy) |
| 62 | + if not loc.found: |
| 63 | + return loc |
| 64 | + |
| 65 | + if (peak := loc.peak_select.split("_")[1]) == "offpeak": |
| 66 | + peak = "off_peak" |
| 67 | + log_found = loc.log_type.split("_")[0] |
| 68 | + loc.key_string = f"{loc.measurement}_{peak}_{log_found}" |
| 69 | + if "gas" in loc.measurement or loc.log_type == "point_meter": |
| 70 | + loc.key_string = f"{loc.measurement}_{log_found}" |
| 71 | + # Only for P1 Actual -------------------# |
| 72 | + if "phase" in loc.measurement: |
| 73 | + loc.key_string = f"{loc.measurement}" |
| 74 | + # --------------------------------------# |
| 75 | + loc.net_string = f"net_electricity_{log_found}" |
| 76 | + val = loc.logs.find(loc.locator).text |
| 77 | + loc.f_val = power_data_local_format(loc.attrs, loc.key_string, val) |
| 78 | + |
| 79 | + return loc |
| 80 | + |
| 81 | + |
| 82 | +def power_data_energy_diff( |
| 83 | + measurement: str, |
| 84 | + net_string: SensorType, |
| 85 | + f_val: float | int, |
| 86 | + data: GwEntityData, |
| 87 | +) -> GwEntityData: |
| 88 | + """Calculate differential energy.""" |
| 89 | + if ( |
| 90 | + "electricity" in measurement |
| 91 | + and "phase" not in measurement |
| 92 | + and "interval" not in net_string |
| 93 | + ): |
| 94 | + diff = 1 |
| 95 | + if "produced" in measurement: |
| 96 | + diff = -1 |
| 97 | + if net_string not in data["sensors"]: |
| 98 | + tmp_val: float | int = 0 |
| 99 | + else: |
| 100 | + tmp_val = data["sensors"][net_string] |
| 101 | + |
| 102 | + if isinstance(f_val, int): |
| 103 | + tmp_val += f_val * diff |
| 104 | + else: |
| 105 | + tmp_val += float(f_val * diff) |
| 106 | + tmp_val = float(f"{round(tmp_val, 3):.3f}") |
| 107 | + |
| 108 | + data["sensors"][net_string] = tmp_val |
| 109 | + |
| 110 | + return data |
| 111 | + |
| 112 | + |
33 | 113 | class SmileCommon: |
34 | 114 | """The SmileCommon class.""" |
35 | 115 |
|
@@ -110,86 +190,6 @@ def _appl_thermostat_info( |
110 | 190 |
|
111 | 191 | return appl |
112 | 192 |
|
113 | | - def _collect_power_values( |
114 | | - self, data: GwEntityData, loc: Munch, tariff: str, legacy: bool = False |
115 | | - ) -> None: |
116 | | - """Something.""" |
117 | | - for loc.peak_select in ("nl_peak", "nl_offpeak"): |
118 | | - loc.locator = ( |
119 | | - f'./{loc.log_type}[type="{loc.measurement}"]/period/' |
120 | | - f'measurement[@{tariff}="{loc.peak_select}"]' |
121 | | - ) |
122 | | - if legacy: |
123 | | - loc.locator = ( |
124 | | - f"./{loc.meas_list[0]}_{loc.log_type}/measurement" |
125 | | - f'[@directionality="{loc.meas_list[1]}"][@{tariff}="{loc.peak_select}"]' |
126 | | - ) |
127 | | - |
128 | | - loc = self._power_data_peak_value(loc, legacy) |
129 | | - if not loc.found: |
130 | | - continue |
131 | | - |
132 | | - data = self._power_data_energy_diff( |
133 | | - loc.measurement, loc.net_string, loc.f_val, data |
134 | | - ) |
135 | | - key = cast(SensorType, loc.key_string) |
136 | | - data["sensors"][key] = loc.f_val |
137 | | - |
138 | | - def _power_data_peak_value(self, loc: Munch, legacy: bool) -> Munch: |
139 | | - """Helper-function for _power_data_from_location() and _power_data_from_modules().""" |
140 | | - loc.found = True |
141 | | - if loc.logs.find(loc.locator) is None: |
142 | | - loc = check_alternative_location(loc, legacy) |
143 | | - if not loc.found: |
144 | | - return loc |
145 | | - |
146 | | - if (peak := loc.peak_select.split("_")[1]) == "offpeak": |
147 | | - peak = "off_peak" |
148 | | - log_found = loc.log_type.split("_")[0] |
149 | | - loc.key_string = f"{loc.measurement}_{peak}_{log_found}" |
150 | | - if "gas" in loc.measurement or loc.log_type == "point_meter": |
151 | | - loc.key_string = f"{loc.measurement}_{log_found}" |
152 | | - # Only for P1 Actual -------------------# |
153 | | - if "phase" in loc.measurement: |
154 | | - loc.key_string = f"{loc.measurement}" |
155 | | - # --------------------------------------# |
156 | | - loc.net_string = f"net_electricity_{log_found}" |
157 | | - val = loc.logs.find(loc.locator).text |
158 | | - loc.f_val = power_data_local_format(loc.attrs, loc.key_string, val) |
159 | | - |
160 | | - return loc |
161 | | - |
162 | | - def _power_data_energy_diff( |
163 | | - self, |
164 | | - measurement: str, |
165 | | - net_string: SensorType, |
166 | | - f_val: float | int, |
167 | | - data: GwEntityData, |
168 | | - ) -> GwEntityData: |
169 | | - """Calculate differential energy.""" |
170 | | - if ( |
171 | | - "electricity" in measurement |
172 | | - and "phase" not in measurement |
173 | | - and "interval" not in net_string |
174 | | - ): |
175 | | - diff = 1 |
176 | | - if "produced" in measurement: |
177 | | - diff = -1 |
178 | | - if net_string not in data["sensors"]: |
179 | | - tmp_val: float | int = 0 |
180 | | - else: |
181 | | - tmp_val = data["sensors"][net_string] |
182 | | - |
183 | | - if isinstance(f_val, int): |
184 | | - tmp_val += f_val * diff |
185 | | - else: |
186 | | - tmp_val += float(f_val * diff) |
187 | | - tmp_val = float(f"{round(tmp_val, 3):.3f}") |
188 | | - |
189 | | - data["sensors"][net_string] = tmp_val |
190 | | - |
191 | | - return data |
192 | | - |
193 | 193 | def _create_gw_entities(self, appl: Munch) -> None: |
194 | 194 | """Helper-function for creating/updating gw_entities.""" |
195 | 195 | self.gw_entities[appl.entity_id] = {"dev_class": appl.pwclass} |
|
0 commit comments