mambo_power.io.csv_bundle

The CSV bundle: manifest.json plus one CSV per entity table, a bit-exact re-spelling of the native schema for spreadsheet tooling. load(dump(net)) == net on every fixture; a bundle that is not exact is refused with a named ImportReport error. See File formats › CSV bundle.

mambo_power.io.csv_bundle

CSV bundle: one directory, manifest.json plus one CSV per entity table (wave M8, W5).

The bundle is a machine-facing, bit-exact spelling of the native schema (spec A5). It exists so a Network can be inspected and edited with spreadsheet tooling and read back identically: load(dump(net)) == net and every NetworkArrays matrix is array_equal (AC-5).

Layout

manifest.json carries what is not tabular: {"format": "mambo-power-csv", "schema_version": 1, "base_mva": ..., "tables": {file: row count, ...}}. schema_version is Network.schema_version, the native format's own version — the bundle has no version of its own because it is a re-spelling of that schema, not a second one.

One CSV per entity list, headed by the model's field names verbatim and in field order (TABLES lists them): buses.csv, branches.csv, generators.csv, loads.csv, shunts.csv, storage.csv, zones.csv. The three nested fields are flattened in place:

• Bus.geo → geo_lat, geo_lon (both empty ⇔ None);
• Generator.cost → cost_kind, cost_startup, cost_shutdown (all empty ⇔ None) plus the long-format side table generator_costs.csv (generator_id, index, p_mw, value): one row per polynomial coefficient (p_mw empty, value = coefficient, highest order first) or per piecewise breakpoint (p_mw, value = cost), index = 0-based position;
• Load.bid → bid_kind plus load_bids.csv (load_id, index, p_mw, value), the same shape.

Long format was chosen over a JSON cell because it is the spreadsheet-friendly one (research §4): a coefficient is a cell, not a substring, and a breakpoint is a row.

Cell rules

• Empty cell ⇔ None. Consequently an optional string field (area, zone, Zone.name) cannot carry ""; dump raises ValueError rather than write a bundle that would read back differently. Required string fields (ids, bus references) round-trip "" fine — an empty cell there is the empty string.
• Ids and every other string are written and read as text; nothing is ever passed to int(), so "01" and "1" stay distinct.
• Floats are written with repr (shortest round-trip form) and read with float; nan/inf are rejected on read, as the model rejects them.
• Booleans are written true/false; true/false/1/0 are accepted on read in any case.
• Empty tables are written header-only, so the manifest's table set never varies.
• Row order is list order and is preserved. A fully blank row is not a row (an editor's trailing newline does not change the count the manifest states), and a table is read as utf-8-sig so a UTF-8 BOM (Excel's "CSV UTF-8") does not become part of the first header; the writer emits plain UTF-8 with no BOM and no blank rows.

Reading

load_with_report validates the whole bundle and collects every problem before giving up: each is an ImportIssue with one of the CODES (all errors — a bundle is either exact or refused; there is nothing to repair). Errors are raised as ReportError, whose .report carries them. Cross-entity invariants (dangling references, slack count, connectivity) are the model's own and surface as NetworkValidationError exactly as for the native format.

FORMAT module-attribute

FORMAT = 'mambo-power-csv'

The format string in manifest.json.

SCHEMA_VERSION module-attribute

SCHEMA_VERSION: int = Network.model_fields[
    "schema_version"
].default

The schema version the bundle spells: Network.schema_version.

TABLES module-attribute

TABLES: tuple[str, ...] = (
    "buses.csv",
    "branches.csv",
    "generators.csv",
    "generator_costs.csv",
    "loads.csv",
    "load_bids.csv",
    "shunts.csv",
    "storage.csv",
    "zones.csv",
)

Every file a bundle carries besides manifest.json, in manifest order.

CODES module-attribute

CODES: tuple[str, ...] = (
    "CSV_MANIFEST_INVALID",
    "CSV_SCHEMA_VERSION",
    "CSV_MISSING_TABLE",
    "CSV_UNKNOWN_COLUMN",
    "CSV_MISSING_COLUMN",
    "CSV_DUPLICATE_ID",
    "CSV_BAD_VALUE",
    "CSV_ORPHAN_ROW",
)

Every report code this module can emit (all as errors; see the module docstring).

dump

dump(net: Network, directory: str | PathLike[str]) -> None

Write net as a bundle into directory (created if absent; files overwritten).

