Adding a New Network Parser

JAFF's file parser (NetworkParser in src/jaff/core/parsers/network/_engine.py) auto-detects the format of an astrochemical network file and parses each reaction line into a common internal representation. Each supported format is a self-contained plugin: a NetworkFormat subclass that lives in its own subpackage under core/parsers/network/_formats/. Adding a new format means adding one subpackage — the engine and the existing formats are never touched.

How the Parser Works

NetworkParser discovers every registered format through all_formats() and sorts them by their declared priority (lower is matched first — not file or import order). For each non-blank line, it walks the formats in priority order; the first format whose _global_re matches wins, and its handle() method extracts the reaction data and appends it to the shared ParseContext.

flowchart TD
    A[NetworkParser.__init__] --> B[__set_known_replacments\nPre-populate SymPy aliases for\ncommon shorthand symbols]
    B --> C[all_formats\nImport format subpackages,\nsort instances by priority]
    C --> S[build_state\nMerge each format's default_state\ninto ParseContext.state]
    S --> D[__parse_file\nRead all lines into memory]
    D --> E{For each line}
    E --> F{Line empty or whitespace?}
    F -- yes --> E
    F -- no --> G[Iterate formats\nin priority order]
    G --> H{fmt._global_re.match line?}
    H -- no match,\ntry next format --> G
    G -- no format matched --> E
    H -- first match wins --> J[fmt.handle match, ctx]
    J --> K{fmt._local_re.match line?}
    K -- no match --> L[fmt._handle_errors\nctx.raise_error with\nline number + file path]
    K -- yes --> M[Extract named groups\nr · p · tmin · tmax · rate · string]
    M --> N[Normalize species names\nReplace format-specific symbols\ne.g. HE→He  user_crflux→crate]
    N --> O[Append parsedListProps dict\nto ctx.parsed_list]
    O --> E
    E -- all lines done --> P[__normalize_rates\nLowercase all rate strings]
    P --> Q[resolve_symbolic_dependencies\nSubstitute @var / VARIABLES globals]
    Q --> R[get_parsed returns\nparsed_list  +  globals dict]

The NetworkFormat contract

Every format subclasses NetworkFormat (_formats/_base.py) and implements:

_global_re / _local_re take ctx so the regex can depend on live parse state (KROME rebuilds its _local_re from the column counts a @format: header wrote). When a pattern is static, compile it once with @cache.

The Two-Level Regex Design

This split keeps the hot path (_global_re) cheap, while _local_re does the heavy structural matching and populates the named groups the handler reads. The handle method receives the global match (useful for error diagnostics) and recomputes the local match itself.

The Parsed Reaction Dict

Every handler appends a parsedListProps dict (defined in core/parsers/network/_typing/) to ctx.parsed_list, with exactly these keys:

Concluding the reaction type

The parser — not the rate expression — decides the reaction "type". Conclude it structurally so it survives custom/auxiliary rates, and inject the driving agent pseudo-species into the reactant list when the format implies one but the line omits it:

• A radiation-driven reaction carries _PHOTON as a reactant → "photo".
• A cosmic-ray reaction carries _CR / _CRP / _CRPHOT → "cosmic_ray".
• Three or more real (non-_) reactants → "3_body".
• Otherwise "unknown".

These special pseudo-species (leading _) give a reaction its identity and serialization but are excluded from the kinetics. See the existing formats' _reaction_type and agent-injection logic (e.g. kida/reaction.py, krome/reaction.py) for reference implementations.

After all lines are parsed, __normalize_rates lower-cases every "rate" string, and resolve_symbolic_dependencies substitutes any global variables (e.g. from @var or VARIABLES blocks) into the expressions.

Step-by-Step: Adding a New Format

1. Create the format subpackage

Add a folder under src/jaff/core/parsers/network/_formats/, e.g. my_format/, with a reaction.py module. Multi-line-type formats (like KROME's @format: header, @var:, and reaction lines) get one module per line type — see krome/ (header.py, var.py, reaction.py).

import re
from functools import cache

from .. import register
from .._base import NetworkFormat
from .._context import ParseContext


@register
class MyFormatReaction(NetworkFormat):
    """My pipe-delimited reaction line."""

    priority = 55
    name = "my_format"

    @cache
    def _global_re(self, ctx: ParseContext) -> re.Pattern:  # (1)
        return re.compile(r"^(?!\s*[!#@]).*\|.*$")

    @cache
    def _local_re(self, ctx: ParseContext) -> re.Pattern:   # (2)
        return re.compile(
            r"^\s*"
            r"(?P<reactants>[^|]+)\s*\|\s*"
            r"(?P<products>[^|]+)\s*\|\s*"
            r"(?P<tmin>[^|]*)\s*\|\s*"
            r"(?P<tmax>[^|]*)\s*\|\s*"
            r"(?P<rate>.*?)\s*$"
        )

    def handle(self, match: re.Match, ctx: ParseContext) -> None:  # (3)
        local = self._local_re(ctx).match(ctx.line)
        if not local:
            self._handle_errors(match, ctx)

        rr = [r.strip() for r in local.group("reactants").split("+") if r.strip()]
        pp = [p.strip() for p in local.group("products").split("+") if p.strip()]

        tmin_str = local.group("tmin").strip()
        tmax_str = local.group("tmax").strip()
        t_min = float(tmin_str) if tmin_str else None
        t_max = float(tmax_str) if tmax_str else None

        # Replace any format-specific symbols with JAFF canonical names
        rate = local.group("rate").strip().replace("my_crflux", "crate")

        # Conclude the reaction type structurally (inject the agent species
        # first if the format implies one). Three or more real reactants => 3-body.
        rtype = "3_body" if sum(not r.startswith("_") for r in rr) >= 3 else "unknown"

        ctx.parsed_list.append(
            {
                "r": rr,
                "p": pp,
                "tmin": t_min,
                "tmax": t_max,
                "rate": rate,
                "type": rtype,
                "string": ctx.line.strip(),
            }
        )

    def _handle_errors(self, match: re.Match, ctx: ParseContext) -> None:
        ctx.raise_error("Invalid MY_FORMAT reaction detected")

1. _global_re — match any non-comment line that contains |. Keep it broad and fast. @cache because it does not depend on parse state.
2. _local_re — use named groups (?P<name>) to capture every field. Named groups map directly to local.group("name") calls in your handler.
3. handle — receives the global match; recomputes the local match, extracts fields, and appends to ctx.parsed_list. Use ctx.raise_error, ctx.globals, ctx.logger, ctx.line, and ctx.nline instead of instance state — the engine owns no per-line state.

2. Export and register the format

Add the subpackage's __init__.py so the class is imported (which runs its @register decorator):

from .reaction import MyFormatReaction

__all__ = ["MyFormatReaction"]

Then add the subpackage to the import line inside all_formats() in _formats/__init__.py so registration is triggered:

def all_formats() -> list[NetworkFormat]:
    from . import kida, krome, my_format, prizmo, uclchem, udfa  # noqa: F401

    return sorted((cls() for cls in _REGISTRY), key=lambda fmt: fmt.priority)

That is the only shared file you edit — registration is by priority, not import order, so the position in this line does not matter.

3. (Optional) Share live state across line types

If your format has a header line that configures later reaction lines (like KROME's @format:), give both classes the same state_key and let the header seed it via default_state():

@register
class MyHeader(NetworkFormat):
    priority = 15
    name = "my_header"
    state_key = "my_format"           # shared namespace

    def default_state(self) -> dict:
        return {"ncols": 0}

    def handle(self, match, ctx):
        self.state(ctx)["ncols"] = ... # header writes shared state


@register
class MyFormatReaction(NetworkFormat):
    priority = 55
    name = "my_format"
    state_key = "my_format"           # same key → same dict

    def _local_re(self, ctx):
        ncols = self.state(ctx)["ncols"]  # reaction reads live state
        ...

build_state() merges every format's default_state() into ParseContext.state[state_key], and self.state(ctx) returns that live dict. A regex that reads state (like the reaction's _local_re above) must not be @cached — it has to recompile when the state changes.

Known Symbol Replacements

After all lines are parsed, __normalize_rates lowercases every rate string. The __set_known_replacments method in core/parsers/network/_engine.py pre-populates self.__globals with SymPy aliases for common shorthand symbols found in KROME/PRIZMO files:

If your format introduces additional shorthand symbols, add them to __set_known_replacments following the same pattern. Compound aliases (those that reference simpler ones) must be listed before the simpler aliases they depend on so that resolve_symbolic_dependencies substitutes correctly.

Checklist

• New subpackage under core/parsers/network/_formats/ with a @register-ed NetworkFormat subclass
• priority chosen so the format matches at the correct point relative to others
• _global_re is a fast filter; _local_re uses named groups for all fields (reactants, products, tmin, tmax, rate)
• Static regexes are @cached; any state-dependent _local_re is left uncached
• handle appends a valid parsedListProps dict (all seven keys, including "type") to ctx.parsed_list
• Reaction "type" concluded structurally; agent pseudo-species (_PHOTON/_CR) injected when the format implies one
• _handle_errors calls ctx.raise_error with a descriptive message
• Subpackage added to the from . import … line in _formats/__init__.py
• Format-specific symbols replaced with JAFF canonical names in the handler or via __set_known_replacments
• Tests added in tests/ with at least one valid reaction line and one malformed line

See Also

• Contributing Guide
• Code Style Guide
• Testing Guide