Browse Source

added new project to communuty section.

pull/200/head
shreshthkapai 3 months ago
parent
commit
723052ef1c
  1. 1
      week2/community-contributions/medical_prescription_to_google_calender/credentials.json
  2. 35
      week2/community-contributions/medical_prescription_to_google_calender/src/calendar_auth.py
  3. 64
      week2/community-contributions/medical_prescription_to_google_calender/src/create_calender_events.py
  4. 26
      week2/community-contributions/medical_prescription_to_google_calender/src/main.py
  5. 71
      week2/community-contributions/medical_prescription_to_google_calender/src/ocr.py
  6. 120
      week2/community-contributions/medical_prescription_to_google_calender/src/parsing_json.py
  7. 141
      week2/community-contributions/medical_prescription_to_google_calender/src/preprocess.py

1
week2/community-contributions/medical_prescription_to_google_calender/credentials.json

@ -0,0 +1 @@
"Your google auth credentials."

35
week2/community-contributions/medical_prescription_to_google_calender/src/calendar_auth.py

@ -0,0 +1,35 @@
import os
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build # Add this import
SCOPES = ["https://www.googleapis.com/auth/calendar.events"]
def authenticate_google_calender():
creds = None
token_path = r"C:\Users\Legion\Desktop\projects\medical_prescription_to_google_calender\token.json"
if os.path.exists(token_path):
creds = Credentials.from_authorized_user_file(token_path, SCOPES)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(r"C:\Users\Legion\Desktop\projects\medical_prescription_to_google_calender\credentials.json", SCOPES)
creds = flow.run_local_server(port=0)
with open(token_path, "w") as token_file:
token_file.write(creds.to_json())
# Build and return the service instead of just credentials
try:
service = build('calendar', 'v3', credentials=creds)
return service
except Exception as e:
print(f"Error building service: {e}")
return None
if __name__ == "__main__":
authenticate_google_calender()

64
week2/community-contributions/medical_prescription_to_google_calender/src/create_calender_events.py

@ -0,0 +1,64 @@
from googleapiclient.discovery import build
from calendar_auth import authenticate_google_calender
from parsing_json import format_calendar_events
from datetime import datetime, timedelta
def create_event(service, event_details):
"""Creates an event in Google Calendar."""
try:
event = service.events().insert(calendarId='primary', body=event_details).execute()
print(f"Event created: {event.get('htmlLink')}")
except Exception as e:
print(f"Error creating event: {str(e)}")
def convert_time_to_24hr(time_str):
"""Converts time from '10:30 am' format to '10:30:00'"""
if time_str and time_str.lower() != 'none':
try:
parsed_time = datetime.strptime(time_str, '%I:%M %p')
return parsed_time.strftime('%H:%M:%S')
except ValueError:
return '09:00:00'
return '09:00:00'
def convert_to_gcal_events(formatted_events):
"""Converts formatted events into Google Calendar's format."""
gcal_events = []
for event in formatted_events:
gcal_event = {
'summary': event['summary'],
'reminders': {
'useDefault': False,
'overrides': [{'method': 'popup', 'minutes': 10}]
}
}
# Check if it's an all-day event (has 'date') or timed event (has 'dateTime')
if 'date' in event['start']:
# All-day event (like tests and follow-ups)
gcal_event['start'] = {
'date': event['start']['date'],
'timeZone': 'Asia/Kolkata'
}
gcal_event['end'] = {
'date': event['end']['date'],
'timeZone': 'Asia/Kolkata'
}
else:
# Timed event (like medicine schedules)
start_dt = datetime.strptime(event['start']['dateTime'], '%Y-%m-%dT%H:%M:%S')
end_dt = start_dt + timedelta(minutes=30)
gcal_event['start'] = {
'dateTime': start_dt.isoformat(),
'timeZone': 'Asia/Kolkata'
}
gcal_event['end'] = {
'dateTime': end_dt.isoformat(),
'timeZone': 'Asia/Kolkata'
}
gcal_events.append(gcal_event)
return gcal_events

26
week2/community-contributions/medical_prescription_to_google_calender/src/main.py

@ -0,0 +1,26 @@
from ocr import *
from calendar_auth import *
from create_calender_events import *
from parsing_json import *
from preprocess import *
image_path = r"C:\Users\Legion\Desktop\projects\medical_prescription_to_google_calender\test_data\prescription_page-0001.jpg"
extracted_text = extract_text_from_image(image_path=image_path)
print(extracted_text)
cleaned_text = clean_text(extracted_text)
print(cleaned_text)
structured_data = preprocess_extracted_text(cleaned_text)
print(structured_data)
final_structured_data = process_dates(structured_data)
print(final_structured_data)
formatted_calender_events = format_calendar_events(final_structured_data)
print(formatted_calender_events)
validated_events = [validate_event(event) for event in formatted_calender_events]
for event in validated_events[:5]:
print(json.dumps(event, indent=2))
service = authenticate_google_calender()
gcal_events = convert_to_gcal_events(validated_events)
for event in gcal_events:
create_event(service, event)

