Coverage for src/lumen/fhir/terminology.py: 91%

1""" 

2This module contains core terminology logic for FHIR CodeSystems, ValueSets, and ConceptMaps. 

3It handles operations like code lookup, value set expansion, and concept translation. 

4""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import Iterator 

9from typing import Any, cast 

10 

11JsonDict = dict[str, Any] 

12 

13 

14class TerminologyError(Exception): 

15 """An unsupported or invalid terminology request (mapped to OperationOutcome).""" 

16 

17 def __init__(self, code: str, message: str) -> None: 

18 self.code = code 

19 super().__init__(message) 

20 

21 

22def _dicts(value: Any) -> list[JsonDict]: 

23 """Return ``value`` as a list of dicts, dropping non-dict / non-list inputs.""" 

24 if not isinstance(value, list): 

25 return [] 

26 return [cast("JsonDict", item) for item in cast("list[Any]", value) if isinstance(item, dict)] 

27 

28 

29# ── CodeSystem / $lookup ─────────────────────────────────────────────────────── 

30 

31 

32def _iter_concepts(concepts: list[JsonDict]) -> Iterator[JsonDict]: 

33 """Yield every concept in a (possibly nested) CodeSystem.concept tree.""" 

34 for concept in concepts: 

35 yield concept 

36 yield from _iter_concepts(_dicts(concept.get("concept"))) 

37 

38 

39def lookup_concept(code_system: JsonDict, code: str) -> JsonDict | None: 

40 """Return the concept with ``code`` from a CodeSystem, or None.""" 

41 for concept in _iter_concepts(_dicts(code_system.get("concept"))): 

42 if concept.get("code") == code: 

43 return concept 

44 return None 

45 

46 

47# ── ValueSet / $expand ───────────────────────────────────────────────────────── 

48 

49 

50def referenced_systems(value_set: JsonDict) -> set[str]: 

51 """Collect every CodeSystem url referenced by a ValueSet's compose rules.""" 

52 systems: set[str] = set() 

53 compose = value_set.get("compose") 

54 if not isinstance(compose, dict): 

55 return systems 

56 compose = cast("JsonDict", compose) 

57 for key in ("include", "exclude"): 

58 for rule in _dicts(compose.get(key)): 

59 system = rule.get("system") 

60 if isinstance(system, str): 

61 systems.add(system) 

62 return systems 

63 

64 

65def _explicit_rule_codes( 

66 explicit: list[Any], 

67 system: str, 

68 code_system: JsonDict | None, 

69) -> list[tuple[str, str, str | None]]: 

70 out: list[tuple[str, str, str | None]] = [] 

71 for concept in _dicts(explicit): 

72 code = concept.get("code") 

73 if not isinstance(code, str): 

74 continue 

75 display = concept.get("display") 

76 if display is None and code_system is not None: 

77 resolved = lookup_concept(code_system, code) 

78 display = resolved.get("display") if resolved else None 

79 out.append((system, code, display if isinstance(display, str) else None)) 

80 return out 

81 

82 

83def _rule_codes(rule: JsonDict, systems: dict[str, JsonDict]) -> list[tuple[str, str, str | None]]: 

84 """Expand one include/exclude rule to (system, code, display) tuples.""" 

85 if "valueSet" in rule: 

86 raise TerminologyError("not-supported", "ValueSet import in compose is not supported") 

87 if "filter" in rule: 

88 raise TerminologyError("not-supported", "compose filters are not supported") 

89 

90 system = rule.get("system") 

91 if not isinstance(system, str): 

92 raise TerminologyError("invalid", "compose rule without a system is not supported") 

93 

94 explicit = rule.get("concept") 

95 if isinstance(explicit, list): 

96 code_system = systems.get(system) 

97 return _explicit_rule_codes(cast("list[Any]", explicit), system, code_system) 

98 

99 # No explicit concepts → the entire CodeSystem. 

100 code_system = systems.get(system) 

101 if code_system is None: 

102 raise TerminologyError("not-found", f"CodeSystem {system} not found for expansion") 

103 out: list[tuple[str, str, str | None]] = [] 

104 for concept in _iter_concepts(_dicts(code_system.get("concept"))): 

105 code = concept.get("code") 

106 if isinstance(code, str): 

107 display = concept.get("display") 

108 out.append((system, code, display if isinstance(display, str) else None)) 

109 return out 

110 

111 

112def expand_valueset(value_set: JsonDict, systems: dict[str, JsonDict]) -> list[JsonDict]: 

113 """Return the ``expansion.contains`` list for a ValueSet given its CodeSystems.""" 

114 compose = value_set.get("compose") 

115 if not isinstance(compose, dict): 

116 return [] 

117 compose = cast("JsonDict", compose) 

118 

119 included: list[tuple[str, str, str | None]] = [] 

120 for rule in _dicts(compose.get("include")): 

121 included.extend(_rule_codes(rule, systems)) 

122 

123 excluded: set[tuple[str, str]] = set() 

124 for rule in _dicts(compose.get("exclude")): 

125 excluded.update((s, c) for s, c, _ in _rule_codes(rule, systems)) 

126 

127 contains: list[JsonDict] = [] 

128 seen: set[tuple[str, str]] = set() 

129 for system, code, display in included: 

130 key = (system, code) 

131 if key in excluded or key in seen: 

132 continue 

133 seen.add(key) 

134 entry: JsonDict = {"system": system, "code": code} 

135 code_system_version = systems.get(system, {}).get("version") 

136 if isinstance(code_system_version, str): 

137 entry["version"] = code_system_version 

138 if display is not None: 

139 entry["display"] = display 

140 contains.append(entry) 

141 return contains 

142 

143 

144# ── ConceptMap / $translate ──────────────────────────────────────────────────── 

145 

146 

147def _translate_group_elements( 

148 group: JsonDict, 

149 code: str, 

150 group_target: str | None, 

151 matches: list[JsonDict], 

152) -> None: 

153 for element in _dicts(group.get("element")): 

154 if element.get("code") != code: 

155 continue 

156 for tgt in _dicts(element.get("target")): 

157 tgt_code = tgt.get("code") 

158 if not isinstance(tgt_code, str): 

159 continue 

160 concept: JsonDict = {"system": group_target, "code": tgt_code} 

161 display = tgt.get("display") 

162 if isinstance(display, str): 

163 concept["display"] = display 

164 equivalence = tgt.get("equivalence") 

165 matches.append( 

166 { 

167 "equivalence": equivalence if isinstance(equivalence, str) else "relatedto", 

168 "concept": concept, 

169 } 

170 ) 

171 

172 

173def translate( 

174 concept_map: JsonDict, 

175 system: str | None, 

176 code: str, 

177 target_system: str | None = None, 

178) -> list[JsonDict]: 

179 """Return target matches for a source code, each as {equivalence, concept}. 

180 

181 ``equivalence`` carries the R4B relationship (the ``equivalence`` field on a 

182 ConceptMap target element). 

183 """ 

184 matches: list[JsonDict] = [] 

185 for group in _dicts(concept_map.get("group")): 

186 if system is not None and group.get("source") != system: 

187 continue 

188 group_target = group.get("target") 

189 if target_system is not None and group_target != target_system: 

190 continue 

191 _translate_group_elements(group, code, group_target, matches) 

192 return matches