Skip to content

ReportGenerator

vartriage.ReportGenerator

Writes clinical reports in JSON, CSV, PDF, or clinical formats.

Uses a temp file + atomic rename so the output path is never left in a half-written state. JSON and CSV stream from iterators without buffering; PDF materializes all variants for pagination. Clinical formats (clinical-pdf, clinical-html, clinical-docx) delegate to the ClinicalReportGenerator.

Parameters

config : ReportConfig Settings including the desired output format. clinical_config : ClinicalReportConfig, optional Configuration for clinical report generation. Required when the output format starts with "clinical-". reference_checksums : dict[str, str], optional Mapping of reference file paths to SHA-256 checksums. Passed through to ClinicalReportGenerator for the audit trail.

Source code in vartriage/reporting/generator.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class ReportGenerator:
    """Writes clinical reports in JSON, CSV, PDF, or clinical formats.

    Uses a temp file + atomic rename so the output path is never left
    in a half-written state. JSON and CSV stream from iterators without
    buffering; PDF materializes all variants for pagination. Clinical
    formats (clinical-pdf, clinical-html, clinical-docx) delegate to
    the ClinicalReportGenerator.

    Parameters
    ----------
    config : ReportConfig
        Settings including the desired output format.
    clinical_config : ClinicalReportConfig, optional
        Configuration for clinical report generation. Required when
        the output format starts with "clinical-".
    reference_checksums : dict[str, str], optional
        Mapping of reference file paths to SHA-256 checksums.
        Passed through to ClinicalReportGenerator for the audit
        trail.
    """

    def __init__(
        self,
        config: ReportConfig,
        clinical_config: ClinicalReportConfig | None = None,
        reference_checksums: dict[str, str] | None = None,
    ) -> None:
        self._config = config
        self._clinical_config = clinical_config
        self._reference_checksums: dict[str, str] = (
            reference_checksums if reference_checksums is not None else {}
        )

    def generate(
        self,
        variants: Iterator[ClassifiedVariant] | Sequence[ClassifiedVariant],
        output_path: Path,
        source_vcf_path: Path | None = None,
        mito_results: list[Any] | None = None,
        qc_report: Any = None,
    ) -> Path:
        """Write classified variants to the configured format.

        Writes to a temp file alongside the target, then atomically
        replaces it on success. On failure, cleans up the temp file.

        For VCF format, delegates entirely to ``write_vcf`` which
        handles its own atomic write and tabix indexing internally.

        Parameters
        ----------
        variants : Union[Iterator[ClassifiedVariant], Sequence[ClassifiedVariant]]
            Classified variants in priority order. May be empty.
            JSON/CSV consume iterators incrementally; PDF materializes.
        output_path : Path
            Where the final report lands.
        source_vcf_path : Path, optional
            Path to the original input VCF. Required when format
            is ``"vcf"``; ignored for other formats.

        Returns
        -------
        Path
            The written report path.

        Raises
        ------
        IOError
            On write or encoding failure, or if format is "vcf"
            and ``source_vcf_path`` is None.
        """
        output_path = resolve_path(Path(output_path))
        output_path.parent.mkdir(parents=True, exist_ok=True)

        fmt = self._config.output_format

        if fmt.startswith("clinical-"):
            return self._generate_clinical(
                variants,
                output_path,
                mito_results=mito_results,
                qc_report=qc_report,
            )

        if fmt == "vcf":
            if source_vcf_path is None:
                raise OSError("VCF output format requires source_vcf_path")
            from vartriage.reporting.vcf_writer import write_vcf

            # VCF output materializes all variants into memory for the
            # lookup dict, unlike JSON/CSV which stream incrementally.
            # For whole-genome inputs this can reach hundreds of MB.
            materialized = list(variants)
            write_vcf(materialized, source_vcf_path, output_path)
            return output_path

        tmp_fd = None
        tmp_path: Path | None = None
        try:
            tmp_fd, tmp_name = tempfile.mkstemp(
                dir=output_path.parent,
                prefix=".report_",
                suffix=f".{fmt}.tmp",
            )
            os.close(tmp_fd)
            tmp_fd = None
            tmp_path = Path(tmp_name)

            if fmt == "json":
                write_json(variants, tmp_path)
            elif fmt == "csv":
                write_csv(variants, tmp_path)
            elif fmt == "pdf":
                materialized = list(variants)
                self._write_pdf(materialized, tmp_path)
            else:
                raise OSError(f"Unsupported output format: {fmt}")

            os.replace(str(tmp_path), str(output_path))
            tmp_path = None

            return output_path

        except OSError:
            raise
        except Exception as exc:
            raise OSError(f"Failed to generate {fmt.upper()} report: {exc}") from exc
        finally:
            if tmp_path is not None:
                with contextlib.suppress(OSError):
                    tmp_path.unlink(missing_ok=True)

    def _write_pdf(
        self,
        variants: Sequence[ClassifiedVariant],
        output_path: Path,
    ) -> Path:
        """Render PDF, trying reportlab first then the text fallback.

        Parameters
        ----------
        variants : Sequence[ClassifiedVariant]
            Materialized variant list.
        output_path : Path
            Temp path for the PDF.

        Returns
        -------
        Path
            Path to the rendered PDF.

        Raises
        ------
        IOError
            If no PDF backend is available or rendering fails.
        """
        try:
            from vartriage.reporting.pdf_writer import (
                HAS_REPORTLAB,
                ReportlabPDFRenderer,
            )
        except ImportError:
            pass
        else:
            if HAS_REPORTLAB:
                renderer = ReportlabPDFRenderer()
                return renderer.render(list(variants), output_path)

        from vartriage.reporting.pdf_fallback import PDFFallbackRenderer

        renderer_fallback = PDFFallbackRenderer()
        return renderer_fallback.render(list(variants), output_path)

    def _generate_clinical(
        self,
        variants: Iterator[ClassifiedVariant] | Sequence[ClassifiedVariant],
        output_path: Path,
        mito_results: list[Any] | None = None,
        qc_report: Any = None,
    ) -> Path:
        """Delegate to ClinicalReportGenerator for clinical formats.

        Parameters
        ----------
        variants : Union[Iterator[ClassifiedVariant], Sequence[ClassifiedVariant]]
            Classified variants to include in the report.
        output_path : Path
            Where the final report lands.

        Returns
        -------
        Path
            The written report path.

        Raises
        ------
        IOError
            If clinical_config was not provided.
        """
        if self._clinical_config is None:
            raise OSError(
                "Clinical format requires a ClinicalReportConfig to be provided"
            )

        from vartriage import __version__
        from vartriage.reporting.clinical.generator import ClinicalReportGenerator

        clinical_gen = ClinicalReportGenerator(
            config=self._clinical_config,
            pipeline_version=__version__,
            reference_checksums=self._reference_checksums,
        )
        return clinical_gen.generate(
            variants,
            output_path,
            mito_results=mito_results,
            qc_report=qc_report,
        )

