Implement ReportGenerator class
This commit is contained in:
199
src/py_dvt_ate/reporting/generator.py
Normal file
199
src/py_dvt_ate/reporting/generator.py
Normal file
@@ -0,0 +1,199 @@
|
||||
"""Report generator orchestrating the full report generation pipeline.
|
||||
|
||||
This module provides the main ReportGenerator class that coordinates
|
||||
data gathering, chart generation, HTML rendering, and PDF conversion.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Protocol
|
||||
from uuid import UUID
|
||||
|
||||
from py_dvt_ate.data.repository import ITestRepository
|
||||
from py_dvt_ate.reporting.charts import ChartGenerator
|
||||
from py_dvt_ate.reporting.exceptions import ReportGenerationError
|
||||
from py_dvt_ate.reporting.models import ReportConfig, ReportData
|
||||
from py_dvt_ate.reporting.renderers import HTMLRenderer, PDFRenderer
|
||||
|
||||
|
||||
class IReportGenerator(Protocol):
|
||||
"""Protocol for report generators."""
|
||||
|
||||
def generate(self, run_id: UUID, output_path: Path | None = None) -> Path:
|
||||
"""Generate a PDF report for a test run.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run.
|
||||
output_path: Optional output path. If None, uses default location.
|
||||
|
||||
Returns:
|
||||
Path to the generated PDF file.
|
||||
"""
|
||||
...
|
||||
|
||||
def generate_bytes(self, run_id: UUID) -> bytes:
|
||||
"""Generate a PDF report and return as bytes.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run.
|
||||
|
||||
Returns:
|
||||
PDF document as bytes.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class ReportGenerator:
|
||||
"""Generates PDF reports from test run data.
|
||||
|
||||
This class orchestrates the full report generation pipeline:
|
||||
1. Fetch test run data from repository
|
||||
2. Generate charts from measurements
|
||||
3. Render HTML from templates
|
||||
4. Convert HTML to PDF
|
||||
|
||||
Example:
|
||||
>>> from py_dvt_ate.data.repository import SQLiteRepository
|
||||
>>> from py_dvt_ate.reporting import ReportGenerator, ReportConfig
|
||||
>>>
|
||||
>>> repo = SQLiteRepository("./data/py_dvt_ate.db")
|
||||
>>> config = ReportConfig(company_name="My Company")
|
||||
>>> generator = ReportGenerator(repo, config)
|
||||
>>> pdf_path = generator.generate(run_id)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
repository: ITestRepository,
|
||||
config: ReportConfig | None = None,
|
||||
reports_dir: Path | None = None,
|
||||
) -> None:
|
||||
"""Initialise the report generator.
|
||||
|
||||
Args:
|
||||
repository: Test data repository for fetching run data.
|
||||
config: Report configuration. Uses defaults if not provided.
|
||||
reports_dir: Directory for generated reports. Defaults to ./data/reports.
|
||||
"""
|
||||
self.repository = repository
|
||||
self.config = config or ReportConfig()
|
||||
self.reports_dir = reports_dir or Path("./data/reports")
|
||||
|
||||
self._html_renderer = HTMLRenderer()
|
||||
self._pdf_renderer = PDFRenderer()
|
||||
self._chart_generator = ChartGenerator(dpi=self.config.chart_dpi)
|
||||
|
||||
def _gather_data(self, run_id: UUID) -> ReportData:
|
||||
"""Gather all data needed for the report.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run.
|
||||
|
||||
Returns:
|
||||
ReportData containing run, results, measurements, and charts.
|
||||
|
||||
Raises:
|
||||
ReportGenerationError: If data gathering fails.
|
||||
"""
|
||||
try:
|
||||
run = self.repository.get_run(run_id)
|
||||
results = self.repository.get_results(run_id)
|
||||
measurements = self.repository.get_measurements_dataframe(run_id)
|
||||
|
||||
# Generate charts if enabled
|
||||
charts: dict[str, str] = {}
|
||||
if self.config.include_charts:
|
||||
charts = self._chart_generator.generate_all(run, results, measurements)
|
||||
|
||||
return ReportData(
|
||||
run=run,
|
||||
results=results,
|
||||
measurements=measurements,
|
||||
charts=charts,
|
||||
config=self.config,
|
||||
)
|
||||
|
||||
except ValueError as e:
|
||||
msg = f"Failed to gather data for run {run_id}: {e}"
|
||||
raise ReportGenerationError(msg) from e
|
||||
|
||||
def _generate_filename(self, run_id: UUID, test_name: str) -> str:
|
||||
"""Generate a filename for the report.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run.
|
||||
test_name: Name of the test.
|
||||
|
||||
Returns:
|
||||
Filename string with timestamp.
|
||||
"""
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
safe_name = test_name.replace(" ", "_").replace("/", "_")
|
||||
return f"{safe_name}_{str(run_id)[:8]}_{timestamp}.pdf"
|
||||
|
||||
def generate(self, run_id: UUID, output_path: Path | None = None) -> Path:
|
||||
"""Generate a PDF report for a test run.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run.
|
||||
output_path: Optional output path. If None, uses default location.
|
||||
|
||||
Returns:
|
||||
Path to the generated PDF file.
|
||||
|
||||
Raises:
|
||||
ReportGenerationError: If report generation fails.
|
||||
"""
|
||||
try:
|
||||
# Gather data
|
||||
data = self._gather_data(run_id)
|
||||
|
||||
# Determine output path
|
||||
if output_path is None:
|
||||
self.reports_dir.mkdir(parents=True, exist_ok=True)
|
||||
filename = self._generate_filename(run_id, data.run.test_name)
|
||||
output_path = self.reports_dir / filename
|
||||
|
||||
# Render HTML
|
||||
html = self._html_renderer.render(data)
|
||||
|
||||
# Convert to PDF
|
||||
self._pdf_renderer.render_to_file(html, output_path)
|
||||
|
||||
return output_path
|
||||
|
||||
except ReportGenerationError:
|
||||
raise
|
||||
except Exception as e:
|
||||
msg = f"Failed to generate report for run {run_id}: {e}"
|
||||
raise ReportGenerationError(msg) from e
|
||||
|
||||
def generate_bytes(self, run_id: UUID) -> bytes:
|
||||
"""Generate a PDF report and return as bytes.
|
||||
|
||||
Useful for streaming downloads without writing to disk.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run.
|
||||
|
||||
Returns:
|
||||
PDF document as bytes.
|
||||
|
||||
Raises:
|
||||
ReportGenerationError: If report generation fails.
|
||||
"""
|
||||
try:
|
||||
# Gather data
|
||||
data = self._gather_data(run_id)
|
||||
|
||||
# Render HTML
|
||||
html = self._html_renderer.render(data)
|
||||
|
||||
# Convert to PDF bytes
|
||||
return self._pdf_renderer.render_to_bytes(html)
|
||||
|
||||
except ReportGenerationError:
|
||||
raise
|
||||
except Exception as e:
|
||||
msg = f"Failed to generate report bytes for run {run_id}: {e}"
|
||||
raise ReportGenerationError(msg) from e
|
||||
Reference in New Issue
Block a user