diff --git a/sdata/sclass/lca.py b/sdata/sclass/lca.py new file mode 100644 index 0000000..61c245a --- /dev/null +++ b/sdata/sclass/lca.py @@ -0,0 +1,486 @@ +# -*- coding: utf-8 -*- +"""Kanonisches LCA-Systemformat: Tabellen-Schemas + :class:`LCASystem`-Container. + +sdata pflegt hier die *kanonischen* Tabellen-Schemas eines matrixbasierten +LCA-Systems (Heijungs & Suh 2002): die sechs Teiltabellen ``flows``, +``processes``, ``elementary_flows``, ``exchanges_a``, ``exchanges_b`` und +``uncertainty`` samt der Verteilungs-Parametrisierungs-Spaltenliste. Konsumenten +(z. B. lepus-lca) pinnen dieses Schema und konvertieren an ihrer Datengrenze zu +numpy/scipy — sdata selbst rechnet nicht, sondern hält Format und Integrität. + +**Matrix-Konvention (Format-Vertrag).** Die Vorzeichen der Austauschkoeffizienten +tragen die Semantik: **Outputs positiv (+), Inputs negativ (−)**. Es gibt im +nativen Format *keine* ``flip``-/``negative``-Flags — deren Übersetzung wandert in +eine spätere bw_processing-Interop-Brücke (sdata-RFC 0015). Ein Gut ist Output, +wo sein Koeffizient positiv ist; ein Abfall ist Behandlungs-Output, wo sein +Koeffizient negativ ist. + +**Zeilenordnung ist normativ.** Die Reihenfolge der Zeilen in ``flows`` / +``processes`` / ``elementary_flows`` *ist* die Index-/Matrixordnung des +Rechenkerns (RFC-002 D7: keine hash-abhängige Iteration). Sie ist damit +identitätsstiftend und geht nachweislich in :meth:`LCASystem.content_checksum` +ein: eine Zeilen-Permutation liefert eine **andere** Content-Prüfsumme, obwohl +die Zeilenmenge unverändert bleibt (siehe Tests). + +Die ``uncertainty``-Tabelle ist ein Format-Vorgriff **ohne Arithmetik**: sie +parametrisiert Zellen-Unschärfe *explizit* — getrennte Spaltensätze je Lesart +(z. B. ``lognormal_underlying_mu``/``lognormal_underlying_sigma`` **oder** +``lognormal_geometric_gmean``/``lognormal_geometric_gsd``), statt der vier +verwechselbaren ``stats_arrays``-Parametrisierungen (Heijungs 2024, S. 915–922). +Konsumenten reichen die Datensätze unangetastet durch, bis sie sie brauchen. + +**Vorzeichen-Vertrag der ``uncertainty``-Tabelle.** Eine Verteilung parametrisiert +stets den **Betrag** ``|x|`` der adressierten Zelle — nie den vorzeichenbehafteten +Wert. Das Zellvorzeichen trägt allein der Nominaleintrag in ``exchanges_a`` / +``exchanges_b`` (Matrix-Konvention: Output +, Input −); der Konsument prägt es beim +Rechnen als ``sign(nominal)·d`` auf (``d`` = Ziehung). Das ist zwingend, weil rein +positive Familien (Lognormal) sonst eine **negative** Zelle — etwa einen Input — +still ins Positive kippen würden. Für positive Zellen ist ``sign = +1`` (kein +Unterschied). Praktisch heißt das: für einen negativen Input mit lognormaler +Unschärfe füllt man ``lognormal_*`` mit den Parametern des **positiven Betrags**, +nicht des negativen Werts. + +**Ergebnis-Rückrichtung (RESULTS).** Neben dem *Eingabe*-System hält sdata das +*Ergebnis*-Format desselben Vertragsstils: :class:`LCAResults` mit den Teiltabellen +``results`` (Kennzahlen je Zielgröße), ``draws`` (die ``g``-Draws als lange Tabelle) +und ``provenance`` (Seeds, Generator, Hashes). Es ist die Datei-Grenze, über die +ein Rechner (lepus-lca) unter Unsicherheit propagierte Ergebnisse an einen +Konsumenten (dp2g) zurückreicht — dieselbe Ordnungs-/Dialekt-Disziplin wie das +Eingabe-System. Die statistisch korrekte Benennung ist normativ (Heijungs 2024 +Tab. 14.2): ``interpercentile_lower``/``_upper`` sind **Interperzentil**-Grenzen +(Streuung, kein Konfidenzintervall). +""" +import hashlib +import os +from typing import Any, Dict, List, Optional + +from sdata.schema import AttrSpec, TableSchema, ValidationReport +from sdata.sclass.dataframe import DataFrame +from sdata.sclass.dataframegroup import DataFrameGroup + +__all__ = [ + "CSV_DIALECT", + "ELEMENTARY_FLOWS_SCHEMA", + "EXCHANGES_A_SCHEMA", + "EXCHANGES_B_SCHEMA", + "FLOWS_SCHEMA", + "FLOW_KINDS", + "FLOW_KIND_GOOD", + "FLOW_KIND_WASTE", + "LCAResults", + "LCASystem", + "LCATableGroup", + "LCA_SCHEMAS", + "LCA_TABLE_ORDER", + "MATRICES", + "MATRIX_A", + "MATRIX_B", + "PROCESSES_SCHEMA", + "RESULTS_DRAWS_SCHEMA", + "RESULTS_PROVENANCE_SCHEMA", + "RESULTS_SCHEMAS", + "RESULTS_SUMMARY_SCHEMA", + "RESULTS_TABLE_ORDER", + "UNCERTAINTY_DIST_COLUMNS", + "UNCERTAINTY_SCHEMA", +] + +#: **CSV-Dialekt-Vertrag** von :meth:`LCASystem.to_csv_dir` / :meth:`from_csv_dir`. +#: Feldtrenner ``","``, Dezimalpunkt ``"."``, Kodierung UTF-8. NaN-Politik: eine +#: leere Zelle *ist* der fehlende Wert (``na_rep=""`` beim Schreiben, leere Zelle +#: → NaN beim Lesen). Der Dialekt ist **gepinnt** und nicht überschreibbar: die +#: Ordnungs- und :meth:`content_checksum`-Garantien gelten ausschliesslich für +#: genau diesen Dialekt. Fremde Dialekte (z. B. ``;``-getrennt, Dezimalkomma) +#: werden **nicht** still korrekt gelesen — sie fallen als falsche Spaltenform +#: auf, statt lautlos fehlzuinterpretieren. +CSV_DIALECT: Dict[str, str] = {"sep": ",", "decimal": ".", "encoding": "utf-8"} + +# --- Konvention (Vertrag) --------------------------------------------------- +#: Gut/Abfall-Label eines ökonomischen Flusses (Definition 6/7, S. 90). +FLOW_KIND_GOOD = "GOOD" +FLOW_KIND_WASTE = "WASTE" +FLOW_KINDS = [FLOW_KIND_GOOD, FLOW_KIND_WASTE] + +#: Bezeichner der beiden Systemmatrizen in ``uncertainty.matrix``. +MATRIX_A = "A" +MATRIX_B = "B" +MATRICES = [MATRIX_A, MATRIX_B] + +#: Explizite Verteilungs-Parametrisierungs-Spalten der ``uncertainty``-Tabelle. +#: Getrennte Spaltensätze je Lesart — nie eine mehrdeutige Sammelspalte. Alle +#: sind optional; je Zeile werden genau die zur ``dist``-Familie passenden +#: Spalten gefüllt (der Konsument liest die zu seiner Lesart gehörigen). +#: +#: **Betrags-Vertrag:** die Parameter beschreiben den **Betrag** ``|x|`` der Zelle; +#: das Vorzeichen liefert der Nominaleintrag in ``exchanges_a``/``exchanges_b`` +#: (``sign(nominal)``, Anwendung ``sign(nominal)·Ziehung``). Für einen negativen +#: Input trägt man also die Parameter des positiven Betrags ein. +UNCERTAINTY_DIST_COLUMNS: List[AttrSpec] = [ + # Normal + AttrSpec("normal_mean", dtype="float", description="Normal: Mittelwert μ"), + AttrSpec("normal_sd", dtype="float", description="Normal: Standardabweichung σ"), + # Lognormal — Lesart 1: Parameter der zugrundeliegenden Normalverteilung + AttrSpec("lognormal_underlying_mu", dtype="float", + description="Lognormal: μ der zugrundeliegenden Normalverteilung (log-Raum)"), + AttrSpec("lognormal_underlying_sigma", dtype="float", + description="Lognormal: σ der zugrundeliegenden Normalverteilung (log-Raum)"), + # Lognormal — Lesart 2: geometrische Kenngrößen im Datenraum + AttrSpec("lognormal_geometric_gmean", dtype="float", + description="Lognormal: geometrischer Mittelwert (Datenraum)"), + AttrSpec("lognormal_geometric_gsd", dtype="float", + description="Lognormal: geometrische Standardabweichung (Datenraum)"), + # Uniform + AttrSpec("uniform_min", dtype="float", description="Uniform: untere Grenze"), + AttrSpec("uniform_max", dtype="float", description="Uniform: obere Grenze"), + # Triangular + AttrSpec("triangular_min", dtype="float", description="Triangular: untere Grenze"), + AttrSpec("triangular_mode", dtype="float", description="Triangular: Modus"), + AttrSpec("triangular_max", dtype="float", description="Triangular: obere Grenze"), +] + +# --- Tabellen-Schemas (Pflichtspalten: required=True) ----------------------- +FLOWS_SCHEMA = TableSchema("flows", [ + AttrSpec("id", dtype="str", required=True, + description="eindeutige Fluss-ID (Zeilenordnung = Zeilen von A)"), + AttrSpec("unit", dtype="str", required=True, description="Einheit des Flusses"), + AttrSpec("kind", dtype="str", required=True, allowed=FLOW_KINDS, + description="GOOD (Gut) | WASTE (Abfall) — Cut-off/Multifunktionalität"), + AttrSpec("name", dtype="str", description="lesbarer Name"), + AttrSpec("region", dtype="str", description="Region (optional)"), + AttrSpec("time_slice", dtype="str", description="Zeitscheibe (optional)"), +]) + +PROCESSES_SCHEMA = TableSchema("processes", [ + AttrSpec("id", dtype="str", required=True, + description="eindeutige Prozess-ID (Spaltenordnung = Spalten von A/B)"), + AttrSpec("name", dtype="str", description="lesbarer Name"), + AttrSpec("reference_output", dtype="str", + description="Fluss-ID des Referenz-/Funktionsoutputs (optional)"), +]) + +ELEMENTARY_FLOWS_SCHEMA = TableSchema("elementary_flows", [ + AttrSpec("id", dtype="str", required=True, + description="eindeutige Elementarfluss-ID (Zeilenordnung = Zeilen von B)"), + AttrSpec("unit", dtype="str", required=True, description="Einheit"), + AttrSpec("compartment", dtype="str", required=True, + description="Kompartiment (z. B. to air / from ground)"), + AttrSpec("name", dtype="str", description="lesbarer Name"), +]) + +EXCHANGES_A_SCHEMA = TableSchema("exchanges_a", [ + AttrSpec("row_id", dtype="str", required=True, description="Fluss-ID (Zeile von A)"), + AttrSpec("col_id", dtype="str", required=True, description="Prozess-ID (Spalte von A)"), + AttrSpec("value", dtype="float", required=True, + description="Koeffizient — Output +, Input − (COO; Duplikat (i,j) verboten)"), +]) + +EXCHANGES_B_SCHEMA = TableSchema("exchanges_b", [ + AttrSpec("row_id", dtype="str", required=True, + description="Elementarfluss-ID (Zeile von B)"), + AttrSpec("col_id", dtype="str", required=True, description="Prozess-ID (Spalte von B)"), + AttrSpec("value", dtype="float", required=True, + description="Koeffizient — Emission +, Ressourcenentnahme − (COO)"), +]) + +UNCERTAINTY_SCHEMA = TableSchema("uncertainty", [ + AttrSpec("matrix", dtype="str", required=True, allowed=MATRICES, + description="A | B — welche Systemmatrix die Zelle trägt"), + AttrSpec("row_id", dtype="str", required=True, description="Zeilen-Fluss-ID der Zelle"), + AttrSpec("col_id", dtype="str", required=True, description="Spalten-Prozess-ID der Zelle"), + AttrSpec("dist", dtype="str", required=True, + description="Verteilungsfamilie (normal | lognormal | uniform | triangular | …); " + "parametrisiert den Betrag |x|, Vorzeichen aus dem Nominaleintrag"), + *UNCERTAINTY_DIST_COLUMNS, +]) + +#: Kanonische Reihenfolge der Teiltabellen (bestimmt die Prüfsummen-Reihenfolge). +LCA_TABLE_ORDER = ( + "flows", "processes", "elementary_flows", + "exchanges_a", "exchanges_b", "uncertainty", +) + +#: Tabellenname → :class:`~sdata.schema.TableSchema`. +LCA_SCHEMAS: Dict[str, TableSchema] = { + "flows": FLOWS_SCHEMA, + "processes": PROCESSES_SCHEMA, + "elementary_flows": ELEMENTARY_FLOWS_SCHEMA, + "exchanges_a": EXCHANGES_A_SCHEMA, + "exchanges_b": EXCHANGES_B_SCHEMA, + "uncertainty": UNCERTAINTY_SCHEMA, +} + +# --- Ergebnis-Schemas (RESULTS, Rückrichtung) ------------------------------- +#: Kennzahlen **je Zielgröße**. Die Benennung ist normativ (Heijungs 2024 +#: Tab. 14.2, L6): ``nominal`` (deterministischer Wert, **nicht** „true value"), +#: ``mean``/``variance`` (MC-Momente) und das **Interperzentil**-Band +#: ``interpercentile_lower``/``_upper`` an den Stufen ``lower_percentile``/ +#: ``upper_percentile`` (Streuung — **kein** Konfidenzintervall). ``n_failed`` +#: (exakt singuläre Draws) und ``n_guarded`` (Fast-Singulär-Wächter) sind getrennt. +RESULTS_SUMMARY_SCHEMA = TableSchema("results", [ + AttrSpec("target", dtype="str", required=True, + description="Zielgröße-ID (Elementarfluss / Impact-Kategorie)"), + AttrSpec("nominal", dtype="float", required=True, + description="deterministischer Wert am nominalen System (nicht der true value)"), + AttrSpec("mean", dtype="float", required=True, description="MC-Mittelwert (Welford)"), + AttrSpec("variance", dtype="float", required=True, description="MC-Varianz (Welford)"), + AttrSpec("interpercentile_lower", dtype="float", required=True, + description="untere Interperzentil-Grenze (Streuung, kein CI)"), + AttrSpec("interpercentile_upper", dtype="float", required=True, + description="obere Interperzentil-Grenze (Streuung, kein CI)"), + AttrSpec("lower_percentile", dtype="float", required=True, + description="untere Perzentil-Stufe (z. B. 0.025)"), + AttrSpec("upper_percentile", dtype="float", required=True, + description="obere Perzentil-Stufe (z. B. 0.975)"), + AttrSpec("n", dtype="int", required=True, description="angeforderte Draw-Zahl"), + AttrSpec("n_failed", dtype="int", required=True, + description="exakt singuläre Draws (verworfen)"), + AttrSpec("n_guarded", dtype="int", required=True, + description="fast-singuläre Draws (Wächter, verworfen)"), +]) + +#: Die ``g``-Draws als **lange** Tabelle (eine Zeile je (Zielgröße, Draw-Index)); +#: optionale Teiltabelle (Quantil-Rekonstruktion / Nachrechnen). +RESULTS_DRAWS_SCHEMA = TableSchema("draws", [ + AttrSpec("target", dtype="str", required=True, description="Zielgröße-ID"), + AttrSpec("draw", dtype="int", required=True, description="Draw-Index (0-basiert)"), + AttrSpec("value", dtype="float", required=True, description="Ziel-g dieses Draws"), +]) + +#: Provenienz als **Schlüssel/Wert**-Tabelle (Seeds, Generator, Hashes, Warnung). +RESULTS_PROVENANCE_SCHEMA = TableSchema("provenance", [ + AttrSpec("key", dtype="str", required=True, + description="Provenienz-Schlüssel (seed | generator | content_hash | …)"), + AttrSpec("value", dtype="str", required=True, description="Provenienz-Wert"), +]) + +#: Kanonische Reihenfolge der Ergebnis-Teiltabellen (Prüfsummen-Reihenfolge). +RESULTS_TABLE_ORDER = ("results", "draws", "provenance") + +#: Ergebnis-Tabellenname → :class:`~sdata.schema.TableSchema`. +RESULTS_SCHEMAS: Dict[str, TableSchema] = { + "results": RESULTS_SUMMARY_SCHEMA, + "draws": RESULTS_DRAWS_SCHEMA, + "provenance": RESULTS_PROVENANCE_SCHEMA, +} + + +class LCATableGroup(DataFrameGroup): + """Gemeinsamer Container-Vertrag für die LCA-Tabellengruppen (System + Ergebnis). + + Trägt die schema-gebundene Ablage (:meth:`set_table`), die deterministische + :meth:`content_checksum` (Zeilen-/Spaltenordnung normativ), die + Schema-Validierung (:meth:`validate_tables`/:meth:`is_valid`) und den + **gepinnten** CSV-Roundtrip (:meth:`to_csv_dir`/:meth:`from_csv_dir`, + :data:`CSV_DIALECT`). Die konkreten Klassen setzen nur ``TABLE_ORDER`` und + ``SCHEMAS`` — die Methoden arbeiten generisch darüber. + """ + + #: Kanonische Tabellenreihenfolge (bestimmt die Prüfsummen-Reihenfolge). + TABLE_ORDER: tuple = () + #: Tabellenname → :class:`TableSchema`. + SCHEMAS: Dict[str, TableSchema] = {} + + # ------------------------------------------------------------- public API + def set_table(self, key: str, df: Any, *, overwrite: bool = True) -> DataFrame: + """Lege die Teiltabelle ``key`` unter ihrem Schema an (oder überschreibe sie). + + :param key: einer der kanonischen Namen (``TABLE_ORDER``). + :param df: ein :class:`DataFrame` oder ein pandas ``DataFrame``. + :param overwrite: bestehende Tabelle ersetzen (Default True). + :return: die gespeicherte :class:`DataFrame`. + :raises ValueError: bei unbekanntem ``key``. + """ + if key not in self.SCHEMAS: + raise ValueError( + f"unbekannte Tabelle {key!r}; erlaubt: {sorted(self.SCHEMAS)}") + sdf = df if isinstance(df, DataFrame) else DataFrame(df=df, name=key) + self.SCHEMAS[key].apply(sdf) # Spalten-Metadaten füllen + self.add(sdf, key=key, overwrite=overwrite) + return sdf + + def table(self, key: str) -> Optional[DataFrame]: + """Die gespeicherte :class:`DataFrame` der Teiltabelle ``key`` (oder ``None``).""" + return self.get(key) + + def content_checksum(self) -> str: + """SHA-256 über **alle** Teiltabellen — die Content-Identität der Gruppe. + + Gehasht wird je vorhandener Tabelle (in ``TABLE_ORDER``) der Tabellenname + plus die kanonische CSV-Form der Daten (:attr:`DataFrame.content_bytes` — + Header + Werte in Zeilenordnung, ohne Index, UTF-8). Damit gehen + **Zeilenordnung und Spaltenordnung** in die Prüfsumme ein: eine + Zeilen-Permutation ergibt eine andere Prüfsumme, obwohl die Zeilenmenge + identisch bleibt (Zeilenordnung ist normativ). + + Reine Datenprüfsumme (keine Metadaten): stabil, wenn sich nur + Annotationen ändern — konsistent mit RFC 0004 (``DataFrame.content_bytes``). + + :return: Hex-Digest (64 Zeichen). + """ + h = hashlib.sha256() + for key in self.TABLE_ORDER: + member = self.get(key) + if member is None: + continue + h.update(key.encode("utf-8")) + h.update(b"\x00") + h.update(member.content_bytes) + h.update(b"\x00") + return h.hexdigest() + + def validate_tables(self, *, only_present: bool = True) -> Dict[str, ValidationReport]: + """Validiere jede Teiltabelle gegen ihr :class:`TableSchema`. + + Reicht die :class:`~sdata.schema.TableSchema`-Validierung durch (Spalten, + dtypes, Einheiten-Annotation), rechnet ``ok`` aber gegen die **Pflicht**- + spalten (``required``): eine fehlende Pflichtspalte macht die Tabelle + ungültig, eine fehlende *optionale* Spalte bleibt gültig. Absente Spalten + werden weiterhin in ``missing`` gelistet (informativ). + + :param only_present: nur die tatsächlich vorhandenen Tabellen prüfen + (Default). Mit ``False`` liefern fehlende Tabellen einen Report, dessen + Pflichtspalten alle als ``missing`` gemeldet werden. + :return: ``{tabellenname: ValidationReport}`` (Report ist truthy, wenn ok). + """ + reports: Dict[str, ValidationReport] = {} + for key, schema in self.SCHEMAS.items(): + member = self.get(key) + required = {c.name for c in schema.columns if c.required} + if member is None: + if only_present: + continue + reports[key] = ValidationReport(ok=False, missing=sorted(required)) + continue + report = schema.validate(member) + missing_required = [m for m in report.missing if m in required] + report.ok = not (missing_required or report.type_errors or report.unit_errors) + reports[key] = report + return reports + + def is_valid(self, **kwargs: Any) -> bool: + """``True``, wenn alle geprüften Teiltabellen ihr Schema erfüllen.""" + return all(self.validate_tables(**kwargs).values()) + + # ------------------------------------------------------------ CSV-Roundtrip + def to_csv_dir(self, path: str) -> List[str]: + """Schreibe je Teiltabelle eine ``.csv`` nach ``path`` (Daten, kein Index). + + Ordnung (Zeilen und Spalten) und Werte bleiben erhalten; die qualifizierende + Spalten-Semantik lebt im Schema, nicht in der CSV. + + **Dialekt-Vertrag (:data:`CSV_DIALECT`).** Geschrieben wird fest mit + Feldtrenner ``","``, Dezimalpunkt ``"."`` und UTF-8; NaN als leere Zelle + (``na_rep=""``). Die Roundtrip- und :meth:`content_checksum`-Garantien + gelten **nur** für diesen Dialekt (siehe :data:`CSV_DIALECT`); er ist + bewusst nicht konfigurierbar, damit die Prüfsumme stabil bleibt. + + :param path: Zielverzeichnis (wird bei Bedarf angelegt). + :return: Liste der geschriebenen Dateipfade (in ``TABLE_ORDER``). + """ + os.makedirs(path, exist_ok=True) + written: List[str] = [] + for key in self.TABLE_ORDER: + member = self.get(key) + if member is None: + continue + filepath = os.path.join(path, f"{key}.csv") + member.df.to_csv( + filepath, index=False, na_rep="", + sep=CSV_DIALECT["sep"], decimal=CSV_DIALECT["decimal"], + encoding=CSV_DIALECT["encoding"]) + written.append(filepath) + return written + + @classmethod + def from_csv_dir(cls, path: str, *, name: str = "lca_table_group") -> "LCATableGroup": + """Lese ein per :meth:`to_csv_dir` geschriebenes Verzeichnis zurück. + + Es werden genau die kanonischen ``.csv`` (``TABLE_ORDER``) geladen, + die vorhanden sind; jede wird unter ihrem Schema abgelegt. + + **Dialekt-Vertrag (:data:`CSV_DIALECT`).** Gelesen wird fest mit + Feldtrenner ``","``, Dezimalpunkt ``"."`` und UTF-8 — passend zu + :meth:`to_csv_dir`. Der Dialekt ist **nicht** überschreibbar (kein + ``**read_csv_kwargs``-Durchgriff): eine Datei in fremdem Dialekt (z. B. + ``;``-getrennt oder mit Dezimalkomma) wird dadurch nicht still korrekt + gelesen, sondern fällt als falsche Spaltenform auf — der Vertrag schützt + die Ordnungs-/Prüfsummen-Garantie vor lautlosem Fehllesen. + + :param path: Quellverzeichnis. + :param name: Name des rekonstruierten Objekts. + :return: eine Instanz der aufrufenden Klasse. + """ + import pandas as pd + group = cls(name=name) + for key in cls.TABLE_ORDER: + filepath = os.path.join(path, f"{key}.csv") + if os.path.exists(filepath): + # float_precision="round_trip": der Default-Parser von read_csv + # verliert bei vollpräzisen float64 Stellen (pandas ≥ 3), sodass die + # content_checksum nach dem Roundtrip nicht mehr stimmte. Der + # round_trip-Parser liest das kürzeste round-trippable repr **exakt** + # zurück — Teil des Dialekt-Vertrags, damit die Prüfsummen-Garantie + # auch für MC-Ergebnisfloats (LCAResults) hält. + group.set_table(key, pd.read_csv( + filepath, sep=CSV_DIALECT["sep"], decimal=CSV_DIALECT["decimal"], + encoding=CSV_DIALECT["encoding"], float_precision="round_trip")) + return group + + +class LCASystem(LCATableGroup): + """Container eines matrixbasierten LCA-Systems (sechs benannte Teiltabellen). + + Erbt von :class:`LCATableGroup` (schema-gebundene Ablage, deterministische + :meth:`content_checksum`, gepinnter CSV-Roundtrip); die Teiltabellen liegen + als vollwertige :class:`~sdata.sclass.dataframe.DataFrame` (mit + Spalten-Metadaten) unter ihren kanonischen Schlüsseln (:data:`LCA_TABLE_ORDER`). + Serialisierung (``to_dict``/``from_dict``, ``to_json``/``from_json``) und die + volle DataFrameGroup-API bleiben erhalten. + + * :meth:`set_table` legt eine Tabelle unter Schema an; + * :meth:`content_checksum` bildet eine deterministische Prüfsumme über alle + Teiltabellen (Zeilen-Permutation ⇒ andere Prüfsumme); + * :meth:`validate_tables` prüft jede vorhandene Tabelle gegen ihr Schema; + * :meth:`to_csv_dir` / :meth:`from_csv_dir` sind ein verlustfreier + CSV-Roundtrip mit **gepinntem** Dialekt (:data:`CSV_DIALECT`). + """ + + SDATA_CLS = "sdata.sclass.lca.LCASystem" + + #: Kanonische Tabellenreihenfolge (bestimmt die Prüfsummen-Reihenfolge). + TABLE_ORDER = LCA_TABLE_ORDER + #: Tabellenname → :class:`TableSchema`. + SCHEMAS = LCA_SCHEMAS + + @classmethod + def from_csv_dir(cls, path: str, *, name: str = "lca_system") -> "LCASystem": + """Lese ein per :meth:`to_csv_dir` geschriebenes System-Verzeichnis zurück.""" + return super().from_csv_dir(path, name=name) # type: ignore[return-value] + + +class LCAResults(LCATableGroup): + """Container eines LCA-**Ergebnisses** unter Unsicherheit (Rückrichtung). + + Die Datei-Grenze, über die ein Rechner (lepus-lca) MC-propagierte Ergebnisse + an einen Konsumenten (dp2g) zurückreicht — derselbe Vertragsstil wie + :class:`LCASystem` (Ordnung normativ, :data:`CSV_DIALECT` gepinnt, + deterministische :meth:`content_checksum`). Die drei Teiltabellen + (:data:`RESULTS_TABLE_ORDER`): + + * ``results`` — Kennzahlen je Zielgröße (nominal, Mittel, Varianz, + Interperzentil-Band, ``n``/``n_failed``/``n_guarded``); + * ``draws`` — die ``g``-Draws als lange Tabelle (optional; Quantil-Nachrechnung); + * ``provenance`` — Schlüssel/Wert (Seeds, Generator, Hashes, Warnung). + """ + + SDATA_CLS = "sdata.sclass.lca.LCAResults" + + #: Kanonische Ergebnis-Tabellenreihenfolge (Prüfsummen-Reihenfolge). + TABLE_ORDER = RESULTS_TABLE_ORDER + #: Ergebnis-Tabellenname → :class:`TableSchema`. + SCHEMAS = RESULTS_SCHEMAS + + @classmethod + def from_csv_dir(cls, path: str, *, name: str = "lca_results") -> "LCAResults": + """Lese ein per :meth:`to_csv_dir` geschriebenes Ergebnis-Verzeichnis zurück.""" + return super().from_csv_dir(path, name=name) # type: ignore[return-value] diff --git a/tests/test_sclass_lca.py b/tests/test_sclass_lca.py new file mode 100644 index 0000000..805babe --- /dev/null +++ b/tests/test_sclass_lca.py @@ -0,0 +1,375 @@ +# -*- coding: utf-8 -*- +"""Kanonisches LCA-Systemformat: Schemas + :class:`LCASystem`-Container. + +Prüft den Format-Vertrag aus ``sdata/sclass/lca.py``: + +* die sechs Tabellen-Schemas tragen genau die spezifizierten Spalten (inkl. der + getrennten Verteilungs-Parametrisierungs-Spalten der ``uncertainty``-Tabelle); +* die Zeilenordnung ist normativ — eine Zeilen-Permutation ergibt eine **andere** + :meth:`LCASystem.content_checksum` (die Prüfsumme identifiziert die Quelle); +* Roundtrip über JSON (dict) und CSV ist verlustfrei (Ordnung und Werte); +* die Schema-Validierung schlägt bei einer fehlenden **Pflicht**spalte an, nicht + bei einer fehlenden optionalen Spalte. +""" +import pandas as pd +import pytest + +from sdata.sclass.lca import ( + CSV_DIALECT, + ELEMENTARY_FLOWS_SCHEMA, + EXCHANGES_A_SCHEMA, + EXCHANGES_B_SCHEMA, + FLOW_KINDS, + FLOWS_SCHEMA, + LCA_SCHEMAS, + LCA_TABLE_ORDER, + MATRICES, + PROCESSES_SCHEMA, + RESULTS_DRAWS_SCHEMA, + RESULTS_PROVENANCE_SCHEMA, + RESULTS_SCHEMAS, + RESULTS_SUMMARY_SCHEMA, + RESULTS_TABLE_ORDER, + UNCERTAINTY_SCHEMA, + LCAResults, + LCASystem, +) + + +def _flows(): + return pd.DataFrame({ + "id": ["glass", "copper", "electricity"], + "unit": ["kg", "kg", "MJ"], + "kind": ["GOOD", "GOOD", "GOOD"], + "name": ["Glass", "Copper", "Electricity"], + "region": ["GLO", "GLO", "GLO"], + "time_slice": ["2024", "2024", "2024"], + }) + + +def _processes(): + return pd.DataFrame({ + "id": ["prod_glass", "prod_copper"], + "name": ["production of glass", "production of copper"], + "reference_output": ["glass", "copper"], + }) + + +def _exchanges_a(): + return pd.DataFrame({ + "row_id": ["glass", "copper", "electricity"], + "col_id": ["prod_glass", "prod_copper", "prod_glass"], + "value": [1000.0, 100.0, -10.0], # Output +, Input − + }) + + +def _system(): + s = LCASystem(name="demo") + s.set_table("flows", _flows()) + s.set_table("processes", _processes()) + s.set_table("exchanges_a", _exchanges_a()) + return s + + +# --------------------------------------------------------------- Schema-Layout +def test_schema_columns_match_layout(): + """Die sechs Schemas tragen exakt die Spalten des Plan-Layouts.""" + assert [c.name for c in FLOWS_SCHEMA.columns] == \ + ["id", "unit", "kind", "name", "region", "time_slice"] + assert [c.name for c in PROCESSES_SCHEMA.columns] == \ + ["id", "name", "reference_output"] + assert [c.name for c in ELEMENTARY_FLOWS_SCHEMA.columns] == \ + ["id", "unit", "compartment", "name"] + assert [c.name for c in EXCHANGES_A_SCHEMA.columns] == ["row_id", "col_id", "value"] + assert [c.name for c in EXCHANGES_B_SCHEMA.columns] == ["row_id", "col_id", "value"] + assert list(LCA_TABLE_ORDER) == list(LCA_SCHEMAS.keys()) + + +def test_flow_kind_and_matrix_vocab(): + assert FLOW_KINDS == ["GOOD", "WASTE"] + assert MATRICES == ["A", "B"] + kind = FLOWS_SCHEMA._by_name["kind"] + assert kind.required and kind.allowed == ["GOOD", "WASTE"] + + +def test_required_columns_are_the_mandatory_ones(): + def req(sch): + return [c.name for c in sch.columns if c.required] + assert req(FLOWS_SCHEMA) == ["id", "unit", "kind"] + assert req(PROCESSES_SCHEMA) == ["id"] # name/reference_output optional + assert req(ELEMENTARY_FLOWS_SCHEMA) == ["id", "unit", "compartment"] + assert req(EXCHANGES_A_SCHEMA) == ["row_id", "col_id", "value"] + assert req(EXCHANGES_B_SCHEMA) == ["row_id", "col_id", "value"] + + +def test_uncertainty_has_explicit_separate_parametrizations(): + """Getrennte Spaltensätze je Lognormal-Lesart — nie eine Sammelspalte.""" + cols = {c.name for c in UNCERTAINTY_SCHEMA.columns} + assert {"matrix", "row_id", "col_id", "dist"} <= cols + # Lognormal, beide Lesarten getrennt parametrisiert + assert {"lognormal_underlying_mu", "lognormal_underlying_sigma"} <= cols + assert {"lognormal_geometric_gmean", "lognormal_geometric_gsd"} <= cols + # weitere Familien + assert {"normal_mean", "normal_sd"} <= cols + assert {"uniform_min", "uniform_max"} <= cols + m = UNCERTAINTY_SCHEMA._by_name["matrix"] + assert m.required and m.allowed == ["A", "B"] + + +# --------------------------------------------------- Zeilenordnung ist normativ +def test_row_permutation_changes_content_checksum(): + """Eine Zeilen-Permutation ⇒ **andere** Content-Prüfsumme (Ordnung normativ).""" + s = _system() + before = s.content_checksum() + permuted = _flows().iloc[[2, 0, 1]].reset_index(drop=True) # gleiche Menge, andere Ordnung + s.set_table("flows", permuted) + after = s.content_checksum() + assert before != after + + +def test_column_permutation_changes_content_checksum(): + """Auch die Spaltenordnung geht in die Prüfsumme ein.""" + s = _system() + before = s.content_checksum() + s.set_table("flows", _flows()[["unit", "id", "kind", "name", "region", "time_slice"]]) + assert s.content_checksum() != before + + +def test_content_checksum_is_stable_and_data_only(): + """Gleiche Daten ⇒ gleiche Prüfsumme (objektunabhängig); Metadaten zählen nicht.""" + assert _system().content_checksum() == _system().content_checksum() + assert len(_system().content_checksum()) == 64 + s = _system() + before = s.content_checksum() + s.table("flows").set_column("id", unit="-", label="annotiert") # nur Metadaten + assert s.content_checksum() == before + + +def test_content_checksum_depends_on_all_subtables(): + s = _system() + before = s.content_checksum() + s.set_table("exchanges_a", _exchanges_a().iloc[[2, 1, 0]].reset_index(drop=True)) + assert s.content_checksum() != before + + +# ------------------------------------------------------------- JSON-Roundtrip +def test_json_roundtrip_lossless_order_and_values(): + pytest.importorskip("pyarrow") # dict/JSON bettet Parquet ein + s = _system() + back = LCASystem.from_dict(s.to_dict()) + assert back.content_checksum() == s.content_checksum() + for key in ("flows", "processes", "exchanges_a"): + assert list(back.table(key).df.columns) == list(s.table(key).df.columns) + assert back.table(key).content_bytes == s.table(key).content_bytes + + +def test_json_string_roundtrip(): + pytest.importorskip("pyarrow") + s = _system() + back = LCASystem.from_json(s.to_json()) + assert back.content_checksum() == s.content_checksum() + + +# -------------------------------------------------------------- CSV-Roundtrip +def test_csv_dir_roundtrip_lossless_order_and_values(tmp_path): + s = _system() + written = s.to_csv_dir(str(tmp_path)) + assert [p.split("/")[-1] for p in written] == \ + ["flows.csv", "processes.csv", "exchanges_a.csv"] + back = LCASystem.from_csv_dir(str(tmp_path)) + assert back.content_checksum() == s.content_checksum() + for key in ("flows", "processes", "exchanges_a"): + assert list(back.table(key).df.columns) == list(s.table(key).df.columns) + assert back.table(key).content_bytes == s.table(key).content_bytes + + +# ----------------------------------------------------- CSV-Dialekt als Vertrag +def test_csv_dialect_contract_is_comma_dot_utf8(): + """Der gepinnte Dialekt ist Teil des öffentlichen Vertrags.""" + assert CSV_DIALECT == {"sep": ",", "decimal": ".", "encoding": "utf-8"} + + +def test_from_csv_dir_does_not_accept_dialect_override(): + """Keine Hintertür: ``sep``/``decimal``/``encoding`` sind nicht überschreibbar + (``**read_csv_kwargs`` entfernt) — der Aufruf mit solchen kwargs schlägt fehl.""" + with pytest.raises(TypeError): + LCASystem.from_csv_dir("irgendwo", sep=";") + + +def test_foreign_dialect_is_not_silently_misread(tmp_path): + """Eine Datei in fremdem Dialekt (``;``-getrennt, Dezimalkomma) wird **nicht** + still korrekt gelesen: unter dem gepinnten Komma-Dialekt landet die ganze + Zeile in EINER Spalte — die Vertragsspalten fehlen, statt lautlos falsch zu + erscheinen (kein stilles Fehllesen).""" + (tmp_path / "exchanges_a.csv").write_text( + "row_id;col_id;value\nglass;prod_glass;1000,0\n", encoding="utf-8") + back = LCASystem.from_csv_dir(str(tmp_path)) + cols = list(back.table("exchanges_a").df.columns) + assert cols != ["row_id", "col_id", "value"] # nicht still in die Vertragsform gelesen + assert len(cols) == 1 # alles in einer (falschen) Spalte + + +def test_csv_roundtrip_stable_under_pinned_dialect(tmp_path): + """Der gepinnte Dialekt hält den Roundtrip byte-/prüfsummenstabil (Regression).""" + s = _system() + s.to_csv_dir(str(tmp_path)) + assert LCASystem.from_csv_dir(str(tmp_path)).content_checksum() == s.content_checksum() + + +# ----------------------------------------------------------- Schema-Validierung +def test_validation_flags_missing_required_column(): + """Fehlende **Pflicht**spalte (``id``) ⇒ Tabelle ungültig.""" + s = LCASystem(name="x") + s.set_table("flows", _flows().drop(columns=["id"])) + rep = s.validate_tables()["flows"] + assert not rep.ok + assert "id" in rep.missing + + +def test_validation_tolerates_missing_optional_column(): + """Fehlende **optionale** Spalte (``region``) ⇒ Tabelle bleibt gültig.""" + s = LCASystem(name="x") + s.set_table("flows", _flows().drop(columns=["region"])) + rep = s.validate_tables()["flows"] + assert rep.ok + assert s.is_valid() + + +def test_full_system_is_valid(): + assert _system().is_valid() + + +def test_validate_tables_absent_with_only_present_false(): + s = LCASystem(name="x") # leeres System + reports = s.validate_tables(only_present=False) + assert set(reports) == set(LCA_SCHEMAS) + assert not reports["flows"].ok # Pflichtspalten fehlen komplett + assert reports == s.validate_tables(only_present=False) + + +# ---------------------------------------------------------------- set_table API +def test_set_table_rejects_unknown_key(): + s = LCASystem(name="x") + with pytest.raises(ValueError): + s.set_table("not_a_table", _flows()) + + +def test_set_table_applies_schema_metadata(): + s = LCASystem(name="x") + s.set_table("flows", _flows()) + # das Schema hat die Spalten-Metadaten vervollständigt (Beschreibung gesetzt) + assert s.table("flows").get_column("kind").description + + +# ====================================================================== +# RESULTS — Ergebnis-Rückrichtung (LCAResults), gleicher Vertragsstil +# ====================================================================== +def _results_summary(): + return pd.DataFrame({ + "target": ["CO2", "CH4"], + "nominal": [2.0, 0.1], + "mean": [2.13, 0.11], + "variance": [0.09, 0.001], + "interpercentile_lower": [1.71, 0.05], + "interpercentile_upper": [2.63, 0.18], + "lower_percentile": [0.025, 0.025], + "upper_percentile": [0.975, 0.975], + "n": [1000, 1000], + "n_failed": [0, 0], + "n_guarded": [3, 3], + }) + + +def _results_draws(): + return pd.DataFrame({ + "target": ["CO2", "CO2", "CH4", "CH4"], + "draw": [0, 1, 0, 1], + "value": [2.05, 1.98, 0.10, 0.12], + }) + + +def _results_provenance(): + return pd.DataFrame({ + "key": ["seed", "generator", "content_hash"], + "value": ["12345", "Philox", "abc123"], + }) + + +def _results(): + r = LCAResults(name="demo_results") + r.set_table("results", _results_summary()) + r.set_table("draws", _results_draws()) + r.set_table("provenance", _results_provenance()) + return r + + +def test_results_schema_columns_match_layout(): + """Die drei Ergebnis-Schemas tragen exakt die spezifizierten Spalten.""" + assert [c.name for c in RESULTS_SUMMARY_SCHEMA.columns] == [ + "target", "nominal", "mean", "variance", + "interpercentile_lower", "interpercentile_upper", + "lower_percentile", "upper_percentile", "n", "n_failed", "n_guarded"] + assert [c.name for c in RESULTS_DRAWS_SCHEMA.columns] == ["target", "draw", "value"] + assert [c.name for c in RESULTS_PROVENANCE_SCHEMA.columns] == ["key", "value"] + assert list(RESULTS_TABLE_ORDER) == list(RESULTS_SCHEMAS.keys()) + + +def test_results_naming_is_interpercentile_not_confidence(): + """L6-Disziplin: das Streuungsband heißt Interperzentil, nie Konfidenz.""" + cols = {c.name for c in RESULTS_SUMMARY_SCHEMA.columns} + assert {"interpercentile_lower", "interpercentile_upper"} <= cols + assert not any("confidence" in c.lower() for c in cols) + assert "nominal" in cols and "true_value" not in cols + + +def test_results_required_columns(): + def req(sch): + return [c.name for c in sch.columns if c.required] + # alle Kennzahl-Spalten sind Pflicht (das Ergebnis ist ohne sie unvollständig) + assert req(RESULTS_SUMMARY_SCHEMA) == [c.name for c in RESULTS_SUMMARY_SCHEMA.columns] + assert req(RESULTS_DRAWS_SCHEMA) == ["target", "draw", "value"] + assert req(RESULTS_PROVENANCE_SCHEMA) == ["key", "value"] + + +def test_results_full_group_is_valid(): + assert _results().is_valid() + + +def test_results_row_order_is_normative(): + """Zeilen-Permutation ⇒ andere Content-Prüfsumme (Ordnung normativ).""" + r = _results() + before = r.content_checksum() + r.set_table("results", _results_summary().iloc[[1, 0]].reset_index(drop=True)) + assert r.content_checksum() != before + + +def test_results_csv_dir_roundtrip_lossless(tmp_path): + r = _results() + written = r.to_csv_dir(str(tmp_path)) + assert [p.split("/")[-1] for p in written] == \ + ["results.csv", "draws.csv", "provenance.csv"] + back = LCAResults.from_csv_dir(str(tmp_path)) + assert back.content_checksum() == r.content_checksum() + for key in RESULTS_TABLE_ORDER: + assert back.table(key).content_bytes == r.table(key).content_bytes + + +def test_results_uses_pinned_dialect_no_override(): + with pytest.raises(TypeError): + LCAResults.from_csv_dir("irgendwo", sep=";") + + +def test_results_validation_flags_missing_required_column(): + r = LCAResults(name="x") + r.set_table("results", _results_summary().drop(columns=["variance"])) + rep = r.validate_tables()["results"] + assert not rep.ok + assert "variance" in rep.missing + + +def test_results_summary_only_is_valid_draws_optional(): + """Nur die Kennzahl-Tabelle (ohne draws/provenance) ist ein gültiges Ergebnis.""" + r = LCAResults(name="summary_only") + r.set_table("results", _results_summary()) + assert r.is_valid() + assert r.table("draws") is None