Coverage for src/lumen/fhirpath/engine/util.py: 89%

1from decimal import Decimal 

2import json 

3from collections import OrderedDict 

4from functools import reduce 

5from lumen.fhirpath.engine.nodes import ResourceNode, FP_Quantity 

6 

7 

8class set_paths: 

9 def __init__(self, func, parsedPath, model=None, options=None): 

10 self.func = func 

11 self.parsedPath = parsedPath 

12 self.model = model 

13 self.options = options 

14 

15 def __call__(self, resource, context=None): 

16 return self.func( 

17 resource, self.parsedPath, context or {}, self.model, self.options 

18 ) 

19 

20 

21def get_data(value): 

22 if isinstance(value, ResourceNode): 

23 value = value.data 

24 

25 if isinstance(value, float): 

26 return Decimal(str(value)) 

27 return value 

28 

29 

30def parse_value(value): 

31 def parse_complex_value(v): 

32 if isinstance(v, FP_Quantity): 

33 return v 

34 num_value, unit = v.get("value"), v.get("code") 

35 return FP_Quantity(num_value, f"'{unit}'") if num_value and unit else None 

36 

37 return ( 

38 parse_complex_value(value.data) 

39 if getattr(value, "get_type_info", lambda: None)() 

40 and value.get_type_info().name == "Quantity" 

41 else value 

42 ) 

43 

44 

45def is_number(value): 

46 return isinstance(value, (int, Decimal, complex)) and not isinstance(value, bool) 

47 

48 

49def is_capitalized(x): 

50 return isinstance(x, str) and x[0] == x[0].upper() 

51 

52 

53def is_empty(x): 

54 return isinstance(x, list) and len(x) == 0 

55 

56 

57def is_some(x): 

58 return x is not None and not is_empty(x) 

59 

60 

61def is_nullable(x): 

62 return x is None or is_empty(x) 

63 

64 

65def is_true(x): 

66 return x == True or isinstance(x, list) and len(x) == 1 and x[0] == True 

67 

68 

69def arraify(x, instead_none=None): 

70 if isinstance(x, list): 

71 return x 

72 if is_some(x): 

73 return [x] 

74 return [] if instead_none is None else [instead_none] 

75 

76 

77def flatten(x): 

78 def func(acc, x): 

79 if isinstance(x, list): 

80 acc = acc + x 

81 else: 

82 acc.append(x) 

83 

84 return acc 

85 

86 return reduce(func, x, []) 

87 

88 

89def uniq(arr): 

90 # Strong type fast implementation for unique values that preserves ordering 

91 ordered_dict = OrderedDict() 

92 for x in arr: 

93 try: 

94 key = json.dumps(x, sort_keys=True) 

95 except TypeError: 

96 key = str(x) 

97 ordered_dict[key] = x 

98 return list(ordered_dict.values()) 

99 

100 

101def val_data_converted(val): 

102 if isinstance(val, ResourceNode): 

103 val = val.convert_data() 

104 

105 return val 

106 

107 

108def process_user_invocation_table(table): 

109 return { 

110 name: { 

111 **entity, 

112 "fn": lambda ctx, inputs, *args, __fn__=entity["fn"]: __fn__( 

113 [get_data(i) for i in inputs], *args 

114 ), 

115 } 

116 for name, entity in table.items() 

117 }