Ali2206 commited on
Commit
cbf903d
·
verified ·
1 Parent(s): a7e68bf

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +27 -45
app.py CHANGED
@@ -9,13 +9,14 @@ from collections import defaultdict
9
  from typing import List, Dict, Tuple
10
 
11
  # Configuration
12
- PERSISTENT_DIR = "/data/hf_cache"
13
- os.makedirs(os.path.join(PERSISTENT_DIR, "reports"), exist_ok=True)
 
14
 
15
  class PatientHistoryAnalyzer:
16
  def __init__(self):
17
- self.max_token_length = 2000 # Conservative limit
18
- self.max_text_length = 500 # Characters per field
19
 
20
  def clean_text(self, text: str) -> str:
21
  """Clean and normalize text fields"""
@@ -36,7 +37,7 @@ class PatientHistoryAnalyzer:
36
  'diagnoses': defaultdict(list),
37
  'tests': defaultdict(list),
38
  'doctors': set(),
39
- 'all_entries': [] # For full history analysis
40
  }
41
 
42
  for _, row in df.iterrows():
@@ -53,7 +54,6 @@ class PatientHistoryAnalyzer:
53
  data['doctors'].add(entry['doctor'])
54
  data['all_entries'].append(entry)
55
 
56
- # Categorize entries
57
  form_lower = entry['form'].lower()
58
  if 'medication' in form_lower or 'drug' in form_lower:
59
  data['medications'][entry['item']].append(entry)
@@ -71,37 +71,34 @@ class PatientHistoryAnalyzer:
71
  """Generate analysis prompts that respect token limits"""
72
  prompts = []
73
 
74
- # 1. Current Status Prompt (most recent data)
75
  current_prompt = self._create_current_status_prompt(patient_data)
76
  prompts.append({
77
  'type': 'current_status',
78
- 'content': current_prompt,
79
- 'token_estimate': len(current_prompt.split()) # Rough estimate
80
  })
81
 
82
- # 2. Historical Analysis Prompt (if needed)
83
  if len(patient_data['all_entries']) > 10:
84
  history_prompt = self._create_historical_prompt(patient_data)
85
  prompts.append({
86
  'type': 'historical',
87
- 'content': history_prompt,
88
- 'token_estimate': len(history_prompt.split())
89
  })
90
 
91
- # 3. Medication-Specific Prompt (if complex medication history)
92
  if len(patient_data['medications']) > 3:
93
  meds_prompt = self._create_medication_prompt(patient_data)
94
  prompts.append({
95
  'type': 'medications',
96
- 'content': meds_prompt,
97
- 'token_estimate': len(meds_prompt.split())
98
  })
99
 
100
  return prompts
101
 
102
  def _create_current_status_prompt(self, data: Dict) -> str:
103
  """Create prompt for current patient status"""
104
- recent_entries = data['timeline'][-10:] # Last 10 entries
105
 
106
  prompt_lines = [
107
  "**Comprehensive Patient Status Analysis**",
@@ -154,7 +151,7 @@ class PatientHistoryAnalyzer:
154
  "",
155
  "**Historical Timeline (condensed):**",
156
  *[f"- {entry['date'][:7]}: {entry['form']} - {entry['response']}"
157
- for entry in data['all_entries'][:-10]], # All except recent 10
158
  "",
159
  "**Required Output Format:**",
160
  "### Historical Patterns",
@@ -196,11 +193,9 @@ class PatientHistoryAnalyzer:
196
  ""
197
  ]
198
 
199
- # Add each analysis section
200
  for result in analysis_results:
201
  report.extend(["", "---", "", result])
202
 
