Coverage for src/lumen/fhirpath/engine/evaluators/__init__.py: 80%

1from collections import abc 

2from decimal import Decimal 

3from functools import reduce 

4 

5import re 

6import json 

7import lumen.fhirpath.engine as engine 

8import lumen.fhirpath.engine.util as util 

9import lumen.fhirpath.engine.nodes as nodes 

10 

11 

12def boolean_literal(ctx, parentData, node): 

13 if node["text"] == "true": 

14 return [True] 

15 return [False] 

16 

17 

18def number_literal(ctx, parentData, node): 

19 text = node["text"] 

20 decimal_number = Decimal(text) 

21 if "." in text: 

22 return [decimal_number] 

23 int_number = int(decimal_number) 

24 return [int_number] if decimal_number == int_number else [decimal_number] 

25 

26 

27def identifier(ctx, parentData, node): 

28 return [re.sub(r"(^\"|\"$)", "", node["text"])] 

29 

30 

31def invocation_term(ctx, parentData, node): 

32 return engine.do_eval(ctx, parentData, node["children"][0]) 

33 

34 

35def invocation_expression(ctx, parentData, node): 

36 return list( 

37 reduce( 

38 lambda accumulator, children: engine.do_eval(ctx, accumulator, children), 

39 node["children"], 

40 parentData, 

41 ) 

42 ) 

43 

44 

45def param_list(ctx, parentData, node): 

46 # we do not eval param list because sometimes it should be passed as 

47 # lambda/macro (for example in case of where(...) 

48 return node 

49 

50 

51def union_expression(ctx, parentData, node): 

52 return engine.infix_invoke(ctx, "|", parentData, node["children"]) 

53 

54 

55def index_invocation(ctx, parentData, node): 

56 return util.arraify(ctx["$index"]) 

57 

58 

59def this_invocation(ctx, parentData, node): 

60 return util.arraify(ctx["$this"]) 

61 

62 

63def total_invocation(ctx, parentData, node): 

64 return util.arraify(ctx["$total"]) 

65 

66 

67def op_expression(ctx, parentData, node): 

68 op = node["terminalNodeText"][0] 

69 return engine.infix_invoke(ctx, op, parentData, node["children"]) 

70 

71 

72def alias_op_expression(mapFn): 

73 def func(ctx, parentData, node): 

74 op = node["terminalNodeText"][0] 

75 

76 if not op in mapFn: 

77 raise Exception("Do not know how to alias " + op + " by " + json.dumps(mapFn)) 

78 

79 alias = mapFn[op] 

80 return engine.infix_invoke(ctx, alias, parentData, node["children"]) 

81 

82 return func 

83 

84 

85def term_expression(ctx, parentData, node): 

86 return engine.do_eval(ctx, parentData, node["children"][0]) 

87 

88 

89def null_literal(ctx, parentData, node): 

90 return [] 

91 

92 

93def parenthesized_term(ctx, parentData, node): 

94 return engine.do_eval(ctx, parentData, node["children"][0]) 

95 

96 

97def literal_term(ctx, parentData, node): 

98 term = node["children"][0] 

99 

100 if term: 

101 return engine.do_eval(ctx, parentData, term) 

102 

103 return [node["text"]] 

104 

105 

106def external_constant_term(ctx, parent_data, node): 

107 ext_constant = node["children"][0] 

108 ext_identifier = ext_constant["children"][0] 

109 varName = identifier(ctx, parent_data, ext_identifier)[0].replace("`", "") 

110 

111 if varName not in ctx["vars"]: 

112 raise ValueError(f'Attempting to access an undefined environment variable: {varName}') 

113 

114 value = ctx["vars"][varName] 

115 

116 # For convenience, we all variable values to be passed in without their array 

117 # wrapper. However, when evaluating, we need to put the array back in. 

118 

119 if value is None: 

120 return [] 

121 

122 if not isinstance(value, list): 

123 return [value] 

124 

125 return value 

