bolade
319 lines
9.9 KiB
Python
319 lines
9.9 KiB
Python
"""
|
|
Report Generator Service
|
|
|
|
This service handles the generation of medical reports from uploaded files.
|
|
It processes data, generates graphs, and creates PDF reports.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
import pandas as pd
|
|
from jinja2 import Environment, FileSystemLoader
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
from app.services.context import context_list
|
|
from app.services.graph_generator import GraphGenerator
|
|
|
|
|
|
class ReportGeneratorService:
|
|
"""Service for generating medical performance reports"""
|
|
|
|
def __init__(
|
|
self,
|
|
template_dir: str = "app/report_gen",
|
|
graphs_dir: str = "graphs",
|
|
reports_dir: str = "reports",
|
|
):
|
|
"""
|
|
Initialize the report generator service.
|
|
|
|
Args:
|
|
template_dir: Directory containing Jinja2 templates
|
|
graphs_dir: Directory to save generated graphs
|
|
reports_dir: Directory to save generated reports
|
|
"""
|
|
self.template_dir = template_dir
|
|
self.graphs_dir = Path(graphs_dir)
|
|
self.reports_dir = Path(reports_dir)
|
|
self.graph_generator = GraphGenerator(charts_dir=str(graphs_dir))
|
|
self.env = Environment(loader=FileSystemLoader(template_dir))
|
|
|
|
# Ensure directories exist
|
|
self.graphs_dir.mkdir(exist_ok=True)
|
|
self.reports_dir.mkdir(exist_ok=True)
|
|
|
|
def process_pnoe_data(self, pnoe_csv_path: str) -> pd.DataFrame:
|
|
"""
|
|
Load and process Pnoe CSV data.
|
|
|
|
Args:
|
|
pnoe_csv_path: Path to Pnoe CSV file
|
|
|
|
Returns:
|
|
Processed DataFrame with smoothed columns
|
|
"""
|
|
# Load data
|
|
df = pd.read_csv(pnoe_csv_path, delimiter=";")
|
|
df = df.apply(pd.to_numeric, errors="ignore")
|
|
|
|
# Calculate derived columns
|
|
df["VO2 Pulse"] = df["VO2(ml/min)"] / df["HR(bpm)"]
|
|
df["VO2 Breath"] = df["VO2(ml/min)"] / df["BF(bpm)"]
|
|
df["CHO"] = df["EE(kcal/min)"] * df["CARBS(%)"] / 100
|
|
df["FAT"] = df["EE(kcal/min)"] * df["FAT(%)"] / 100
|
|
|
|
# Smooth columns
|
|
window_size = 10
|
|
columns_to_smooth = [
|
|
"VO2(ml/min)",
|
|
"VCO2(ml/min)",
|
|
"HR(bpm)",
|
|
"VT(l)",
|
|
"BF(bpm)",
|
|
"VE(l/min)",
|
|
"VO2 Pulse",
|
|
"VO2 Breath",
|
|
"CHO",
|
|
"FAT",
|
|
]
|
|
|
|
for col in columns_to_smooth:
|
|
if col in df.columns:
|
|
df[f"{col}_smoothed"] = (
|
|
df[col].rolling(window=window_size, min_periods=1).mean()
|
|
)
|
|
|
|
return df
|
|
|
|
def generate_graphs(self, df: pd.DataFrame) -> List[Dict[str, str]]:
|
|
"""
|
|
Generate all required graphs from processed data.
|
|
|
|
Args:
|
|
df: Processed DataFrame with smoothed columns
|
|
|
|
Returns:
|
|
List of dictionaries containing graph names and paths
|
|
"""
|
|
graphs_generated = []
|
|
|
|
# List of graphs to generate
|
|
graph_methods = [
|
|
("respiratory", self.graph_generator.generate_respiratory_chart),
|
|
("fuel_utilization", self.graph_generator.generate_fuel_utilization_chart),
|
|
("vo2_pulse", self.graph_generator.generate_vo2_pulse_chart),
|
|
("vo2_breath", self.graph_generator.generate_vo2_breath_chart),
|
|
("fat_metabolism", self.graph_generator.generate_fat_metabolism_chart),
|
|
("recovery", self.graph_generator.generate_recovery_chart),
|
|
]
|
|
|
|
for name, method in graph_methods:
|
|
try:
|
|
path = method(df, save_as_base64=False)
|
|
graphs_generated.append({"name": name, "path": str(path)})
|
|
except Exception as e:
|
|
print(f"Warning: Could not generate {name} chart: {e}")
|
|
|
|
return graphs_generated
|
|
|
|
def calculate_analysis_metrics(self, df: pd.DataFrame) -> Dict[str, Any]:
|
|
"""
|
|
Calculate basic analysis metrics from processed data.
|
|
|
|
Args:
|
|
df: Processed DataFrame with smoothed columns
|
|
|
|
Returns:
|
|
Dictionary containing analysis metrics
|
|
"""
|
|
return {
|
|
"vo2_max": float(df["VO2(ml/min)_smoothed"].max())
|
|
if "VO2(ml/min)_smoothed" in df.columns
|
|
else 0,
|
|
"peak_vt": float(df["VT(l)_smoothed"].max())
|
|
if "VT(l)_smoothed" in df.columns
|
|
else 0,
|
|
"max_hr": float(df["HR(bpm)_smoothed"].max())
|
|
if "HR(bpm)_smoothed" in df.columns
|
|
else 0,
|
|
}
|
|
|
|
def generate_html(self, patient_info: Dict[str, Any]) -> str:
|
|
"""
|
|
Generate HTML content for the report.
|
|
|
|
Args:
|
|
patient_info: Dictionary containing patient information
|
|
(patient_name, age, height, weight, focus)
|
|
|
|
Returns:
|
|
Complete HTML document as string
|
|
"""
|
|
html_pages = []
|
|
|
|
# Header context
|
|
header_context = {
|
|
"patient_name": patient_info.get("patient_name", ""),
|
|
"age": patient_info.get("age", ""),
|
|
"height": patient_info.get("height", ""),
|
|
"weight": patient_info.get("weight", ""),
|
|
"focus": patient_info.get("focus", "Endurance"),
|
|
}
|
|
|
|
# Footer context
|
|
footer_context = [
|
|
{
|
|
"contact_email": "info@ishplabs.com",
|
|
"website": "www.ishplabs.com",
|
|
"social": "@ishplabs",
|
|
"page_number": i + 1,
|
|
}
|
|
for i in range(len(context_list))
|
|
]
|
|
|
|
# Render header
|
|
header_html = self.env.get_template("header.html").render(header_context)
|
|
|
|
# Render footers
|
|
footer_html_list = [
|
|
self.env.get_template("footer.html").render(context)
|
|
for context in footer_context
|
|
]
|
|
|
|
# Render pages
|
|
for i, context in enumerate(context_list):
|
|
template = self.env.get_template(f"page_{i + 1}.html").render(context)
|
|
|
|
if (i + 1) > 2:
|
|
full_html = f"""
|
|
<div class="page flex flex-col justify-between">
|
|
<div>
|
|
{header_html}
|
|
</div>
|
|
<main class="flex-grow p-4">
|
|
{template}
|
|
</main>
|
|
<div class="border-t text-center text-sm text-gray-600">
|
|
{footer_html_list[i]}
|
|
</div>
|
|
</div>
|
|
"""
|
|
html_pages.append(full_html)
|
|
else:
|
|
html_pages.append(template)
|
|
|
|
# Combine with page breaks
|
|
final_html = "<div class='page-break'></div>".join(html_pages)
|
|
|
|
# Wrap in full HTML document
|
|
html_doc = f"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<link href="https://cdn.jsdelivr.net/npm/tailwindcss/dist/tailwind.min.css" rel="stylesheet">
|
|
<style>
|
|
html, body {{
|
|
height: 100%;
|
|
margin: 0;
|
|
padding: 0;
|
|
}}
|
|
.page-break {{ page-break-after: always; }}
|
|
.page {{
|
|
height: 100vh;
|
|
min-height: 100vh;
|
|
display: flex;
|
|
flex-direction: column;
|
|
}}
|
|
.page main {{
|
|
flex: 1;
|
|
overflow: hidden;
|
|
}}
|
|
* {{
|
|
margin: 0;
|
|
padding: 0;
|
|
box-sizing: border-box;
|
|
}}
|
|
img {{
|
|
max-height: 300px;
|
|
}}
|
|
.chart-large {{
|
|
max-height: 500px !important;
|
|
}}
|
|
</style>
|
|
</head>
|
|
<body class="m-0 p-0">
|
|
{final_html}
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return html_doc
|
|
|
|
def html_to_pdf(self, html_content: str, pdf_path: str) -> None:
|
|
"""
|
|
Convert HTML content to PDF file.
|
|
|
|
Args:
|
|
html_content: HTML content as string
|
|
pdf_path: Path where PDF should be saved
|
|
"""
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch()
|
|
page = browser.new_page()
|
|
page.set_content(html_content)
|
|
page.pdf(path=pdf_path, format="A4", print_background=True)
|
|
browser.close()
|
|
|
|
def generate_report(
|
|
self,
|
|
spirometry_pdf_path: str,
|
|
pnoe_csv_path: str,
|
|
seca_excel_path: str,
|
|
patient_info: Dict[str, Any],
|
|
output_filename: str = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Generate complete medical report from uploaded files.
|
|
|
|
Args:
|
|
spirometry_pdf_path: Path to Spirometry PDF file
|
|
pnoe_csv_path: Path to Pnoe CSV file
|
|
seca_excel_path: Path to SECA Excel file
|
|
patient_info: Dictionary containing patient information
|
|
output_filename: Optional custom output filename
|
|
n
|
|
Returns:
|
|
Dictionary containing report path, graphs generated, and analysis data
|
|
"""
|
|
# Process data
|
|
df = self.process_pnoe_data(pnoe_csv_path)
|
|
|
|
# Generate graphs
|
|
graphs_generated = self.generate_graphs(df)
|
|
|
|
# Calculate analysis metrics
|
|
analysis_data = self.calculate_analysis_metrics(df)
|
|
analysis_data["graphs_count"] = len(graphs_generated)
|
|
|
|
# Generate HTML
|
|
html_content = self.generate_html(patient_info)
|
|
|
|
# Generate PDF
|
|
if output_filename is None:
|
|
patient_name = patient_info.get("patient_name", "Unknown")
|
|
session_id = patient_info.get("session_id", "default")
|
|
output_filename = (
|
|
f"report_{patient_name.replace(' ', '_')}_{session_id}.pdf"
|
|
)
|
|
|
|
report_path = self.reports_dir / output_filename
|
|
self.html_to_pdf(html_content, str(report_path))
|
|
|
|
return {
|
|
"report_path": str(report_path),
|
|
"graphs_generated": graphs_generated,
|
|
"analysis_data": analysis_data,
|
|
}
|