diff --git a/notebooks/voila.ipynb b/notebooks/voila.ipynb index 04ce194..3519f75 100644 --- a/notebooks/voila.ipynb +++ b/notebooks/voila.ipynb @@ -168,6 +168,50 @@ " f\"Return code: {result.returncode}\\n\"\n", " f\"Stdout:\\n{result.stdout}\\n\"\n", " f\"Stderr:\\n{result.stderr}\"\n", + " ) from error\n", + "\n", + "\n", + "def _build_report_cmd(*args):\n", + " \"\"\"Build the uvx command for the report-generation CLI.\"\"\"\n", + " if os.environ.get(\"SAR_PATTERN_VALIDATION_BACKEND_MODE\") == \"local\":\n", + " package_spec = str(PROJECT_ROOT)\n", + " else:\n", + " package_spec = GITHUB_PACKAGE_SPEC\n", + " return [\n", + " \"uvx\",\n", + " \"--no-cache\",\n", + " \"--from\",\n", + " package_spec,\n", + " \"sar-pattern-validation-report\",\n", + " *args,\n", + " ]\n", + "\n", + "\n", + "def run_report_generation(*args):\n", + " \"\"\"Run the report-generation CLI and return parsed JSON output.\"\"\"\n", + " cmd = _build_report_cmd(*args)\n", + "\n", + " env = os.environ.copy()\n", + " env[\"MPLBACKEND\"] = \"agg\"\n", + "\n", + " result = subprocess.run(cmd, capture_output=True, text=True, env=env)\n", + " _log_path = WORKSPACE_ROOT / \"system_state\" / \"voila_report.log\"\n", + " _log_path.parent.mkdir(exist_ok=True, parents=True)\n", + " with open(_log_path, \"w\") as _f:\n", + " _f.write(\n", + " f\"CMD: {' '.join(cmd)}\\n\\nSTDOUT:\\n{result.stdout}\\n\\nSTDERR:\\n{result.stderr}\\n\"\n", + " )\n", + " raw_output = result.stdout\n", + "\n", + " try:\n", + " return json.loads(raw_output)\n", + " except json.JSONDecodeError as error:\n", + " raise RuntimeError(\n", + " \"sar-pattern-validation-report did not return valid JSON.\\n\"\n", + " f\"Command: {' '.join(cmd)}\\n\"\n", + " f\"Return code: {result.returncode}\\n\"\n", + " f\"Stdout:\\n{result.stdout}\\n\"\n", + " f\"Stderr:\\n{result.stderr}\"\n", " ) from error" ] }, @@ -258,6 +302,9 @@ " WORKSPACE_ROOT / \"system_state\" / \"workflow_results.json\"\n", " )\n", " FILTERED_DB_CSV_PATH = str(WORKSPACE_ROOT / \"system_state\" / \"filtered_db.csv\")\n", + " WORKFLOW_RESULTS_JSON = str(\n", + " WORKSPACE_ROOT / \"system_state\" / \"workflow_results.json\"\n", + " )\n", " REPORT_OUTPUT_DIR = str(WORKSPACE_ROOT / \"report\")\n", " REPORT_TEMPLATE_DIR = str(PROJECT_ROOT / \"report_template\")" ] @@ -793,6 +840,7 @@ " self.radio_button_grid._on_filter_changed = self._update_run_button\n", "\n", " self.workflow_results: WorkflowResults\n", + " self.report_generation_running = False\n", "\n", " # Thread control\n", " self._progress_thread = None\n", @@ -897,8 +945,11 @@ " self.reset_report_button.disabled = False\n", " else:\n", " self.reset_report_button.disabled = True\n", - " report_path_list = list(report_output_dir.glob(\"*.pdf\"))\n", - " if len(report_path_list) != 0:\n", + " workflow_results_json = Path(FilePaths.WORKFLOW_RESULTS_JSON.value)\n", + " if (\n", + " workflow_results_json.exists()\n", + " and not self.report_generation_running\n", + " ):\n", " self.export_report_button.disabled = False\n", " else:\n", " self.export_report_button.disabled = True\n", @@ -1076,19 +1127,6 @@ " *meas_area_args,\n", " \"--noise_floor\",\n", " f\"{self.noise_floor.value}\",\n", - " \"--report\",\n", - " \"--report_output_dir\",\n", - " f\"{FilePaths.REPORT_OUTPUT_DIR.value}\",\n", - " \"--report_template_dir\",\n", - " f\"{FilePaths.REPORT_TEMPLATE_DIR.value}\",\n", - " \"--report_antenna_type\",\n", - " f\"{_filter_options.antenna_type.value}\",\n", - " \"--report_frequency_mhz\",\n", - " f\"{int(_filter_options.frequency.value)}\",\n", - " \"--report_distance_mm\",\n", - " f\"{int(_filter_options.distance.value)}\",\n", - " \"--report_mass_g\",\n", - " f\"{int(_filter_options.mass.value)}\",\n", " )\n", "\n", " self.logger.info(\"SAR Pattern Validation done.\")\n", @@ -1489,21 +1527,127 @@ " ph.value = ph_data\n", "\n", " def handle_export_button_click(self, button: widgets.Button):\n", - " timestr = time.strftime(\"%Y%m%d-%H%M%S\")\n", + " \"\"\"Generate (or append) a 1-page report for the current results.\"\"\"\n", + " button.disabled = True\n", + " self.report_generation_running = True\n", + " self._clear_feedback_banner()\n", "\n", - " report_output_dir = Path(FilePaths.REPORT_OUTPUT_DIR.value)\n", - " report_path = list(report_output_dir.glob(\"*.pdf\"))[0]\n", + " if not hasattr(self, \"workflow_results\") or self.workflow_results is None:\n", + " self._set_feedback_banner(\n", + " \"No workflow results available. Run 'Compare Patterns' first.\",\n", + " severity=\"error\",\n", + " )\n", + " return\n", + "\n", + " _filter_options = self.radio_button_grid.filter_options\n", + " if not _filter_options.antenna_type or not _filter_options.frequency:\n", + " self._set_feedback_banner(\n", + " \"Filter selections incomplete — cannot generate report.\",\n", + " severity=\"error\",\n", + " )\n", + " return\n", + "\n", + " try:\n", + " _measured_path = list(\n", + " Path(FilePaths.MEASURED_FILE_DIR.value).glob(\"*.csv\")\n", + " )[0]\n", + " _ref_path = str(\n", + " self.radio_button_grid.filtered_db[\n", + " DBColumnNames.FILE_PATH.value\n", + " ].to_list()[0]\n", + " )\n", + "\n", + " meas_area_args = []\n", + " if self.measurement_area_x.value and self.measurement_area_y.value:\n", + " meas_area_args = [\n", + " \"--measurement-area-x-mm\",\n", + " str(float(self.measurement_area_x.value)),\n", + " \"--measurement-area-y-mm\",\n", + " str(float(self.measurement_area_y.value)),\n", + " ]\n", + "\n", + " report_output = run_report_generation(\n", + " \"--workflow-results-json\",\n", + " FilePaths.WORKFLOW_RESULTS_JSON_PATH.value,\n", + " \"--measured-file-path\",\n", + " str(_measured_path),\n", + " \"--reference-file-path\",\n", + " _ref_path,\n", + " \"--power-level-dbm\",\n", + " str(self.power_level.value),\n", + " \"--noise-floor\",\n", + " str(self.noise_floor.value),\n", + " *meas_area_args,\n", + " \"--report-output-dir\",\n", + " FilePaths.REPORT_OUTPUT_DIR.value,\n", + " \"--report-template-dir\",\n", + " FilePaths.REPORT_TEMPLATE_DIR.value,\n", + " \"--antenna-type\",\n", + " str(_filter_options.antenna_type.value),\n", + " \"--frequency-mhz\",\n", + " str(int(_filter_options.frequency.value)),\n", + " \"--distance-mm\",\n", + " str(int(_filter_options.distance.value)),\n", + " \"--mass-g\",\n", + " str(int(_filter_options.mass.value)),\n", + " )\n", "\n", - " outputs_dir = Path(os.environ.get(\"DY_SIDECAR_PATH_OUTPUTS\"))\n", - " output_1 = outputs_dir / \"output_1\"\n", - " outfile_path = output_1 / f\"report_{timestr}.pdf\"\n", + " if report_output.get(\"status\") != \"success\":\n", + " _msg = report_output.get(\"error\", {}).get(\n", + " \"message\", \"Report generation failed.\"\n", + " )\n", + " raise RuntimeError(_msg)\n", "\n", - " shutil.copy2(report_path, outfile_path)\n", + " # Copy PDF to outputs directory as report.pdf\n", + " report_output_dir = Path(FilePaths.REPORT_OUTPUT_DIR.value)\n", + " pdf_list = list(report_output_dir.glob(\"*.pdf\"))\n", + " if pdf_list:\n", + " outputs_dir_env = os.environ.get(\"DY_SIDECAR_PATH_OUTPUTS\")\n", + " if outputs_dir_env:\n", + " outputs_dir = Path(outputs_dir_env)\n", + " output_1 = outputs_dir / \"output_1\"\n", + " output_1.mkdir(parents=True, exist_ok=True)\n", + " outfile_path = output_1 / \"report.pdf\"\n", + " shutil.copy2(pdf_list[0], outfile_path)\n", + "\n", + " self._set_feedback_banner(\n", + " \"Report page added successfully.\", severity=\"info\"\n", + " )\n", + "\n", + " except Exception as e:\n", + " self._set_feedback_banner(\n", + " f\"Report generation failed: {e}\", severity=\"error\"\n", + " )\n", + " self.logger.warning(f\"Report generation failed: {e}\")\n", + " finally:\n", + " button.disabled = False\n", + " self.report_generation_running = False\n", "\n", " def handle_reset_report_button_click(self, button: widgets.Button):\n", + " \"\"\"Show confirmation before deleting the current report.\"\"\"\n", + " button.disabled = True\n", + " self._clear_feedback_banner()\n", + "\n", " report_output_dir = Path(FilePaths.REPORT_OUTPUT_DIR.value)\n", - " if report_output_dir.is_dir():\n", - " shutil.rmtree(report_output_dir)\n", + " if not report_output_dir.is_dir():\n", + " self._set_feedback_banner(\"No report to reset.\", severity=\"warning\")\n", + " return\n", + "\n", + " report_output_dir_inner = Path(FilePaths.REPORT_OUTPUT_DIR.value)\n", + " if report_output_dir_inner.is_dir():\n", + " shutil.rmtree(report_output_dir_inner)\n", + " # Also remove the exported PDF from outputs\n", + " outputs_dir_env = os.environ.get(\"DY_SIDECAR_PATH_OUTPUTS\")\n", + " if outputs_dir_env:\n", + " out_pdf = Path(outputs_dir_env) / \"output_1\" / \"report.pdf\"\n", + " if out_pdf.is_file():\n", + " out_pdf.unlink()\n", + " self._set_feedback_banner(\n", + " \"Report deleted. You can start a new report.\",\n", + " severity=\"info\",\n", + " )\n", + " self.feedback_banner.value = \"\"\n", + " button.disabled = False\n", "\n", " def create_maintenance_ui(self) -> widgets.HTML:\n", " \"\"\"\n", diff --git a/pyproject.toml b/pyproject.toml index 6e6d22f..13d0fb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ e2e = [ [project.scripts] sar-pattern-validation = "sar_pattern_validation.workflow_cli:main" +sar-pattern-validation-report = "sar_pattern_validation.report_cli:main" [build-system] requires = ["hatchling"] diff --git a/src/sar_pattern_validation/report_cli.py b/src/sar_pattern_validation/report_cli.py new file mode 100644 index 0000000..f4d3ea8 --- /dev/null +++ b/src/sar_pattern_validation/report_cli.py @@ -0,0 +1,189 @@ +""" +CLI entry point for standalone report generation. + +Generates (or appends to) the SAR Pattern Validation PDF report from +previously saved workflow results, without re-running the gamma workflow. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path +from typing import Any + +from sar_pattern_validation.report import DEFAULT_TEMPLATE_DIR, generate_report +from sar_pattern_validation.workflow_config import WorkflowConfig +from sar_pattern_validation.workflows import WorkflowResult + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Generate or append a page to the SAR validation report." + ) + parser.add_argument( + "--workflow-results-json", + type=str, + required=True, + help="Path to the workflow results JSON file.", + ) + parser.add_argument( + "--measured-file-path", + type=str, + required=True, + help="Path to the measured CSV (used for filename display in report).", + ) + parser.add_argument( + "--reference-file-path", + type=str, + default="reference.csv", + help="Path to the reference CSV.", + ) + parser.add_argument( + "--power-level-dbm", + type=float, + default=30.0, + ) + parser.add_argument( + "--noise-floor", + type=float, + default=0.05, + ) + parser.add_argument( + "--measurement-area-x-mm", + type=float, + default=None, + ) + parser.add_argument( + "--measurement-area-y-mm", + type=float, + default=None, + ) + parser.add_argument( + "--report-output-dir", + type=str, + default="report", + help="Output directory for the generated LaTeX report.", + ) + parser.add_argument( + "--report-template-dir", + type=str, + default=None, + help="Override path to the LaTeX report template directory.", + ) + parser.add_argument( + "--antenna-type", + type=str, + default="dipole", + ) + parser.add_argument( + "--frequency-mhz", + type=int, + default=0, + ) + parser.add_argument( + "--distance-mm", + type=int, + default=0, + ) + parser.add_argument( + "--mass-g", + type=int, + default=0, + ) + return parser + + +def _load_workflow_result(json_path: str) -> WorkflowResult: + """Load WorkflowResult from a JSON file (as saved by workflow_cli).""" + data: dict[str, Any] = json.loads(Path(json_path).read_text(encoding="utf-8")) + + # Convert string paths back to Path objects for image fields + path_fields = { + "gamma_image_path", + "failure_image_path", + "registered_overlay_path", + "loaded_images_path", + "reference_image_path", + "measured_image_path", + "aligned_measured_path", + } + for key in path_fields: + val = data.get(key) + if val is not None and val != "None": + data[key] = Path(val) + else: + data[key] = None + + # Remove fields not in WorkflowResult dataclass (e.g. extra metadata) + import dataclasses + + valid_fields = {f.name for f in dataclasses.fields(WorkflowResult)} + filtered = {k: v for k, v in data.items() if k in valid_fields} + + return WorkflowResult(**filtered) + + +def main(argv: list[str] | None = None) -> int: + """CLI entrypoint for report generation.""" + logging.basicConfig( + level=logging.INFO, + stream=sys.stderr, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + parser = _build_parser() + args = parser.parse_args(argv if argv is not None else sys.argv[1:]) + + try: + workflow_result = _load_workflow_result(args.workflow_results_json) + + workflow_config = WorkflowConfig( + measured_file_path=args.measured_file_path, + reference_file_path=args.reference_file_path, + power_level_dbm=args.power_level_dbm, + noise_floor=args.noise_floor, + measurement_area_x_mm=args.measurement_area_x_mm, + measurement_area_y_mm=args.measurement_area_y_mm, + ) + + template_dir = ( + Path(args.report_template_dir) + if args.report_template_dir + else DEFAULT_TEMPLATE_DIR + ) + + report_path = generate_report( + workflow_result=workflow_result, + workflow_config=workflow_config, + output_dir=args.report_output_dir, + template_dir=template_dir, + antenna_type=args.antenna_type, + frequency_mhz=args.frequency_mhz, + distance_mm=args.distance_mm, + mass_g=args.mass_g, + ) + + payload = { + "status": "success", + "report_path": str(report_path), + } + print(json.dumps(payload, indent=2)) + return 0 + + except Exception as exc: + error_payload = { + "status": "error", + "error": { + "type": type(exc).__name__, + "message": str(exc), + }, + } + print(json.dumps(error_payload, indent=2)) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_report_cli.py b/tests/test_report_cli.py new file mode 100644 index 0000000..ea9596d --- /dev/null +++ b/tests/test_report_cli.py @@ -0,0 +1,188 @@ +"""Tests for the report_cli module.""" + +import json +from pathlib import Path + +import pytest + +from sar_pattern_validation.report_cli import _load_workflow_result, main + + +@pytest.fixture() +def workflow_results_json(tmp_path: Path) -> Path: + """Create a sample workflow results JSON file.""" + data = { + "pass_rate_percent": 97.5, + "evaluated_pixel_count": 1000, + "passed_pixel_count": 975, + "failed_pixel_count": 25, + "gamma_image_path": None, + "failure_image_path": None, + "registered_overlay_path": None, + "loaded_images_path": None, + "reference_image_path": None, + "measured_image_path": None, + "aligned_measured_path": None, + "measured_peak_wkg": 0.246, + "measured_pssar": 24.63, + "reference_pssar": 24.71, + "scaling_error": -0.0035, + "dose_to_agreement": 5.0, + "distance_to_agreement": 2.0, + "min_inscribed_square_mm": 22.0, + "mask_fits_min_inscribed_square": True, + "issues": [], + } + path = tmp_path / "workflow_results.json" + path.write_text(json.dumps(data), encoding="utf-8") + return path + + +def test_load_workflow_result(workflow_results_json: Path): + result = _load_workflow_result(str(workflow_results_json)) + assert result.pass_rate_percent == 97.5 + assert result.measured_pssar == 24.63 + assert result.gamma_image_path is None + + +def test_load_workflow_result_with_paths(tmp_path: Path): + data = { + "pass_rate_percent": 100.0, + "evaluated_pixel_count": 500, + "passed_pixel_count": 500, + "failed_pixel_count": 0, + "gamma_image_path": "/tmp/gamma.png", + "failure_image_path": "None", + "registered_overlay_path": None, + "loaded_images_path": None, + "reference_image_path": "/tmp/ref.png", + "measured_image_path": None, + "aligned_measured_path": None, + "measured_peak_wkg": 0.5, + "measured_pssar": 30.0, + "reference_pssar": 30.0, + "scaling_error": 0.0, + "dose_to_agreement": 5.0, + "distance_to_agreement": 2.0, + "min_inscribed_square_mm": 22.0, + "mask_fits_min_inscribed_square": True, + "issues": [], + } + path = tmp_path / "results.json" + path.write_text(json.dumps(data), encoding="utf-8") + + result = _load_workflow_result(str(path)) + assert result.gamma_image_path == Path("/tmp/gamma.png") + assert result.failure_image_path is None # "None" string → None + assert result.reference_image_path == Path("/tmp/ref.png") + + +def test_main_generates_report(tmp_path: Path, workflow_results_json: Path, capsys): + """main() generates a report .tex file and prints success JSON.""" + from sar_pattern_validation.report import DEFAULT_TEMPLATE_DIR + + output_dir = tmp_path / "report_out" + + exit_code = main( + [ + "--workflow-results-json", + str(workflow_results_json), + "--measured-file-path", + "data/measurements/D900_Flat_HSL_15mm_10dBm_10g_3.csv", + "--reference-file-path", + "data/database/dipole_900MHz_Flat_15mm_10g.csv", + "--power-level-dbm", + "10.0", + "--noise-floor", + "0.001", + "--report-output-dir", + str(output_dir), + "--report-template-dir", + str(DEFAULT_TEMPLATE_DIR), + "--antenna-type", + "dipole", + "--frequency-mhz", + "900", + "--distance-mm", + "15", + "--mass-g", + "10", + ] + ) + + assert exit_code == 0 + captured = capsys.readouterr() + payload = json.loads(captured.out) + assert payload["status"] == "success" + assert "report_path" in payload + + # Verify the .tex file was created + tex_path = output_dir / "main.tex" + assert tex_path.is_file() + text = tex_path.read_text(encoding="utf-8") + assert "dipole" in text + assert "900" in text + + +def test_main_appends_to_existing_report( + tmp_path: Path, workflow_results_json: Path, capsys +): + """Calling main() twice appends a second page.""" + from sar_pattern_validation.report import DEFAULT_TEMPLATE_DIR + + output_dir = tmp_path / "report_out" + + base_args = [ + "--workflow-results-json", + str(workflow_results_json), + "--measured-file-path", + "data/measurements/measured.csv", + "--report-output-dir", + str(output_dir), + "--report-template-dir", + str(DEFAULT_TEMPLATE_DIR), + ] + + # First call + main([*base_args, "--antenna-type", "dipole", "--frequency-mhz", "900"]) + capsys.readouterr() + + # Second call + exit_code = main( + [ + *base_args, + "--antenna-type", + "patch", + "--frequency-mhz", + "2450", + ] + ) + assert exit_code == 0 + + tex_path = output_dir / "main.tex" + text = tex_path.read_text(encoding="utf-8") + assert "dipole" in text + assert "patch" in text + assert text.count(r"\end{document}") == 1 + + # Two case figure directories + assert (output_dir / "figures" / "case_000").is_dir() + assert (output_dir / "figures" / "case_001").is_dir() + + +def test_main_returns_error_on_missing_json(tmp_path: Path, capsys): + """main() returns error JSON when workflow results file is missing.""" + exit_code = main( + [ + "--workflow-results-json", + str(tmp_path / "nonexistent.json"), + "--measured-file-path", + "measured.csv", + ] + ) + + assert exit_code == 1 + captured = capsys.readouterr() + payload = json.loads(captured.out) + assert payload["status"] == "error" + assert "error" in payload