| """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", |
| } |
|
|
