utils.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642
  1. # SPDX-License-Identifier: Apache-2.0
  2. # Copyright 2020 Contributors to OpenLEADR
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. # http://www.apache.org/licenses/LICENSE-2.0
  7. # Unless required by applicable law or agreed to in writing, software
  8. # distributed under the License is distributed on an "AS IS" BASIS,
  9. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. # See the License for the specific language governing permissions and
  11. # limitations under the License.
  12. from datetime import datetime, timedelta, timezone
  13. from dataclasses import is_dataclass, asdict
  14. from collections import OrderedDict
  15. from openleadr import enums, objects
  16. import asyncio
  17. import re
  18. import ssl
  19. import hashlib
  20. import uuid
  21. import logging
  22. logger = logging.getLogger('openleadr')
  23. DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
  24. DATETIME_FORMAT_NO_MICROSECONDS = "%Y-%m-%dT%H:%M:%SZ"
  25. def generate_id(*args, **kwargs):
  26. """
  27. Generate a string that can be used as an identifier in OpenADR messages.
  28. """
  29. return str(uuid.uuid4())
  30. def flatten_xml(message):
  31. """
  32. Flatten the entire XML structure.
  33. """
  34. lines = [line.strip() for line in message.split("\n") if line.strip() != ""]
  35. for line in lines:
  36. line = re.sub(r'\n', '', line)
  37. line = re.sub(r'\s\s+', ' ', line)
  38. return "".join(lines)
  39. def normalize_dict(ordered_dict):
  40. """
  41. Main conversion function for the output of xmltodict to the OpenLEADR
  42. representation of OpenADR contents.
  43. :param ordered_dict dict: The OrderedDict, dict or dataclass that you wish to convert.
  44. """
  45. if is_dataclass(ordered_dict):
  46. ordered_dict = asdict(ordered_dict)
  47. def normalize_key(key):
  48. if key.startswith('oadr'):
  49. key = key[4:]
  50. elif key.startswith('ei'):
  51. key = key[2:]
  52. # Don't normalize the measurement descriptions
  53. if key in enums._MEASUREMENT_NAMESPACES:
  54. return key
  55. key = re.sub(r'([a-z])([A-Z])', r'\1_\2', key)
  56. if '-' in key:
  57. key = key.replace('-', '_')
  58. return key.lower()
  59. d = {}
  60. for key, value in ordered_dict.items():
  61. # Interpret values from the dict
  62. if key.startswith("@"):
  63. continue
  64. key = normalize_key(key)
  65. if isinstance(value, (OrderedDict, dict)):
  66. d[key] = normalize_dict(value)
  67. elif isinstance(value, list):
  68. d[key] = []
  69. for item in value:
  70. if isinstance(item, (OrderedDict, dict)):
  71. dict_item = normalize_dict(item)
  72. d[key].append(normalize_dict(dict_item))
  73. else:
  74. d[key].append(item)
  75. elif key in ("duration", "startafter", "max_period", "min_period"):
  76. d[key] = parse_duration(value)
  77. elif ("date_time" in key or key == "dtstart") and isinstance(value, str):
  78. d[key] = parse_datetime(value)
  79. elif value in ('true', 'false'):
  80. d[key] = parse_boolean(value)
  81. elif isinstance(value, str):
  82. if re.match(r'^-?\d+$', value):
  83. d[key] = int(value)
  84. elif re.match(r'^-?[\d.]+$', value):
  85. d[key] = float(value)
  86. else:
  87. d[key] = value
  88. else:
  89. d[key] = value
  90. # Do our best to make the dictionary structure as pythonic as possible
  91. if key.startswith("x_ei_"):
  92. d[key[5:]] = d.pop(key)
  93. key = key[5:]
  94. # Group all targets as a list of dicts under the key "target"
  95. if key == 'target':
  96. targets = d.pop(key)
  97. new_targets = []
  98. if targets:
  99. for ikey in targets:
  100. if isinstance(targets[ikey], list):
  101. new_targets.extend([{ikey: value} for value in targets[ikey]])
  102. else:
  103. new_targets.append({ikey: targets[ikey]})
  104. d[key + "s"] = new_targets
  105. key = key + "s"
  106. # Also add a targets_by_type element to this dict
  107. # to access the targets in a more convenient way.
  108. d['targets_by_type'] = group_targets_by_type(new_targets)
  109. # Group all reports as a list of dicts under the key "pending_reports"
  110. if key == "pending_reports":
  111. if isinstance(d[key], dict) and 'report_request_id' in d[key] \
  112. and isinstance(d[key]['report_request_id'], list):
  113. d['pending_reports'] = [{'request_id': rrid}
  114. for rrid in d['pending_reports']['report_request_id']]
  115. # Group all events al a list of dicts under the key "events"
  116. elif key == "event" and isinstance(d[key], list):
  117. events = d.pop("event")
  118. new_events = []
  119. for event in events:
  120. new_event = event['event']
  121. new_event['response_required'] = event['response_required']
  122. new_events.append(new_event)
  123. d["events"] = new_events
  124. # If there's only one event, also put it into a list
  125. elif key == "event" and isinstance(d[key], dict) and "event" in d[key]:
  126. oadr_event = d.pop('event')
  127. ei_event = oadr_event['event']
  128. ei_event['response_required'] = oadr_event['response_required']
  129. d['events'] = [ei_event]
  130. elif key in ("request_event", "created_event") and isinstance(d[key], dict):
  131. d = d[key]
  132. # Plurarize some lists
  133. elif key in ('report_request', 'report', 'specifier_payload'):
  134. if isinstance(d[key], list):
  135. d[key + 's'] = d.pop(key)
  136. else:
  137. d[key + 's'] = [d.pop(key)]
  138. elif key in ('report_description', 'event_signal'):
  139. descriptions = d.pop(key)
  140. if not isinstance(descriptions, list):
  141. descriptions = [descriptions]
  142. for description in descriptions:
  143. # We want to make the identification of the measurement universal
  144. for measurement in enums._MEASUREMENT_NAMESPACES:
  145. if measurement in description:
  146. name, item = measurement, description.pop(measurement)
  147. break
  148. else:
  149. break
  150. item['description'] = item.pop('item_description', None)
  151. item['unit'] = item.pop('item_units', None)
  152. if 'si_scale_code' in item:
  153. item['scale'] = item.pop('si_scale_code')
  154. if 'pulse_factor' in item:
  155. item['pulse_factor'] = item.pop('pulse_factor')
  156. description['measurement'] = {'name': name,
  157. **item}
  158. d[key + 's'] = descriptions
  159. # Promote the contents of the Qualified Event ID
  160. elif key == "qualified_event_id" and isinstance(d['qualified_event_id'], dict):
  161. qeid = d.pop('qualified_event_id')
  162. d['event_id'] = qeid['event_id']
  163. d['modification_number'] = qeid['modification_number']
  164. # Durations are encapsulated in their own object, remove this nesting
  165. elif isinstance(d[key], dict) and "duration" in d[key] and len(d[key]) == 1:
  166. d[key] = d[key]["duration"]
  167. # In general, remove all double nesting
  168. elif isinstance(d[key], dict) and key in d[key] and len(d[key]) == 1:
  169. d[key] = d[key][key]
  170. # In general, remove the double nesting of lists of items
  171. elif isinstance(d[key], dict) and key[:-1] in d[key] and len(d[key]) == 1:
  172. if isinstance(d[key][key[:-1]], list):
  173. d[key] = d[key][key[:-1]]
  174. else:
  175. d[key] = [d[key][key[:-1]]]
  176. # Payload values are wrapped in an object according to their type. We don't need that.
  177. elif key in ("signal_payload", "current_value"):
  178. value = d[key]
  179. if isinstance(d[key], dict):
  180. if 'payload_float' in d[key] and 'value' in d[key]['payload_float'] \
  181. and d[key]['payload_float']['value'] is not None:
  182. d[key] = float(d[key]['payload_float']['value'])
  183. elif 'payload_int' in d[key] and 'value' in d[key]['payload_int'] \
  184. and d[key]['payload_int'] is not None:
  185. d[key] = int(d[key]['payload_int']['value'])
  186. # Report payloads contain an r_id and a type-wrapped payload_float
  187. elif key == 'report_payload':
  188. if 'payload_float' in d[key] and 'value' in d[key]['payload_float']:
  189. v = d[key].pop('payload_float')
  190. d[key]['value'] = float(v['value'])
  191. elif 'payload_int' in d[key] and 'value' in d[key]['payload_int']:
  192. v = d[key].pop('payload_float')
  193. d[key]['value'] = int(v['value'])
  194. # All values other than 'false' must be interpreted as True for testEvent (rule 006)
  195. elif key == 'test_event' and not isinstance(d[key], bool):
  196. d[key] = True
  197. # Promote the 'text' item
  198. elif isinstance(d[key], dict) and "text" in d[key] and len(d[key]) == 1:
  199. if key == 'uid':
  200. d[key] = int(d[key]["text"])
  201. else:
  202. d[key] = d[key]["text"]
  203. # Promote a 'date-time' item
  204. elif isinstance(d[key], dict) and "date_time" in d[key] and len(d[key]) == 1:
  205. d[key] = d[key]["date_time"]
  206. # Promote 'properties' item, discard the unused? 'components' item
  207. elif isinstance(d[key], dict) and "properties" in d[key] and len(d[key]) <= 2:
  208. d[key] = d[key]["properties"]
  209. # Remove all empty dicts
  210. elif isinstance(d[key], dict) and len(d[key]) == 0:
  211. d.pop(key)
  212. return d
  213. def parse_datetime(value):
  214. """
  215. Parse an ISO8601 datetime into a datetime.datetime object.
  216. """
  217. matches = re.match(r'(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})\.?(\d{1,6})?\d*Z', value)
  218. if matches:
  219. year, month, day, hour, minute, second, micro = (int(value) for value in matches.groups())
  220. return datetime(year, month, day, hour, minute, second, micro, tzinfo=timezone.utc)
  221. else:
  222. logger.warning(f"parse_datetime: {value} did not match format")
  223. return value
  224. def parse_duration(value):
  225. """
  226. Parse a RFC5545 duration.
  227. """
  228. # TODO: implement the full regex:
  229. # matches = re.match(r'(\+|\-)?P((\d+Y)?(\d+M)?(\d+D)?T?(\d+H)?(\d+M)?(\d+S)?)|(\d+W)', value)
  230. if isinstance(value, timedelta):
  231. return value
  232. matches = re.match(r'P(\d+(?:D|W))?T?(\d+H)?(\d+M)?(\d+S)?', value)
  233. if not matches:
  234. return False
  235. days = hours = minutes = seconds = 0
  236. _days, _hours, _minutes, _seconds = matches.groups()
  237. if _days:
  238. if _days.endswith("D"):
  239. days = int(_days[:-1])
  240. elif _days.endswith("W"):
  241. days = int(_days[:-1]) * 7
  242. if _hours:
  243. hours = int(_hours[:-1])
  244. if _minutes:
  245. minutes = int(_minutes[:-1])
  246. if _seconds:
  247. seconds = int(_seconds[:-1])
  248. return timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
  249. def parse_boolean(value):
  250. if value == 'true':
  251. return True
  252. else:
  253. return False
  254. def datetimeformat(value, format=DATETIME_FORMAT):
  255. """
  256. Format a given datetime as a UTC ISO3339 string.
  257. """
  258. if not isinstance(value, datetime):
  259. return value
  260. return value.astimezone(timezone.utc).strftime(format)
  261. def timedeltaformat(value):
  262. """
  263. Format a timedelta to a RFC5545 Duration.
  264. """
  265. if not isinstance(value, timedelta):
  266. return value
  267. days = value.days
  268. hours, seconds = divmod(value.seconds, 3600)
  269. minutes, seconds = divmod(seconds, 60)
  270. formatted = "P"
  271. if days:
  272. formatted += f"{days}D"
  273. if hours or minutes or seconds:
  274. formatted += "T"
  275. if hours:
  276. formatted += f"{hours}H"
  277. if minutes:
  278. formatted += f"{minutes}M"
  279. if seconds:
  280. formatted += f"{seconds}S"
  281. return formatted
  282. def booleanformat(value):
  283. """
  284. Format a boolean value
  285. """
  286. if isinstance(value, bool):
  287. if value is True:
  288. return "true"
  289. elif value is False:
  290. return "false"
  291. elif value in ("true", "false"):
  292. return value
  293. else:
  294. raise ValueError(f"A boolean value must be provided, not {value}.")
  295. def ensure_bytes(obj):
  296. """
  297. Converts a utf-8 str object to bytes.
  298. """
  299. if obj is None:
  300. return obj
  301. if isinstance(obj, bytes):
  302. return obj
  303. if isinstance(obj, str):
  304. return bytes(obj, 'utf-8')
  305. else:
  306. raise TypeError("Must be bytes or str")
  307. def ensure_str(obj):
  308. """
  309. Converts bytes to a utf-8 string.
  310. """
  311. if obj is None:
  312. return None
  313. if isinstance(obj, str):
  314. return obj
  315. if isinstance(obj, bytes):
  316. return obj.decode('utf-8')
  317. else:
  318. raise TypeError("Must be bytes or str")
  319. def certificate_fingerprint_from_der(der_bytes):
  320. hash = hashlib.sha256(der_bytes).digest().hex()
  321. return ":".join([hash[i-2:i].upper() for i in range(-20, 0, 2)])
  322. def certificate_fingerprint(certificate_str):
  323. """
  324. Calculate the fingerprint for the given certificate, as defined by OpenADR.
  325. """
  326. der_bytes = ssl.PEM_cert_to_DER_cert(ensure_str(certificate_str))
  327. return certificate_fingerprint_from_der(der_bytes)
  328. def extract_pem_cert(tree):
  329. """
  330. Extract a given X509 certificate inside an XML tree and return the standard
  331. form of a PEM-encoded certificate.
  332. :param tree lxml.etree: The tree that contains the X509 element. This is
  333. usually the KeyInfo element from the XMLDsig Signature
  334. part of the message.
  335. """
  336. cert = tree.find('.//{http://www.w3.org/2000/09/xmldsig#}X509Certificate').text
  337. return "-----BEGIN CERTIFICATE-----\n" + cert + "-----END CERTIFICATE-----\n"
  338. def find_by(dict_or_list, key, value, *args):
  339. """
  340. Find a dict inside a dict or list by key, value properties.
  341. """
  342. search_params = [(key, value)]
  343. if args:
  344. search_params += [(args[i], args[i+1]) for i in range(0, len(args), 2)]
  345. if isinstance(dict_or_list, dict):
  346. dict_or_list = dict_or_list.values()
  347. for item in dict_or_list:
  348. if not isinstance(item, dict):
  349. _item = item.__dict__
  350. else:
  351. _item = item
  352. for key, value in search_params:
  353. if isinstance(value, tuple):
  354. if _item[key] not in value:
  355. break
  356. else:
  357. if _item[key] != value:
  358. break
  359. else:
  360. return item
  361. else:
  362. return None
  363. def group_by(list_, key, pop_key=False):
  364. """
  365. Return a dict that groups values
  366. """
  367. grouped = {}
  368. key_path = key.split(".")
  369. for item in list_:
  370. value = item
  371. for key in key_path:
  372. value = value.get(key)
  373. if value not in grouped:
  374. grouped[value] = []
  375. grouped[value].append(item)
  376. return grouped
  377. def cron_config(interval, randomize_seconds=False):
  378. """
  379. Returns a dict with cron settings for the given interval
  380. """
  381. if interval < timedelta(minutes=1):
  382. second = f"*/{interval.seconds}"
  383. minute = "*"
  384. hour = "*"
  385. elif interval < timedelta(hours=1):
  386. second = "0"
  387. minute = f"*/{int(interval.total_seconds()/60)}"
  388. hour = "*"
  389. elif interval < timedelta(hours=24):
  390. second = "0"
  391. minute = "0"
  392. hour = f"*/{int(interval.total_seconds()/3600)}"
  393. else:
  394. second = "0"
  395. minute = "0"
  396. hour = "0"
  397. cron_config = {"second": second, "minute": minute, "hour": hour}
  398. if randomize_seconds:
  399. jitter = min(int(interval.total_seconds() / 10), 300)
  400. cron_config['jitter'] = jitter
  401. return cron_config
  402. def get_cert_fingerprint_from_request(request):
  403. ssl_object = request.transport.get_extra_info('ssl_object')
  404. if ssl_object:
  405. der_bytes = ssl_object.getpeercert(binary_form=True)
  406. if der_bytes:
  407. return certificate_fingerprint_from_der(der_bytes)
  408. def group_targets_by_type(list_of_targets):
  409. targets_by_type = {}
  410. for target in list_of_targets:
  411. for key, value in target.items():
  412. if value is None:
  413. continue
  414. if key not in targets_by_type:
  415. targets_by_type[key] = []
  416. targets_by_type[key].append(value)
  417. return targets_by_type
  418. def ungroup_targets_by_type(targets_by_type):
  419. ungrouped_targets = []
  420. for target_type, targets in targets_by_type.items():
  421. if isinstance(targets, list):
  422. for target in targets:
  423. ungrouped_targets.append({target_type: target})
  424. elif isinstance(targets, str):
  425. ungrouped_targets.append({target_type: targets})
  426. return ungrouped_targets
  427. def validate_report_measurement_dict(measurement):
  428. from openleadr.enums import _ACCEPTABLE_UNITS, _MEASUREMENT_DESCRIPTIONS
  429. if 'name' not in measurement \
  430. or 'description' not in measurement \
  431. or 'unit' not in measurement:
  432. raise ValueError("The measurement dict must contain the following keys: "
  433. "'name', 'description', 'unit'. Please correct this.")
  434. name = measurement['name']
  435. description = measurement['description']
  436. unit = measurement['unit']
  437. # Validate the item name and description match
  438. if name in _MEASUREMENT_DESCRIPTIONS:
  439. required_description = _MEASUREMENT_DESCRIPTIONS[name]
  440. if description != required_description:
  441. if description.lower() == required_description.lower():
  442. logger.warning(f"The description for the measurement with name {name} "
  443. f"was not in the correct case; you provided {description} but "
  444. f"it should be {required_description}. "
  445. "This was automatically corrected.")
  446. measurement['description'] = required_description
  447. else:
  448. raise ValueError(f"The measurement's description {description} "
  449. f"did not match the expected description for this type "
  450. f" ({required_description}). Please correct this, or use "
  451. "'customUnit' as the name.")
  452. if unit not in _ACCEPTABLE_UNITS[name]:
  453. raise ValueError(f"The unit {unit} is not acceptable for measurement {name}. Allowed "
  454. f"units are {_ACCEPTABLE_UNITS[name]}.")
  455. else:
  456. if name != 'customUnit':
  457. logger.warning(f"You provided a measurement with an unknown name {name}. "
  458. "This was corrected to 'customUnit'. Please correct this in your "
  459. "report definition.")
  460. measurement['report_description']['name'] = 'customUnit'
  461. if 'power' in name:
  462. if 'power_attributes' in measurement:
  463. power_attributes = measurement['power_attributes']
  464. if 'voltage' not in power_attributes \
  465. or 'ac' not in power_attributes \
  466. or 'hertz' not in power_attributes:
  467. raise ValueError("The power_attributes of the measurement must contain the "
  468. "following keys: 'voltage' (int), 'ac' (bool), 'hertz' (int).")
  469. else:
  470. raise ValueError("A 'power' related measurement must contain a "
  471. "'power_attributes' section that contains the following "
  472. "keys: 'voltage' (int), 'ac' (boolean), 'hertz' (int)")
  473. def get_active_period_from_intervals(intervals, as_dict=True):
  474. if is_dataclass(intervals[0]):
  475. intervals = [asdict(i) for i in intervals]
  476. period_start = min([i['dtstart'] for i in intervals])
  477. period_duration = max([i['dtstart'] + i['duration'] - period_start for i in intervals])
  478. if as_dict:
  479. return {'dtstart': period_start,
  480. 'duration': period_duration}
  481. else:
  482. from openleadr.objects import ActivePeriod
  483. return ActivePeriod(dtstart=period_start, duration=period_duration)
  484. def determine_event_status(active_period):
  485. if is_dataclass(active_period):
  486. active_period = asdict(active_period)
  487. now = datetime.now(timezone.utc)
  488. if active_period['dtstart'].tzinfo is None:
  489. active_period['dtstart'] = active_period['dtstart'].astimezone(timezone.utc)
  490. active_period_start = active_period['dtstart']
  491. active_period_end = active_period['dtstart'] + active_period['duration']
  492. if now >= active_period_end:
  493. return 'completed'
  494. if now >= active_period_start:
  495. return 'active'
  496. if active_period.get('ramp_up_duration') is not None:
  497. ramp_up_start = active_period_start - active_period['ramp_up_duration']
  498. if now >= ramp_up_start:
  499. return 'near'
  500. return 'far'
  501. async def delayed_call(func, delay):
  502. if isinstance(delay, timedelta):
  503. delay = delay.total_seconds()
  504. await asyncio.sleep(delay)
  505. if asyncio.iscoroutinefunction(func):
  506. await func()
  507. elif asyncio.iscoroutine(func):
  508. await func
  509. else:
  510. func()
  511. def hasmember(obj, member):
  512. """
  513. Check if a dict or dataclass has the given member
  514. """
  515. if is_dataclass(obj):
  516. if hasattr(obj, member):
  517. return True
  518. else:
  519. if member in obj:
  520. return True
  521. return False
  522. def getmember(obj, member):
  523. """
  524. Get a member from a dict or dataclass
  525. """
  526. if is_dataclass(obj):
  527. return getattr(obj, member)
  528. else:
  529. return obj[member]
  530. def setmember(obj, member, value):
  531. """
  532. Set a member of a dict of dataclass
  533. """
  534. if is_dataclass(obj):
  535. setattr(obj, member, value)
  536. else:
  537. obj[member] = value
  538. def get_next_event_from_deque(deque):
  539. unused_elements = []
  540. event = None
  541. for i in range(len(deque)):
  542. msg = deque.popleft()
  543. if isinstance(msg, objects.Event) or (isinstance(msg, dict) and 'event_descriptor' in msg):
  544. event = msg
  545. break
  546. else:
  547. unused_elements.append(msg)
  548. deque.extend(unused_elements)
  549. return event