utils.py 22 KB

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