Neemah's picture
Update app.py
ab0bf12 verified
Raw
History Blame Contribute Delete
41.6 kB
import os
import time
import csv
import hashlib
from datetime import datetime
import numpy as np
import pydicom
import torch
import gradio as gr
from PIL import Image
from transformers import pipeline
from huggingface_hub import HfApi, hf_hub_download
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
from reportlab.lib.units import inch
MODEL_ID = "google/medgemma-1.5-4b-it"
PROMPT = (
"You are a senior consultant radiologist. Write a detailed, structured chest X-ray report.\n\n"
"Use this exact format:\n"
"1) Technique: (projection PA/AP, penetration; if uncertain, say so)\n"
"2) Findings:\n"
" - Airways\n"
" - Lungs & pleura (consolidation, atelectasis, effusion, pneumothorax, interstitial markings, and any other relevant findings)\n"
" - Cardiomediastinal silhouette (heart size, mediastinum, hila)\n"
" - Diaphragm\n"
" - Bones & soft tissues\n"
" - Devices/lines (if any)\n"
"3) Impression: (bullet points, most likely diagnosis + key differentials)\n"
"4) Urgent alerts: (state 'None' if none)\n\n"
"Be specific about location and limitations. Do not invent clinical history. "
"If something is not clearly visible, say 'cannot be confidently assessed'."
)
RATING_MAP = {
"1 - Very Unlikely": 1,
"2 - Unlikely": 2,
"3 - Neutral": 3,
"4 - Likely": 4,
"5 - Very Likely": 5,
}
LOCAL_TMP_DIR = "/tmp"
REVIEWS_CSV_NAME = "radiologist_reviews.csv"
LOCAL_REVIEWS_CSV = os.path.join(LOCAL_TMP_DIR, REVIEWS_CSV_NAME)
# Environment variables
hf_token = os.environ.get("HF_TOKEN")
if not hf_token:
raise RuntimeError("HF_TOKEN is missing. Add it in your Space secrets.")
dataset_repo = os.environ.get("HF_DATASET_REPO")
if not dataset_repo:
raise RuntimeError("HF_DATASET_REPO is missing. Add it in your Space variables.")
# HF API client
api = HfApi(token=hf_token)
# Load model once
use_cuda = torch.cuda.is_available()
dtype = torch.bfloat16 if use_cuda else torch.float32
pipe = pipeline(
"image-text-to-text",
model=MODEL_ID,
dtype=dtype,
device=0 if use_cuda else -1,
token=hf_token,
)
def generate_report(img: Image.Image):
if img is None:
return "Please upload a chest X-ray image.", None
img = img.convert("L").convert("RGB")
messages = [{
"role": "user",
"content": [
{"type": "image", "image": img},
{"type": "text", "text": PROMPT},
],
}]
out = pipe(
text=messages,
max_new_tokens=500,
do_sample=False,
temperature=0.0,
top_p=1.0,
)
report = out[0]["generated_text"][-1]["content"]
return report, None
def text_to_pdf(report_text: str):
if not report_text or not report_text.strip():
return None
filename = f"radiology_report_{int(time.time())}.pdf"
path = os.path.join("/tmp", filename)
c = canvas.Canvas(path, pagesize=letter)
width, height = letter
left = 0.75 * inch
top = height - 0.75 * inch
line_height = 14
max_width = width - 2 * left
c.setFont("Helvetica-Bold", 14)
c.drawString(left, top, "Chest X-ray Report")
c.setFont("Helvetica", 11)
y = top - 0.45 * inch
for paragraph in report_text.split("\n"):
words = paragraph.split(" ")
line = ""
for w in words:
test = (line + " " + w).strip()
if c.stringWidth(test, "Helvetica", 11) <= max_width:
line = test
else:
if y < 0.75 * inch:
c.showPage()
c.setFont("Helvetica", 11)
y = height - 0.75 * inch
if line:
c.drawString(left, y, line)
y -= line_height
line = w
if y < 0.75 * inch:
c.showPage()
c.setFont("Helvetica", 11)
y = height - 0.75 * inch
if line:
c.drawString(left, y, line)
y -= line_height
c.save()
return path
def dicom_to_png(dicom_file):
if dicom_file is None:
return None, "Please upload a DICOM file."
try:
ds = pydicom.dcmread(dicom_file.name)
pixel_array = ds.pixel_array.astype(float)
# Normalize to 0-255
pixel_min = pixel_array.min()
pixel_max = pixel_array.max()
if pixel_max - pixel_min == 0:
return None, "Image has no contrast — cannot convert."
normalized = (pixel_array - pixel_min) / (pixel_max - pixel_min) * 255.0
img_uint8 = normalized.astype(np.uint8)
# Handle grayscale vs RGB
if img_uint8.ndim == 2:
img = Image.fromarray(img_uint8, mode="L").convert("RGB")
elif img_uint8.ndim == 3:
img = Image.fromarray(img_uint8)
else:
return None, "Unsupported pixel array dimensions."
# Save as PNG
out_path = os.path.join(LOCAL_TMP_DIR, f"converted_{int(time.time())}.png")
img.save(out_path, format="PNG")
return out_path, "Conversion successful. Download your PNG below."
except Exception as e:
return None, f"Conversion failed: {str(e)}"
def load_image_from_upload(upload) -> tuple[Image.Image | None, str]:
"""
Accepts either a PIL Image (from gr.Image) or a file object (from gr.File).
If DICOM, converts internally. Returns (PIL Image, error_message).
"""
if upload is None:
return None, "Please upload a chest X-ray image."
# If already a PIL Image (standard image upload)
if isinstance(upload, Image.Image):
return upload, ""
# Otherwise it's a file object — check extension
filepath = upload.name if hasattr(upload, "name") else str(upload)
ext = os.path.splitext(filepath)[-1].lower()
if ext == ".dcm":
try:
ds = pydicom.dcmread(filepath)
pixel_array = ds.pixel_array.astype(float)
pixel_min, pixel_max = pixel_array.min(), pixel_array.max()
if pixel_max - pixel_min == 0:
return None, "DICOM image has no contrast — cannot convert."
normalized = (pixel_array - pixel_min) / (pixel_max - pixel_min) * 255.0
img_uint8 = normalized.astype(np.uint8)
if img_uint8.ndim == 2:
img = Image.fromarray(img_uint8, mode="L").convert("RGB")
elif img_uint8.ndim == 3:
img = Image.fromarray(img_uint8)
else:
return None, "Unsupported DICOM pixel format."
return img, ""
except Exception as e:
return None, f"DICOM conversion failed: {str(e)}"
else:
# Regular image file (.png, .jpg etc.)
try:
img = Image.open(filepath).convert("RGB")
return img, ""
except Exception as e:
return None, f"Could not open image: {str(e)}"
def download_existing_csv() -> str:
os.makedirs(LOCAL_TMP_DIR, exist_ok=True)
try:
csv_path = hf_hub_download(
repo_id=dataset_repo,
repo_type="dataset",
filename=REVIEWS_CSV_NAME,
token=hf_token,
local_dir=LOCAL_TMP_DIR,
local_dir_use_symlinks=False,
)
return csv_path
except Exception:
with open(LOCAL_REVIEWS_CSV, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
"Review_ID",
"Timestamp_utc",
"Report_hash",
"Technique_label", "Technique_value",
"Findings_label", "Findings_value",
"Impression_label","Impression_value",
"Image_filename",
])
return LOCAL_REVIEWS_CSV
def upload_csv_to_dataset(csv_path: str):
api.upload_file(
path_or_fileobj=csv_path,
path_in_repo=REVIEWS_CSV_NAME,
repo_id=dataset_repo,
repo_type="dataset",
commit_message="Update radiologist review ratings CSV",
)
def get_next_review_id(csv_path: str) -> int:
"""Count existing data rows and return next ID."""
with open(csv_path, "r", encoding="utf-8") as f:
reader = csv.reader(f)
rows = list(reader)
# rows[0] is header, rest are data
return len(rows) # len includes header, so len=1 means 0 reviews → next ID is 1
def get_report_hash(report_text: str) -> str:
"""MD5 hash of the report text to detect duplicates."""
return hashlib.md5(report_text.strip().encode("utf-8")).hexdigest()
def is_duplicate_review(csv_path: str, report_hash: str) -> bool:
"""Check if this report hash already exists in the CSV."""
with open(csv_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
if row.get("Report_hash") == report_hash:
return True
return False
def upload_image_to_dataset(img: Image.Image, review_id: int) -> str:
"""Save image locally and upload to dataset repo. Returns filename."""
filename = f"review_{review_id}.jpg"
local_path = os.path.join(LOCAL_TMP_DIR, filename)
img.convert("RGB").save(local_path, format="JPEG")
api.upload_file(
path_or_fileobj=local_path,
path_in_repo=f"images/{filename}",
repo_id=dataset_repo,
repo_type="dataset",
commit_message=f"Upload image for review {review_id}",
)
return filename
def save_review(report_text: str, technique_label: str, findings_label: str, impression_label: str, img: Image.Image = None):
if not report_text or not report_text.strip():
return "No report available to review."
if not technique_label or not findings_label or not impression_label:
return "Please select a rating for all three sections before saving."
try:
csv_path = download_existing_csv()
# Duplicate check
report_hash = get_report_hash(report_text)
if is_duplicate_review(csv_path, report_hash):
return "This report has already been reviewed. Duplicate not saved."
# Get next ID
review_id = get_next_review_id(csv_path)
# Upload image if provided
image_filename = "N/A"
if img is not None:
image_filename = upload_image_to_dataset(img, review_id)
with open(csv_path, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
review_id,
datetime.utcnow().isoformat(),
report_hash,
technique_label, RATING_MAP[technique_label],
findings_label, RATING_MAP[findings_label],
impression_label, RATING_MAP[impression_label],
image_filename,
])
upload_csv_to_dataset(csv_path)
return f"Review #{review_id} saved successfully to dataset repo: {dataset_repo}"
except Exception as e:
return f"Failed to save review: {str(e)}"
# DARK_CSS = """
# body, .gradio-container {
# background: #393e46 !important;
# color: white !important;
# }
# /* Fix width nd center the app */
# .gradio-container {
# max-width: 900px !important;
# margin-left: auto !important;
# margin-right: auto !important;
# width: 100% !important;
# }
# /* Markdown text */
# .gr-markdown, .gr-markdown h1, .gr-markdown h2, .gr-markdown p, h1, p, h2, h3, .gr-markdown h3 {
# color: white !important;
# }
# /* Panels / component containers */
# .gr-panel, .gr-box, .gr-form {
# background: #393e46 !important;
# border-color: #2a2a2a !important;
# }
# /* Textbox */
# textarea, .wrap textarea {
# background: #393e46 !important;
# color: white !important;
# border: 1px solid #2a2a2a !important;
# }
# /* Labels */
# label, .gr-label, .wrap label {
# color: white !important;
# }
# /* Buttons */
# .gr-button {
# background: white !important;
# color: black !important;
# border: none !important;
# }
# button {
# background: #393e46 !important;
# color: white !important;
# border: 2px solid #e70000 !important;
# }
# .gr-button:hover {
# background: #393e46 !important;
# }
# button:hover {
# background: #e70000 !important;
# color: black !important;
# }
# /* File component area */
# input[type="file"] {
# color: white !important;
# }
# .gradio-file-upload svg {
# fill: #e70000 !important;
# stroke: #e70000 !important;
# }
# """
DARK_CSS = """
/* ── Base ── */
body, .gradio-container {
background: #0a1628 !important;
color: #e8edf5 !important;
font-family: 'Segoe UI', system-ui, -apple-system, sans-serif !important;
}
/* Centre & constrain */
.gradio-container {
max-width: 920px !important;
margin-left: auto !important;
margin-right: auto !important;
width: 100% !important;
}
/* ── Headings & body text ── */
.gr-markdown,
.gr-markdown h1, .gr-markdown h2, .gr-markdown h3,
.gr-markdown p, h1, h2, h3, p {
color: #e8edf5 !important;
}
h1 {
font-size: 1.75rem !important;
font-weight: 700 !important;
letter-spacing: -0.3px !important;
color: #ffffff !important;
}
/* ── Labels ── */
label, .gr-label, .wrap label {
color: #93aac9 !important;
font-size: 0.8rem !important;
font-weight: 600 !important;
letter-spacing: 0.06em !important;
text-transform: uppercase !important;
}
/* ── Panels / containers ── */
.gr-panel, .gr-box, .gr-form {
background: #112040 !important;
border: 1px solid #1e3560 !important;
border-radius: 12px !important;
}
/* ── Textbox — report text stays near-black on white ── */
textarea, .wrap textarea {
background: #ffffff !important;
color: #111827 !important;
border: 1px solid #1e3560 !important;
border-radius: 10px !important;
font-size: 0.9rem !important;
line-height: 1.7 !important;
}
textarea:focus, .wrap textarea:focus {
border-color: #4a90d9 !important;
box-shadow: 0 0 0 3px rgba(74, 144, 217, 0.2) !important;
outline: none !important;
}
/* ── Buttons — base ── */
button {
background: transparent !important;
color: #e8edf5 !important;
border: 1.5px solid #2e5080 !important;
border-radius: 8px !important;
font-weight: 600 !important;
font-size: 0.875rem !important;
letter-spacing: 0.02em !important;
padding: 10px 20px !important;
transition: all 0.18s ease !important;
}
/* ── Primary action (Generate) — accent fill ── */
button.primary, .gr-button-primary, #component-gen-btn {
background: #1a6de0 !important;
color: #ffffff !important;
border-color: #1a6de0 !important;
box-shadow: 0 2px 12px rgba(26, 109, 224, 0.35) !important;
}
button.primary:hover:not(:disabled),
.gr-button-primary:hover:not(:disabled),
#component-gen-btn:hover:not(:disabled) {
background: #1558c0 !important;
border-color: #1558c0 !important;
box-shadow: 0 4px 18px rgba(26, 109, 224, 0.5) !important;
transform: translateY(-1px) !important;
}
/* ── All other button hovers ── */
button:hover:not(:disabled) {
background: #1e3560 !important;
border-color: #4a90d9 !important;
color: #ffffff !important;
transform: translateY(-1px) !important;
box-shadow: 0 3px 10px rgba(0,0,0,0.25) !important;
}
button:active:not(:disabled) {
transform: translateY(0) !important;
box-shadow: none !important;
}
/* ── Disabled state ── */
button:disabled {
opacity: 0.3 !important;
cursor: not-allowed !important;
transform: none !important;
box-shadow: none !important;
}
/* ── Radio buttons ── */
.gr-radio, .wrap .gr-radio {
color: #e8edf5 !important;
}
input[type="radio"]:checked + span {
color: #4a90d9 !important;
font-weight: 600 !important;
}
/* ── File upload zone ── */
input[type="file"] {
color: #93aac9 !important;
}
.gradio-file-upload, [data-testid="file-drop-area"] {
background: #0d1f3c !important;
border: 1.5px dashed #2e5080 !important;
border-radius: 12px !important;
transition: border-color 0.2s, background 0.2s !important;
}
.gradio-file-upload:hover, [data-testid="file-drop-area"]:hover {
border-color: #4a90d9 !important;
background: #112040 !important;
}
.gradio-file-upload svg {
fill: #4a90d9 !important;
stroke: #4a90d9 !important;
}
/* ── Image preview ── */
.gr-image, [data-testid="image"] {
border: 1px solid #1e3560 !important;
border-radius: 12px !important;
overflow: hidden !important;
}
/* ── Scrollbar ── */
::-webkit-scrollbar { width: 5px; height: 5px; }
::-webkit-scrollbar-t
"""
with gr.Blocks() as demo:
gr.Markdown("# Chest X-ray Auto Report Generator\n Radiologist Assistant")
image_in = gr.File(
label="Upload chest X-ray (PNG / JPG / DICOM .dcm)",
file_types=[".png", ".jpg", ".jpeg", ".dcm"]
)
image_preview = gr.Image(
label="Image Preview",
type="pil",
interactive=False,
height=400,
visible = False,
)
report_loading = gr.HTML(
"""
<div style="
height: 420px;
border: 1px solid #d1d5db;
border-radius: 8px;
background: black;
display: flex;
align-items: center;
justify-content: center;
flex-direction: column;
gap: 10px;
padding: 16px;
">
<img src="https://media.giphy.com/media/v1.Y2lkPTc5MGI3NjExYW5pdDh0ZHdoNXplaTlqdno1OWNrbzc3cDcxaTRkc3FoaTN1anF2aiZlcD12MV9naWZzX3NlYXJjaCZjdD1n/dt0KXLj7bzwZuRQBwY/giphy.gif"
style="width:200px; height:auto;" />
<div style="font-size:25px; font-weight:800; color:white;">Generating...</div>
</div>
""",
visible=False,
)
report_out = gr.Textbox(
label="Generated radiology report",
lines=18,
interactive=False,
visible=True,
)
review_done = gr.State(False)
review_section = gr.Column(visible=False)
LIKERT_CHOICES = [
"1 - Very Unlikely",
"2 - Unlikely",
"3 - Neutral",
"4 - Likely",
"5 - Very Likely",
]
with review_section:
gr.Markdown("Radiologist Review")
gr.Markdown("Rate each section of the generated report:")
technique_radio = gr.Radio(
choices=LIKERT_CHOICES,
label="1. Technique — accuracy & completeness",
value=None
)
findings_radio = gr.Radio(
choices=LIKERT_CHOICES,
label="2. Findings — accuracy & completeness",
value=None
)
impression_radio = gr.Radio(
choices=LIKERT_CHOICES,
label="3. Impression — accuracy & completeness",
value=None
)
save_review_btn = gr.Button("Save Review")
review_status = gr.Textbox(label="Review Status", interactive=False)
def preview_only(upload):
img, error = load_image_from_upload(upload)
if img is None:
return gr.update(value=None, visible=False), gr.update(interactive=False)
return gr.update(value=img, visible=True), gr.update(interactive=True)
def clear_all():
return (
gr.update(value=None, visible=False), # image_preview
gr.update(value="", visible=True), # report_out
gr.update(visible=False), # loading
gr.update(visible=False), # review section
gr.update(value=None), # technique
gr.update(value=None), # findings
gr.update(value=None), # impression
gr.update(value=""), # status
False,
)
gen_btn = gr.Button("Generate report", interactive=False)
pdf_btn = gr.DownloadButton("Download as PDF", value=None, visible=True)
def generate_report_ui(upload):
img, error = load_image_from_upload(upload)
if img is None:
yield (
gr.update(value=None),
gr.update(value="error", interactive=False, visible=True),
gr.update(visible=False),
gr.update(visible=False),
gr.update(value=None),
gr.update(value=None),
gr.update(value=None),
gr.update(value=""),
False,
)
return
# Show loader
yield (
gr.update(value=img, visible=True),
gr.update(value="", interactive=False, visible=False),
gr.update(visible=True),
gr.update(visible=False),
gr.update(value=None),
gr.update(value=None),
gr.update(value=None),
gr.update(value=""),
False,
)
report, _ = generate_report(img)
# Show report + review section
yield (
gr.update(value=img, visible=True),
gr.update(value=report, interactive=True, visible=True),
gr.update(visible=False),
gr.update(visible=True),
gr.update(value=None),
gr.update(value=None),
gr.update(value=None),
gr.update(value=""),
False,
)
gen_btn.click(
fn=generate_report_ui,
inputs=image_in,
outputs=[
image_preview,
report_out,
report_loading,
review_section,
technique_radio,
findings_radio,
impression_radio,
review_status,
review_done,
],
)
def guarded_pdf_download(report_text, review_done_flag):
if not review_done_flag:
gr.Warning("Please submit a review before downloading the report.")
return None
return text_to_pdf(report_text)
pdf_btn.click(
fn=guarded_pdf_download,
inputs=[report_out, review_done],
outputs=pdf_btn,
)
def save_review_ui(report_text, technique, findings, impression, upload):
img, _ = load_image_from_upload(upload)
message = save_review(report_text, technique, findings, impression, img)
if "successfully" in message.lower():
return message, True
else:
return message, False
save_review_btn.click(
fn=save_review_ui,
inputs=[report_out, technique_radio, findings_radio, impression_radio, image_in],
outputs=[review_status, review_done],
)
image_in.change(
fn=preview_only,
inputs=image_in,
outputs=[image_preview, gen_btn],
)
#Reset everything when image is cleared
image_in.clear(
fn=clear_all,
outputs=[
image_preview,
report_out,
report_loading,
review_section,
technique_radio,
findings_radio,
impression_radio,
review_status,
review_done,
]
)
demo.launch(
theme=gr.themes.Monochrome(),
css=DARK_CSS,
)
#Normal Project
# import os
# import time
# import csv
# import hashlib
# from datetime import datetime
# import numpy as np
# import pydicom
# import torch
# import gradio as gr
# from PIL import Image
# from transformers import pipeline
# from huggingface_hub import HfApi, hf_hub_download
# from reportlab.pdfgen import canvas
# from reportlab.lib.pagesizes import letter
# from reportlab.lib.units import inch
# MODEL_ID = "google/medgemma-1.5-4b-it"
# PROMPT = (
# "You are a senior consultant radiologist. Write a detailed, structured chest X-ray report.\n\n"
# "Use this exact format:\n"
# "1) Technique: (projection PA/AP, penetration; if uncertain, say so)\n"
# "2) Findings:\n"
# " - Airways\n"
# " - Lungs & pleura (consolidation, atelectasis, effusion, pneumothorax, interstitial markings, and any other relevant findings)\n"
# " - Cardiomediastinal silhouette (heart size, mediastinum, hila)\n"
# " - Diaphragm\n"
# " - Bones & soft tissues\n"
# " - Devices/lines (if any)\n"
# "3) Impression: (bullet points, most likely diagnosis + key differentials)\n"
# "4) Urgent alerts: (state 'None' if none)\n\n"
# "Be specific about location and limitations. Do not invent clinical history. "
# "If something is not clearly visible, say 'cannot be confidently assessed'."
# )
# RATING_MAP = {
# "1 - Very Unlikely": 1,
# "2 - Unlikely": 2,
# "3 - Neutral": 3,
# "4 - Likely": 4,
# "5 - Very Likely": 5,
# }
# LOCAL_TMP_DIR = "/tmp"
# REVIEWS_CSV_NAME = "radiologist_reviews.csv"
# LOCAL_REVIEWS_CSV = os.path.join(LOCAL_TMP_DIR, REVIEWS_CSV_NAME)
# # Environment variables
# hf_token = os.environ.get("HF_TOKEN")
# if not hf_token:
# raise RuntimeError("HF_TOKEN is missing. Add it in your Space secrets.")
# dataset_repo = os.environ.get("HF_DATASET_REPO")
# if not dataset_repo:
# raise RuntimeError("HF_DATASET_REPO is missing. Add it in your Space variables.")
# # HF API client
# api = HfApi(token=hf_token)
# # Load model once
# use_cuda = torch.cuda.is_available()
# dtype = torch.bfloat16 if use_cuda else torch.float32
# pipe = pipeline(
# "image-text-to-text",
# model=MODEL_ID,
# dtype=dtype,
# device=0 if use_cuda else -1,
# token=hf_token,
# )
# def generate_report(img: Image.Image):
# if img is None:
# return "Please upload a chest X-ray image.", None
# img = img.convert("L").convert("RGB")
# messages = [{
# "role": "user",
# "content": [
# {"type": "image", "image": img},
# {"type": "text", "text": PROMPT},
# ],
# }]
# out = pipe(
# text=messages,
# max_new_tokens=500,
# do_sample=False,
# temperature=0.0,
# top_p=1.0,
# )
# report = out[0]["generated_text"][-1]["content"]
# return report, None
# def text_to_pdf(report_text: str):
# if not report_text or not report_text.strip():
# return None
# filename = f"radiology_report_{int(time.time())}.pdf"
# path = os.path.join("/tmp", filename)
# c = canvas.Canvas(path, pagesize=letter)
# width, height = letter
# left = 0.75 * inch
# top = height - 0.75 * inch
# line_height = 14
# max_width = width - 2 * left
# c.setFont("Helvetica-Bold", 14)
# c.drawString(left, top, "Chest X-ray Report")
# c.setFont("Helvetica", 11)
# y = top - 0.45 * inch
# for paragraph in report_text.split("\n"):
# words = paragraph.split(" ")
# line = ""
# for w in words:
# test = (line + " " + w).strip()
# if c.stringWidth(test, "Helvetica", 11) <= max_width:
# line = test
# else:
# if y < 0.75 * inch:
# c.showPage()
# c.setFont("Helvetica", 11)
# y = height - 0.75 * inch
# if line:
# c.drawString(left, y, line)
# y -= line_height
# line = w
# if y < 0.75 * inch:
# c.showPage()
# c.setFont("Helvetica", 11)
# y = height - 0.75 * inch
# if line:
# c.drawString(left, y, line)
# y -= line_height
# c.save()
# return path
# def dicom_to_png(dicom_file):
# if dicom_file is None:
# return None, "Please upload a DICOM file."
# try:
# ds = pydicom.dcmread(dicom_file.name)
# pixel_array = ds.pixel_array.astype(float)
# # Normalize to 0-255
# pixel_min = pixel_array.min()
# pixel_max = pixel_array.max()
# if pixel_max - pixel_min == 0:
# return None, "Image has no contrast — cannot convert."
# normalized = (pixel_array - pixel_min) / (pixel_max - pixel_min) * 255.0
# img_uint8 = normalized.astype(np.uint8)
# # Handle grayscale vs RGB
# if img_uint8.ndim == 2:
# img = Image.fromarray(img_uint8, mode="L").convert("RGB")
# elif img_uint8.ndim == 3:
# img = Image.fromarray(img_uint8)
# else:
# return None, "Unsupported pixel array dimensions."
# # Save as PNG
# out_path = os.path.join(LOCAL_TMP_DIR, f"converted_{int(time.time())}.png")
# img.save(out_path, format="PNG")
# return out_path, "Conversion successful. Download your PNG below."
# except Exception as e:
# return None, f"Conversion failed: {str(e)}"
# def load_image_from_upload(upload) -> tuple[Image.Image | None, str]:
# """
# Accepts either a PIL Image (from gr.Image) or a file object (from gr.File).
# If DICOM, converts internally. Returns (PIL Image, error_message).
# """
# if upload is None:
# return None, "Please upload a chest X-ray image."
# # If already a PIL Image (standard image upload)
# if isinstance(upload, Image.Image):
# return upload, ""
# # Otherwise it's a file object — check extension
# filepath = upload.name if hasattr(upload, "name") else str(upload)
# ext = os.path.splitext(filepath)[-1].lower()
# if ext == ".dcm":
# try:
# ds = pydicom.dcmread(filepath)
# pixel_array = ds.pixel_array.astype(float)
# pixel_min, pixel_max = pixel_array.min(), pixel_array.max()
# if pixel_max - pixel_min == 0:
# return None, "DICOM image has no contrast — cannot convert."
# normalized = (pixel_array - pixel_min) / (pixel_max - pixel_min) * 255.0
# img_uint8 = normalized.astype(np.uint8)
# if img_uint8.ndim == 2:
# img = Image.fromarray(img_uint8, mode="L").convert("RGB")
# elif img_uint8.ndim == 3:
# img = Image.fromarray(img_uint8)
# else:
# return None, "Unsupported DICOM pixel format."
# return img, ""
# except Exception as e:
# return None, f"DICOM conversion failed: {str(e)}"
# else:
# # Regular image file (.png, .jpg etc.)
# try:
# img = Image.open(filepath).convert("RGB")
# return img, ""
# except Exception as e:
# return None, f"Could not open image: {str(e)}"
# def download_existing_csv() -> str:
# os.makedirs(LOCAL_TMP_DIR, exist_ok=True)
# try:
# csv_path = hf_hub_download(
# repo_id=dataset_repo,
# repo_type="dataset",
# filename=REVIEWS_CSV_NAME,
# token=hf_token,
# local_dir=LOCAL_TMP_DIR,
# local_dir_use_symlinks=False,
# )
# return csv_path
# except Exception:
# with open(LOCAL_REVIEWS_CSV, "w", newline="", encoding="utf-8") as f:
# writer = csv.writer(f)
# writer.writerow([
# "Review_ID",
# "Timestamp_utc",
# "Report_hash",
# "Technique_label", "Technique_value",
# "Findings_label", "Findings_value",
# "Impression_label","Impression_value",
# "Image_filename",
# ])
# return LOCAL_REVIEWS_CSV
# def upload_csv_to_dataset(csv_path: str):
# api.upload_file(
# path_or_fileobj=csv_path,
# path_in_repo=REVIEWS_CSV_NAME,
# repo_id=dataset_repo,
# repo_type="dataset",
# commit_message="Update radiologist review ratings CSV",
# )
# def get_next_review_id(csv_path: str) -> int:
# """Count existing data rows and return next ID."""
# with open(csv_path, "r", encoding="utf-8") as f:
# reader = csv.reader(f)
# rows = list(reader)
# # rows[0] is header, rest are data
# return len(rows) # len includes header, so len=1 means 0 reviews → next ID is 1
# def get_report_hash(report_text: str) -> str:
# """MD5 hash of the report text to detect duplicates."""
# return hashlib.md5(report_text.strip().encode("utf-8")).hexdigest()
# def is_duplicate_review(csv_path: str, report_hash: str) -> bool:
# """Check if this report hash already exists in the CSV."""
# with open(csv_path, "r", encoding="utf-8") as f:
# reader = csv.DictReader(f)
# for row in reader:
# if row.get("Report_hash") == report_hash:
# return True
# return False
# def upload_image_to_dataset(img: Image.Image, review_id: int) -> str:
# """Save image locally and upload to dataset repo. Returns filename."""
# filename = f"review_{review_id}.jpg"
# local_path = os.path.join(LOCAL_TMP_DIR, filename)
# img.convert("RGB").save(local_path, format="JPEG")
# api.upload_file(
# path_or_fileobj=local_path,
# path_in_repo=f"images/{filename}",
# repo_id=dataset_repo,
# repo_type="dataset",
# commit_message=f"Upload image for review {review_id}",
# )
# return filename
# def save_review(report_text: str, technique_label: str, findings_label: str, impression_label: str, img: Image.Image = None):
# if not report_text or not report_text.strip():
# return "No report available to review."
# if not technique_label or not findings_label or not impression_label:
# return "Please select a rating for all three sections before saving."
# try:
# csv_path = download_existing_csv()
# # Duplicate check
# report_hash = get_report_hash(report_text)
# if is_duplicate_review(csv_path, report_hash):
# return "This report has already been reviewed. Duplicate not saved."
# # Get next ID
# review_id = get_next_review_id(csv_path)
# # Upload image if provided
# image_filename = "N/A"
# if img is not None:
# image_filename = upload_image_to_dataset(img, review_id)
# with open(csv_path, "a", newline="", encoding="utf-8") as f:
# writer = csv.writer(f)
# writer.writerow([
# review_id,
# datetime.utcnow().isoformat(),
# report_hash,
# technique_label, RATING_MAP[technique_label],
# findings_label, RATING_MAP[findings_label],
# impression_label, RATING_MAP[impression_label],
# image_filename,
# ])
# upload_csv_to_dataset(csv_path)
# return f"Review #{review_id} saved successfully to dataset repo: {dataset_repo}"
# except Exception as e:
# return f"Failed to save review: {str(e)}"
# DARK_CSS = """
# body, .gradio-container {
# background: #393e46 !important;
# color: white !important;
# }
# /* Markdown text */
# .gr-markdown, .gr-markdown h1, .gr-markdown h2, .gr-markdown p, h1, p, h2, h3, .gr-markdown h3 {
# color: white !important;
# }
# /* Panels / component containers */
# .gr-panel, .gr-box, .gr-form {
# background: #393e46 !important;
# border-color: #2a2a2a !important;
# }
# /* Textbox */
# textarea, .wrap textarea {
# background: #393e46 !important;
# color: white !important;
# border: 1px solid #2a2a2a !important;
# }
# /* Labels */
# label, .gr-label, .wrap label {
# color: white !important;
# }
# /* Buttons */
# .gr-button {
# background: white !important;
# color: black !important;
# border: none !important;
# }
# button {
# background: #393e46 !important;
# color: white !important;
# border: 2px solid #e70000 !important;
# }
# .gr-button:hover {
# background: #393e46 !important;
# }
# button:hover {
# background: #e70000 !important;
# color: black !important;
# }
# /* File component area */
# input[type="file"] {
# color: white !important;
# }
# """
# with gr.Blocks() as demo:
# gr.Markdown("# Chest X-ray Auto Report Generator\n Radiologist Assistant")
# image_in = gr.File(
# label="Upload chest X-ray (PNG / JPG / DICOM .dcm)",
# file_types=[".png", ".jpg", ".jpeg", ".dcm"]
# )
# report_loading = gr.HTML(
# """
# <div style="
# height: 420px;
# border: 1px solid #d1d5db;
# border-radius: 8px;
# background: black;
# display: flex;
# align-items: center;
# justify-content: center;
# flex-direction: column;
# gap: 10px;
# padding: 16px;
# ">
# <img src="https://media.giphy.com/media/v1.Y2lkPTc5MGI3NjExYW5pdDh0ZHdoNXplaTlqdno1OWNrbzc3cDcxaTRkc3FoaTN1anF2aiZlcD12MV9naWZzX3NlYXJjaCZjdD1n/dt0KXLj7bzwZuRQBwY/giphy.gif"
# style="width:200px; height:auto;" />
# <div style="font-size:25px; font-weight:800; color:white;">Generating...</div>
# </div>
# """,
# visible=False,
# )
# report_out = gr.Textbox(
# label="Generated radiology report (editable)",
# lines=18,
# interactive=False,
# visible=True
# )
# review_done = gr.State(False)
# review_section = gr.Column(visible=False)
# LIKERT_CHOICES = [
# "1 - Very Unlikely",
# "2 - Unlikely",
# "3 - Neutral",
# "4 - Likely",
# "5 - Very Likely",
# ]
# with review_section:
# gr.Markdown("Radiologist Review")
# gr.Markdown("Rate each section of the generated report:")
# technique_radio = gr.Radio(
# choices=LIKERT_CHOICES,
# label="1. Technique — accuracy & completeness",
# value=None
# )
# findings_radio = gr.Radio(
# choices=LIKERT_CHOICES,
# label="2. Findings — accuracy & completeness",
# value=None
# )
# impression_radio = gr.Radio(
# choices=LIKERT_CHOICES,
# label="3. Impression — accuracy & completeness",
# value=None
# )
# save_review_btn = gr.Button("Save Review")
# review_status = gr.Textbox(label="Review Status", interactive=False)
# gen_btn = gr.Button("Generate report")
# pdf_btn = gr.DownloadButton("Download as PDF", value=None, visible=True)
# def generate_report_ui(upload):
# img, error = load_image_from_upload(upload)
# if img is None:
# yield (
# gr.update(value="error", interactive=False, visible=True),
# gr.update(visible=False),
# gr.update(visible=False),
# gr.update(value=None),
# gr.update(value=None),
# gr.update(value=None),
# gr.update(value=""),
# False,
# )
# return
# # Show loader
# yield (
# gr.update(value="", interactive=False, visible=False),
# gr.update(visible=True),
# gr.update(visible=False),
# gr.update(value=None),
# gr.update(value=None),
# gr.update(value=None),
# gr.update(value=""),
# False,
# )
# report, _ = generate_report(img)
# # Show report + review section
# yield (
# gr.update(value=report, interactive=True, visible=True),
# gr.update(visible=False),
# gr.update(visible=True),
# gr.update(value=None),
# gr.update(value=None),
# gr.update(value=None),
# gr.update(value=""),
# False,
# )
# gen_btn.click(
# fn=generate_report_ui,
# inputs=image_in,
# outputs=[
# report_out,
# report_loading,
# review_section,
# technique_radio,
# findings_radio,
# impression_radio,
# review_status,
# review_done,
# ],
# )
# def guarded_pdf_download(report_text, review_done_flag):
# if not review_done_flag:
# gr.Warning("Please submit a review before downloading the report.")
# return None
# return text_to_pdf(report_text)
# pdf_btn.click(
# fn=guarded_pdf_download,
# inputs=[report_out, review_done],
# outputs=pdf_btn,
# )
# def save_review_ui(report_text, technique, findings, impression, upload):
# img, _ = load_image_from_upload(upload)
# message = save_review(report_text, technique, findings, impression, img)
# if "successfully" in message.lower():
# return message, True
# else:
# return message, False
# save_review_btn.click(
# fn=save_review_ui,
# inputs=[report_out, technique_radio, findings_radio, impression_radio, image_in],
# outputs=[review_status, review_done],
# )
# demo.launch(
# theme=gr.themes.Monochrome(),
# css=DARK_CSS,
# )