sanggusti commited on
Commit
9690499
·
1 Parent(s): 2aadf09

Grok Engineering, fingers crossed

Browse files
Files changed (3) hide show
  1. utils/meldrx.py +8 -12
  2. utils/oneclick.py +33 -40
  3. utils/responseparser.py +26 -7
utils/meldrx.py CHANGED
@@ -59,18 +59,14 @@ class MeldRxAPI:
59
  headers["Authorization"] = f"Bearer {self.access_token}"
60
  return headers
61
 
62
- def get_patients(self) -> Optional[Dict[str, Any]]:
63
- url = f"{self.fhir_base_url}/Patient"
64
- if not self.access_token and not self.authenticate():
65
- print("Cannot proceed without authentication.")
66
- return None
67
- try:
68
- response = self.session.get(url, headers=self._get_headers())
69
- response.raise_for_status()
70
- return response.json() if response.text else {}
71
- except requests.RequestException as e:
72
- print(f"Failed to retrieve patients: {e}")
73
- return None
74
 
75
  def get_authorization_url(self, scope: str = "patient/*.read openid profile", state: str = "random_state") -> str:
76
  code_verifier = self._generate_code_verifier()
 
59
  headers["Authorization"] = f"Bearer {self.access_token}"
60
  return headers
61
 
62
+ def get_patients(self):
63
+ # This should return the FHIR Bundle as per your sample output
64
+ headers = {"Authorization": f"Bearer {self.access_token}"}
65
+ response = requests.get(
66
+ f"https://app.meldrx.com/api/fhir/{self.workspace_id}/Patient",
67
+ headers=headers
68
+ )
69
+ return response.json() if response.status_code == 200 else None
 
 
 
 
70
 
71
  def get_authorization_url(self, scope: str = "patient/*.read openid profile", state: str = "random_state") -> str:
72
  code_verifier = self._generate_code_verifier()
