utils.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651
  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 = (int(value)for value in matches.groups()[:-1])
  220. micro = matches.groups()[-1]
  221. if micro is None:
  222. micro = 0
  223. else:
  224. if len(micro) == 6:
  225. micro = int(micro)
  226. else:
  227. micro = int(micro + "0" * (6 - len(micro)))
  228. return datetime(year, month, day, hour, minute, second, micro, tzinfo=timezone.utc)
  229. else:
  230. logger.warning(f"parse_datetime: {value} did not match format")
  231. return value
  232. def parse_duration(value):
  233. """
  234. Parse a RFC5545 duration.
  235. """
  236. if isinstance(value, timedelta):
  237. return value
  238. regex = r'(\+|\-)?P(?:(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)D)?T?(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)|(?:(\d+)W)'
  239. matches = re.match(regex, value)
  240. if not matches:
  241. raise ValueError(f"The duration '{value}' did not match the requested format")
  242. years, months, days, hours, minutes, seconds, weeks = (int(g) if g else 0 for g in matches.groups()[1:])
  243. if years != 0:
  244. logger.warning("Received a duration that specifies years, which is not a determinate duration. "
  245. "It will be interpreted as 1 year = 365 days.")
  246. days = days + 365 * years
  247. if months != 0:
  248. logger.warning("Received a duration that specifies months, which is not a determinate duration "
  249. "It will be interpreted as 1 month = 30 days.")
  250. days = days + 30 * months
  251. duration = timedelta(weeks=weeks, days=days, hours=hours, minutes=minutes, seconds=seconds)
  252. if matches.groups()[0] == "-":
  253. duration = -1 * duration
  254. return duration
  255. def parse_boolean(value):
  256. if value == 'true':
  257. return True
  258. else:
  259. return False
  260. def datetimeformat(value, format=DATETIME_FORMAT):
  261. """
  262. Format a given datetime as a UTC ISO3339 string.
  263. """
  264. if not isinstance(value, datetime):
  265. return value
  266. return value.astimezone(timezone.utc).strftime(format)
  267. def timedeltaformat(value):
  268. """
  269. Format a timedelta to a RFC5545 Duration.
  270. """
  271. if not isinstance(value, timedelta):
  272. return value
  273. days = value.days
  274. hours, seconds = divmod(value.seconds, 3600)
  275. minutes, seconds = divmod(seconds, 60)
  276. formatted = "P"
  277. if days:
  278. formatted += f"{days}D"
  279. if hours or minutes or seconds:
  280. formatted += "T"
  281. if hours:
  282. formatted += f"{hours}H"
  283. if minutes:
  284. formatted += f"{minutes}M"
  285. if seconds:
  286. formatted += f"{seconds}S"
  287. return formatted
  288. def booleanformat(value):
  289. """
  290. Format a boolean value
  291. """
  292. if isinstance(value, bool):
  293. if value is True:
  294. return "true"
  295. elif value is False:
  296. return "false"
  297. elif value in ("true", "false"):
  298. return value
  299. else:
  300. raise ValueError(f"A boolean value must be provided, not {value}.")
  301. def ensure_bytes(obj):
  302. """
  303. Converts a utf-8 str object to bytes.
  304. """
  305. if obj is None:
  306. return obj
  307. if isinstance(obj, bytes):
  308. return obj
  309. if isinstance(obj, str):
  310. return bytes(obj, 'utf-8')
  311. else:
  312. raise TypeError("Must be bytes or str")
  313. def ensure_str(obj):
  314. """
  315. Converts bytes to a utf-8 string.
  316. """
  317. if obj is None:
  318. return None
  319. if isinstance(obj, str):
  320. return obj
  321. if isinstance(obj, bytes):
  322. return obj.decode('utf-8')
  323. else:
  324. raise TypeError("Must be bytes or str")
  325. def certificate_fingerprint_from_der(der_bytes):
  326. hash = hashlib.sha256(der_bytes).digest().hex()
  327. return ":".join([hash[i-2:i].upper() for i in range(-20, 0, 2)])
  328. def certificate_fingerprint(certificate_str):
  329. """
  330. Calculate the fingerprint for the given certificate, as defined by OpenADR.
  331. """
  332. der_bytes = ssl.PEM_cert_to_DER_cert(ensure_str(certificate_str))
  333. return certificate_fingerprint_from_der(der_bytes)
  334. def extract_pem_cert(tree):
  335. """
  336. Extract a given X509 certificate inside an XML tree and return the standard
  337. form of a PEM-encoded certificate.
  338. :param tree lxml.etree: The tree that contains the X509 element. This is
  339. usually the KeyInfo element from the XMLDsig Signature
  340. part of the message.
  341. """
  342. cert = tree.find('.//{http://www.w3.org/2000/09/xmldsig#}X509Certificate').text
  343. return "-----BEGIN CERTIFICATE-----\n" + cert + "-----END CERTIFICATE-----\n"
  344. def find_by(dict_or_list, key, value, *args):
  345. """
  346. Find a dict inside a dict or list by key, value properties.
  347. """
  348. search_params = [(key, value)]
  349. if args:
  350. search_params += [(args[i], args[i+1]) for i in range(0, len(args), 2)]
  351. if isinstance(dict_or_list, dict):
  352. dict_or_list = dict_or_list.values()
  353. for item in dict_or_list:
  354. if not isinstance(item, dict):
  355. _item = item.__dict__
  356. else:
  357. _item = item
  358. for key, value in search_params:
  359. if isinstance(value, tuple):
  360. if key not in _item or _item[key] not in value:
  361. break
  362. else:
  363. if key not in _item or _item[key] != value:
  364. break
  365. else:
  366. return item
  367. else:
  368. return None
  369. def group_by(list_, key, pop_key=False):
  370. """
  371. Return a dict that groups values
  372. """
  373. grouped = {}
  374. key_path = key.split(".")
  375. for item in list_:
  376. value = item
  377. for key in key_path:
  378. value = value.get(key)
  379. if value not in grouped:
  380. grouped[value] = []
  381. grouped[value].append(item)
  382. return grouped
  383. def cron_config(interval, randomize_seconds=False):
  384. """
  385. Returns a dict with cron settings for the given interval
  386. """
  387. if interval < timedelta(minutes=1):
  388. second = f"*/{interval.seconds}"
  389. minute = "*"
  390. hour = "*"
  391. elif interval < timedelta(hours=1):
  392. second = "0"
  393. minute = f"*/{int(interval.total_seconds()/60)}"
  394. hour = "*"
  395. elif interval < timedelta(hours=24):
  396. second = "0"
  397. minute = "0"
  398. hour = f"*/{int(interval.total_seconds()/3600)}"
  399. else:
  400. second = "0"
  401. minute = "0"
  402. hour = "0"
  403. cron_config = {"second": second, "minute": minute, "hour": hour}
  404. if randomize_seconds:
  405. jitter = min(int(interval.total_seconds() / 10), 300)
  406. cron_config['jitter'] = jitter
  407. return cron_config
  408. def get_cert_fingerprint_from_request(request):
  409. ssl_object = request.transport.get_extra_info('ssl_object')
  410. if ssl_object:
  411. der_bytes = ssl_object.getpeercert(binary_form=True)
  412. if der_bytes:
  413. return certificate_fingerprint_from_der(der_bytes)
  414. def group_targets_by_type(list_of_targets):
  415. targets_by_type = {}
  416. for target in list_of_targets:
  417. for key, value in target.items():
  418. if value is None:
  419. continue
  420. if key not in targets_by_type:
  421. targets_by_type[key] = []
  422. targets_by_type[key].append(value)
  423. return targets_by_type
  424. def ungroup_targets_by_type(targets_by_type):
  425. ungrouped_targets = []
  426. for target_type, targets in targets_by_type.items():
  427. if isinstance(targets, list):
  428. for target in targets:
  429. ungrouped_targets.append({target_type: target})
  430. elif isinstance(targets, str):
  431. ungrouped_targets.append({target_type: targets})
  432. return ungrouped_targets
  433. def validate_report_measurement_dict(measurement):
  434. from openleadr.enums import _ACCEPTABLE_UNITS, _MEASUREMENT_DESCRIPTIONS
  435. if 'name' not in measurement \
  436. or 'description' not in measurement \
  437. or 'unit' not in measurement:
  438. raise ValueError("The measurement dict must contain the following keys: "
  439. "'name', 'description', 'unit'. Please correct this.")
  440. name = measurement['name']
  441. description = measurement['description']
  442. unit = measurement['unit']
  443. # Validate the item name and description match
  444. if name in _MEASUREMENT_DESCRIPTIONS:
  445. required_description = _MEASUREMENT_DESCRIPTIONS[name]
  446. if description != required_description:
  447. if description.lower() == required_description.lower():
  448. logger.warning(f"The description for the measurement with name '{name}' "
  449. f"was not in the correct case; you provided '{description}' but "
  450. f"it should be '{required_description}'. "
  451. "This was automatically corrected.")
  452. measurement['description'] = required_description
  453. else:
  454. raise ValueError(f"The measurement's description '{description}' "
  455. f"did not match the expected description for this type "
  456. f" ('{required_description}'). Please correct this, or use "
  457. "'customUnit' as the name.")
  458. if unit not in _ACCEPTABLE_UNITS[name]:
  459. raise ValueError(f"The unit '{unit}' is not acceptable for measurement '{name}'. Allowed "
  460. f"units are: '" + "', '".join(_ACCEPTABLE_UNITS[name]) + "'.")
  461. else:
  462. if name != 'customUnit':
  463. logger.warning(f"You provided a measurement with an unknown name {name}. "
  464. "This was corrected to 'customUnit'. Please correct this in your "
  465. "report definition.")
  466. measurement['name'] = 'customUnit'
  467. if 'power' in name:
  468. if 'power_attributes' in measurement:
  469. power_attributes = measurement['power_attributes']
  470. if 'voltage' not in power_attributes \
  471. or 'ac' not in power_attributes \
  472. or 'hertz' not in power_attributes:
  473. raise ValueError("The power_attributes of the measurement must contain the "
  474. "following keys: 'voltage' (int), 'ac' (bool), 'hertz' (int).")
  475. else:
  476. raise ValueError("A 'power' related measurement must contain a "
  477. "'power_attributes' section that contains the following "
  478. "keys: 'voltage' (int), 'ac' (boolean), 'hertz' (int)")
  479. def get_active_period_from_intervals(intervals, as_dict=True):
  480. if is_dataclass(intervals[0]):
  481. intervals = [asdict(i) for i in intervals]
  482. period_start = min([i['dtstart'] for i in intervals])
  483. period_duration = max([i['dtstart'] + i['duration'] - period_start for i in intervals])
  484. if as_dict:
  485. return {'dtstart': period_start,
  486. 'duration': period_duration}
  487. else:
  488. from openleadr.objects import ActivePeriod
  489. return ActivePeriod(dtstart=period_start, duration=period_duration)
  490. def determine_event_status(active_period):
  491. if is_dataclass(active_period):
  492. active_period = asdict(active_period)
  493. now = datetime.now(timezone.utc)
  494. if active_period['dtstart'].tzinfo is None:
  495. active_period['dtstart'] = active_period['dtstart'].astimezone(timezone.utc)
  496. active_period_start = active_period['dtstart']
  497. active_period_end = active_period['dtstart'] + active_period['duration']
  498. if now >= active_period_end:
  499. return 'completed'
  500. if now >= active_period_start:
  501. return 'active'
  502. if active_period.get('ramp_up_duration') is not None:
  503. ramp_up_start = active_period_start - active_period['ramp_up_duration']
  504. if now >= ramp_up_start:
  505. return 'near'
  506. return 'far'
  507. async def delayed_call(func, delay):
  508. try:
  509. if isinstance(delay, timedelta):
  510. delay = delay.total_seconds()
  511. await asyncio.sleep(delay)
  512. if asyncio.iscoroutinefunction(func):
  513. await func()
  514. elif asyncio.iscoroutine(func):
  515. await func
  516. else:
  517. func()
  518. except asyncio.CancelledError:
  519. pass
  520. def hasmember(obj, member):
  521. """
  522. Check if a dict or dataclass has the given member
  523. """
  524. if is_dataclass(obj):
  525. if hasattr(obj, member):
  526. return True
  527. else:
  528. if member in obj:
  529. return True
  530. return False
  531. def getmember(obj, member):
  532. """
  533. Get a member from a dict or dataclass
  534. """
  535. if is_dataclass(obj):
  536. return getattr(obj, member)
  537. else:
  538. return obj[member]
  539. def setmember(obj, member, value):
  540. """
  541. Set a member of a dict of dataclass
  542. """
  543. if is_dataclass(obj):
  544. setattr(obj, member, value)
  545. else:
  546. obj[member] = value
  547. def get_next_event_from_deque(deque):
  548. unused_elements = []
  549. event = None
  550. for i in range(len(deque)):
  551. msg = deque.popleft()
  552. if isinstance(msg, objects.Event) or (isinstance(msg, dict) and 'event_descriptor' in msg):
  553. event = msg
  554. break
  555. else:
  556. unused_elements.append(msg)
  557. deque.extend(unused_elements)
  558. return event