topmark.pipeline.machine.serializers

topmark / pipeline / machine / serializers

Pure JSON/NDJSON serializers for TopMark processing and probe machine-readable output.

This module is intentionally console- and Click-free: it takes already-shaped processing/probe envelopes or record streams and produces serialized strings (or streams of strings).

Responsibilities: - JSON: serialize a single, already-shaped processing or probe envelope mapping. - NDJSON: serialize a stream of already-shaped processing or probe record mappings as newline-delimited JSON (one JSON object per line).

Creation of envelopes (adding meta/kind, selecting container keys, etc.) happens in topmark.pipeline.machine.envelopes, using shared helpers from topmark.core.machine.envelopes. Payload normalization for JSON-compatibility is handled by topmark.core.machine.schemas.normalize_payload.

serialize_probe_results

serialize_probe_results(
    *, meta, config, resolved_toml, results, fmt
)

Serialize resolution probe results in a machine-readable format.

Probe results may include normal file-backed probe contexts and synthetic contexts for explicit inputs filtered during discovery before file-type probing.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Shared machine-readable output metadata payload.	required
config	FrozenConfig	Effective configuration for the run.	required
resolved_toml	ResolvedTopmarkTomlSources	ResolvedTopmarkTomlSources,	required
results	list[ProcessingContext]	Ordered list of probe contexts to serialize.	required
fmt	OutputFormat	Output format (OutputFormat.JSON or OutputFormat.NDJSON).	required

Returns:

Type	Description
str | Iterator[str]	Serialized JSON string or iterator of NDJSON strings.

Raises:

Type	Description
ValueError	If fmt is not a supported machine-readable format.

Source code in src/topmark/pipeline/machine/serializers.py

def serialize_probe_results(
    *,
    meta: MetaPayload,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    results: list[ProcessingContext],
    fmt: OutputFormat,
) -> str | Iterator[str]:
    """Serialize resolution probe results in a machine-readable format.

    Probe results may include normal file-backed probe contexts and synthetic
    contexts for explicit inputs filtered during discovery before file-type
    probing.

    Args:
        meta: Shared machine-readable output metadata payload.
        config: Effective configuration for the run.
        resolved_toml: ResolvedTopmarkTomlSources,
        results: Ordered list of probe contexts to serialize.
        fmt: Output format (`OutputFormat.JSON` or `OutputFormat.NDJSON`).

    Returns:
        Serialized JSON string or iterator of NDJSON strings.

    Raises:
        ValueError: If `fmt` is not a supported machine-readable format.
    """
    if fmt == OutputFormat.JSON:
        envelope: dict[str, object] = build_probe_results_json_envelope(
            meta=meta,
            config=config,
            resolved_toml=resolved_toml,
            results=results,
        )
        return serialize_json_object(envelope)

    if fmt == OutputFormat.NDJSON:
        records: Iterator[dict[str, object]] = iter_probe_results_ndjson_records(
            meta=meta,
            config=config,
            resolved_toml=resolved_toml,
            results=results,
        )
        return iter_ndjson_strings(records)

    raise ValueError(f"Unsupported machine-readable output format: {fmt!r}")

serialize_processing_results

serialize_processing_results(
    *,
    meta,
    config,
    resolved_toml,
    results,
    fmt,
    summary_mode,
)

Serialize processing results for check / strip in a machine-readable format.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Shared machine-readable output metadata payload.	required
config	FrozenConfig	Effective configuration for the run.	required
resolved_toml	ResolvedTopmarkTomlSources	ResolvedTopmarkTomlSources,	required
results	list[ProcessingContext]	Ordered list of per-file check/strip processing contexts.	required
fmt	OutputFormat	Output format (OutputFormat.JSON or OutputFormat.NDJSON).	required
summary_mode	bool	If True, emit aggregated outcome summaries instead of per-file entries.	required

Returns:

Type	Description
str | Iterator[str]	If fmt is JSON: a single pretty-printed JSON string (no trailing newline).
str | Iterator[str]	If fmt is NDJSON: an iterable of JSON strings (one per record), where each yielded string has no trailing newline. The caller controls line joining and whether a final newline is printed.

Raises:

Type	Description
ValueError	If fmt is not a supported machine-readable format.

Source code in src/topmark/pipeline/machine/serializers.py