203
- # Add summary section
204
  report.extend([
205
  "",
206
  "## Overall Clinical Summary",
@@ -215,9 +210,11 @@ class PatientHistoryAnalyzer:
215
 
216
  full_report = "\n".join(report)
217
 
218
- # Save to file
219
  timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
220
- report_path = os.path.join(PERSISTENT_DIR, "reports", f"patient_report_{timestamp}.md")
 
 
221
  with open(report_path, 'w') as f:
222
  f.write(full_report)
223
 
@@ -226,26 +223,21 @@ class PatientHistoryAnalyzer:
226
  def analyze(self, file_path: str) -> Tuple[str, str]:
227
  """Main analysis workflow"""
228
  try:
229
- # Process data
230
  patient_data = self.process_excel(file_path)
231
-
232
- # Generate prompts (simulating LLM analysis)
233
  prompts = self.generate_analysis_prompt(patient_data)
234
 
235
- # Simulate LLM responses (in a real system, you'd call your LLM here)
236
  simulated_responses = [
237
- "### Summary of Current Status\nPatient shows improvement in blood pressure control but new concerns about medication side effects...",
238
- "### Historical Patterns\nChronic back pain has been a consistent issue across 5 providers over 3 years...",
239
- "### Medication Summary\nCurrent regimen includes 4 medications with one potential interaction between..."
240
  ]
241
 
242
- # Generate final report
243
  return self.generate_report(simulated_responses)
244
 
245
  except Exception as e:
246
  return f"Error during analysis: {str(e)}", ""
247
 
248
- # Gradio Interface
249
  def create_interface():
250
  analyzer = PatientHistoryAnalyzer()
251
 
@@ -261,10 +253,6 @@ def create_interface():
261
  file_types=[".xlsx"],
262
  type="filepath"
263
  )
264
- additional_instructions = gr.Textbox(
265
- label="Special Instructions (Optional)",
266
- placeholder="E.g. 'Focus on pain management history'"
267
- )
268
  analyze_btn = gr.Button("Analyze Full History", variant="primary")
269
 
270
  with gr.Column(scale=2):
@@ -281,13 +269,13 @@ def create_interface():
281
  gr.Markdown("""
282
  ## How to Use This Tool
283
 
284
- 1. **Upload** your patient's Excel file containing all medical encounters
285
- 2. **Click Analyze** to process the complete history
286
  3. **Review** the comprehensive analysis
287
  4. **Download** the full report
288
 
289
  ### File Requirements
290
- Excel file must contain these columns:
291
  - Booking Number
292
  - Form Name
293
  - Form Item
@@ -295,13 +283,6 @@ def create_interface():
295
  - Interview Date
296
  - Interviewer
297
  - Description
298
-
299
- ### Analysis Includes
300
- - Current health status
301
- - Medication history
302
- - Diagnostic consistency
303
- - Treatment patterns
304
- - Clinical recommendations
305
  """)
306
 
307
  analyze_btn.click(
@@ -319,7 +300,8 @@ if __name__ == "__main__":
319
  demo.launch(
320
  server_name="0.0.0.0",
321
  server_port=7860,
322
- show_error=True
 
323
  )
324
  except Exception as e:
325
  print(f"Error launching application: {str(e)}")
 
9
  from typing import List, Dict, Tuple
10
 
11
  # Configuration
12
+ WORKING_DIR = os.getcwd()
13
+ REPORT_DIR = os.path.join(WORKING_DIR, "reports")
14
+ os.makedirs(REPORT_DIR, exist_ok=True)
15
 
16
  class PatientHistoryAnalyzer:
17
  def __init__(self):
18
+ self.max_token_length = 2000
19
+ self.max_text_length = 500
20
 
21
  def clean_text(self, text: str) -> str:
22
  """Clean and normalize text fields"""
 
37
  'diagnoses': defaultdict(list),
38
  'tests': defaultdict(list),
39
  'doctors': set(),
40
+ 'all_entries': []
41
  }
42
 
43
  for _, row in df.iterrows():
 
54
  data['doctors'].add(entry['doctor'])
55
  data['all_entries'].append(entry)
56
 
 
57
  form_lower = entry['form'].lower()
58
  if 'medication' in form_lower or 'drug' in form_lower:
59
  data['medications'][entry['item']].append(entry)
 
71
  """Generate analysis prompts that respect token limits"""
72
  prompts = []
73
 
74
+ # Current Status Prompt
75
  current_prompt = self._create_current_status_prompt(patient_data)
76
  prompts.append({
77
  'type': 'current_status',
78
+ 'content': current_prompt
 
79
  })
80
 
81
+ # Historical Analysis Prompt
82
  if len(patient_data['all_entries']) > 10:
83
  history_prompt = self._create_historical_prompt(patient_data)
84
  prompts.append({
85
  'type': 'historical',
86
+ 'content': history_prompt
 
87
  })
88
 
89
+ # Medication-Specific Prompt
90
  if len(patient_data['medications']) > 3:
91
  meds_prompt = self._create_medication_prompt(patient_data)
92
  prompts.append({
93
  'type': 'medications',
94
+ 'content': meds_prompt
 
95
  })
96
 
97
  return prompts
98
 
99
  def _create_current_status_prompt(self, data: Dict) -> str:
100
  """Create prompt for current patient status"""
101
+ recent_entries = data['timeline'][-10:]
102
 
103
  prompt_lines = [
104
  "**Comprehensive Patient Status Analysis**",
 
151
  "",
152
  "**Historical Timeline (condensed):**",
153
  *[f"- {entry['date'][:7]}: {entry['form']} - {entry['response']}"
154
+ for entry in data['all_entries'][:-10]],
155
  "",
156
  "**Required Output Format:**",
157
  "### Historical Patterns",
 
193
  ""
194
  ]
195
 
 
196
  for result in analysis_results:
197
  report.extend(["", "---", "", result])
198
 
 
199
  report.extend([
200
  "",
201
  "## Overall Clinical Summary",
 
210
 
211
  full_report = "\n".join(report)
212
 
213
+ # Save to file in working directory
214
  timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
215
+ report_filename = f"patient_report_{timestamp}.md"
216
+ report_path = os.path.join(REPORT_DIR, report_filename)
217
+
218
  with open(report_path, 'w') as f:
219
  f.write(full_report)
220
 
 
223
  def analyze(self, file_path: str) -> Tuple[str, str]:
224
  """Main analysis workflow"""
225
  try:
 
226
  patient_data = self.process_excel(file_path)
 
 
227
  prompts = self.generate_analysis_prompt(patient_data)
228
 
229
+ # Simulate LLM responses
230
  simulated_responses = [
231
+ "### Summary of Current Status\nPatient shows improvement in blood pressure control...",
232
+ "### Historical Patterns\nChronic back pain has been a consistent issue...",
233
+ "### Medication Summary\nCurrent regimen includes 4 medications..."
234
  ]
235
 
 
236
  return self.generate_report(simulated_responses)
237
 
238
  except Exception as e:
239
  return f"Error during analysis: {str(e)}", ""
240
 
 
241
  def create_interface():
242
  analyzer = PatientHistoryAnalyzer()
243
 
 
253
  file_types=[".xlsx"],
254
  type="filepath"
255
  )
 
 
 
 
256
  analyze_btn = gr.Button("Analyze Full History", variant="primary")
257
 
258
  with gr.Column(scale=2):
 
269
  gr.Markdown("""
270
  ## How to Use This Tool
271
 
272
+ 1. **Upload** your patient's Excel file
273
+ 2. **Click Analyze** to process the history
274
  3. **Review** the comprehensive analysis
275
  4. **Download** the full report
276
 
277
  ### File Requirements
278
+ Excel file must contain:
279
  - Booking Number
280
  - Form Name
281
  - Form Item
 
283
  - Interview Date
284
  - Interviewer
285
  - Description
 
 
 
 
 
 
 
286
  """)
287
 
288
  analyze_btn.click(
 
300
  demo.launch(
301
  server_name="0.0.0.0",
302
  server_port=7860,
303
+ show_error=True,
304
+ allowed_paths=[WORKING_DIR, REPORT_DIR]
305
  )
306
  except Exception as e:
307
  print(f"Error launching application: {str(e)}")