# 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 openleadr.utils import generate_id, group_targets_by_type from openleadr.messaging import create_message, parse_message, validate_xml_schema from openleadr import enums from pprint import pprint from termcolor import colored from datetime import datetime, timezone, timedelta import pytest from pprint import pprint, pformat from lxml import etree from dataclasses import asdict import re DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" def create_dummy_event(ven_id): """ Creates a dummy event """ now = datetime.now(timezone.utc) event_id = generate_id() active_period = {"dtstart": (now + timedelta(minutes=1)), "duration": timedelta(minutes=9)} event_descriptor = {"event_id": event_id, "modification_number": 1, "modification_date_time": now, "priority": 1, "market_context": "http://MarketContext1", "created_date_time": now, "event_status": "near", "test_event": False, "vtn_comment": "This is an event"} event_signals = [{"intervals": [{"duration": timedelta(minutes=1), "uid": 0, "signal_payload": 8.0}, {"duration": timedelta(minutes=1), "uid": 1, "signal_payload": 10.0}, {"duration": timedelta(minutes=1), "uid": 2, "signal_payload": 12.0}, {"duration": timedelta(minutes=1), "uid": 3, "signal_payload": 14.0}, {"duration": timedelta(minutes=1), "uid": 4, "signal_payload": 16.0}, {"duration": timedelta(minutes=1), "uid": 5, "signal_payload": 18.0}, {"duration": timedelta(minutes=1), "uid": 6, "signal_payload": 20.0}, {"duration": timedelta(minutes=1), "uid": 7, "signal_payload": 10.0}, {"duration": timedelta(minutes=1), "uid": 8, "signal_payload": 20.0}], "signal_name": "LOAD_CONTROL", "signal_type": "x-loadControlCapacity", "signal_id": generate_id(), "measurement": {"name": "voltage", "description": "Voltage", "unit": "V", "scale": "none"}, "current_value": 0.0}] event_targets = [{"ven_id": 'VEN001'}, {"ven_id": 'VEN002'}] event = {'active_period': active_period, 'event_descriptor': event_descriptor, 'event_signals': event_signals, 'targets': event_targets, 'targets_by_type': group_targets_by_type(event_targets), 'response_required': 'always'} return event reports = [{'report_id': generate_id(), 'duration': timedelta(seconds=3600), 'report_descriptions': [{'r_id': generate_id(), 'report_subject': {'end_device_asset': {'mrid': 'meter001'}}, 'report_data_source': {'resource_id': 'resource001'}, 'report_type': 'usage', 'measurement': asdict(measurement), 'reading_type': 'Direct Read', 'market_context': 'http://MarketContext1', 'sampling_rate': {'min_period': timedelta(seconds=10), 'max_period': timedelta(seconds=30), 'on_change': False}} for measurement in enums.MEASUREMENTS.values], 'report_specifier_id': generate_id(), 'report_name': 'METADATA_HISTORY_USAGE', 'report_request_id': None, 'created_date_time': datetime.now(timezone.utc)}] for report in reports: for rd in report['report_descriptions']: rd['measurement'].pop('acceptable_units') rd['measurement'].pop('ns') if rd['measurement']['power_attributes'] is None: rd['measurement'].pop('power_attributes') if rd['measurement']['scale'] is None: rd['measurement'].pop('scale') if rd['measurement']['pulse_factor'] is None: rd['measurement'].pop('pulse_factor') testcases = [ ('oadrCanceledOpt', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, opt_id=generate_id())), ('oadrCanceledPartyRegistration', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, registration_id=generate_id(), ven_id='123ABC')), ('oadrCanceledReport', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, pending_reports=[{'request_id': generate_id()}, {'request_id': generate_id()}])), ('oadrCanceledReport', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, pending_reports=[{'request_id': generate_id()}, {'request_id': generate_id()}], ven_id='123ABC')), ('oadrCancelOpt', dict(request_id=generate_id(), ven_id='123ABC', opt_id=generate_id())), ('oadrCancelPartyRegistration', dict(request_id=generate_id(), ven_id='123ABC', registration_id=generate_id())), ('oadrCancelReport', dict(request_id=generate_id(), ven_id='123ABC', report_request_id=generate_id(), report_to_follow=True)), ('oadrCreatedEvent', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, event_responses=[{'response_code': 200, 'response_description': 'OK', 'request_id': generate_id(), 'event_id': generate_id(), 'modification_number': 1, 'opt_type': 'optIn'}, {'response_code': 200, 'response_description': 'OK', 'request_id': generate_id(), 'event_id': generate_id(), 'modification_number': 1, 'opt_type': 'optIn'}, {'response_code': 200, 'response_description': 'OK', 'request_id': generate_id(), 'event_id': generate_id(), 'modification_number': 1, 'opt_type': 'optIn'}], ven_id='123ABC')), ('oadrCreatedReport', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, pending_reports=[{'request_id': generate_id()}, {'request_id': generate_id()}], ven_id='123ABC')), ('oadrCreatedEvent', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, event_responses=[{'response_code': 200, 'response_description': 'OK', 'request_id': generate_id(), 'event_id': generate_id(), 'modification_number': 1, 'opt_type': 'optIn'}, {'response_code': 200, 'response_description': 'OK', 'request_id': generate_id(), 'event_id': generate_id(), 'modification_number': 1, 'opt_type': 'optOut'}], ven_id='123ABC')), ('oadrCreatedPartyRegistration', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, registration_id=generate_id(), ven_id='123ABC', profiles=[{'profile_name': '2.0b', 'transports': [{'transport_name': 'simpleHttp'}]}], vtn_id='VTN123')), ('oadrCreatedReport', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, pending_reports=[{'request_id': generate_id()}, {'request_id': generate_id()}])), ('oadrCreateOpt', dict(opt_id=generate_id(), opt_type='optIn', opt_reason='participating', created_date_time=datetime.now(timezone.utc), request_id=generate_id(), event_id=generate_id(), modification_number=1, targets=[{'ven_id': '123ABC'}], targets_by_type=group_targets_by_type([{'ven_id': '123ABC'}]), ven_id='VEN123')), ('oadrCreatePartyRegistration', dict(request_id=generate_id(), ven_id='123ABC', profile_name='2.0b', transport_name='simpleHttp', transport_address='http://localhost', report_only=False, xml_signature=False, ven_name='test', http_pull_model=True)), ('oadrCreateReport', dict(request_id=generate_id(), ven_id='123ABC', report_requests=[{'report_request_id': 'd2b7bade5f', 'report_specifier': {'granularity': timedelta(seconds=900), 'report_back_duration': timedelta(seconds=900), 'report_interval': {'dtstart': datetime(2019, 11, 19, 11, 0, 18, 672768, tzinfo=timezone.utc), 'duration': timedelta(seconds=7200)}, 'report_specifier_id': '9c8bdc00e7', 'specifier_payloads': [{'r_id': 'd6e2e07485', 'reading_type': 'Direct Read'}]}}])), ('oadrDistributeEvent', dict(request_id=generate_id(), response={'request_id': 123, 'response_code': 200, 'response_description': 'OK'}, events=[create_dummy_event(ven_id='123ABC')], vtn_id='VTN123')), ('oadrDistributeEvent', dict(request_id=generate_id(), response={'request_id': 123, 'response_code': 200, 'response_description': 'OK'}, events=[create_dummy_event(ven_id='123ABC'), create_dummy_event(ven_id='123ABC')], vtn_id='VTN123')), ('oadrPoll', dict(ven_id='123ABC')), ('oadrQueryRegistration', dict(request_id=generate_id())), ('oadrRegisteredReport', dict(ven_id='VEN123', response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, report_requests=[{'report_request_id': generate_id(), 'report_specifier': {'report_specifier_id': generate_id(), 'granularity': timedelta(minutes=15), 'report_back_duration': timedelta(minutes=15), 'report_interval': {'dtstart': datetime.now(timezone.utc), 'duration': timedelta(hours=2)}, 'specifier_payloads': [{'r_id': generate_id(), 'reading_type': 'Direct Read'}]}}, {'report_request_id': generate_id(), 'report_specifier': {'report_specifier_id': generate_id(), 'granularity': timedelta(minutes=15), 'report_back_duration': timedelta(minutes=15), 'report_interval': {'dtstart': datetime.now(timezone.utc), 'duration': timedelta(hours=2)}, 'specifier_payloads': [{'r_id': generate_id(), 'reading_type': 'Direct Read'}]}}])), ('oadrRequestEvent', dict(request_id=generate_id(), ven_id='123ABC')), ('oadrRequestReregistration', dict(ven_id='123ABC')), ('oadrRegisterReport', dict(request_id=generate_id(), reports=[{'report_id': generate_id(), 'report_descriptions': [{ 'r_id': generate_id(), 'report_subject': {'end_device_asset': {'mrid': 'meter001'}}, 'report_data_source': {'resource_id': '123ABC'}, 'report_type': 'reading', 'reading_type': 'Direct Read', 'market_context': 'http://localhost', 'sampling_rate': {'min_period': timedelta(minutes=1), 'max_period': timedelta(minutes=1), 'on_change': True}}], 'report_request_id': generate_id(), 'report_specifier_id': generate_id(), 'report_name': 'HISTORY_USAGE', 'created_date_time': datetime.now(timezone.utc)}], ven_id='123ABC', report_request_id=generate_id())), ('oadrRegisterReport', {'request_id': '8a4f859883', 'reports': [{'report_id': generate_id(), 'duration': timedelta(seconds=7200), 'report_descriptions': [{'r_id': generate_id(), 'report_data_source': {'resource_id': 'resource1'}, 'report_type': 'x-resourceStatus', 'reading_type': 'x-notApplicable', 'market_context': 'http://MarketContext1', 'sampling_rate': {'min_period': timedelta(seconds=60), 'max_period': timedelta(seconds=60), 'on_change': False}}], 'report_request_id': generate_id(), 'report_specifier_id': '789ed6cd4e_telemetry_status', 'report_name': 'METADATA_TELEMETRY_STATUS', 'created_date_time': datetime(2019, 11, 20, 15, 4, 52, 638621, tzinfo=timezone.utc)}, {'report_id': generate_id(), 'duration': timedelta(seconds=7200), 'report_descriptions': [{'r_id': 'resource1_energy', 'report_data_source': {'resource_id': 'resource1'}, 'report_type': 'usage', 'measurement': {'name': 'energyReal', 'description': 'RealEnergy', 'ns': 'power', 'unit': 'Wh', 'scale': 'n'}, 'reading_type': 'Direct Read', 'market_context': 'http://MarketContext1', 'sampling_rate': {'min_period': timedelta(seconds=60), 'max_period': timedelta(seconds=60), 'on_change': False}}, {'r_id': 'resource1_power', 'report_data_source': {'resource_id': 'resource1'}, 'report_type': 'usage', 'measurement': {'name': 'powerReal', 'description': 'RealPower', 'ns': 'power', 'unit': 'W', 'scale': 'n', 'power_attributes': {'hertz': 50, 'voltage': 230, 'ac': True}}, 'reading_type': 'Direct Read', 'market_context': 'http://MarketContext1', 'sampling_rate': {'min_period': timedelta(seconds=60), 'max_period': timedelta(seconds=60), 'on_change': False}}], 'report_request_id': generate_id(), 'report_specifier_id': '789ed6cd4e_telemetry_usage', 'report_name': 'METADATA_TELEMETRY_USAGE', 'created_date_time': datetime(2019, 11, 20, 15, 4, 52, 638621, tzinfo=timezone.utc)}, {'report_id': generate_id(), 'duration': timedelta(seconds=7200), 'report_descriptions': [{'r_id': 'resource1_energy', 'report_data_source': {'resource_id': 'resource1'}, 'report_type': 'usage', 'measurement': {'name': 'energyReal', 'description': 'RealEnergy', 'ns': 'power', 'unit': 'Wh', 'scale': 'n'}, 'reading_type': 'Direct Read', 'market_context': 'http://MarketContext1', 'sampling_rate': {'min_period': timedelta(seconds=60), 'max_period': timedelta(seconds=60), 'on_change': False}}, {'r_id': 'resource1_power', 'report_data_source': {'resource_id': 'resource1'}, 'report_type': 'usage', 'measurement': {'name': 'powerReal', 'description': 'RealPower', 'ns': 'power', 'unit': 'W', 'scale': 'n', 'power_attributes': {'hertz': 50, 'voltage': 230, 'ac': True}}, 'reading_type': 'Direct Read', 'market_context': 'http://MarketContext1', 'sampling_rate': {'min_period': timedelta(seconds=60), 'max_period': timedelta(seconds=60), 'on_change': False}}], 'report_request_id': generate_id(), 'report_specifier_id': '789ed6cd4e_history_usage', 'report_name': 'METADATA_HISTORY_USAGE', 'created_date_time': datetime(2019, 11, 20, 15, 4, 52, 638621, tzinfo=timezone.utc)}], 'ven_id': 's3cc244ee6'}), ('oadrRegisterReport', {'ven_id': 'ven123', 'request_id': generate_id(), 'reports': reports}), ('oadrResponse', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, ven_id='123ABC')), ('oadrResponse', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': None}, ven_id='123ABC')), ('oadrUpdatedReport', dict(response={'response_code': 200, 'response_description': 'OK', 'request_id': generate_id()}, ven_id='123ABC', cancel_report={'request_id': generate_id(), 'report_request_id': [generate_id(), generate_id(), generate_id()], 'report_to_follow': False, 'ven_id': '123ABC'})), ('oadrUpdateReport', dict(request_id=generate_id(), reports=[{'report_id': generate_id(), 'report_name': enums.REPORT_NAME.values[0], 'created_date_time': datetime.now(timezone.utc), 'report_request_id': generate_id(), 'report_specifier_id': generate_id(), 'report_descriptions': [{'r_id': generate_id(), 'report_subject': {'end_device_asset': {'mrid': 'meter001'}}, 'report_data_source': {'resource_id': '123ABC'}, 'report_type': enums.REPORT_TYPE.values[0], 'reading_type': enums.READING_TYPE.values[0], 'market_context': 'http://localhost', 'sampling_rate': {'min_period': timedelta(minutes=1), 'max_period': timedelta(minutes=2), 'on_change': False}} ]}], ven_id='123ABC')) ] @pytest.mark.parametrize('message_type,data', testcases) def test_message(message_type, data): # file = open('representations.rst', 'a') # print(f".. _{message_type}:", file=file) # print("", file=file) # print(message_type, file=file) # print("="*len(message_type), file=file) # print("", file=file) # print("OpenADR payload:", file=file) # print("", file=file) # print(".. code-block:: xml", file=file) # print(" ", file=file) message = create_message(message_type, **data) # message = re.sub(r"\s\s+","",message) # message = message.replace("\n","") # xml_lines = etree.tostring(etree.fromstring(message.replace('\n', '').encode('utf-8')), pretty_print=True).decode('utf-8').splitlines() # for line in xml_lines: # print(" " + line, file=file) # print("", file=file) # print("OpenLEADR representation:", file=file) # print(" ", file=file) # print(".. code-block:: python3", file=file) # print(" ", file=file) validate_xml_schema(message) parsed = parse_message(message)[1] # dict_lines = pformat(parsed).splitlines() # for line in dict_lines: # print(" " + line, file=file) # print("", file=file) # print("", file=file) if message_type == 'oadrRegisterReport': for report in data['reports']: for rd in report['report_descriptions']: if 'measurement' in rd: rd['measurement'].pop('ns') if message_type == 'oadrDistributeEvent': for event in data['events']: for signal in event['event_signals']: if 'measurement' in signal: signal['measurement'].pop('ns') assert parsed == data