def serialize_processing_results(
    *,
    meta: MetaPayload,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    results: list[ProcessingContext],
    fmt: OutputFormat,
    summary_mode: bool,
) -> str | Iterator[str]:
    """Serialize processing results for `check` / `strip` in a machine-readable format.

    Args:
        meta: Shared machine-readable output metadata payload.
        config: Effective configuration for the run.
        resolved_toml: ResolvedTopmarkTomlSources,
        results: Ordered list of per-file check/strip processing contexts.
        fmt: Output format (`OutputFormat.JSON` or `OutputFormat.NDJSON`).
        summary_mode: If True, emit aggregated outcome summaries instead of per-file entries.

    Returns:
        - If `fmt` is JSON: a single pretty-printed JSON string (no trailing newline).
        - If `fmt` is NDJSON: an iterable of JSON strings (one per record), where each
            yielded string has no trailing newline. The caller controls line joining and
            whether a final newline is printed.

    Raises:
        ValueError: If `fmt` is not a supported machine-readable format.
    """
    if fmt == OutputFormat.JSON:
        return serialize_processing_results_json(
            meta=meta,
            config=config,
            resolved_toml=resolved_toml,
            results=results,
            summary_mode=summary_mode,
        )

    if fmt == OutputFormat.NDJSON:
        return serialize_processing_results_ndjson(
            meta=meta,
            config=config,
            resolved_toml=resolved_toml,
            results=results,
            summary_mode=summary_mode,
        )

    # Defensive guard
    raise ValueError(f"Unsupported machine-readable output format: {fmt!r}")

serialize_processing_results_json

serialize_processing_results_json(
    *, meta, config, resolved_toml, results, summary_mode
)

Serialize processing results for check / strip in a machine-readable format.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Shared machine-readable output metadata payload.	required
config	FrozenConfig	Effective configuration for the run.	required
resolved_toml	ResolvedTopmarkTomlSources	ResolvedTopmarkTomlSources,	required
results	list[ProcessingContext]	Ordered list of per-file check/strip processing contexts.	required
summary_mode	bool	If True, emit aggregated outcome summaries instead of per-file entries.	required

Returns:

Type	Description
str	Single pretty-printed JSON string (no trailing newline).

Source code in src/topmark/pipeline/machine/serializers.py

def serialize_processing_results_json(
    *,
    meta: MetaPayload,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    results: list[ProcessingContext],
    summary_mode: bool,
) -> str:
    """Serialize processing results for `check` / `strip` in a machine-readable format.

    Args:
        meta: Shared machine-readable output metadata payload.
        config: Effective configuration for the run.
        resolved_toml: ResolvedTopmarkTomlSources,
        results: Ordered list of per-file check/strip processing contexts.
        summary_mode: If True, emit aggregated outcome summaries instead of per-file entries.

    Returns:
        Single pretty-printed JSON string (no trailing newline).
    """
    envelope: dict[str, object] = build_processing_results_json_envelope(
        meta=meta,
        config=config,
        resolved_toml=resolved_toml,
        results=results,
        summary_mode=summary_mode,
    )
    return serialize_json_object(envelope)

serialize_processing_results_ndjson

serialize_processing_results_ndjson(
    *, meta, config, resolved_toml, results, summary_mode
)

Serialize processing results for check / strip in a machine-readable format.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Shared machine-readable output metadata payload.	required
config	FrozenConfig	Effective configuration for the run.	required
resolved_toml	ResolvedTopmarkTomlSources	ResolvedTopmarkTomlSources,	required
results	list[ProcessingContext]	Ordered list of per-file check/strip processing contexts.	required
summary_mode	bool	If True, emit aggregated outcome summaries instead of per-file entries.	required

Returns:

Type	Description
Iterator[str]	Iterator of JSON strings (one per record), where each yielded string has
Iterator[str]	no trailing newline. The caller controls line joining and whether a final
Iterator[str]	newline is printed.

Source code in src/topmark/pipeline/machine/serializers.py

def serialize_processing_results_ndjson(
    *,
    meta: MetaPayload,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    results: list[ProcessingContext],
    summary_mode: bool,
) -> Iterator[str]:
    """Serialize processing results for `check` / `strip` in a machine-readable format.

    Args:
        meta: Shared machine-readable output metadata payload.
        config: Effective configuration for the run.
        resolved_toml: ResolvedTopmarkTomlSources,
        results: Ordered list of per-file check/strip processing contexts.
        summary_mode: If True, emit aggregated outcome summaries instead of per-file entries.

    Returns:
        Iterator of JSON strings (one per record), where each yielded string has
        no trailing newline. The caller controls line joining and whether a final
        newline is printed.
    """
    iter_records: Iterator[dict[str, object]] = iter_processing_results_ndjson_records(
        meta=meta,
        config=config,
        resolved_toml=resolved_toml,
        results=results,
        summary_mode=summary_mode,
    )
    return iter_ndjson_strings(iter_records)