126 

127 

128def match(m): 

129 code = m.group(1) 

130 return chr(int(code[1:], 16)) 

131 

132 

133def string_literal(ctx, parentData, node): 

134 # Remove the beginning and ending quotes. 

135 rtn = re.sub(r"^['\"]|['\"]$", "", node["text"]) 

136 

137 rtn = rtn.replace("\\'", "'") 

138 rtn = rtn.replace("\\`", "`") 

139 rtn = rtn.replace('\\"', '"') 

140 rtn = rtn.replace("\\r", "\r") 

141 rtn = rtn.replace("\\n", "\n") 

142 rtn = rtn.replace("\\t", "\t") 

143 rtn = rtn.replace("\\f", "\f") 

144 rtn = rtn.replace("\\\\", "\\") 

145 rtn = re.sub(r"\\(u\d{4})", match, rtn) 

146 

147 return [rtn] 

148 

149 

150def quantity_literal(ctx, parentData, node): 

151 valueNode = node["children"][0] 

152 value = Decimal(valueNode["terminalNodeText"][0]) 

153 unitNode = valueNode["children"][0] 

154 if len(unitNode["terminalNodeText"]) > 0: 

155 unit = unitNode["terminalNodeText"][0] 

156 # Sometimes the unit is in a child node of the child 

157 elif "children" in unitNode and len(unitNode["children"]) > 0: 

158 unit = unitNode["children"][0]["terminalNodeText"][0] 

159 else: 

160 unit = None 

161 

162 return [nodes.FP_Quantity(value, unit)] 

163 

164 

165def date_time_literal(ctx, parentData, node): 

166 dateStr = node["text"][1:] 

167 return [nodes.FP_DateTime(dateStr)] 

168 

169 

170_TIME_TZ_RE = re.compile(r"(Z|[+-]\d{2}:\d{2})$") 

171 

172 

173def time_literal(ctx, parentData, node): 

174 timeStr = node["text"][2:] 

175 if _TIME_TZ_RE.search(timeStr): 

176 raise ValueError(f"Time literals may not contain timezone information: @T{timeStr}") 

177 return [nodes.FP_Time(timeStr)] 

178 

179 

180def create_reduce_member_invocation(model, key): 

181 def func(acc, res): 

182 res = nodes.ResourceNode.create_node(res) 

183 childPath = f"{res.path}.{key}" if res.path else f"_.{key}" 

184 fullPath = f"{res.propName}.{key}" if res.propName else childPath # The full path to the node (weill evenutally be) e.g. Patient.name[0].given 

185 fullPath = fullPath.replace("_", "") 

186 

187 actualTypes = None 

188 toAdd = None 

189 toAdd_ = None 

190 

191 if isinstance(model, dict): 

192 childPath = model["pathsDefinedElsewhere"].get(childPath, childPath) 

193 actualTypes = model["choiceTypePaths"].get(childPath) 

194 

195 if isinstance(res.data, nodes.FP_Quantity): 

196 fp_q = res.data 

197 if key == "value": 

198 toAdd = fp_q.value 

199 elif key == "unit": 

200 u_display = getattr(fp_q, "unit_display", None) 

201 raw_unit = fp_q.unit 

202 if u_display: 

203 toAdd = u_display 

204 elif raw_unit and raw_unit.startswith("'") and raw_unit.endswith("'"): 

205 toAdd = raw_unit[1:-1] 

206 else: 

207 toAdd = raw_unit 

208 elif key == "code": 

209 raw_unit = fp_q.unit 

210 if raw_unit and raw_unit.startswith("'") and raw_unit.endswith("'"): 

211 toAdd = raw_unit[1:-1] 

212 else: 

213 toAdd = raw_unit 

214 elif key == "system": 

215 toAdd = "http://unitsofmeasure.org" 

216 

217 if actualTypes and isinstance(res.data, abc.Mapping): 

218 # Use actualTypes to find the field's value 

219 for actualType in actualTypes: 

