Coverage for openleadr/objects.py : 99%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: Apache-2.0
3# Copyright 2020 Contributors to OpenLEADR
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
9# http://www.apache.org/licenses/LICENSE-2.0
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
17from dataclasses import dataclass, field, asdict, is_dataclass
18from typing import List, Dict
19from datetime import datetime, timezone, timedelta
20from openleadr import utils
21from openleadr import enums
24@dataclass
25class AggregatedPNode:
26 node: str
29@dataclass
30class EndDeviceAsset:
31 mrid: str
34@dataclass
35class MeterAsset:
36 mrid: str
39@dataclass
40class PNode:
41 node: str
44@dataclass
45class FeatureCollection:
46 id: str
47 location: dict
50@dataclass
51class ServiceArea:
52 feature_collection: FeatureCollection
55@dataclass
56class ServiceDeliveryPoint:
57 node: str
60@dataclass
61class ServiceLocation:
62 node: str
65@dataclass
66class TransportInterface:
67 point_of_receipt: str
68 point_of_delivery: str
71@dataclass
72class Target:
73 aggregated_p_node: AggregatedPNode = None
74 end_device_asset: EndDeviceAsset = None
75 meter_asset: MeterAsset = None
76 p_node: PNode = None
77 service_area: ServiceArea = None
78 service_delivery_point: ServiceDeliveryPoint = None
79 service_location: ServiceLocation = None
80 transport_interface: TransportInterface = None
81 group_id: str = None
82 group_name: str = None
83 resource_id: str = None
84 ven_id: str = None
85 party_id: str = None
87 def __repr__(self):
88 targets = {key: value for key, value in asdict(self).items() if value is not None}
89 targets_str = ", ".join(f"{key}={value}" for key, value in targets.items())
90 return f"Target('{targets_str}')"
93@dataclass
94class EventDescriptor:
95 event_id: str
96 modification_number: int
97 market_context: str
98 event_status: str
100 created_date_time: datetime = None
101 modification_date_time: datetime = None
102 priority: int = 0
103 test_event: bool = False
104 vtn_comment: str = None
106 def __post_init__(self):
107 if self.modification_date_time is None:
108 self.modification_date_time = datetime.now(timezone.utc)
109 if self.created_date_time is None:
110 self.created_date_time = datetime.now(timezone.utc)
111 if self.modification_number is None:
112 self.modification_number = 0
115@dataclass
116class ActivePeriod:
117 dtstart: datetime
118 duration: timedelta
119 tolerance: dict = None
120 notification_period: dict = None
121 ramp_up_period: dict = None
122 recovery_period: dict = None
125@dataclass
126class Interval:
127 dtstart: datetime
128 duration: timedelta
129 signal_payload: float
130 uid: int = None
133@dataclass
134class SamplingRate:
135 min_period: timedelta = None
136 max_period: timedelta = None
137 on_change: bool = False
140@dataclass
141class PowerAttributes:
142 hertz: int = 50
143 voltage: int = 230
144 ac: bool = True
147@dataclass
148class Measurement:
149 name: str
150 description: str
151 unit: str
152 acceptable_units: List[str] = field(repr=False, default_factory=list)
153 scale: str = None
154 power_attributes: PowerAttributes = None
155 pulse_factor: int = None
156 ns: str = 'power'
158 def __post_init__(self):
159 if self.name not in enums._MEASUREMENT_NAMESPACES:
160 self.name = 'customUnit'
161 self.ns = enums._MEASUREMENT_NAMESPACES[self.name]
164@dataclass
165class EventSignal:
166 intervals: List[Interval]
167 signal_name: str
168 signal_type: str
169 signal_id: str
170 current_value: float = None
171 targets: List[Target] = None
172 targets_by_type: Dict = None
173 measurement: Measurement = None
175 def __post_init__(self):
176 if self.signal_type not in enums.SIGNAL_TYPE.values:
177 raise ValueError(f"""The signal_type must be one of '{"', '".join(enums.SIGNAL_TYPE.values)}', """
178 f"""you specified: '{self.signal_type}'.""")
179 if self.signal_name not in enums.SIGNAL_NAME.values and not self.signal_name.startswith('x-'):
180 raise ValueError(f"""The signal_name must be one of '{"', '".join(enums.SIGNAL_TYPE.values)}', """
181 f"""or it must begin with 'x-'. You specified: '{self.signal_name}'""")
182 if self.targets is None and self.targets_by_type is None:
183 return
184 elif self.targets_by_type is None:
185 list_of_targets = [asdict(target) if is_dataclass(target) else target for target in self.targets]
186 targets_by_type = utils.group_targets_by_type(list_of_targets)
187 if len(targets_by_type) > 1:
188 raise ValueError("In OpenADR, the EventSignal target may only be of type endDeviceAsset. "
189 f"You provided types: '{', '.join(targets_by_type)}'")
190 elif self.targets is None:
191 self.targets = [Target(**target) for target in utils.ungroup_targets_by_type(self.targets_by_type)]
192 elif self.targets is not None and self.targets_by_type is not None:
193 list_of_targets = [asdict(target) if is_dataclass(target) else target for target in self.targets]
194 if utils.group_targets_by_type(list_of_targets) != self.targets_by_type:
195 raise ValueError("You assigned both 'targets' and 'targets_by_type' in your event, "
196 "but the two were not consistent with each other. "
197 f"You supplied 'targets' = {self.targets} and "
198 f"'targets_by_type' = {self.targets_by_type}")
201@dataclass
202class Event:
203 event_descriptor: EventDescriptor
204 event_signals: List[EventSignal]
205 targets: List[Target] = None
206 targets_by_type: Dict = None
207 active_period: ActivePeriod = None
208 response_required: str = 'always'
210 def __post_init__(self):
211 if self.active_period is None:
212 dtstart = min([i['dtstart']
213 if isinstance(i, dict) else i.dtstart
214 for s in self.event_signals for i in s.intervals])
215 duration = max([i['dtstart'] + i['duration']
216 if isinstance(i, dict) else i.dtstart + i.duration
217 for s in self.event_signals for i in s.intervals]) - dtstart
218 self.active_period = ActivePeriod(dtstart=dtstart,
219 duration=duration)
220 if self.targets is None and self.targets_by_type is None:
221 raise ValueError("You must supply either 'targets' or 'targets_by_type'.")
222 elif self.targets_by_type is None:
223 list_of_targets = [asdict(target) if is_dataclass(target) else target for target in self.targets]
224 self.targets_by_type = utils.group_targets_by_type(list_of_targets)
225 elif self.targets is None:
226 self.targets = [Target(**target) for target in utils.ungroup_targets_by_type(self.targets_by_type)]
227 elif self.targets is not None and self.targets_by_type is not None:
228 list_of_targets = [asdict(target) if is_dataclass(target) else target for target in self.targets]
229 if utils.group_targets_by_type(list_of_targets) != self.targets_by_type:
230 raise ValueError("You assigned both 'targets' and 'targets_by_type' in your event, "
231 "but the two were not consistent with each other. "
232 f"You supplied 'targets' = {self.targets} and "
233 f"'targets_by_type' = {self.targets_by_type}")
234 # Set the event status
235 self.event_descriptor.event_status = utils.determine_event_status(self.active_period)
238@dataclass
239class Response:
240 response_code: int
241 response_description: str
242 request_id: str
245@dataclass
246class ReportDescription:
247 r_id: str # Identifies a specific datapoint in a report
248 market_context: str
249 reading_type: str
250 report_subject: Target
251 report_data_source: Target
252 report_type: str
253 sampling_rate: SamplingRate
254 measurement: Measurement = None
257@dataclass
258class ReportPayload:
259 r_id: str
260 value: float
261 confidence: int = None
262 accuracy: int = None
265@dataclass
266class ReportInterval:
267 dtstart: datetime
268 report_payload: ReportPayload
269 duration: timedelta = None
272@dataclass
273class Report:
274 report_specifier_id: str # This is what the VEN calls this report
275 report_name: str # Usually one of the default ones (enums.REPORT_NAME)
276 report_request_id: str = None # Usually empty
277 report_descriptions: List[ReportDescription] = None
278 created_date_time: datetime = None
280 dtstart: datetime = None # For delivering values
281 duration: timedelta = None # For delivering values
282 intervals: List[ReportInterval] = None # For delivering values
283 data_collection_mode: str = 'incremental'
285 def __post_init__(self):
286 if self.created_date_time is None:
287 self.created_date_time = datetime.now(timezone.utc)
288 if self.report_descriptions is None:
289 self.report_descriptions = []
292@dataclass
293class SpecifierPayload:
294 r_id: str
295 reading_type: str
296 measurement: Measurement = None
299@dataclass
300class ReportSpecifier:
301 report_specifier_id: str # This is what the VEN called this report
302 granularity: timedelta
303 specifier_payloads: List[SpecifierPayload]
304 report_interval: Interval = None
305 report_back_duration: timedelta = None
308@dataclass
309class ReportRequest:
310 report_request_id: str
311 report_specifier: ReportSpecifier