Pipeline

pipeline

Smelt pipeline — detect, count, compact.

check(text=None, *, parsed=None)

Analyze text without transforming it.

The report carries two distinct savings figures:

• strategy_estimates maps each registry strategy to the reduction it achieves in isolation, measured against the unmutated input. These estimates are independent and non-additive: summing them overstates the achievable gain, because strategies overlap (e.g. minify already removes whitespace that collapse_whitespace would also target).
• savings_pct is the real cumulative gain — the reduction obtained by chaining the default strategy set (resolve_strategies(None, None), i.e. the safe preset, exactly what :func:smelt applies with no explicit strategies). It equals what a user would actually get from smelt(text).

original and compacted stay identical: check never transforms its input, it only measures.

Source code in packages/axm-smelt/src/axm_smelt/core/pipeline.py

def check(
    text: str | None = None,
    *,
    parsed: dict[str, JsonValue] | list[JsonValue] | None = None,
) -> SmeltReport:
    """Analyze *text* without transforming it.

    The report carries two distinct savings figures:

    - ``strategy_estimates`` maps each registry strategy to the reduction it
      achieves *in isolation*, measured against the unmutated input. These
      estimates are **independent and non-additive**: summing them overstates
      the achievable gain, because strategies overlap (e.g. ``minify`` already
      removes whitespace that ``collapse_whitespace`` would also target).
    - ``savings_pct`` is the **real cumulative gain** — the reduction obtained
      by *chaining* the default strategy set (``resolve_strategies(None, None)``,
      i.e. the ``safe`` preset, exactly what :func:`smelt` applies with no
      explicit strategies). It equals what a user would actually get from
      ``smelt(text)``.

    ``original`` and ``compacted`` stay identical: ``check`` never transforms
    its input, it only measures.
    """
    from axm_smelt.strategies import _REGISTRY

    if parsed is not None:
        text = json.dumps(parsed, separators=(",", ":"))
    elif text is None:
        msg = "Either text or parsed must be provided"
        raise ValueError(msg)

    fmt, detected_parsed = detect_format_parsed(text)
    tokens, backend = count_with_backend(text)

    if detected_parsed is not None:
        ctx = SmeltContext(text=text, format=fmt, parsed=detected_parsed)
    else:
        ctx = SmeltContext(text=text, format=fmt)

    estimates: dict[str, float] = {}
    for name, cls in _REGISTRY.items():
        strategy = cls()
        result = _safe_apply(strategy, ctx)
        if result.text != ctx.text:
            result_tokens, b = count_with_backend(result.text)
            backend = _worst(backend, b)
            savings = (1 - result_tokens / tokens) * 100 if tokens > 0 else 0.0
            if savings > 0:
                estimates[name] = round(savings, 2)

    strats = resolve_strategies(None, None)
    chained_ctx, _applied, b_strat = _apply_strategies(ctx, strats, tokens)
    backend = _worst(backend, b_strat)
    chained_tokens, b_chain = count_with_backend(chained_ctx.text)
    backend = _worst(backend, b_chain)
    cumulative = (1 - chained_tokens / tokens) * 100 if tokens > 0 else 0.0

    return SmeltReport(
        original=text,
        compacted=text,
        original_tokens=tokens,
        compacted_tokens=tokens,
        savings_pct=cumulative,
        format=fmt,
        strategies_applied=[],
        strategy_estimates=estimates,
        counter_backend=backend,
    )

resolve_input(text, parsed)

Normalize inputs into (text, parsed).

Source code in packages/axm-smelt/src/axm_smelt/core/pipeline.py

def resolve_input(
    text: str | None,
    parsed: dict[str, JsonValue] | list[JsonValue] | None,
) -> tuple[str, dict[str, JsonValue] | list[JsonValue] | None]:
    """Normalize inputs into ``(text, parsed)``."""
    if parsed is not None:
        return json.dumps(parsed, separators=(",", ":")), parsed
    if text is None:
        msg = "Either text or parsed must be provided"
        raise ValueError(msg)
    return text, None

resolve_strategies(strategies, preset)

Return strategy instances from explicit names, a preset, or the default.

Source code in packages/axm-smelt/src/axm_smelt/core/pipeline.py

def resolve_strategies(
    strategies: list[str] | None,
    preset: str | None,
) -> list[SmeltStrategy]:
    """Return strategy instances from explicit names, a preset, or the default."""
    if strategies:
        return [get_strategy(s) for s in strategies]
    if preset:
        return get_preset(preset)
    return get_preset("safe")

smelt(text=None, strategies=None, preset=None, *, parsed=None)

Run the compaction pipeline and return a report.

Baseline for savings_pct: the pipeline's working text — the compact serialization the strategies actually operate on. On the text= path this is the provided raw string; on the parsed= path it is the compact dump json.dumps(parsed, separators=(",", ":")) (identical to report.original).

Reporting savings against the working text keeps savings_pct honest: it measures only what the strategies achieved, never the pretty-vs-compact gap of :func:resolve_input (which no strategy performed). A parsed= input on which no strategy applies therefore reports 0 — not a phantom reduction. This single honest baseline also seeds the keep-if-reduced guard, so a strategy whose output is heavier than the working text can never be accepted.

Source code in packages/axm-smelt/src/axm_smelt/core/pipeline.py

def smelt(
    text: str | None = None,
    strategies: list[str] | None = None,
    preset: str | None = None,
    *,
    parsed: dict[str, JsonValue] | list[JsonValue] | None = None,
) -> SmeltReport:
    """Run the compaction pipeline and return a report.

    Baseline for ``savings_pct``: the pipeline's **working text** — the
    compact serialization the strategies actually operate on. On the
    ``text=`` path this is the provided raw string; on the ``parsed=`` path
    it is the compact dump ``json.dumps(parsed, separators=(",", ":"))``
    (identical to ``report.original``).

    Reporting savings against the working text keeps ``savings_pct`` honest:
    it measures *only what the strategies achieved*, never the
    pretty-vs-compact gap of :func:`resolve_input` (which no strategy
    performed). A ``parsed=`` input on which no strategy applies therefore
    reports ``0`` — not a phantom reduction. This single honest baseline
    also seeds the keep-if-reduced guard, so a strategy whose output is
    heavier than the working text can never be accepted.
    """
    text, parsed = resolve_input(text, parsed)

    fmt, detected_parsed = detect_format_parsed(text)
    if parsed is not None:
        detected_parsed = parsed
    original_tokens, b1 = count_with_backend(text)

    strats = resolve_strategies(strategies, preset)

    if detected_parsed is not None:
        ctx = SmeltContext(text=text, format=fmt, parsed=detected_parsed)
    else:
        ctx = SmeltContext(text=text, format=fmt)

    ctx, applied, b_strat = _apply_strategies(ctx, strats, original_tokens)

    compacted = ctx.text
    compacted_tokens, b3 = count_with_backend(compacted)
    backend = _worst(_worst(b1, b_strat), b3)
    savings = (
        (1 - compacted_tokens / original_tokens) * 100 if original_tokens > 0 else 0.0
    )

    return SmeltReport(
        original=text,
        compacted=compacted,
        original_tokens=original_tokens,
        compacted_tokens=compacted_tokens,
        savings_pct=savings,
        format=fmt,
        strategies_applied=applied,
        counter_backend=backend,
    )