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
 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
def check(
    text: str | None = None,
    *,
    parsed: dict[str, Any] | list[Any] | 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 = count(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 = count(result.text)
            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,
    )

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
17
18
19
20
21
22
23
24
25
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
def smelt(
    text: str | None = None,
    strategies: list[str] | None = None,
    preset: str | None = None,
    *,
    parsed: dict[str, Any] | list[Any] | None = None,
) -> SmeltReport:
    """Run the compaction pipeline and return a report."""
    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, _parsed = detect_format_parsed(text)
    if parsed is not None:
        _parsed = parsed
    original_tokens = count(text)

    strats: list[SmeltStrategy]
    if strategies:
        strats = [get_strategy(s) for s in strategies]
    elif preset:
        strats = get_preset(preset)
    else:
        strats = get_preset("safe")

    ctx = SmeltContext(text=text, format=fmt)
    if _parsed is not None:
        ctx._parsed = _parsed
    applied: list[str] = []
    current_tokens = original_tokens
    for s in strats:
        result = s.apply(ctx)
        if result.text != ctx.text:
            result_tokens = count(result.text)
            if result_tokens < current_tokens or (
                result_tokens == current_tokens and len(result.text) < len(ctx.text)
            ):
                applied.append(s.name)
                ctx = result
                current_tokens = result_tokens

    compacted = ctx.text
    compacted_tokens = count(compacted)
    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,
    )