71
week2/community-contributions/medical_prescription_to_google_calender/src/ocr.py

@ -0,0 +1,71 @@
import os
from openai import OpenAI
from dotenv import load_dotenv
import base64
from PIL import Image
import re
load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")
MODEL = "gpt-4o"
openai = OpenAI()
def encode_image(image_path):
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode("utf-8")
def extract_text_from_image(image_path):
response = openai.chat.completions.create(
model = MODEL,
max_tokens = 1000,
messages=[
{
"role": "system", "content": """You are an OCR assistant that extracts text from medical
prescription images. Extract all the text exactly as it
appears in the prescription image. Dont include images. Only
extract text."""
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "Extract text from this image: "
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{encode_image(image_path)}"
}
}
]
}
]
)
return response.choices[0].message.content
import re
def clean_text(text):
# Remove all hyphens
text = re.sub(r'-', ' ', text)
# Remove excessive non-word characters but keep necessary punctuation
text = re.sub(r'[^\w\s.,()%/]', '', text)
# Remove multiple spaces and ensure single spaces
text = re.sub(r'\s+', ' ', text)
# Replace multiple newlines with a single newline
text = re.sub(r'\n+', '\n', text)
# Ensure spacing around punctuation marks
text = re.sub(r'([.,])([^\s])', r'\1 \2', text)
return text.strip()

120
week2/community-contributions/medical_prescription_to_google_calender/src/parsing_json.py

@ -0,0 +1,120 @@
import json
import re
from datetime import datetime, timedelta
# Default number of days to schedule indefinitely recurring events (1 year)
DEFAULT_DURATION_DAYS = 365
# Function to assign a default time for general terms like "before breakfast", etc.
def assign_time(timing):
time_mappings = {
"random": "09:00 AM",
"daily": "09:00 AM",
"before breakfast": "07:00 AM",
"after breakfast": "08:30 AM",
"before lunch": "12:00 PM",
"after lunch": "01:30 PM",
"before dinner": "07:00 PM",
"after dinner": "08:30 PM",
}
return time_mappings.get(timing.lower(), timing)
# Function to extract the recurrence pattern
def get_recurrence_interval(timing):
""" Extracts interval days from 'every X days', 'once a week', or 'once a month'. """
timing = timing.lower().strip()
if "every alternate day" in timing:
return 2 # Every other day (every 2 days)
elif match := re.search(r"every (\d+) days", timing):
return int(match.group(1)) # Extract number of days
elif "once a week" in timing:
return 7 # Every 7 days (once a week)
elif "once a month" in timing:
return "monthly" # Special case for monthly scheduling
elif timing in ["daily", "every day"]:
return 1 # Every day
else:
return None # Not a recurring event
# Function to convert AM/PM time format to 24-hour format
def convert_to_24hr(time_str):
return datetime.strptime(time_str, "%I:%M %p").strftime("%H:%M")
# Function to generate Google Calendar events
def format_calendar_events(processed_data):
events = []
start_date = datetime.today().date()
# Medicines
if "medicines" in processed_data:
for med in processed_data["medicines"]:
if med.get("name"):
event_time = assign_time(med.get("timing", "09:00 AM"))
interval_days = get_recurrence_interval(med["timing"])
# If no interval, assume daily (default behavior)
if interval_days is None:
interval_days = 1
# Generate events for 1 year if no duration is given
event_date = start_date
for _ in range(365 if interval_days != "monthly" else 12):
if interval_days == "monthly":
event_date = (event_date.replace(day=1) + timedelta(days=32)).replace(day=1) # Jump to the next month
else:
event_date += timedelta(days=interval_days)
event = {
"summary": f"Take {med['name']} ({med.get('dosage', 'Dosage not specified')})",
"start": {
"dateTime": f"{event_date.isoformat()}T{convert_to_24hr(event_time)}:00",
"timeZone": "Asia/Kolkata"
},
"end": {
"dateTime": f"{event_date.isoformat()}T{convert_to_24hr(event_time)}:59",
"timeZone": "Asia/Kolkata"
}
}
events.append(event)
# Tests
if "tests" in processed_data:
for test in processed_data["tests"]:
if test.get("name") and test.get("dueDate"): # Use 'dueDate' instead of 'date'
event = {
"summary": f"Medical Test: {test['name']}",
"start": {"date": test["dueDate"]}, # Fix here
"end": {"date": test["dueDate"]}, # Fix here
"timeZone": "Asia/Kolkata"
}
events.append(event)
# Follow-ups
if "follow_ups" in processed_data:
for follow_up in processed_data["follow_ups"]:
if follow_up.get("date"):
event = {
"summary": "Doctor Follow-up Appointment",
"start": {"date": follow_up["date"]},
"end": {"date": follow_up["date"]},
"timeZone": "Asia/Kolkata"
}
events.append(event)
return events
# Function to validate events before sending to Google Calendar
def validate_event(event):
required_fields = {
"summary": "Untitled Event",
"start": {"dateTime": datetime.today().isoformat(), "timeZone": "Asia/Kolkata"},
"end": {"dateTime": (datetime.today() + timedelta(minutes=30)).isoformat(), "timeZone": "Asia/Kolkata"}
}
for field, default_value in required_fields.items():
if field not in event or event[field] is None:
event[field] = default_value
return event