utils/oneclick.py CHANGED
@@ -91,63 +91,56 @@ def generate_discharge_paper_one_click(
91
 
92
  logger.info(f"Found {len(extractor.patients)} patients in the data")
93
 
94
- matching_patients = []
95
- all_patient_ids = []
96
- all_patient_names = []
97
-
 
 
98
  for i in range(len(extractor.patients)):
99
  extractor.set_patient_by_index(i)
100
  patient_data = extractor.get_patient_dict()
101
-
102
  patient_id_from_data = str(patient_data.get('id', '')).strip().lower()
103
- first_name_from_data = str(patient_data.get('first_name', '')).strip().lower()
104
- last_name_from_data = str(patient_data.get('last_name', '')).strip().lower()
105
-
106
- all_patient_ids.append(patient_id_from_data)
107
- all_patient_names.append(f"{first_name_from_data} {last_name_from_data}".strip())
108
-
109
- patient_id_input = str(patient_id).strip().lower()
110
- first_name_input = str(first_name).strip().lower()
111
- last_name_input = str(last_name).strip().lower()
112
 
113
- logger.debug(f"Patient {i}: ID={patient_id_from_data}, Name={first_name_from_data} {last_name_from_data}")
114
- logger.debug(f"Comparing - Input: ID={patient_id_input}, First={first_name_input}, Last={last_name_input}")
115
-
116
- # Match logic: ID takes precedence, then first/last name
117
- matches = False
118
  if patient_id_input and patient_id_from_data == patient_id_input:
119
- matches = True
120
- elif not patient_id_input and first_name_input and last_name_input:
121
- if first_name_input == first_name_from_data and last_name_input == last_name_from_data:
122
- matches = True
123
- elif not patient_id_input and not first_name_input and not last_name_input:
124
- continue # Skip if no criteria provided
125
-
126
- if matches:
127
- matching_patients.append(patient_data)
128
- logger.info(f"Found matching patient: ID={patient_id_from_data}, "
129
- f"Name={first_name_from_data} {last_name_from_data}")
130
-
131
- if not matching_patients:
 
 
 
 
 
 
132
  search_criteria = f"ID: {patient_id or 'N/A'}, First: {first_name or 'N/A'}, Last: {last_name or 'N/A'}"
 
 
 
133
  logger.warning(f"No patients matched criteria: {search_criteria}")
134
- logger.info(f"Available patient IDs: {all_patient_ids}")
135
- logger.info(f"Available patient names: {all_patient_names}")
136
  return None, (f"No patients found matching criteria: {search_criteria}\n"
137
  f"Available IDs: {', '.join(all_patient_ids)}\n"
138
  f"Available Names: {', '.join(all_patient_names)}"), None, None, None
 
 
139
 
140
- patient_data = matching_patients[0] # Take the first match
141
- logger.info(f"Selected patient data: {patient_data}")
142
-
143
- basic_summary = format_discharge_summary(patient_data)
144
- ai_summary, verified_summary = generate_ai_discharge_summary(patient_data, client)
145
 
146
  if not ai_summary or not verified_summary:
147
  return None, "Failed to generate or verify AI summary", basic_summary, None, None
148
 
149
  pdf_gen = PDFGenerator()
150
- filename = f"discharge_{patient_data.get('id', 'unknown')}_{patient_data.get('last_name', 'patient')}.pdf"
151
  pdf_path = pdf_gen.generate_pdf_from_text(ai_summary, filename)
152
 
153
  if pdf_path:
 
91
 
92
  logger.info(f"Found {len(extractor.patients)} patients in the data")
93
 
94
+ matching_patient = None
95
+ patient_id_input = str(patient_id).strip().lower()
96
+ first_name_input = str(first_name).strip().lower()
97
+ last_name_input = str(last_name).strip().lower()
98
+
99
+ # Strict matching by patient_id first
100
  for i in range(len(extractor.patients)):
101
  extractor.set_patient_by_index(i)
102
  patient_data = extractor.get_patient_dict()
 
103
  patient_id_from_data = str(patient_data.get('id', '')).strip().lower()
 
 
 
 
 
 
 
 
 
104
 
 
 
 
 
 
105
  if patient_id_input and patient_id_from_data == patient_id_input:
106
+ matching_patient = patient_data
107
+ logger.info(f"Exact match found for patient ID: {patient_id_from_data}")
108
+ break
109
+
110
+ # Fallback to name-based matching if no ID match and names are provided
111
+ if not matching_patient and (first_name_input or last_name_input):
112
+ for i in range(len(extractor.patients)):
113
+ extractor.set_patient_by_index(i)
114
+ patient_data = extractor.get_patient_dict()
115
+ first_name_from_data = str(patient_data.get('first_name', '')).strip().lower()
116
+ last_name_from_data = str(patient_data.get('last_name', '')).strip().lower()
117
+
118
+ if (first_name_input == first_name_from_data and
119
+ last_name_input == last_name_from_data):
120
+ matching_patient = patient_data
121
+ logger.info(f"Match found by name: {first_name_from_data} {last_name_from_data}")
122
+ break
123
+
124
+ if not matching_patient:
125
  search_criteria = f"ID: {patient_id or 'N/A'}, First: {first_name or 'N/A'}, Last: {last_name or 'N/A'}"
126
+ all_patient_ids = [str(p.get('id', '')) for p in extractor.get_all_patients()]
127
+ all_patient_names = [f"{p.get('first_name', '')} {p.get('last_name', '')}".strip()
128
+ for p in extractor.get_all_patients()]
129
  logger.warning(f"No patients matched criteria: {search_criteria}")
 
 
130
  return None, (f"No patients found matching criteria: {search_criteria}\n"
131
  f"Available IDs: {', '.join(all_patient_ids)}\n"
132
  f"Available Names: {', '.join(all_patient_names)}"), None, None, None
133
+
134
+ logger.info(f"Selected patient data: {matching_patient}")
135
 
136
+ basic_summary = format_discharge_summary(matching_patient)
137
+ ai_summary, verified_summary = generate_ai_discharge_summary(matching_patient, client)
 
 
 
138
 
139
  if not ai_summary or not verified_summary:
140
  return None, "Failed to generate or verify AI summary", basic_summary, None, None
141
 
142
  pdf_gen = PDFGenerator()
143
+ filename = f"discharge_{matching_patient.get('id', 'unknown')}_{matching_patient.get('last_name', 'patient')}.pdf"
144
  pdf_path = pdf_gen.generate_pdf_from_text(ai_summary, filename)
145
 
146
  if pdf_path:
utils/responseparser.py CHANGED
@@ -256,13 +256,24 @@ class PatientDataExtractor:
256
  def get_patient_dict(self) -> Dict[str, str]:
257
  """Return a dictionary of patient data mapped to discharge form fields."""
258
  data = self.get_all_patient_data()
 
 
259
  latest_encounter = data["encounters"][-1] if data["encounters"] else {}
 
 
 
 
260
  latest_condition = data["conditions"][-1] if data["conditions"] else {}
261
- medications_str = "; ".join([m["description"] for m in data["medications"] if m["description"]])
 
 
 
 
262
  return {
263
  "id": data["id"],
264
  "first_name": data["first_name"],
265
  "last_name": data["last_name"],
 
266
  "dob": data["dob"],
267
  "age": data["age"],
268
  "sex": data["gender"],
@@ -271,17 +282,25 @@ class PatientDataExtractor:
271
  "state": data["state"],
272
  "zip_code": data["zip_code"],
273
  "phone": data["phone"],
274
- "doctor_first_name": "", # Could be extracted from Practitioner resource if available
 
 
 
 
275
  "doctor_last_name": "",
276
- "hospital_name": "",
277
  "doctor_address": "",
278
  "doctor_city": "",
279
  "doctor_state": "",
280
  "doctor_zip": "",
281
- "admission_date": latest_encounter.get("start", ""),
282
- "discharge_date": latest_encounter.get("end", ""),
283
- "diagnosis": latest_condition.get("description", ""),
284
- "medications": medications_str if medications_str else "None specified",
 
 
 
 
285
  }
286
 
287
  def get_all_patient_data(self) -> Dict[str, Union[str, List, Dict]]:
 
256
  def get_patient_dict(self) -> Dict[str, str]:
257
  """Return a dictionary of patient data mapped to discharge form fields."""
258
  data = self.get_all_patient_data()
259
+
260
+ # Get the latest encounter for admission/discharge dates
261
  latest_encounter = data["encounters"][-1] if data["encounters"] else {}
262
+ admission_date = latest_encounter.get("start", "")
263
+ discharge_date = latest_encounter.get("end", "")
264
+
265
+ # Get the latest condition for diagnosis
266
  latest_condition = data["conditions"][-1] if data["conditions"] else {}
267
+ diagnosis = latest_condition.get("description", "")
268
+
269
+ # Format medications as a string
270
+ medications_str = "; ".join([m["description"] for m in data["medications"] if m["description"]]) or "None specified"
271
+
272
  return {
273
  "id": data["id"],
274
  "first_name": data["first_name"],
275
  "last_name": data["last_name"],
276
+ "name_prefix": data["name_prefix"],
277
  "dob": data["dob"],
278
  "age": data["age"],
279
  "sex": data["gender"],
 
282
  "state": data["state"],
283
  "zip_code": data["zip_code"],
284
  "phone": data["phone"],
285
+ "admission_date": admission_date,
286
+ "discharge_date": discharge_date,
287
+ "diagnosis": diagnosis,
288
+ "medications": medications_str,
289
+ "doctor_first_name": "", # Could be extracted from Practitioner resource if linked
290
  "doctor_last_name": "",
291
+ "hospital_name": "", # Could be extracted from Organization resource if linked
292
  "doctor_address": "",
293
  "doctor_city": "",
294
  "doctor_state": "",
295
  "doctor_zip": "",
296
+ "middle_initial": "",
297
+ "referral_source": "",
298
+ "admission_method": "",
299
+ "discharge_reason": "",
300
+ "date_of_death": "",
301
+ "procedures": "",
302
+ "preparer_name": "",
303
+ "preparer_job_title": ""
304
  }
305
 
306
  def get_all_patient_data(self) -> Dict[str, Union[str, List, Dict]]: