Use pyspssio as the sole SPSS adapter

Author: TonisOrmisson

src/openstatspec/api.py

@@ -19,9 +19,9 @@ def capability_matrix() -> Mapping[str, Any]:
             "print_format": "supported", "write_format": "unobservable",
             "source_encoding": "preserved only when UTF-8",
             "measurement_level": "supported", "user_missing_rules": "supported",
-            "documents": "supported", "multiple_response_sets": "lossy",
-            "variable_alignment": "lossy", "variable_sets": "unobservable",
-            "custom_attributes": "unobservable", "variable_role": "unobservable",
+            "documents": "unobservable", "multiple_response_sets": "supported",
+            "variable_alignment": "supported", "variable_sets": "fail-closed",
+            "custom_attributes": "supported", "variable_role": "supported",
         },
         "sql_profiles": declared_profiles(),
     }

src/openstatspec/core/results.py

@@ -9,9 +9,13 @@
 class Diagnostic:
     code: str
     detail: str
+    details: Mapping[str, Any] = field(default_factory=dict)
 
-    def as_dict(self) -> dict[str, str]:
-        return {"code": self.code, "detail": self.detail}
+    def as_dict(self) -> dict[str, Any]:
+        result: dict[str, Any] = {"code": self.code, "detail": self.detail}
+        if self.details:
+            result["details"] = dict(self.details)
+        return result
 
 
 @dataclass(frozen=True)
@@ -36,5 +40,8 @@ def __len__(self) -> int:
 
 def result(values: Mapping[str, Any]) -> OperationResult:
     payload = dict(values)
-    diagnostics = tuple(Diagnostic(**item) for item in payload.pop("loss_report", ()))
+    diagnostics = tuple(
+        Diagnostic(code=item["code"], detail=item["detail"], details=item.get("details", {}))
+        for item in payload.pop("loss_report", ())
+    )
     return OperationResult(payload, diagnostics)

src/openstatspec/spss/sav.py

@@ -9,7 +9,7 @@
 from collections.abc import Iterable, Mapping
 from typing import Any
 
-from sqlalchemy import delete, BigInteger, Boolean, Column, Float, Integer, MetaData, String, Table, Text, create_engine, insert, select, update
+from sqlalchemy import delete, BigInteger, Boolean, Column, Float, Integer, MetaData, String, Table, Text, create_engine, insert, inspect, select, text, update
 from sqlalchemy.dialects import mysql, postgresql, sqlite
 from .profiles import preflight, validate_connection_url
 
@@ -48,6 +48,8 @@ def catalog(metadata: MetaData) -> tuple[Table, Table, Table, Table]:
         Column("case_count", BigInteger, nullable=False),
         Column("file_label", Text, nullable=False, default=""),
         Column("documents", Text, nullable=False, default="[]"),
+        Column("file_attributes", Text, nullable=False, default="{}"),
+        Column("case_weight_variable", String(255)),
         Column("multiple_response_sets", Text, nullable=False, default="{}"),
     )
     variables = Table(
@@ -62,8 +64,11 @@ def catalog(metadata: MetaData) -> tuple[Table, Table, Table, Table]:
         Column("label", Text, nullable=False, default=""),
         Column("format", String(64)),
         Column("measure", String(32)),
+        Column("role", String(32)),
         Column("alignment", String(32)),
         Column("display_width", Integer),
+        Column("attributes", Text, nullable=False, default="{}"),
+        Column("compat_name", String(255)),
         Column("value_labels", Text, nullable=False, default="{}"),
         Column("missing_ranges", Text, nullable=False, default="[]"),
     )
@@ -96,6 +101,38 @@ def catalog(metadata: MetaData) -> tuple[Table, Table, Table, Table]:
 def _now() -> str:
     return datetime.now(UTC).isoformat()
 