220 field = f"{key}{actualType}" 

221 toAdd = res.data.get(field) 

222 toAdd_ = res.data.get(f"_{field}") 

223 if toAdd is not None or toAdd_ is not None: 

224 childPath += actualType 

225 break 

226 elif isinstance(res.data, abc.Mapping): 

227 toAdd = res.data.get(key) 

228 toAdd_ = res.data.get(f"_{key}") 

229 if key == "extension": 

230 childPath = "Extension" 

231 else: 

232 if key == "length": 

233 toAdd = len(res.data) 

234 

235 childPath = ( 

236 model["path2Type"].get(childPath, childPath) 

237 if isinstance(model, dict) and "path2Type" in model 

238 else childPath 

239 ) 

240 

241 if util.is_some(toAdd): 

242 if isinstance(toAdd, list): 

243 mapped = [nodes.ResourceNode.create_node(x, childPath, propName=f"{fullPath}[{i}]", index=i) for i, x in enumerate(toAdd)] 

244 for m in mapped: 

245 m.data = m.convert_data() 

246 acc = acc + mapped 

247 else: 

248 m = nodes.ResourceNode.create_node(toAdd, childPath, propName=fullPath) 

249 m.data = m.convert_data() 

250 acc.append(m) 

251 if util.is_some(toAdd_): 

252 if isinstance(toAdd_, list): 

253 mapped = [nodes.ResourceNode.create_node(x, childPath, propName=f"{fullPath}[{i}]", index=i) for i, x in enumerate(toAdd_)] 

254 for m in mapped: 

255 m.data = m.convert_data() 

256 acc = acc + mapped 

257 else: 

258 m = nodes.ResourceNode.create_node(toAdd_, childPath, propName=fullPath) 

259 m.data = m.convert_data() 

260 acc.append(m) 

261 return acc 

262 

263 return func 

264 

265 

266def member_invocation(ctx, parentData, node): 

267 key = engine.do_eval(ctx, parentData, node["children"][0])[0].replace("`", "") 

268 model = ctx["model"] 

269 

270 if isinstance(parentData, list): 

271 if util.is_capitalized(key): 

272 try: 

273 filtered = [x for x in parentData if x["resourceType"] == key] 

274 mapped = [nodes.ResourceNode.create_node(x, key) for x in filtered] 

275 

276 return mapped 

277 except TypeError: 

278 pass 

279 

280 return list(reduce(create_reduce_member_invocation(model, key), parentData, [])) 

281 

282 return [] 

283 

284 

285def indexer_expression(ctx, parentData, node): 

286 coll_node = node["children"][0] 

287 idx_node = node["children"][1] 

288 

289 coll = engine.do_eval(ctx, parentData, coll_node) 

290 idx = engine.do_eval(ctx, parentData, idx_node) 

291 

292 if util.is_empty(idx): 

293 return [] 

294 

295 idxNum = int(idx[0]) 

296 

297 if coll is not None and util.is_some(idxNum) and len(coll) > idxNum and idxNum >= 0: 

298 return [coll[idxNum]] 

299 

300 return [] 

301 

302 

303def functn(ctx, parentData, node): 

304 return [engine.do_eval(ctx, parentData, x) for x in node["children"]] 

305 

306 

307def function_invocation(ctx, parentData, node): 

308 args = engine.do_eval(ctx, parentData, node["children"][0]) 

309 fn_name = args[0] 

310 args = args[1:] 

311 

312 raw_params = None 

313 if isinstance(args, list) and len(args) > 0 and "children" in args[0]: 

314 raw_params = args[0]["children"] 

315 

316 return engine.doInvoke(ctx, fn_name, parentData, raw_params) 

317 

318 

319def is_choice_suffixed(model, res_path, key): 

320 if not isinstance(model, dict) or not res_path: 

321 return False 

322 choice_type_paths = model.get("choiceTypePaths", {}) 

323 for choice_path, types in choice_type_paths.items(): 