Raises ValueError if an optional string field holds "" (see the module docstring's cell rules); nothing else about a valid Network can fail to serialise.

All-or-nothing: every table is rendered before anything is written, the files go into a fresh temporary sibling directory (.<name>.tmp-<random>, so nothing pre-existing is ever removed), and when directory already holds a bundle the two are swapped as directories -- the old one is renamed aside, the new one renamed into place, the old one then removed (foreign files in it -- a README, a notebook -- are carried over first). So an exception anywhere -- the "" refusal, a full disk, a table another program holds open (Windows refuses the first rename before anything has moved) -- leaves whatever bundle was there before byte-for-byte untouched, and nothing beside it (M8 critic findings 7, 20, 26). A directory that exists without a bundle in it (the working directory, say, which Windows cannot rename) is filled in place: there is nothing old to protect.

Raises NotADirectoryError, before anything is written, when directory names a file.

Source code in src/mambo_power/io/csv_bundle.py

def dump(net: Network, directory: str | PathLike[str]) -> None:
    """Write ``net`` as a bundle into ``directory`` (created if absent; files overwritten).

    Raises :class:`ValueError` if an optional string field holds ``""`` (see the module
    docstring's cell rules); nothing else about a valid :class:`Network` can fail to serialise.

    All-or-nothing: every table is rendered before anything is written, the files go into a
    fresh temporary sibling directory (``.<name>.tmp-<random>``, so nothing pre-existing is
    ever removed), and when ``directory`` already holds a bundle the two are swapped *as
    directories* -- the old one is renamed aside, the new one renamed into place, the old one
    then removed (foreign files in it -- a README, a notebook -- are carried over first). So an
    exception anywhere -- the ``""`` refusal, a full disk, a table another program holds open
    (Windows refuses the first rename before anything has moved) -- leaves whatever bundle was
    there before byte-for-byte untouched, and nothing beside it (M8 critic findings 7, 20, 26).
    A ``directory`` that exists without a bundle in it (the working directory, say, which
    Windows cannot rename) is filled in place: there is nothing old to protect.

    Raises :class:`NotADirectoryError`, before anything is written, when ``directory`` names a
    file.
    """
    target = Path(directory).resolve()
    if target.exists() and not target.is_dir():
        raise NotADirectoryError(f"{directory!s} is a file, not a bundle directory")
    rendered = _render(net)
    manifest = {
        "format": FORMAT,
        "schema_version": net.schema_version,
        "base_mva": net.base_mva,
        "tables": {file: len(rendered[file][1]) for file in TABLES},
    }
    bundle_files = (*TABLES, _MANIFEST)
    target.parent.mkdir(parents=True, exist_ok=True)
    staging = Path(tempfile.mkdtemp(prefix=f".{target.name}.tmp-", dir=target.parent))
    try:
        for file, (header, rows) in rendered.items():
            _write_csv(staging / file, header, rows)
        (staging / _MANIFEST).write_text(
            json.dumps(manifest, indent=2) + "\n", encoding="utf-8", newline="\n"
        )
        if not target.is_dir() or not any((target / file).exists() for file in bundle_files):
            target.mkdir(exist_ok=True)
            for file in bundle_files:
                os.replace(staging / file, target / file)
            return
        old = staging.with_name(staging.name.replace(".tmp-", ".old-", 1))
        os.rename(target, old)  # fails whole on Windows if a file inside is open
        try:
            os.rename(staging, target)
        except BaseException:
            os.rename(old, target)
            raise
        for entry in old.iterdir():  # foreign files survive; bundle files are replaced
            if entry.name not in bundle_files:
                os.rename(entry, target / entry.name)
        _remove_tree(old)
    finally:
        if staging.exists():
            _remove_tree(staging)

load_with_report

load_with_report(
    directory: str | PathLike[str],
) -> tuple[Network, ImportReport]

Read a bundle. The report is always empty on success (a bundle is exact or refused).

Raises ReportError carrying every CODES issue found when the bundle cannot be read; NetworkValidationError when the tables read cleanly but the network they describe breaks a cross-entity invariant.

Source code in src/mambo_power/io/csv_bundle.py

def load_with_report(directory: str | PathLike[str]) -> tuple[Network, ImportReport]:
    """Read a bundle. The report is always empty on success (a bundle is exact or refused).

    Raises :class:`~mambo_power.io.report.ReportError` carrying every :data:`CODES` issue found
    when the bundle cannot be read; :class:`~mambo_power.model.NetworkValidationError` when the
    tables read cleanly but the network they describe breaks a cross-entity invariant.
    """
    reader = _Reader(Path(directory))
    manifest = reader.manifest()
    if manifest is None:
        _raise(reader.issues)
    counts = manifest["tables"]
    lists: dict[str, list[BaseModel] | None] = {
        table.attr: reader.entities(table, counts) for table in _ENTITY_TABLES
    }
    if reader.issues:
        _raise(reader.issues)
    net = Network(
        schema_version=manifest["schema_version"],
        base_mva=manifest["base_mva"],
        **{attr: entities for attr, entities in lists.items()},  # type: ignore[arg-type]
    )
    return net, ImportReport()

load

load(directory: str | PathLike[str]) -> Network

Read a bundle; load_with_report without the (always empty) report.

Source code in src/mambo_power/io/csv_bundle.py

def load(directory: str | PathLike[str]) -> Network:
    """Read a bundle; :func:`load_with_report` without the (always empty) report."""
    return load_with_report(directory)[0]