Coverage for src/lumen/fhirpath/engine/nodes.py: 48%
653 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1from collections import abc
2import copy
3from datetime import datetime, timedelta, timezone
4from dateutil.relativedelta import relativedelta
5from dateutil import parser, tz
6from decimal import ROUND_HALF_UP, ROUND_UP, Decimal
7import math
8import json
9import re
10import time
11from typing import Optional
14timeRE = (
15 r"^T?(\d{2})(?::(\d{2}))?(?::(\d{2}))?(?:\.(\d+))?(Z|(\+|-)\d{2}(:\d{2})?)?$"
16)
17dateTimeRE = r"^(?P<year>\d{4})(?:-(?P<month>\d{2})(?:-(?P<day>\d{2}))?)?(?:T(?P<hour>\d{2})(?::(?P<minute>\d{2}))?(?::(?P<second>\d{2}))?(?:\.(?P<millisecond>\d+))?(?P<timezone>Z|(\+|-)\d{2}:\d{2})?)?$"
18_dateTimeWithBareT_RE = r"^(?P<year>\d{4})(?:-(?P<month>\d{2})(?:-(?P<day>\d{2}))?)?T$"
21class _NegatedKey:
22 """Wraps any value so that comparisons are inverted.
24 Used by polarity_expression (unary -) on non-numeric sort keys
25 (strings, ResourceNodes) inside sort(-field) expressions.
26 Comparisons are inverted so sorted() produces descending order.
27 None is treated as greatest (goes first in descending).
28 """
30 __slots__ = ("_val",)
32 def __init__(self, val: object) -> None:
33 self._val = val
35 def _to_cmp(self) -> object:
36 from lumen.fhirpath.engine.nodes import ResourceNode
37 v = self._val
38 if isinstance(v, ResourceNode):
39 v = v.data
40 return v
42 def __lt__(self, other: object) -> bool:
43 if not isinstance(other, _NegatedKey):
44 return NotImplemented
45 a, b = self._to_cmp(), other._to_cmp()
46 if a is None and b is None:
47 return False
48 if a is None:
49 return False # None is "greatest" → comes first in desc → not less
50 if b is None:
51 return True
52 try:
53 return a > b # type: ignore[operator]
54 except TypeError:
55 return str(a) > str(b)
57 def __le__(self, other: object) -> bool:
58 if not isinstance(other, _NegatedKey):
59 return NotImplemented
60 return self == other or self < other
62 def __gt__(self, other: object) -> bool:
63 if not isinstance(other, _NegatedKey):
64 return NotImplemented
65 return other < self
67 def __ge__(self, other: object) -> bool:
68 if not isinstance(other, _NegatedKey):
69 return NotImplemented
70 return other <= self
72 def __eq__(self, other: object) -> bool:
73 if not isinstance(other, _NegatedKey):
74 return NotImplemented
75 return self._to_cmp() == other._to_cmp()
77 def __repr__(self) -> str:
78 return f"_NegatedKey({self._val!r})"
80class FP_Type:
81 """
82 Class FP_Type is the superclass for FHIRPath types that required special handling
83 """
85 def equals(self):
86 """
87 Tests whether this object is equal to another. Returns either True,
88 false, or undefined (where in the FHIRPath specification empty would be
89 returned). The undefined return value indicates that the values were the
90 same to the shared precision, but that they had differnent levels of
91 precision.
92 """
93 return False
95 def equivalentTo(self):
96 """
97 Tests whether this object is equivalant to another. Returns either True,
98 false, or undefined (where in the FHIRPath specification empty would be
99 returned).
100 """
101 return False
103 def toString(self):
104 return str(self)
106 def toJSON(self):
107 return str(self)
109 def compare(self):
110 raise NotImplementedError()
113class FP_Quantity(FP_Type):
114 """
115 A map of the UCUM units that must be paired with integer values when doing arithmetic.
116 """
118 timeUnitsToUCUM = {
119 "years": "'a'",
120 "months": "'mo'",
121 "weeks": "'wk'",
122 "days": "'d'",
123 "hours": "'h'",
124 "minutes": "'min'",
125 "seconds": "'s'",
126 "milliseconds": "'ms'",
127 "year": "'a'",
128 "month": "'mo'",
129 "week": "'wk'",
130 "day": "'d'",
131 "hour": "'h'",
132 "minute": "'min'",
133 "second": "'s'",
134 "millisecond": "'ms'",
135 "'a'": "'a'",
136 "'mo'": "'mo'",
137 "'wk'": "'wk'",
138 "'d'": "'d'",
139 "'h'": "'h'",
140 "'min'": "'min'",
141 "'s'": "'s'",
142 "'ms'": "'ms'",
143 }
145 mapUCUMCodeToTimeUnits = {
146 "'a'": "year",
147 "'mo'": "month",
148 "'wk'": "week",
149 "'d'": "day",
150 "'h'": "hour",
151 "'min'": "minute",
152 "'s'": "second",
153 "'ms'": "millisecond",
154 }
156 """
157 A map of the UCUM units that must be paired with integer values when doing arithmetic.
158 """
159 integerUnits = {
160 "'a'": True,
161 "'mo'": True,
162 "'wk'": True,
163 "'d'": True,
164 "'h'": True,
165 "'min'": True,
166 }
168 _years_and_months = [
169 "'a'",
170 "year",
171 "years",
172 "'mo'",
173 "month",
174 "months",
175 ]
177 _weeks_days_and_time = [
178 "'wk'",
179 "week",
180 "weeks",
181 "'d'",
182 "day",
183 "days",
184 "'h'",
185 "hour",
186 "hours",
187 "'min'",
188 "minute",
189 "minutes",
190 "'s'",
191 "second",
192 "seconds",
193 "'ms'",
194 "millisecond",
195 "milliseconds",
196 ]
198 _arithmetic_duration_units = {
199 "years": "year",
200 "months": "month",
201 "weeks": "week",
202 "days": "day",
203 "hours": "hour",
204 "minutes": "minute",
205 "seconds": "second",
206 "milliseconds": "millisecond",
207 "year": "year",
208 "month": "month",
209 "week": "week",
210 "day": "day",
211 "hour": "hour",
212 "minute": "minute",
213 "second": "second",
214 "millisecond": "millisecond",
215 "'wk'": "week",
216 "'d'": "day",
217 "'h'": "hour",
218 "'min'": "minute",
219 "'s'": "second",
220 "'ms'": "millisecond",
221 # Quoted calendar unit names (e.g. 1 'month')
222 "'year'": "year",
223 "'month'": "month",
224 "'week'": "week",
225 "'day'": "day",
226 "'hour'": "hour",
227 "'minute'": "minute",
228 "'second'": "second",
229 "'millisecond'": "millisecond",
230 }
232 _year_month_conversion_factor = {"'a'": 12, "'mo'": 1}
233 _m_cm_mm_conversion_factor = {"'m'": 1.0, "'cm'": 0.01, "'mm'": 0.001}
234 _lbs_kg_conversion_factor = {"'kg'": 1.0, "lbs": 0.453592, "'[lb_av]'": 0.453592}
235 _g_mg_conversion_factor = {"'g'": 1.0, "'mg'": 0.001}
237 datetime_multipliers = {
238 **{key: Decimal("604800") for key in ["'wk'", "week", "weeks"]},
239 **{key: Decimal("86400") for key in ["'d'", "day", "days"]},
240 **{key: Decimal("3600") for key in ["'h'", "hour", "hours"]},
241 **{key: Decimal("60") for key in ["'min'", "minute", "minutes"]},
242 **{key: Decimal("1") for key in ["'s'", "second", "seconds"]},
243 **{key: Decimal("0.001") for key in ["'ms'", "millisecond", "milliseconds"]},
244 }
246 def __init__(self, value: Decimal, unit: str, unit_display: str | None = None):
247 super().__init__()
248 self.asStr = f"{value} {unit}"
249 self.value = value
250 self.unit = unit
251 self.unit_display = unit_display
253 def __str__(self):
254 return self.asStr
256 def __repr__(self):
257 return f"{type(self)}<{self.asStr}>"
259 def __hash__(self):
260 if self.unit in self._years_and_months:
261 value_in_months = self.value
262 if self.unit in ["'a'", "year", "years"]:
263 value_in_months *= 12
264 return hash(("months", value_in_months))
265 elif self.unit in self._weeks_days_and_time:
266 value_in_seconds = self.value * self.datetime_multipliers[self.unit]
267 return hash(("seconds", value_in_seconds))
268 else:
269 return hash((self.value, self.unit))
271 def __eq__(self, other):
272 if isinstance(other, FP_Quantity):
273 if self.unit in self._years_and_months and other.unit in self._years_and_months:
274 if self.unit == "'a'" or other.unit == "'a'":
275 return False
276 return self._compare_years_and_months(other)
277 elif self.unit in self._weeks_days_and_time and other.unit in self._weeks_days_and_time:
278 self_value_in_seconds = self.value * self.datetime_multipliers[self.unit]
279 other_value_in_seconds = other.value * self.datetime_multipliers[other.unit]
280 return self_value_in_seconds == other_value_in_seconds
281 else:
282 if self.unit != other.unit:
283 converted = FP_Quantity.conv_unit_to(self.unit, self.value, other.unit)
284 if converted is not None:
285 return other.value == converted.value and other.unit == converted.unit
287 return self.value == other.value and self.unit == other.unit
288 else:
289 return super().__eq__(other)
291 def compare(self, other):
292 if not isinstance(other, FP_Quantity):
293 raise TypeError("Can only compare FP_Quantity to another FP_Quantity")
294 if self == other:
295 return 0
296 if self.unit in self._years_and_months and other.unit in self._years_and_months:
297 self_val = self.value * 12 if self.unit in ["'a'", "year", "years"] else self.value
298 other_val = other.value * 12 if other.unit in ["'a'", "year", "years"] else other.value
299 return -1 if self_val < other_val else 1
300 elif self.unit in self._weeks_days_and_time and other.unit in self._weeks_days_and_time:
301 self_val = self.value * self.datetime_multipliers[self.unit]
302 other_val = other.value * other.datetime_multipliers[other.unit]
303 return -1 if self_val < other_val else 1
304 else:
305 if self.unit != other.unit:
306 converted = FP_Quantity.conv_unit_to(self.unit, self.value, other.unit)
307 if converted is not None:
308 return -1 if converted.value < other.value else 1
309 return -1 if self.value < other.value else 1
311 def deep_equal(self, other):
312 if isinstance(other, FP_Quantity):
313 if self.unit in self._years_and_months and other.unit in self._years_and_months:
314 return self._compare_years_and_months(other, year_units=["'a'", "year", "years"])
315 else:
316 if self.unit != other.unit:
317 converted = FP_Quantity.conv_unit_to(self.unit, self.value, other.unit)
318 reverse_converted = FP_Quantity.conv_unit_to(
319 converted.unit, converted.value, self.unit
320 )
321 if converted is not None:
322 return (
323 self.value == reverse_converted.value
324 and self.unit == reverse_converted.unit
325 )
326 return self.__eq__(other)
327 else:
328 return super().__eq__(other)
330 def conv_unit_to(fromUnit, value, toUnit):
331 from_year_month_magnitude = FP_Quantity._year_month_conversion_factor.get(fromUnit)
332 to_year_month_magnitude = FP_Quantity._year_month_conversion_factor.get(toUnit)
333 if from_year_month_magnitude and to_year_month_magnitude:
334 return FP_Quantity(from_year_month_magnitude * value / to_year_month_magnitude, toUnit)
336 elif (
337 fromUnit in FP_Quantity._weeks_days_and_time
338 and toUnit in FP_Quantity._weeks_days_and_time
339 ):
340 value_in_seconds = value * FP_Quantity.datetime_multipliers.get(fromUnit)
341 new_value = value_in_seconds / FP_Quantity.datetime_multipliers.get(toUnit)
342 return FP_Quantity(new_value, toUnit)
344 from_m_cm_mm_magnitude = FP_Quantity._m_cm_mm_conversion_factor.get(fromUnit)
345 to_m_cm_mm_magnitude = FP_Quantity._m_cm_mm_conversion_factor.get(toUnit)
346 if from_m_cm_mm_magnitude and to_m_cm_mm_magnitude:
347 if (
348 fromUnit in FP_Quantity._m_cm_mm_conversion_factor
349 or toUnit in FP_Quantity._m_cm_mm_conversion_factor
350 ):
351 from_magnitude, to_magnitude = Decimal(from_m_cm_mm_magnitude), Decimal(
352 to_m_cm_mm_magnitude
353 )
354 return FP_Quantity(from_magnitude * value / to_magnitude, toUnit)
356 from_lbs_kg_magnitude = FP_Quantity._lbs_kg_conversion_factor.get(fromUnit)
357 to_lbs_kg_magnitude = FP_Quantity._lbs_kg_conversion_factor.get(toUnit)
358 if from_lbs_kg_magnitude and to_lbs_kg_magnitude:
359 converted_value = (Decimal(from_lbs_kg_magnitude) * value) / Decimal(
360 to_lbs_kg_magnitude
361 )
362 rounded_value = converted_value.quantize(Decimal("1."), rounding=ROUND_UP)
363 return FP_Quantity(rounded_value, toUnit)
365 from_g_mg_magnitude = FP_Quantity._g_mg_conversion_factor.get(fromUnit)
366 to_g_mg_magnitude = FP_Quantity._g_mg_conversion_factor.get(toUnit)
367 if from_g_mg_magnitude and to_g_mg_magnitude:
368 value = Decimal(value) * Decimal(FP_Quantity._g_mg_conversion_factor[fromUnit])
369 result = (value / Decimal(FP_Quantity._g_mg_conversion_factor[toUnit])).quantize(
370 Decimal("1."), rounding=ROUND_HALF_UP
371 )
372 return FP_Quantity(result, toUnit)
373 return None
375 def _compare_years_and_months(self, other, year_units=None):
376 year_units = year_units or ["year", "years"]
377 self_value_in_months = self.value
378 other_value_in_months = other.value
380 if self.unit in year_units:
381 self_value_in_months *= 12
382 if other.unit in year_units:
383 other_value_in_months *= 12
384 return self_value_in_months == other_value_in_months
387class FP_TimeBase(FP_Type):
388 datetime_multipliers = [
389 {"key": "year", "value": (365 * 12 * 24 * 60 * 60)},
390 {"key": "month", "value": (12 * 24 * 60 * 60)},
391 {"key": "day", "value": (24 * 60 * 60)},
392 {"key": "hour", "value": (60 * 60)},
393 {"key": "minute", "value": 60},
394 {"key": "second", "value": 1},
395 {"key": "tz", "value": (60 * 60)},
396 ]
398 def _extractAsMatchList(self, matchData, matchGroupsIndices, is_date=True):
399 result = []
400 for matchGroupIndex in matchGroupsIndices:
401 if is_date:
402 group = matchData.group(matchGroupIndex["key"])
403 else:
404 index = matchGroupIndex["index"]
405 group = matchData.group(index) if index <= matchData.lastindex else None
406 result.append(group if group is not None else None)
407 return result
409 def _calculatePrecision(self, dt_list):
410 return sum(all(x not in i for x in ["+", "-"]) for i in dt_list if i is not None)
412 def _getMatchAsList(self):
413 raise NotImplementedError()
415 def _getDateTimeInt(self):
416 raise NotImplementedError()
418 def equals(self, otherDateTime):
419 """
420 From the 2020 August:
421 For DateTime and Time equality, the comparison is performed by
422 considering each precision in order, beginning with years (or hours for
423 time values), and respecting timezone offsets. If the values are the
424 same, comparison proceeds to the next precision; if the values are
425 different, the comparison stops and the result is false. If one input has
426 a value for the precision and the other does not, the comparison stops
427 and the result is empty ({ }); if neither input has a value for the
428 precision, or the last precision has been reached, the comparison stops
429 and the result is true.
430 Note: Per the spec above
431 :return:
432 2012-01 = 2012 returns empty
433 2012-01 = 2011 returns false
434 2012-01 ~ 2012 returns false
435 """
436 if type(otherDateTime) != type(self):
437 return False
439 thisdt_list = self._getMatchAsList()
440 otherdt_list = otherDateTime._getMatchAsList()
442 normalized_thisdt_list = self._normalize_datetime(thisdt_list)
443 normalized_otherdt_list = self._normalize_datetime(otherdt_list)
445 indices_to_remove = [
446 i
447 for i in range(len(normalized_thisdt_list))
448 if normalized_thisdt_list[i] == normalized_otherdt_list[i] == None
449 ]
451 for i in reversed(indices_to_remove):
452 del normalized_thisdt_list[i]
453 del normalized_otherdt_list[i]
455 normalized_thisdt_precision = self._calculatePrecision(normalized_thisdt_list)
456 normalized_otherdt_precision = self._calculatePrecision(normalized_otherdt_list)
458 if normalized_thisdt_precision == normalized_otherdt_precision:
459 # One has timezone, other doesn't → undefined per spec
460 self_has_tz = len(thisdt_list) >= 8 and thisdt_list[7] is not None
461 other_has_tz = len(otherdt_list) >= 8 and otherdt_list[7] is not None
462 if self_has_tz != other_has_tz:
463 return None
464 return self._getDateTimeInt() == otherDateTime._getDateTimeInt()
466 if normalized_thisdt_precision != normalized_otherdt_precision:
467 min_precision = min(normalized_thisdt_precision, normalized_otherdt_precision)
468 for i in range(min_precision):
469 if normalized_thisdt_list[i] is None or normalized_otherdt_list[i] is None:
470 return None
471 if normalized_thisdt_list[i] != normalized_otherdt_list[i]:
472 return False
473 tz_thisdt_list = len(thisdt_list) >= 8 and thisdt_list[7] is not None
474 tz_otherdt_list = len(otherdt_list) >= 8 and otherdt_list[7] is not None
475 if (tz_thisdt_list and not tz_otherdt_list) or (
476 tz_otherdt_list and not tz_thisdt_list
477 ):
478 return None
480 return None
482 def _normalize_datetime(self, dt_list):
483 def to_str(number):
484 return "0" + str(number) if 0 < number < 10 else str(number)
486 if len(dt_list) < 6:
487 year, month, day = (None, None, None)
488 hour, minute, second = (int(dt_list[i]) if dt_list[i] else None for i in range(3))
489 timezone_str = dt_list[4] if len(dt_list) > 4 else None
490 else:
491 year, month, day = (int(dt_list[i]) if dt_list[i] else None for i in range(3))
492 hour, minute, second = (int(dt_list[i]) if dt_list[i] else None for i in range(3, 6))
493 timezone_str = dt_list[7] if len(dt_list) > 7 else None
495 dt = datetime(year or 2023, month or 1, day or 1, hour or 0, minute or 0, second or 0)
496 if timezone_str and timezone_str != "Z":
497 tz_hours, tz_minutes = map(int, timezone_str[1:].split(":"))
498 tz_delta = timedelta(hours=tz_hours, minutes=tz_minutes)
499 dt = dt - tz_delta if timezone_str.startswith("+") else dt + tz_delta
501 return [
502 to_str(dt.year) if year is not None else None,
503 to_str(dt.month) if month is not None else None,
504 to_str(dt.day) if day is not None else None,
505 to_str(dt.hour) if hour is not None else None,
506 to_str(dt.minute) if minute is not None else None,
507 to_str(dt.second) if second is not None else None,
508 ]
510 def compare(self, otherDateTime):
511 if type(otherDateTime) != type(self):
512 raise TypeError
514 thisDateTimeList = self._getMatchAsList()
515 otherDateTimeList = otherDateTime._getMatchAsList()
517 normalized_thisdt_list = self._normalize_datetime(thisDateTimeList)
518 normalized_otherdt_list = self._normalize_datetime(otherDateTimeList)
519 indices_to_remove = [
520 i
521 for i in range(len(normalized_thisdt_list))
522 if normalized_thisdt_list[i] == normalized_otherdt_list[i] == None
523 ]
524 for i in reversed(indices_to_remove):
525 del normalized_thisdt_list[i]
526 del normalized_otherdt_list[i]
528 normalized_thisdt_precision = self._calculatePrecision(normalized_thisdt_list)
529 normalized_otherdt_precision = self._calculatePrecision(normalized_otherdt_list)
531 if normalized_thisdt_precision != normalized_otherdt_precision:
532 min_precision = min(normalized_thisdt_precision, normalized_otherdt_precision)
533 # Compare raw (non-timezone-normalized) date components to avoid UTC-shifted
534 # date mismatches when one side is a date-only value with no timezone.
535 for i in range(min_precision):
536 v_this = thisDateTimeList[i] if i < len(thisDateTimeList) else None
537 v_other = otherDateTimeList[i] if i < len(otherDateTimeList) else None
538 if v_this is None or v_other is None:
539 return None
540 int_this = int(v_this)
541 int_other = int(v_other)
542 if int_this > int_other:
543 return 1
544 if int_this < int_other:
545 return -1
546 return None
548 thisDateTimeInt = self._getDateTimeInt()
549 otherDateTimeInt = otherDateTime._getDateTimeInt()
551 if thisDateTimeInt < otherDateTimeInt:
552 return -1
553 elif thisDateTimeInt == otherDateTimeInt:
554 return 0
555 return 1
557 def plus(self, time_quantity):
558 tz_match = re.search(r"([+-]\d{2}:\d{2})$", self.asStr or "")
559 original_tz_str = tz_match.group(1) if tz_match else None
561 # Inline/re-use the original logic to get the result
562 value = int(time_quantity.value)
563 time_unit = FP_Quantity._arithmetic_duration_units.get(time_quantity.unit)
564 if time_unit is None:
565 valid_units = ", ".join(FP_Quantity._arithmetic_duration_units.keys())
566 raise ValueError(
567 f"For date/time arithmetic, the unit of the quantity must be one of the following time-based units: {valid_units}"
568 )
569 dt_list = self._getMatchAsList()
570 if isinstance(self, FP_DateTime):
571 precision = self._calculatePrecision(dt_list)
572 date_obj = self._convertDatetime(dt_list)
573 if time_unit == "year":
574 result_dt = date_obj + relativedelta(years=value)
575 elif time_unit == "month":
576 result_dt = date_obj + relativedelta(months=value)
577 elif time_unit in ["day", "week"]:
578 if time_unit == "week":
579 value *= 7
580 if precision == 1:
581 years = math.floor(value / 365) if value >= 0 else math.ceil(value / 365)
582 result_dt = date_obj + relativedelta(years=years)
583 elif precision == 2:
584 months = math.floor(value / 30) if value >= 0 else math.ceil(value / 30)
585 result_dt = date_obj + relativedelta(months=months)
586 else:
587 result_dt = date_obj + relativedelta(days=value)
588 elif time_unit in ["hour"]:
589 if precision == 2:
590 months = (
591 math.floor(value / (24 * 30))
592 if value >= 0
593 else math.ceil(value / (24 * 30))
594 )
595 result_dt = date_obj + relativedelta(months=months)
596 elif precision == 3:
597 days = math.floor(value / 24) if value >= 0 else math.ceil(value / 24)
598 result_dt = date_obj + relativedelta(days=days)
599 else:
600 result_dt = date_obj + timedelta(hours=value)
601 elif time_unit in ["minute", "second", "millisecond"]:
602 if precision == 4:
603 if time_unit == "minute":
604 hours = math.floor(value / 60) if value >= 0 else math.ceil(value / 60)
605 result_dt = date_obj + relativedelta(hours=hours)
606 elif time_unit == "second":
607 hours = (
608 math.floor(value / (60 * 60))
609 if value >= 0
610 else math.ceil(value / (60 * 60))
611 )
612 result_dt = date_obj + relativedelta(hours=hours)
613 elif time_unit == "millisecond":
614 hours = (
615 math.floor(value / (60 * 60 * 1000))
616 if value >= 0
617 else math.ceil(value / (60 * 60 * 1000))
618 )
619 result_dt = date_obj + relativedelta(hours=hours)
620 elif precision == 5:
621 if time_unit == "minute":
622 minutes = math.floor(value) if value >= 0 else math.ceil(value)
623 result_dt = date_obj + relativedelta(minutes=minutes)
624 elif time_unit == "second":
625 minutes = math.floor(value / 60) if value >= 0 else math.ceil(value / 60)
626 result_dt = date_obj + relativedelta(minutes=minutes)
627 elif time_unit == "millisecond":
628 minutes = (
629 math.floor(value / (60 * 1000))
630 if value >= 0
631 else math.ceil(value / (60 * 1000))
632 )
633 result_dt = date_obj + relativedelta(minutes=minutes)
634 else:
635 if time_unit == "minute":
636 result_dt = date_obj + timedelta(minutes=value)
637 elif time_unit == "second":
638 result_dt = date_obj + timedelta(seconds=value)
639 elif time_unit == "millisecond":
640 result_dt = date_obj + timedelta(milliseconds=value)
641 result = self._extractDateByPrecision(result_dt, precision)
642 else:
643 # FP_Time logic
644 precision = self._calculateTimePrecision(dt_list)
645 date_obj = self._convertTime(dt_list)
646 if precision == 2:
647 if time_unit == "hour":
648 result_dt = date_obj + relativedelta(minutes=value * 60)
649 elif time_unit == "minute":
650 result_dt = date_obj + relativedelta(minutes=value)
651 elif time_unit == "second":
652 minutes = math.floor(value / 60) if value >= 0 else math.ceil(value / 60)
653 result_dt = date_obj + relativedelta(minutes=minutes)
654 elif time_unit == "millisecond":
655 minutes = (
656 math.floor(value / (60 * 1000))
657 if value >= 0
658 else math.ceil(value / (60 * 1000))
659 )
660 result_dt = date_obj + relativedelta(minutes=minutes)
661 elif precision in [3, 4]:
662 if time_unit == "hour":
663 result_dt = date_obj + relativedelta(minutes=value * 60)
664 elif time_unit == "minute":
665 result_dt = date_obj + relativedelta(minutes=value)
666 elif time_unit == "second":
667 milliseconds = float(str(time_quantity).split()[0]) * 1000
668 result_dt = date_obj + relativedelta(microseconds=milliseconds * 1000)
669 elif time_unit == "millisecond":
670 result_dt = date_obj + relativedelta(microseconds=value * 1000)
671 result = (
672 self._extractTimeByPrecision(result_dt, precision if precision < 3 else 4) + dt_list[4]
673 )
675 if original_tz_str is None or not isinstance(result, str):
676 return result
678 sign = 1 if original_tz_str.startswith("+") else -1
679 tz_h, tz_m = map(int, original_tz_str[1:].split(":"))
680 tz_total_mins = sign * (tz_h * 60 + tz_m)
682 # Strip any existing tz suffix from the result
683 tz_in_result = re.search(r"([+-]\d{2}:\d{2})$", result)
684 base_str = result[: tz_in_result.start()] if tz_in_result else result
686 # Separate milliseconds ("...T07:00:00.123") from the main part
687 ms_match = re.match(r"^(.+?)\.(\d+)$", base_str)
688 if ms_match:
689 date_part = ms_match.group(1)
690 ms_part = ms_match.group(2)
691 else:
692 date_part = base_str
693 ms_part = None
695 dt_formats = (
696 "%Y-%m-%dT%H:%M:%S",
697 "%Y-%m-%dT%H:%M",
698 "%Y-%m-%dT%H",
699 "%Y-%m-%d",
700 "%Y-%m",
701 "%Y",
702 )
703 for fmt in dt_formats:
704 try:
705 dt = datetime.strptime(date_part, fmt)
706 dt_adj = dt + timedelta(minutes=tz_total_mins)
707 result_base = dt_adj.strftime(fmt)
708 if ms_part is not None:
709 result_base = f"{result_base}.{ms_part}"
710 return result_base + original_tz_str
711 except ValueError:
712 continue
714 return result
716 @staticmethod
717 def check_string(cls, str_val):
718 val = cls(str_val)
719 return val
721 @staticmethod
722 def get_match_data(str_val):
723 match = re.match(dateTimeRE, str_val)
724 if match:
725 return FP_DateTime(str_val)
726 match = re.match(timeRE, str_val)
727 if match:
728 return FP_Time(str_val)
729 return None
732class FP_Time(FP_TimeBase):
733 matchGroupsIndices = [
734 {"key": "hour", "index": 1},
735 {"key": "minute", "index": 2},
736 {"key": "second", "index": 3},
737 {"key": "millisecond", "index": 4},
738 {"key": "timezone", "index": 5},
739 ]
741 def __new__(cls, dateStr):
742 if not isinstance(dateStr, str):
743 return None
745 if not re.match(timeRE, dateStr):
746 return None
748 return super(FP_Time, cls).__new__(cls)
750 def __init__(self, timeStr):
751 self.asStr = timeStr if isinstance(timeStr, str) else None
752 self._timeMatchData = re.match(timeRE, self.asStr)
753 self._timeMatchStr = None
754 self._timeAsList = []
755 self._precision = 0
756 self._pyTimeObject = None
758 if self._timeMatchData:
759 self._timeMatchStr = self._timeMatchData.group(0)
760 self._timeAsList = self._extractAsMatchList(
761 self._timeMatchData, self.matchGroupsIndices, is_date=False
762 )
763 self._precision = self._calculatePrecision(self._timeAsList)
764 formats = [
765 "T%H:%M:%S%z",
766 "T%H:%M:%S.%f%z",
767 "T%H:%M:%S",
768 "T%H:%M:%S.%f",
769 "T%H:%M%z",
770 "%H:%M:%S%z",
771 "%H:%M:%S.%f%z",
772 "%H:%M:%S",
773 "%H:%M:%S.%f",
774 "%H:%M%z",
775 "%H:%M",
776 "%H%z",
777 ]
779 for fmt in formats:
780 try:
781 parsed_datetime = datetime.strptime(self.asStr, fmt)
782 if parsed_datetime.tzinfo:
783 parsed_datetime = parsed_datetime.astimezone(timezone.utc)
784 self._pyTimeObject = parsed_datetime.time()
785 break
786 except ValueError:
787 continue
789 def __str__(self):
790 if self._pyTimeObject:
791 time_str = self._pyTimeObject.isoformat()
792 if "." in time_str:
793 time_str = time_str[: time_str.index(".") + 4]
794 return time_str
795 return self.asStr
797 def __eq__(self, other):
798 if isinstance(other, str):
799 return self.getTimeMatchStr()
801 def getTimeMatchStr(self):
802 return self._timeMatchStr
804 def _getMatchAsList(self):
805 return self._timeAsList
807 def _getDateTimeInt(self):
808 """
809 :return: If self.timeMatchData returns DateTime object converted to seconds int, else returns None
810 """
811 if self._pyTimeObject:
812 return timedelta(
813 hours=self._pyTimeObject.hour,
814 minutes=self._pyTimeObject.minute,
815 seconds=self._pyTimeObject.second,
816 microseconds=self._pyTimeObject.microsecond,
817 ).total_seconds()
818 return None
820 def _extractTimeByPrecision(self, date_obj, precision):
821 format = {1: "T%H", 2: "T%H:%M", 3: "T%H:%M:%S", 4: "T%H:%M:%S.%f"}
822 return date_obj.strftime(format.get(precision)) if precision in format else None
824 def _calculateTimePrecision(self, dt_list):
825 return sum(1 for i in dt_list[0:4] if i is not None)
827 def _convertTime(self, time_list):
828 hour = time_list[0] if time_list[0] is not None else 00
829 minute = time_list[1] if time_list[1] is not None else 00
830 second = time_list[2] if time_list[2] is not None else 00
831 millisecond = time_list[3] if time_list[3] is not None else 000
832 return datetime.strptime(f"{hour}:{minute}:{second}.{millisecond}", "%H:%M:%S.%f")
835class FP_DateTime(FP_TimeBase):
836 matchGroupsIndices = [
837 {"key": "year", "index": 0},
838 {"key": "month", "index": 4},
839 {"key": "day", "index": 6},
840 {"key": "hour", "index": 8},
841 {"key": "minute", "index": 9},
842 {"key": "second", "index": 10},
843 {"key": "millisecond", "index": 11},
844 {"key": "timezone", "index": 12},
845 ]
846 minPrecision = 3
848 def __new__(cls, dateStr):
849 if not isinstance(dateStr, str):
850 return None
852 if not re.match(dateTimeRE, dateStr) and not re.match(_dateTimeWithBareT_RE, dateStr):
853 return None
855 return super(FP_DateTime, cls).__new__(cls)
857 def __init__(self, dateStr):
858 self.asStr = dateStr if isinstance(dateStr, str) else None
859 # Strip bare trailing T (e.g. "2015T", "2015-02T") before matching the full datetime RE.
860 # The asStr retains the T so TypeInfo can identify the value as DateTime, not Date.
861 _match_str = re.sub(r"T$", "", self.asStr) if self.asStr and re.match(_dateTimeWithBareT_RE, self.asStr) else self.asStr
862 self._dateTimeMatchData = (
863 re.match(dateTimeRE, _match_str) if isinstance(_match_str, str) else None
864 )
865 self._dateTimeMatchStr = None
866 self._dateTimeAsList = []
867 self._precision = 0
869 if self._dateTimeMatchData:
870 self._dateTimeMatchStr = self._dateTimeMatchData.group(0)
871 self._dateTimeAsList = self._extractAsMatchList(
872 self._dateTimeMatchData, self.matchGroupsIndices
873 )
874 self._precision = self._calculatePrecision(self._dateTimeAsList)
876 def __str__(self):
877 if re.match(r"^\d{4}-\d{2}-\d{2}$", self.asStr):
878 return self.asStr
879 if self.asStr and len(self.asStr) <= 4:
880 return self.asStr
881 if self._getDateTimeObject():
882 iso_str = self._getDateTimeObject().isoformat()
883 if "." in iso_str:
884 iso_str = iso_str[: iso_str.index(".") + 4] + iso_str[iso_str.index(".") + 7 :]
885 return iso_str
886 return self.asStr
888 def __eq__(self, other):
889 if isinstance(other, str):
890 return self.getDateTimeMatchStr()
892 def __deepcopy__(self, memo):
893 return type(self)(copy.deepcopy(self.asStr, memo))
895 def getDateTimeMatchStr(self):
896 return self._dateTimeMatchStr
898 def _getMatchAsList(self):
899 return self._dateTimeAsList
901 def _getDateTimeObject(self):
902 if self._dateTimeMatchData:
903 if "Z" in self.asStr:
904 date_str = self.asStr.replace("Z", "+00:00")
905 else:
906 date_str = self.asStr
907 return parser.parse(date_str)
908 return None
910 def _getDateTimeInt(self):
911 """
912 :return: If self.timeMatchData returns DateTime object converted to seconds int, else returns None
913 """
914 if not self._dateTimeMatchData:
915 return None
917 if self._precision >= FP_DateTime.minPrecision:
918 dateTimeObject = self._getDateTimeObject()
919 return dateTimeObject.timestamp()
921 integer_result = 0
922 for prec in range(self._precision):
923 integer_result += (
924 int(self._dateTimeAsList[prec]) * self.datetime_multipliers[prec]["value"]
925 )
927 return integer_result
929 def _extractDateByPrecision(self, date_obj: datetime, precision):
930 if date_obj.tzinfo is None or date_obj.tzinfo.utcoffset(date_obj) is None:
931 date_obj = date_obj.replace(tzinfo=tz.tzutc())
932 format = {
933 1: "%Y",
934 2: "%Y-%m",
935 3: "%Y-%m-%d",
936 4: "%Y-%m-%dT%H",
937 5: "%Y-%m-%dT%H:%M",
938 6: "%Y-%m-%dT%H:%M:%S",
939 7: "%Y-%m-%dT%H:%M:%S",
940 }
941 formatted_date = date_obj.strftime(format.get(precision, ""))
942 if precision == 7:
943 milliseconds = date_obj.strftime("%f")[:3]
944 tz_offset = date_obj.strftime("%z")
945 tz_formatted = tz_offset[:3] + ":" + tz_offset[3:]
946 formatted_date = f"{formatted_date}.{milliseconds}{tz_formatted}"
947 return formatted_date
949 def _convertDatetime(self, date_list):
950 n_date_list = self._normalize_datetime(date_list)
951 year = n_date_list[0] if n_date_list[0] is not None else "0"
952 month = n_date_list[1] if n_date_list[1] is not None else "01"
953 day = n_date_list[2] if n_date_list[2] is not None else "01"
954 hour = n_date_list[3] if n_date_list[3] is not None else "00"
955 minute = n_date_list[4] if n_date_list[4] is not None else "00"
956 second = n_date_list[5] if n_date_list[5] is not None else "00"
957 millisecond = date_list[6] if date_list[6] is not None else "000"
958 date_string = f"{year}-{month}-{day} {hour}:{minute}:{second}.{millisecond}"
959 return datetime.strptime(date_string, "%Y-%m-%d %H:%M:%S.%f")
962class ResourceNode:
963 """
964 * Constructs a instance for the given node ("data") of a resource. If the
965 * data is the top-level node of a resouce, the path and type parameters will
966 * be ignored in favor of the resource's resourceType field.
967 * @param data the node's data or value (which might be an object with
968 * sub-nodes, an array, or FHIR data type)
969 * @param path the node's path in the resource (e.g. Patient.name). If the
970 * data's type can be determined from data, that will take precedence over
971 * this parameter.
972 * @param _data additional data stored in a property named with "_" prepended.
973 """
975 def __init__(self, data, path, _data=None, propName=None, index=None):
976 """
977 If data is a resource (maybe a contained resource) reset the path
978 information to the resource type.
979 """
980 if isinstance(data, abc.Mapping) and "resourceType" in data:
981 path = data["resourceType"]
983 self.path = path
984 self.data = data
985 self._data = _data
986 self.propName: Optional[str] = propName
987 self.index: Optional[int] = index
989 def __eq__(self, value):
990 if isinstance(value, ResourceNode):
991 return self.data == value.data
992 return self.data == value
994 def __hash__(self):
995 data_hash = hash(json.dumps(self.data, sort_keys=True, default=str))
996 path_hash = hash(self.path)
997 return hash((data_hash, path_hash))
999 def get_type_info(self):
1000 namespace = TypeInfo.FHIR
1002 if self.path is None:
1003 return None
1005 match = re.match(r"^System\.(.*)$", self.path)
1006 if match:
1007 return TypeInfo(namespace=TypeInfo.System, name=match.group(1))
1009 # If it's a simple path name, e.g. Patient.gender, we want "code" or "string"
1010 # as defined in the FHIR profile (which is in the path).
1011 # We can extract the final type name from the path if there are dots,
1012 # e.g. Patient.gender -> "code" (or whatever the path resolves to).
1013 # But if it's already resolved in the parser/engine to childPath, that childPath is self.path.
1014 if "." not in self.path:
1015 return TypeInfo(namespace=namespace, name=self.path)
1016 else:
1017 name = self.path.split(".")[-1]
1018 if TypeInfo.model:
1019 resolved_type = TypeInfo.model.get("path2Type", {}).get(self.path)
1020 if resolved_type:
1021 return TypeInfo(namespace=namespace, name=resolved_type)
1022 # If name is a primitive type (like string, boolean, code etc) we can return that
1023 primitive_types = {
1024 "base64Binary", "boolean", "canonical", "code", "date", "dateTime",
1025 "decimal", "id", "instant", "integer", "markdown", "oid",
1026 "positiveInt", "string", "time", "unsignedInt", "uri", "url", "uuid"
1027 }
1028 if name in primitive_types:
1029 return TypeInfo(namespace=namespace, name=name)
1031 if isinstance(self.data, (str, bool, int, float, Decimal, FP_DateTime, FP_Time, FP_Quantity)):
1032 return TypeInfo.create_by_value_in_namespace(namespace=namespace, value=self.data)
1034 if not TypeInfo.model:
1035 return TypeInfo.create_by_value_in_namespace(namespace=namespace, value=self.data)
1037 return TypeInfo(namespace=namespace, name="BackboneElement")
1039 def toJSON(self):
1040 return json.dumps(self.data)
1042 @staticmethod
1043 def create_node(data, path=None, _data=None, propName=None, index=None):
1044 if isinstance(data, ResourceNode):
1045 return data
1046 return ResourceNode(data, path, _data, propName, index)
1048 def convert_data(self):
1049 data = self.data
1051 # FHIR Quantity: {"value": ..., "unit": ..., "system": "http://unitsofmeasure.org", "code": ...}
1052 if isinstance(data, abc.Mapping) and data.get("system") == "http://unitsofmeasure.org":
1053 raw_value = data["value"]
1054 try:
1055 quantity_value = Decimal(str(raw_value))
1056 except Exception:
1057 quantity_value = raw_value
1058 return FP_Quantity(
1059 quantity_value,
1060 FP_Quantity.timeUnitsToUCUM.get(data["code"], "'" + data["code"] + "'"),
1061 unit_display=data.get("unit"),
1062 )
1064 # Primitive extension pattern: {"value": <actual>, "extension": [...]}
1065 # Identified by having "value" alongside "extension" but no resourceType.
1066 # This is the FHIR XML primitive-with-extension representation.
1067 if (
1068 isinstance(data, abc.Mapping)
1069 and "value" in data
1070 and "extension" in data
1071 and "resourceType" not in data
1072 ):
1073 data = data["value"]
1075 cls = TypeInfo.type_to_class_with_check_string.get(self.path)
1076 if cls:
1077 data = FP_TimeBase.check_string(cls, data) or data
1079 return data
1082class TypeInfo:
1083 model = None
1084 System = "System"
1085 FHIR = "FHIR"
1087 type_to_class_with_check_string = {
1088 "date": FP_DateTime,
1089 "dateTime": FP_DateTime,
1090 "time": FP_Time,
1091 }
1093 def __init__(self, name, namespace):
1094 self.name = name
1095 self.namespace = namespace
1097 @staticmethod
1098 def is_type(type_name, super_type):
1099 while type_name:
1100 if type_name == super_type:
1101 return True
1103 # TODO: Double check it
1104 type_name = TypeInfo.model.get("type2Parent").get(type_name) or TypeInfo.model.get(
1105 "path2Type"
1106 ).get(type_name)
1108 return False
1110 def is_(self, other):
1111 if isinstance(other, TypeInfo) and (
1112 not self.namespace or not other.namespace or self.namespace == other.namespace
1113 ):
1114 if TypeInfo.model and (not self.namespace or self.namespace == TypeInfo.FHIR):
1115 return TypeInfo.is_type(self.name, other.name)
1116 else:
1117 return self.name == other.name
1118 return False
1120 @staticmethod
1121 def create_by_value_in_namespace(namespace, value):
1122 name = type(value).__name__
1124 if isinstance(value, int) and not isinstance(value, bool):
1125 name = "integer"
1126 elif isinstance(value, float) or isinstance(value, Decimal):
1127 name = "decimal"
1128 elif isinstance(value, FP_DateTime):
1129 # Distinguish date from dateTime: presence of 'T' in the string indicates
1130 # DateTime domain (even without a time component, e.g. @2015T).
1131 has_t = "T" in (value.asStr or "")
1132 name = "dateTime" if has_t or value._precision > 3 else "date"
1133 elif isinstance(value, FP_Time):
1134 name = "time"
1135 elif isinstance(value, FP_Quantity):
1136 name = "Quantity"
1137 elif isinstance(value, str):
1138 name = "string"
1139 elif isinstance(value, abc.Mapping):
1140 name = "object"
1142 if name == "bool":
1143 name = "boolean"
1145 if namespace == TypeInfo.System:
1146 if name == "dateTime":
1147 name = "DateTime"
1148 else:
1149 name = name.capitalize()
1151 return TypeInfo(name, namespace)
1153 @staticmethod
1154 def from_value(value):
1155 if isinstance(value, ResourceNode):
1156 return value.get_type_info()
1157 else:
1158 return TypeInfo.create_by_value_in_namespace(TypeInfo.System, value)