1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
# SPDX-License-Identifier: Apache-2.0
# Copyright 2020 Contributors to OpenLEADR
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime, timedelta, timezone
from dataclasses import asdict, is_dataclass
from openleadr import enums, utils
import logging
logger = logging.getLogger('openleadr')
def preflight_message(message_type, message_payload):
"""
Tests message contents before sending them. It will correct benign errors
and warn you about them. Uncorrectable errors will raise an Exception. It
changes the message_payload dict in-place.
:param message_type string: The type of message you are sending
:param message_payload dict: The contents of the message
"""
if f'_preflight_{message_type}' in globals():
message_payload = message_payload.copy()
for key, value in message_payload.items():
if isinstance(value, list):
message_payload[key] = [asdict(item) if is_dataclass(item) else item
for item in value]
else:
message_payload[key] = asdict(value) if is_dataclass(value) else value
globals()[f'_preflight_{message_type}'](message_payload)
return message_payload
def _preflight_oadrRegisterReport(message_payload):
for report in message_payload['reports']:
# Check that the report name is preceded by METADATA_ when registering reports
if report['report_name'] in enums.REPORT_NAME.values \
and not report['report_name'].startswith("METADATA"):
report['report_name'] = 'METADATA_' + report['report_name']
# Check that the measurement name and description match according to the schema
for report_description in report['report_descriptions']:
if 'measurement' in report_description and report_description['measurement'] is not None:
utils.validate_report_measurement_dict(report_description['measurement'])
# Add the correct namespace to the measurement
for report_description in report['report_descriptions']:
if 'measurement' in report_description and report_description['measurement'] is not None:
if report_description['measurement']['name'] in enums._MEASUREMENT_NAMESPACES:
measurement_name = report_description['measurement']['name']
measurement_ns = enums._MEASUREMENT_NAMESPACES[measurement_name]
report_description['measurement']['ns'] = measurement_ns
else:
raise ValueError("The Measurement Name is unknown")
def _preflight_oadrDistributeEvent(message_payload):
if 'parse_duration' not in globals():
from .utils import parse_duration
# Check that the total event_duration matches the sum of the interval durations (rule 8)
for event in message_payload['events']:
active_period_duration = event['active_period']['duration']
signal_durations = []
for signal in event['event_signals']:
signal_durations.append(sum([parse_duration(i['duration'])
for i in signal['intervals']], timedelta(seconds=0)))
if not all([d == active_period_duration for d in signal_durations]):
if not all([d == signal_durations[0] for d in signal_durations]):
raise ValueError("The different EventSignals have different total durations. "
"Please correct this.")
else:
logger.warning(f"The active_period duration for event "
f"{event['event_descriptor']['event_id']} ({active_period_duration})"
f" differs from the sum of the interval's durations "
f"({signal_durations[0]}). The active_period duration has been "
f"adjusted to ({signal_durations[0]}).")
event['active_period']['duration'] = signal_durations[0]
# Check that payload values with signal name SIMPLE are constricted (rule 9)
for event in message_payload['events']:
for event_signal in event['event_signals']:
if event_signal['signal_name'] == "SIMPLE":
for interval in event_signal['intervals']:
if interval['signal_payload'] not in (0, 1, 2, 3):
raise ValueError("Payload Values used with Signal Name SIMPLE "
"must be one of 0, 1, 2 or 3")
# Check that the current_value is 0 for SIMPLE events that are not yet active (rule 14)
for event in message_payload['events']:
for event_signal in event['event_signals']:
if 'current_value' in event_signal and event_signal['current_value'] != 0:
if event_signal['signal_name'] == "SIMPLE" \
and event['event_descriptor']['event_status'] != "ACTIVE":
logger.warning("The current_value for a SIMPLE event "
"that is not yet active must be 0. "
"This will be corrected.")
event_signal['current_value'] = 0
# Add the correct namespace to the measurement
for event in message_payload['events']:
for event_signal in event['event_signals']:
if 'measurement' in event_signal and event_signal['measurement'] is not None:
if event_signal['measurement']['name'] in enums._MEASUREMENT_NAMESPACES:
measurement_name = event_signal['measurement']['name']
measurement_ns = enums._MEASUREMENT_NAMESPACES[measurement_name]
event_signal['measurement']['ns'] = measurement_ns
else:
raise ValueError("The Measurement Name is unknown")
# Check that there is a valid oadrResponseRequired value for each Event
for event in message_payload['events']:
if 'response_required' not in event:
event['response_required'] = 'always'
elif event['response_required'] not in ('never', 'always'):
logger.warning(f"The response_required property in an Event "
f"should be 'never' or 'always', not "
f"{event['response_required']}. Changing to 'always'.")
event['response_required'] = 'always'
# Check that there is a valid oadrResponseRequired value for each Event
for event in message_payload['events']:
if 'created_date_time' not in event['event_descriptor'] \
or not event['event_descriptor']['created_date_time']:
logger.warning("Your event descriptor did not contain a created_date_time. "
"This will be automatically added.")
event['event_descriptor']['created_date_time'] = datetime.now(timezone.utc)
# Check that the target designations are correct and consistent
for event in message_payload['events']:
if 'targets' in event and 'targets_by_type' in event:
if utils.group_targets_by_type(event['targets']) != event['targets_by_type']:
raise ValueError("You assigned both 'targets' and 'targets_by_type' in your event, "
"but the two were not consistent with each other. "
f"You supplied 'targets' = {event['targets']} and "
f"'targets_by_type' = {event['targets_by_type']}")
elif 'targets_by_type' in event and 'targets' not in event:
event['targets'] = utils.ungroup_targets_by_type(event['targets_by_type'])
|