generate(variants, output_path, source_vcf_path=None, mito_results=None, qc_report=None)

Write classified variants to the configured format.

Writes to a temp file alongside the target, then atomically replaces it on success. On failure, cleans up the temp file.

For VCF format, delegates entirely to write_vcf which handles its own atomic write and tabix indexing internally.

Parameters

variants : Union[Iterator[ClassifiedVariant], Sequence[ClassifiedVariant]] Classified variants in priority order. May be empty. JSON/CSV consume iterators incrementally; PDF materializes. output_path : Path Where the final report lands. source_vcf_path : Path, optional Path to the original input VCF. Required when format is "vcf"; ignored for other formats.

Returns

Path The written report path.

Raises

IOError On write or encoding failure, or if format is "vcf" and source_vcf_path is None.

Source code in vartriage/reporting/generator.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def generate(
    self,
    variants: Iterator[ClassifiedVariant] | Sequence[ClassifiedVariant],
    output_path: Path,
    source_vcf_path: Path | None = None,
    mito_results: list[Any] | None = None,
    qc_report: Any = None,
) -> Path:
    """Write classified variants to the configured format.

    Writes to a temp file alongside the target, then atomically
    replaces it on success. On failure, cleans up the temp file.

    For VCF format, delegates entirely to ``write_vcf`` which
    handles its own atomic write and tabix indexing internally.

    Parameters
    ----------
    variants : Union[Iterator[ClassifiedVariant], Sequence[ClassifiedVariant]]
        Classified variants in priority order. May be empty.
        JSON/CSV consume iterators incrementally; PDF materializes.
    output_path : Path
        Where the final report lands.
    source_vcf_path : Path, optional
        Path to the original input VCF. Required when format
        is ``"vcf"``; ignored for other formats.

    Returns
    -------
    Path
        The written report path.

    Raises
    ------
    IOError
        On write or encoding failure, or if format is "vcf"
        and ``source_vcf_path`` is None.
    """
    output_path = resolve_path(Path(output_path))
    output_path.parent.mkdir(parents=True, exist_ok=True)

    fmt = self._config.output_format

    if fmt.startswith("clinical-"):
        return self._generate_clinical(
            variants,
            output_path,
            mito_results=mito_results,
            qc_report=qc_report,
        )

    if fmt == "vcf":
        if source_vcf_path is None:
            raise OSError("VCF output format requires source_vcf_path")
        from vartriage.reporting.vcf_writer import write_vcf

        # VCF output materializes all variants into memory for the
        # lookup dict, unlike JSON/CSV which stream incrementally.
        # For whole-genome inputs this can reach hundreds of MB.
        materialized = list(variants)
        write_vcf(materialized, source_vcf_path, output_path)
        return output_path

    tmp_fd = None
    tmp_path: Path | None = None
    try:
        tmp_fd, tmp_name = tempfile.mkstemp(
            dir=output_path.parent,
            prefix=".report_",
            suffix=f".{fmt}.tmp",
        )
        os.close(tmp_fd)
        tmp_fd = None
        tmp_path = Path(tmp_name)

        if fmt == "json":
            write_json(variants, tmp_path)
        elif fmt == "csv":
            write_csv(variants, tmp_path)
        elif fmt == "pdf":
            materialized = list(variants)
            self._write_pdf(materialized, tmp_path)
        else:
            raise OSError(f"Unsupported output format: {fmt}")

        os.replace(str(tmp_path), str(output_path))
        tmp_path = None

        return output_path

    except OSError:
        raise
    except Exception as exc:
        raise OSError(f"Failed to generate {fmt.upper()} report: {exc}") from exc
    finally:
        if tmp_path is not None:
            with contextlib.suppress(OSError):
                tmp_path.unlink(missing_ok=True)