File size: 1,684 Bytes
cb505ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | """Deterministic CDM capabilities exposed through the Tool Broker."""
from __future__ import annotations
from typing import Any, Dict
from langchain_core.tools import tool
from src.tools.fhir_memory import get_observations_by_patient
def _trend_statistics(values: list[float]) -> Dict[str, Any]:
if len(values) < 2:
return {"trend": "insufficient", "slope": 0.0, "projection": None}
slope = (values[-1] - values[0]) / (len(values) - 1)
trend = "rising" if slope > 0.5 else "falling" if slope < -0.5 else "stable"
return {
"trend": trend,
"slope": float(slope),
"projection": float(values[-1] + slope),
}
@tool
def analyze_trends(patient_id: str) -> Dict[str, Any]:
"""Compute deterministic trend summaries from authorized FHIR observations."""
observations = get_observations_by_patient.invoke({"patient_id": patient_id})
if not isinstance(observations, list):
return {"patient_id": patient_id, "metrics": {}, "status": "unavailable"}
grouped: Dict[str, list[float]] = {}
for observation in observations:
try:
resource = observation.get("resource", observation)
display = resource["code"]["coding"][0]["display"]
value = float(resource["valueQuantity"]["value"])
except (AttributeError, KeyError, IndexError, TypeError, ValueError):
continue
grouped.setdefault(display, []).append(value)
return {
"patient_id": patient_id,
"metrics": {
metric: {"values": values, **_trend_statistics(values)}
for metric, values in grouped.items()
},
"status": "completed",
}
|