+def _migrate_catalog_columns(connection: Any, datasets: Table, variables: Table) -> None:
+    """Add pyspssio metadata columns to catalogs created by earlier adapters.
+
+    This additive migration is intentionally small and portable: no existing
+    source data or dictionary row is rewritten, while a later import can store
+    the newly observable metadata alongside it.
+    """
+    inspector = inspect(connection)
+    additions = {
+        datasets.name: {
+            "file_attributes": "TEXT NOT NULL DEFAULT '{}'",
+            "case_weight_variable": "VARCHAR(255)",
+        },
+        variables.name: {
+            "role": "VARCHAR(32)",
+            "attributes": "TEXT NOT NULL DEFAULT '{}'",
+            "compat_name": "VARCHAR(255)",
+        },
+    }
+    preparer = connection.dialect.identifier_preparer
+    for table_name, columns in additions.items():
+        if not inspector.has_table(table_name):
+            continue
+        existing = {column["name"] for column in inspector.get_columns(table_name)}
+        for name, declaration in columns.items():
+            if name not in existing:
+                connection.execute(text(
+                    f"ALTER TABLE {preparer.quote(table_name)} ADD COLUMN "
+                    f"{preparer.quote(name)} {declaration}"
+                ))
+
+
 
 def _event_rows(
     *, operation_id: str, dataset_id: str | None, direction: str,
@@ -304,6 +341,7 @@ def create_wide_dataset(
     *, database_url: str, dataset_id: str, source_name: str, source_format: str,
     rows: Iterable[Mapping[str, Any]], variables: list[dict[str, Any]], file_label: str = "",
     source_encoding: str | None = None, documents: str = "[]",
+    file_attributes: str = "{}", case_weight_variable: str | None = None,
     source_table_name: str | None = None,
     source_sha256: str = "",
     source_created_at: str | None = None, source_modified_at: str | None = None,
@@ -337,6 +375,7 @@ def create_wide_dataset(
     )
     with engine.begin() as connection:
         metadata.create_all(connection, tables=[datasets, variable_catalog, multiple_response_catalog, documents_catalog, value_labels_catalog, missing_rules_catalog, fidelity_event_catalog, operation_catalog])
+        _migrate_catalog_columns(connection, datasets, variable_catalog)
         connection.execute(insert(operation_catalog).values(
             operation_id=operation_id, direction="import", status="running", dataset_id=dataset_id,
             source=source_name, created_at=_now(), details=json.dumps({"variable_count": len(variables)}, sort_keys=True),
@@ -355,6 +394,7 @@ def create_wide_dataset(
             source_created_at=source_created_at, source_modified_at=source_modified_at,
             imported_at=imported_at,
             file_label=file_label, documents=documents,
+            file_attributes=file_attributes, case_weight_variable=case_weight_variable,
             multiple_response_sets=multiple_response_sets,
         ))
         connection.execute(insert(variable_catalog), [dict(dataset_id=dataset_id, **item) for item in variables])
@@ -411,7 +451,8 @@ def read_wide_dataset(*, database_url: str, dataset_id: str) -> tuple[dict[str,
     datasets, variable_catalog, _, _ = catalog(metadata)
     documents_catalog, value_labels_catalog, missing_rules_catalog = normalized_metadata_tables(metadata)
     with engine.begin() as connection:
-        metadata.create_all(connection, tables=[documents_catalog, value_labels_catalog, missing_rules_catalog])
+        metadata.create_all(connection, tables=[datasets, variable_catalog, documents_catalog, value_labels_catalog, missing_rules_catalog])
+        _migrate_catalog_columns(connection, datasets, variable_catalog)
         dataset = dict(connection.execute(select(datasets).where(datasets.c.dataset_id == dataset_id)).mappings().one())
         data_table = Table(dataset["data_table"], MetaData(), autoload_with=connection)
         variables = [dict(item) for item in connection.execute(

src/openstatspec/sql/wide.py

@@ -1,29 +1,16 @@
-"""Reusable fixtures and semantic comparators for supported SAV/ZSAV round trips.
-
-These helpers deliberately compare only metadata that the Python adapter and
-its current SAV writer expose. Attributes, variable sets, and
-multiple-response sets are not asserted here because they are explicitly
-reported as unsupported fidelity losses.
-"""
+"""Reusable pyspssio fixtures and semantic comparators for SAV/ZSAV round trips."""
 
 import math
-import sys
 from pathlib import Path
 from typing import Any
 
 import pandas as pd
-import pyreadstat
+import pyspssio
 from pandas.testing import assert_frame_equal
 
 
 def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any]:
-    """Write a fixture spanning every currently supported dictionary semantic.
-
-    -sys.float_info.max and sys.float_info.max are how ReadStat represents
-    SPSS LOWEST and HIGHEST missing-range endpoints on write. ReadStat exposes
-    them as NaN and +inf when the SAV/ZSAV file is read back; the comparator
-    canonicalizes those markers.
-    """
+    """Write the metadata pyspssio demonstrably exposes and restores."""
     destination = Path(destination)
     long_text = "Õ🙂漢字" * 90
     frame = pd.DataFrame({
@@ -33,92 +20,63 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any]
         "status": ["NA", "DK", "ok", ""],
         "comment": [long_text, "", "näide", long_text],
         "interview_date": [23123.0, 23124.0, 23125.0, 23126.0],
+        "resp_a": [1.0, 0.0, 1.0, 0.0],
+        "resp_b": [0.0, 1.0, 1.0, 0.0],
     })
-    pyreadstat.write_sav(
-        frame,
-        destination,
-        compress=destination.suffix.lower() == ".zsav",
-        file_label="OpenStatSpec supported-semantics fixture",
-        column_labels={
-            "lowest_range": "LOWEST-style missing range",
-            "highest_range": "HIGHEST-style missing range",
-            "code": "Ordered numeric code",
-            "status": "String missing code",
-            "comment": "Long UTF-8 comment",
-            "interview_date": "SPSS numeric date",
+    metadata = {
+        "var_types": {"status": 8, "comment": 1024},
+        "var_formats": {
+            "lowest_range": "F12.1", "highest_range": "F12.1", "code": "F8.0",
+            "status": "A8", "comment": "A1024", "interview_date": "DATE11",
+        },
+        "var_labels": {
+            "lowest_range": "LOWEST-style missing range", "highest_range": "HIGHEST-style missing range",
+            "code": "Ordered numeric code", "status": "String missing code",
+            "comment": "Long UTF-8 comment", "interview_date": "SPSS numeric date",
         },
-        variable_value_labels={
-            # Deliberately non-sorted insertion order is part of the fixture.
+        "var_value_labels": {
             "code": {3.0: "third", 1.0: "first", 2.0: "second"},
             "status": {"DK": "don't know", "NA": "not answered", "ok": "valid"},
         },
-        missing_ranges={
-            "lowest_range": [{"lo": -sys.float_info.max, "hi": -1.0}, 7.0],
-            "highest_range": [{"lo": 1.0, "hi": sys.float_info.max}, -99.0],
-            "status": ["NA", "DK"],
-        },
-        variable_measure={
-            "lowest_range": "scale",
-            "highest_range": "scale",
-            "code": "ordinal",
-            "status": "nominal",
-            "comment": "nominal",
-            "interview_date": "scale",
-        },
-        variable_display_width={
-            "lowest_range": 12,
-            "highest_range": 12,
-            "code": 8,
-            "status": 12,
-            "comment": 48,
-            "interview_date": 11,
+        "var_missing_values": {
+            "lowest_range": {"lo": -sys_float_max(), "hi": -1.0, "values": [7.0]},
+            "highest_range": {"lo": 1.0, "hi": sys_float_max(), "values": [-99.0]},
+            "status": {"values": ["NA", "DK"]},
         },
-        variable_format={
-            "lowest_range": "F12.1",
-            "highest_range": "F12.1",
-            "code": "F8.0",
-            "status": "A8",
-            "comment": "A1024",
-            "interview_date": "DATE11",
+        "var_measure_levels": {
+            "lowest_range": "scale", "highest_range": "scale", "code": "ordinal",
+            "status": "nominal", "comment": "nominal", "interview_date": "scale",
         },
-        note=["First ordered document", "Teine dokument: Õ🙂"],
-    )
+        "var_alignments": {column: "left" for column in frame.columns},
+        "var_column_widths": {"lowest_range": 12, "highest_range": 12, "code": 8, "status": 12, "comment": 48, "interview_date": 11},
+        "var_roles": {"code": "target", "status": "input"},
+        "var_attributes": {"code": {"Origin": "fixture"}},
+        "mrsets": {"$responses": {"label": "Responses", "counted_value": 1, "variable_list": ["resp_a", "resp_b"]}},
+        "file_attributes": {"Fixture": "OpenStatSpec"},
+        "case_weight_var": "code",
+    }
+    pyspssio.write_sav(str(destination), frame, metadata=metadata)
     return {"long_text": long_text}
 
 
-def _canonical_missing_endpoint(value: Any, *, endpoint: str) -> Any:
-    """Normalize ReadStat's public sentinels for SPSS LOWEST/HIGHEST."""
-    if isinstance(value, float):
-        if math.isnan(value):
-            return "<LOWEST>" if endpoint == "lo" else "<SYSTEM-NAN>"
-        if math.isinf(value):
-            return "<HIGHEST>" if value > 0 else "<LOWEST>"
-    return value
+def sys_float_max() -> float:
+    return float.fromhex("0x1.fffffffffffffp+1023")
 
 
-def _canonical_missing_ranges(metadata: Any) -> dict[str, tuple[tuple[Any, Any], ...]]:
-    result: dict[str, tuple[tuple[Any, Any], ...]] = {}
-    for variable, rules in (getattr(metadata, "missing_ranges", {}) or {}).items():
-        result[variable] = tuple(
-            (
-                _canonical_missing_endpoint(rule.get("lo"), endpoint="lo"),
-                _canonical_missing_endpoint(rule.get("hi"), endpoint="hi"),
-            )
-            for rule in rules
-        )
+def _canonical_missing(metadata: dict[str, Any]) -> dict[str, tuple[tuple[Any, Any, tuple[Any, ...]], ...]]:
+    result: dict[str, tuple[tuple[Any, Any, tuple[Any, ...]], ...]] = {}
+    for name, rule in (metadata.get("var_missing_values") or {}).items():
+        rule = rule or {}
+        if "lo" in rule or "hi" in rule:
+            result[name] = ((rule.get("lo"), rule.get("hi"), tuple(rule.get("values") or ())),)
+        else:
+            result[name] = ((None, None, tuple(rule.get("values") or ())),)
     return result
 
 
-def _ordered_value_labels(metadata: Any) -> dict[str, tuple[tuple[Any, str], ...]]:
-    return {
-        variable: tuple((value, str(label)) for value, label in labels.items())
-        for variable, labels in (getattr(metadata, "variable_value_labels", {}) or {}).items()
-    }
-
-
 def compare_sav_semantics(source: str | Path, exported: str | Path) -> dict[str, Any]:
-    source_frame, source_meta = pyreadstat.read_sav(source, user_missing=True, disable_datetime_conversion=True)
-    exported_frame, exported_meta = pyreadstat.read_sav(exported, user_missing=True, disable_datetime_conversion=True)
+    source_frame, source_metadata = pyspssio.read_sav(str(source), convert_datetimes=False, include_user_missing=True)
+    exported_frame, exported_metadata = pyspssio.read_sav(str(exported), convert_datetimes=False, include_user_missing=True)
     failures: list[str] = []
     if list(source_frame.columns) != list(exported_frame.columns):
         failures.append("variable-order")
@@ -127,19 +85,12 @@ def compare_sav_semantics(source: str | Path, exported: str | Path) -> dict[str,
     except AssertionError:
         failures.append("values-or-case-order")
     for attribute in (
-        "column_labels",
-        "original_variable_types",
-        "variable_measure",
-        "variable_display_width",
-        "variable_storage_width",
-        "notes",
-        "file_label",
-        "file_encoding",
+        "encoding", "case_weight_var", "file_attributes", "mrsets", "var_types", "var_formats",
+        "var_labels", "var_alignments", "var_column_widths", "var_measure_levels", "var_roles",
+        "var_value_labels", "var_attributes",
     ):
-        if getattr(source_meta, attribute, None) != getattr(exported_meta, attribute, None):
+        if source_metadata.get(attribute) != exported_metadata.get(attribute):
             failures.append(attribute)
-    if _canonical_missing_ranges(source_meta) != _canonical_missing_ranges(exported_meta):
-        failures.append("missing_ranges")
-    if _ordered_value_labels(source_meta) != _ordered_value_labels(exported_meta):
-        failures.append("variable_value_labels")
-    return {"equivalent": not failures, "differences": failures}
+    if _canonical_missing(source_metadata) != _canonical_missing(exported_metadata):
+        failures.append("var_missing_values")
+    return {"equivalent": not failures, "differences": failures}

tests/conformance.py

@@ -2,33 +2,35 @@
 
 import openstatspec.cli
 import pandas as pd
-import pyreadstat
+import pyspssio
 
 
 def test_cli_import_inspect_validate_and_export_emit_json(tmp_path, capsys) -> None:
     source = tmp_path / "fixture.sav"
     database = f"sqlite:///{tmp_path / 'dataset.sqlite'}"
     output = tmp_path / "output.zsav"
-    pyreadstat.write_sav(pd.DataFrame({"answer": [1.0]}), source)
+    pyspssio.write_sav(str(source), pd.DataFrame({"answer": [1.0]}))
 
     assert openstatspec.cli.main(["import", str(source), "--database-url", database, "--dataset-id", "fixture"]) == 0
     imported = json.loads(capsys.readouterr().out)
-    preview = openstatspec.cli.main(["inspect", str(source)])
-    assert preview == 0
+    assert openstatspec.cli.main(["inspect", str(source)]) == 0
     inspected = json.loads(capsys.readouterr().out)
     assert inspected["source_format"] == "SAV"
     assert inspected["source_sha256"]
-    assert inspected["loss_report"][0]["code"] == "unobservable-source-dictionary-features"
+    assert {event["code"] for event in inspected["loss_report"]} == {
+        "file-label-and-documents-unobservable", "separate-write-format-unobservable",
+    }
     assert imported["case_count"] == 1
+
     from openstatspec.core.results import OperationResult
     assert isinstance(openstatspec.import_sav(source, database_url=f"sqlite:///{tmp_path / 'typed.sqlite'}", dataset_id="typed"), OperationResult)
 
     assert openstatspec.cli.main(["validate", "--database-url", database, "--dataset-id", "fixture"]) == 0
     assert json.loads(capsys.readouterr().out)["valid"] is True
 
-    assert openstatspec.cli.main(["export", "--database-url", database, "--dataset-id", "fixture", "--output", str(output), "--allow-loss", "unobservable-source-dictionary-features"]) == 0
+    assert openstatspec.cli.main([
+        "export", "--database-url", database, "--dataset-id", "fixture", "--output", str(output),
+        "--allow-loss", "file-label-and-documents-unobservable", "--allow-loss", "separate-write-format-unobservable",
+    ]) == 0
     assert json.loads(capsys.readouterr().out)["destination"] == str(output)
-    assert pyreadstat.read_sav(output)[0]["answer"].tolist() == [1.0]
-
-    assert openstatspec.cli.main(["inspect", str(source)]) == 0
-    assert json.loads(capsys.readouterr().out)["variable_count"] == 1
+    assert pyspssio.read_sav(str(output), convert_datetimes=False)[0]["answer"].tolist() == [1.0]