324 prefix = f"{res_path}." 

325 if choice_path.startswith(prefix): 

326 base_field = choice_path[len(prefix) :] 

327 if "." not in base_field and key.startswith(base_field) and len(key) > len(base_field): 

328 suffix = key[len(base_field) :] 

329 if any(t.lower() == suffix.lower() for t in types): 

330 return True 

331 return False 

332 

333 

334original_create_reduce_member_invocation = create_reduce_member_invocation 

335 

336def create_reduce_member_invocation(model, key): 

337 reducer = original_create_reduce_member_invocation(model, key) 

338 def wrapped_reducer(acc, res): 

339 node = nodes.ResourceNode.create_node(res) 

340 if is_choice_suffixed(model, node.path, key): 

341 raise ValueError(f"Direct access to choice types by type suffix is not allowed: {key}") 

342 return reducer(acc, res) 

343 return wrapped_reducer 

344 

345 

346def polarity_expression(ctx, parentData, node): 

347 sign = node["terminalNodeText"][0] 

348 rtn = engine.do_eval(ctx, parentData, node["children"][0]) 

349 

350 if not rtn: 

351 return rtn 

352 

353 if len(rtn) != 1: 

354 raise Exception("Unary " + sign + " can only be applied to an individual number.") 

355 

356 val = util.get_data(rtn[0]) 

357 

358 if isinstance(val, nodes.FP_Quantity): 

359 if sign == "-": 

360 return [nodes.FP_Quantity(-val.value, val.unit)] 

361 return rtn 

362 

363 if util.is_number(val): 

364 if sign == "-": 

365 rtn[0] = -rtn[0] 

366 return rtn 

367 

368 # Booleans are not numeric; unary - on bool is an error in FHIRPath. 

369 if isinstance(val, bool): 

370 raise Exception("Unary " + sign + " can only be applied to a number.") 

371 

372 # For non-numeric, non-bool values (strings, ResourceNodes) used as sort keys: 

373 # wrap with _NegatedKey so sorted() can invert order via __lt__/__gt__. 

374 if sign == "-": 

375 return [nodes._NegatedKey(val)] 

376 

377 return rtn 

378 

379 

380evaluators = { 

381 "Functn": functn, 

382 "ParamList": param_list, 

383 "Identifier": identifier, 

384 # terms 

385 "NullLiteral": null_literal, 

386 "LiteralTerm": literal_term, 

387 "NumberLiteral": number_literal, 

388 "StringLiteral": string_literal, 

389 "BooleanLiteral": boolean_literal, 

390 "QuantityLiteral": quantity_literal, 

391 "DateTimeLiteral": date_time_literal, 

392 "TimeLiteral": time_literal, 

393 "InvocationTerm": invocation_term, 

394 "ParenthesizedTerm": parenthesized_term, 

395 "ExternalConstantTerm": external_constant_term, 

396 # Invocations 

397 "ThisInvocation": this_invocation, 

398 "MemberInvocation": member_invocation, 

399 "FunctionInvocation": function_invocation, 

400 "IndexInvocation": index_invocation, 

401 "TotalInvocation": total_invocation, 

402 # expressions 

403 "PolarityExpression": polarity_expression, 

404 "IndexerExpression": indexer_expression, 

405 "MembershipExpression": alias_op_expression({"contains": "containsOp", "in": "inOp"}), 

406 "TermExpression": term_expression, 

407 "UnionExpression": union_expression, 

408 "InvocationExpression": invocation_expression, 

409 "InequalityExpression": op_expression, 

410 "AdditiveExpression": op_expression, 

411 "MultiplicativeExpression": op_expression, 

412 "TypeExpression": alias_op_expression({"is": "isOp", "as": "asOp"}), 

413 "EqualityExpression": op_expression, 

414 "OrExpression": op_expression, 

415 "ImpliesExpression": op_expression, 

416 "AndExpression": op_expression, 

417 "XorExpression": op_expression, 

418}