Spaces:
Sleeping
Sleeping
File size: 3,774 Bytes
22195c1 e15864e 22195c1 a009989 22195c1 e15864e 22195c1 | 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 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | from __future__ import annotations
import math
from dataclasses import dataclass, field
from enum import Enum
from typing import Union
import httpx
@dataclass
class PatientProfile:
disease: str
age: int
onset_months: int
diagnosis_months: int = 0
benchmarks: dict[str, str] = field(default_factory=dict)
zip_code: str = ""
country_code: str = "US"
lat: float = 0.0
lon: float = 0.0
radius_miles: int = 20
phases: list[str] = field(default_factory=list)
include_eap: bool = False
include_observational: bool = False
lang: str = "en"
def summary(self) -> str:
lines = [
f"Disease: {self.disease}",
f"Age: {self.age}",
f"Symptom onset: {self.onset_months} months ago",
f"Formal diagnosis: {self.diagnosis_months} months ago",
]
if self.benchmarks:
lines.append("Benchmarks: " + ", ".join(f"{k}={v}" for k, v in self.benchmarks.items()))
lines.append(
f"Location: ZIP {self.zip_code}, {self.country_code} "
f"(lat={self.lat:.4f}, lon={self.lon:.4f})"
)
lines.append(f"Search radius: {self.radius_miles} miles")
if self.phases:
def _phase_label(p: str) -> str:
if p == "0":
return "Early Phase 1"
if p == "na":
return "Not Applicable"
return f"Phase {p}"
labels = [_phase_label(p) for p in self.phases]
lines.append(f"Phases: {', '.join(labels)}")
interests = ["Clinical trials"]
if self.include_observational:
interests.append("Observational studies")
if self.include_eap:
interests.append("Expanded Access Programs (EAP)")
lines.append(f"Study type interest: {', '.join(interests)}")
return "\n".join(lines)
class CriterionVerdict(str, Enum):
PASS = "pass"
FAIL = "fail"
UNKNOWN = "unknown"
@dataclass
class ParsedConstraint:
key: str
operator: str # "<=", ">=", "==", "!=", "in", "not_in", "between"
value: Union[int, float, str, list]
unit: str | None
@dataclass
class EligibilityCriterion:
key: str
type: str # "inclusion" | "exclusion"
description: str
raw_criteria: str # verbatim criterion text preserved for patient transparency
constraint: ParsedConstraint | None
@dataclass
class CriterionAssessment:
criterion: EligibilityCriterion
verdict: CriterionVerdict
reason: str
patient_value: str | None
confidence: str # "high" (deterministic) | "medium" | "low" (LLM)
@dataclass
class TrialEligibilityReport:
nct_id: str
overall_verdict: CriterionVerdict
assessments: list[CriterionAssessment]
missing_data_keys: list[str]
def geocode_zip(zip_code: str, country_code: str = "US") -> tuple[float, float]:
resp = httpx.get(
"https://nominatim.openstreetmap.org/search",
params={"postalcode": zip_code, "country": country_code, "format": "json", "limit": 1},
headers={"User-Agent": "Beacon-ClinicalTrialFinder/1.0"},
timeout=10,
)
resp.raise_for_status()
results = resp.json()
if not results:
raise ValueError(f"Cannot geocode ZIP {zip_code!r} in {country_code!r}")
return float(results[0]["lat"]), float(results[0]["lon"])
def haversine_miles(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
R = 3958.8
φ1, φ2 = math.radians(lat1), math.radians(lat2)
dφ, dλ = math.radians(lat2 - lat1), math.radians(lon2 - lon1)
a = math.sin(dφ / 2) ** 2 + math.cos(φ1) * math.cos(φ2) * math.sin(dλ / 2) ** 2
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|