Skip to content

Pipeline

pipeline

Smelt pipeline — detect, count, compact.

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

Analyze text without transforming it.

Source code in packages/axm-smelt/src/axm_smelt/core/pipeline.py
Python
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 check(
    text: str | None = None,
    *,
    parsed: dict[str, JsonValue] | list[JsonValue] | None = None,
) -> SmeltReport:
    """Analyze *text* without transforming it."""
    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 = detect_format(text)
    tokens, backend = count_with_backend(text)

    ctx = SmeltContext(text=text, format=fmt)
    estimates: dict[str, float] = {}
    for name, cls in _REGISTRY.items():
        strategy = cls()
        result = strategy.apply(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)

    return SmeltReport(
        original=text,
        compacted=text,
        original_tokens=tokens,
        compacted_tokens=tokens,
        savings_pct=0.0,
        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
Python
28
29
30
31
32
33
34
35
36
37
38
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
Python
41
42
43
44
45
46
47
48
49
50
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.

Source code in packages/axm-smelt/src/axm_smelt/core/pipeline.py
Python
 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
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."""
    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,
    )