141
week2/community-contributions/medical_prescription_to_google_calender/src/preprocess.py

@ -0,0 +1,141 @@
import os
from openai import OpenAI
from dotenv import load_dotenv
import json
from datetime import datetime, timedelta
load_dotenv()
openai_api = os.getenv("OPENAI_API_KEY")
MODEL = "gpt-4o-mini"
openai = OpenAI()
system_prompt = """You are a medical assistant that processes prescription text.
Your goal is to extract medicines, tests, and follow-ups in a structured JSON format.
### **Instructions:**
- Extract **medicines**, **dosages**, and **timings** if available.
- **Convert vague timings** into precise values:
- **Before breakfast** `07:30 AM`
- **After lunch** `02:00 PM`
- **Before dinner** `07:00 PM`
- **After dinner** `10:00 PM`
- **30 minutes before breakfast** `07:00 AM`
- If **"daily"** is mentioned without a time, **assign a logical time** between **08:00 AM - 10:00 PM**.
- If the prescription says **"every alternate day"**, return `"interval": 2` instead of just `"daily"`.
### **Tests & Follow-ups:**
- Extract **medical tests** and their required dates.
- Convert relative times (e.g., `"after 3 months"`) into **exact calendar dates**, using the prescription date.
- If the prescription date is missing, use today's date.
- Follow-up should **only be included if required**, not just for general check-ups.
### **Output Format:**
Return **only valid JSON**, structured as follows:
{
"medicines": [
{
"name": "<Medicine Name>",
"dosage": "<Dosage>",
"timing": "<Time>",
"interval": <Interval in days (if applicable)>
}
],
"tests": [
{
"name": "<Test Name>",
"date": "<YYYY-MM-DD>"
}
],
"follow_ups": [
{
"date": "<YYYY-MM-DD>"
}
]
}
"""
def clean_json_string(json_str):
"""Clean and validate JSON string before parsing."""
try:
start = json_str.find('{')
end = json_str.rfind('}') + 1
if start >= 0 and end > 0:
json_str = json_str[start:end]
# Remove any extra whitespace
json_str = json_str.strip()
# Attempt to parse the JSON
return json.loads(json_str)
except json.JSONDecodeError as e:
print(f"Failed to parse JSON. Raw response:\n{json_str}")
print(f"Error: {str(e)}")
return None
def preprocess_extracted_text(extracted_text):
"""Calls GPT-4o-mini to process prescription text into structured JSON."""
try:
response = openai.chat.completions.create(
model=MODEL,
messages=[
{
"role": "system",
"content": system_prompt,
},
{
"role": "user",
"content": f"Process this prescription and return ONLY valid JSON:\n\n{extracted_text}"
}
],
temperature=0.3 # Lower temperature for more consistent JSON output
)
# Get the response content
content = response.choices[0].message.content
# Clean and parse the JSON
parsed_data = clean_json_string(content)
if parsed_data is None:
return {
"medicines": [],
"tests": [],
"follow_ups": []
}
return parsed_data
except Exception as e:
print(f"Error in API call or processing: {str(e)}")
return {
"medicines": [],
"tests": [],
"follow_ups": []
}
def process_dates(data):
"""Adjusts test dates and follow-up based on the prescription date or today's date."""
try:
# Extract prescription date (if available) or use today's date
prescription_date = datetime.strptime("02 JANUARY 2025", "%d %B %Y").date()
# Process test dates
for test in data.get("tests", []):
if isinstance(test, dict) and "date" not in test and "after_months" in test:
test_date = prescription_date + timedelta(days=test["after_months"] * 30)
test["date"] = test_date.strftime("%Y-%m-%d")
# Process follow-up dates
follow_ups = data.get("follow_ups", [])
for follow_up in follow_ups:
if isinstance(follow_up, dict) and "date" not in follow_up and "after_months" in follow_up:
follow_up_date = prescription_date + timedelta(days=follow_up["after_months"] * 30)
follow_up["date"] = follow_up_date.strftime("%Y-%m-%d")
return data
except Exception as e:
print(f"Error processing dates: {str(e)}")
return data
Loading…
Cancel
Save