Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 53 additions & 40 deletions src/cases/service/case_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -270,20 +270,13 @@ def get_page_configuration(self):

def get_case_review(self, case_config_id): # pragma: no cover
"""
A function that implements the business logic for the /case-reviews/{id} API endpoint.
Steps:
1) Load saved DisplayConfig for this case_config_id + user_email.
1) Load DisplayConfig, verify access.
2) Build full unpruned case_details tree.
3) Prune under “BACKGROUND” according to CSV path_config.
4) If CSV included a literal Colorectal Cancer Score leaf, use that directly.
5) Else if CSV toggled CRC risk assessments, look for an observation row:
a) First, any value_as_string starting with "Colorectal Cancer Score"
b) Then, any "Adjusted CRC Risk" row (renaming it)
c) If neither found, emit "N/A"
6) If CSV had no RISK ASSESSMENT entries at all, do not emit any CRC score node.
7) Return Case object with pruned case_details and important_infos.

This function relies on the Tree data structure, which is built by utils/csv_parser.py.
3) Prune under “BACKGROUND” per path_config.
4) Handle CSV-provided literal Colorectal Cancer Score leaves (now possibly multiple):
• collect all, compute min/max.
5) Else if old CRC toggle, fetch from DB as before.
6) Return Case with sorted important_infos.
"""
# --- 1) Load configuration & verify access ---
configuration = self.configuration_repository.get_configuration_by_id(case_config_id)
Expand All @@ -298,8 +291,8 @@ def get_case_review(self, case_config_id): # pragma: no cover
raw_path_cfg = configuration.path_config or []
parent_to_entries: dict[str, list[dict]] = defaultdict(list)

# Track whether CSV provided an explicit literal score or just toggled the section
csv_crc_score_leaf: str | None = None
# Track CSV-provided literal score overrides (now possibly many)
csv_crc_score_leaves: list[str] = []
old_crc_toggle = False

for entry in raw_path_cfg:
Expand All @@ -309,43 +302,39 @@ def get_case_review(self, case_config_id): # pragma: no cover
continue

segments = path_str.split(".")
# We need at least a parent and a leaf
if len(segments) < 2:
continue

parent_key = ".".join(segments[:-1])
leaf_text = segments[-1]

# CSV-provided literal score override?
# collect any literal "Colorectal Cancer Score: X"
if path_str.startswith("RISK ASSESSMENT.Colorectal Cancer Score"):
csv_crc_score_leaf = leaf_text
csv_crc_score_leaves.append(leaf_text)

# Old-style toggle of the entire CRC section?
# old-style toggle of entire CRC section
if path_str == "RISK ASSESSMENT.CRC risk assessments":
old_crc_toggle = True

parent_to_entries[parent_key].append({"leaf": leaf_text, "style": style})

# --- 4) Prune under BACKGROUND according to path_config ---
# --- 4) Prune under BACKGROUND per path_config ---
important_infos: list[dict] = []
for top in case_details:
if top.key != "BACKGROUND":
continue
for child in top.values:
if child.key == "Patient Demographics":
# always keep demographics
continue
continue # always keep
pk = f"BACKGROUND.{child.key}"
if pk not in parent_to_entries:
# no CSV entries for this category ⇒ drop it entirely
child.values = []
continue
entries = parent_to_entries[pk]
# only keep those leaves the CSV listed
keep = {e["leaf"] for e in entries}
child.values = [v for v in child.values if v in keep]

# merge style directives (collapse, highlight, top)
# merge style directives
merged: dict = {}
for e in entries:
s = e["style"]
Expand All @@ -357,32 +346,55 @@ def get_case_review(self, case_config_id): # pragma: no cover
merged["top"] = max(merged.get("top", -1), s["top"])

child.style = merged
# collect any "top"-weighted nodes for separate display
if merged.get("top") is not None:
important_infos.append({
"key": child.key,
"values": child.values,
"weight": merged["top"],
})

# sort by weight and wrap into TreeNodes
# sort and wrap into TreeNodes
important_infos.sort(key=itemgetter("weight"))
sorted_important = [TreeNode(e["key"], e["values"]) for e in important_infos]

# --- 5) Handle AI CRC Risk Score section only if CSV referenced it ---
# --- 5) Handle AI CRC Risk Score section ---
ai_label = "AI CRC Risk Score (<6: Low; 6-11: Medium; >11: High)"

if csv_crc_score_leaf:
# CSV gave us the exact leaf text to show
sorted_important.append(TreeNode(ai_label, [csv_crc_score_leaf]))
if csv_crc_score_leaves:
# parse all numeric scores
nums = []
for leaf in csv_crc_score_leaves:
parts = leaf.split(":")
if len(parts) == 2:
try:
nums.append(float(parts[1].strip()))
except ValueError:
pass

# pick the "predicted" one (e.g. first)
predicted_leaf = csv_crc_score_leaves[0]
display_pred = predicted_leaf.replace(
"Colorectal Cancer Score",
"Predicted Colorectal Cancer Score"
)
sorted_important.append(TreeNode(ai_label, [display_pred]))

# if we have numeric values, emit min/max
if nums:
min_val = int(min(nums))
max_val = int(max(nums))
sorted_important.append(
TreeNode("Min Predicted Colorectal Cancer Score", [str(min_val)])
)
sorted_important.append(
TreeNode("Max Predicted Colorectal Cancer Score", [str(max_val)])
)

elif old_crc_toggle:
# CSV toggled the section but did not supply a literal score; fetch from DB
crc_obs = self.observation_repository.get_observations_by_concept(
configuration.case_id, [45614722]
)

# (a) look first for any literal "Colorectal Cancer Score: X"
# (a) any literal "Colorectal Cancer Score: X"
csv_obs = next(
(
o.value_as_string
Expand All @@ -392,25 +404,26 @@ def get_case_review(self, case_config_id): # pragma: no cover
None
)
if csv_obs:
sorted_important.append(TreeNode(ai_label, [csv_obs]))
display_txt = csv_obs.replace(
"Colorectal Cancer Score",
"Predicted Colorectal Cancer Score"
)
sorted_important.append(TreeNode(ai_label, [display_txt]))
else:
# (b) fallback to any "Adjusted CRC Risk" row
# (b) fallback to Adjusted CRC Risk
for obs in crc_obs:
txt = obs.value_as_string or ""
if txt.startswith("Adjusted CRC Risk"):
# rename to match "AI-Predicted CRC Risk Score"
new_txt = txt.replace(
"Adjusted CRC Risk",
"AI-Predicted CRC Risk Score"
)
sorted_important.append(TreeNode(ai_label, [new_txt]))
break
else:
# (c) CSV toggled but no score found ⇒ explicitly show N/A
sorted_important.append(TreeNode(ai_label, ["N/A"]))

# 6) If neither csv_crc_score_leaf nor old_crc_toggle, we do not append any CRC score node.

# 6) done—return full Case
return Case(
self.person.person_source_value,
str(configuration.case_id),
Expand Down