Search, analysis, and tables

thermoml_io.collection

Search, ranking, and multi-document aggregation for ThermoML data.

DatasetMatch dataclass

A ranked dataset returned by :meth:ThermoMLCollection.search.

observation_count counts only property values matching the requested property/category filters. It is therefore the ranking metric used before limit is applied.

Source code in src/thermoml_io/collection.py

@dataclass(frozen=True, slots=True)
class DatasetMatch:
    """A ranked dataset returned by :meth:`ThermoMLCollection.search`.

    ``observation_count`` counts only property values matching the requested
    property/category filters. It is therefore the ranking metric used before
    ``limit`` is applied.
    """

    document: ThermoMLDocument
    dataset: DataSet
    matching_property_numbers: tuple[int, ...]
    observation_count: int
    data_types: tuple[str, ...]
    matching_variable_numbers: tuple[int, ...] = ()

    @property
    def dataset_key(self) -> str:
        """Return the stable publication-scoped key of the matched dataset."""
        return self.document.dataset_key(self.dataset)

    @property
    def system_key(self) -> str:
        """Return the order-independent chemical system key."""
        return self.document.system_key(self.dataset)

dataset_key: str property

Return the stable publication-scoped key of the matched dataset.

system_key: str property

Return the order-independent chemical system key.

ThermoMLCollection dataclass

An immutable collection of independently sourced ThermoML documents.

Source code in src/thermoml_io/collection.py

@dataclass(frozen=True, slots=True)
class ThermoMLCollection:
    """An immutable collection of independently sourced ThermoML documents."""

    documents: tuple[ThermoMLDocument, ...]
    _component_index: ComponentIndex = field(init=False, repr=False, compare=False)

    def __post_init__(self) -> None:
        object.__setattr__(self, "_component_index", ComponentIndex.from_documents(self.documents))

    @property
    def component_index(self) -> ComponentIndex:
        """Return the collection-wide exact-alias component index."""
        return self._component_index

    def resolve_component(self, query: ComponentQuery) -> ComponentIdentity:
        """Resolve one friendly or namespaced component query.

        Raises
        ------
        ComponentNotFoundError
            If no compound in the collection reports the requested identity.
        AmbiguousComponentError
            If the query matches more than one chemically distinct identity.
        """
        return self._component_index.resolve(query)

    @classmethod
    def from_urls(
        cls,
        urls: tuple[str, ...] | list[str],
        *,
        timeout: float = 30.0,
        max_bytes: int = 100 * 1024 * 1024,
    ) -> ThermoMLCollection:
        """Download several ThermoML documents without persisting their bytes."""
        return cls(
            tuple(load_thermoml_url(url, timeout=timeout, max_bytes=max_bytes) for url in urls)
        )

    def search(
        self,
        *,
        components: (
            ComponentQuery | tuple[ComponentQuery, ...] | list[ComponentQuery] | None
        ) = None,
        required_components: (tuple[ComponentQuery, ...] | list[ComponentQuery] | None) = None,
        component_match: ComponentMatch = "contains",
        system: tuple[ComponentQuery, ...] | list[ComponentQuery] | None = None,
        system_match: SystemMatch = "exact",
        property_name: str | None = None,
        data_type: str | None = None,
        independent_variable: str | None = None,
        limit: int | None = None,
    ) -> tuple[DatasetMatch, ...]:
        """Search and rank experimental datasets.

        Parameters
        ----------
        components:
            One component or a list defining the component query. Queries match
            exact reported names, formulas, CAS numbers, InChI, or InChIKey,
            case-insensitively. Prefixes such as ``"formula:H2"``,
            ``"cas:1333-74-0"``, and ``"inchikey:..."`` restrict the
            identifier type. Friendly aliases are resolved across the complete
            collection before dataset matching.
        required_components:
            Mandatory subset of ``components`` when ``component_match`` is
            ``"within"``. Other listed components are optional, while unlisted
            components are rejected.
        component_match:
            ``"exact"`` requires exactly ``components``; ``"contains"``
            requires every listed component and allows extras; ``"within"``
            restricts systems to subsets of ``components`` and requires
            ``required_components``.
        system:
            Component identities describing a chemical system. By default the
            match is order-independent and exact.
        system_match:
            ``"exact"`` requires the complete system; ``"contains"`` allows
            additional components.
        property_name:
            Case-insensitive substring of the reported ThermoML property name.
        data_type:
            Package classification such as ``"VLE"``, ``"volumetric"``, or
            ``"transport"``. The original ThermoML property group is also
            accepted.
        independent_variable:
            Case-insensitive normalized substring of a reported ThermoML
            variable name, for example ``"pressure"`` or ``"temperature"``.
            Fixed constraints are retained as metadata but do not satisfy this
            filter.
        limit:
            Maximum number of datasets returned. Ranking by the number of
            matching property observations occurs before truncation.

        Returns
        -------
        tuple[DatasetMatch, ...]
            Matches sorted by descending matching observation count, followed
            by deterministic publication and dataset identifiers.

        Examples
        --------
        Search for the ten densest water/carbon-dioxide VLE datasets::

            collection.search(
                system=("H2O", "CO2"), data_type="VLE", limit=10
            )
        """
        if components is not None and system is not None:
            raise ValueError("Specify either components or system, not both.")
        if component_match not in {"exact", "contains", "within"}:
            raise ValueError("component_match must be 'exact', 'contains', or 'within'.")
        if system_match not in {"exact", "contains"}:
            raise ValueError("system_match must be 'exact' or 'contains'.")
        if limit is not None and limit < 0:
            raise ValueError("limit must be non-negative or None.")
        component_queries = (
            (components,)
            if isinstance(components, str | ComponentIdentity)
            else tuple(components or ())
        )
        system_queries = tuple(system or ())
        required_queries = tuple(required_components or ())
        if required_queries and component_match != "within":
            raise ValueError("required_components is only valid with component_match='within'.")
        if component_match == "within" and not component_queries:
            raise ValueError("component_match='within' requires components.")
        resolved_components = self._component_index.resolve_many(component_queries)
        resolved_system = self._component_index.resolve_many(system_queries)
        resolved_required = self._component_index.resolve_many(required_queries)
        queries = resolved_components or resolved_system
        component_identifiers = {item.stable_identifier for item in resolved_components}
        if any(item.stable_identifier not in component_identifiers for item in resolved_required):
            raise ValueError("required_components must be a subset of components.")
        exact = (system is not None and system_match == "exact") or (
            components is not None and component_match == "exact"
        )
        normalized_property = property_name.casefold() if property_name else None
        normalized_type = normalize_term(data_type) if data_type else None
        normalized_variable = normalize_term(independent_variable) if independent_variable else None

        matches: list[DatasetMatch] = []
        for document in self.documents:
            for dataset in document.datasets:
                compounds = document.system_compounds(dataset)
                identities = tuple(
                    ComponentIdentity.from_compound(compound) for compound in compounds
                )
                if component_match == "within" and components is not None:
                    if not all(
                        any(query.matches(identity) for query in resolved_components)
                        for identity in identities
                    ):
                        continue
                    if not all(
                        any(query.matches(identity) for identity in identities)
                        for query in resolved_required
                    ):
                        continue
                elif queries and not all(
                    any(query.matches(identity) for identity in identities) for query in queries
                ):
                    continue
                if exact and len(set(dataset.component_ids)) != len(queries):
                    continue

                selected_variables = tuple(
                    item
                    for item in dataset.variables
                    if normalized_variable and normalized_variable in normalize_term(item.name)
                )
                if independent_variable and not selected_variables:
                    continue

                selected: list[PropertyDefinition] = []
                categories: list[str] = []
                for property_definition in dataset.properties:
                    category = classify_property(property_definition, dataset)
                    if normalized_property and normalized_property not in (
                        property_definition.name.casefold()
                    ):
                        continue
                    if normalized_type and normalized_type not in {
                        normalize_term(category),
                        normalize_term(property_definition.group),
                    }:
                        continue
                    selected.append(property_definition)
                    categories.append(category)
                if (property_name or data_type) and not selected:
                    continue
                selected_numbers = tuple(item.number for item in selected)
                if not property_name and not data_type:
                    selected_numbers = tuple(item.number for item in dataset.properties)
                    categories = [classify_property(item, dataset) for item in dataset.properties]
                selected_variable_numbers = tuple(item.number for item in selected_variables)
                count = 0
                for point in dataset.points:
                    property_count = sum(
                        value.number in selected_numbers for value in point.property_values
                    )
                    if independent_variable:
                        variable_count = sum(
                            value.number in selected_variable_numbers
                            for value in point.variable_values
                        )
                        count += property_count * variable_count
                    else:
                        count += property_count
                if count == 0:
                    continue
                matches.append(
                    DatasetMatch(
                        document=document,
                        dataset=dataset,
                        matching_property_numbers=selected_numbers,
                        observation_count=count,
                        data_types=tuple(dict.fromkeys(categories)),
                        matching_variable_numbers=selected_variable_numbers,
                    )
                )
        matches.sort(
            key=lambda match: (
                -match.observation_count,
                match.document.citation.normalized_doi or match.document.provenance.sha256,
                match.dataset.number,
            )
        )
        if limit is not None:
            matches = matches[:limit]
        return tuple(matches)

    def all_matches(self) -> tuple[DatasetMatch, ...]:
        """Return every non-empty dataset ranked by observation count."""
        return self.search()

component_index: ComponentIndex property

Return the collection-wide exact-alias component index.

resolve_component(query: ComponentQuery) -> ComponentIdentity

Resolve one friendly or namespaced component query.

Raises:

Type	Description
ComponentNotFoundError	If no compound in the collection reports the requested identity.
AmbiguousComponentError	If the query matches more than one chemically distinct identity.

Source code in src/thermoml_io/collection.py

def resolve_component(self, query: ComponentQuery) -> ComponentIdentity:
    """Resolve one friendly or namespaced component query.

    Raises
    ------
    ComponentNotFoundError
        If no compound in the collection reports the requested identity.
    AmbiguousComponentError
        If the query matches more than one chemically distinct identity.
    """
    return self._component_index.resolve(query)

from_urls(urls: tuple[str, ...] | list[str], *, timeout: float = 30.0, max_bytes: int = 100 * 1024 * 1024) -> ThermoMLCollection classmethod

Download several ThermoML documents without persisting their bytes.

Source code in src/thermoml_io/collection.py

@classmethod
def from_urls(
    cls,
    urls: tuple[str, ...] | list[str],
    *,
    timeout: float = 30.0,
    max_bytes: int = 100 * 1024 * 1024,
) -> ThermoMLCollection:
    """Download several ThermoML documents without persisting their bytes."""
    return cls(
        tuple(load_thermoml_url(url, timeout=timeout, max_bytes=max_bytes) for url in urls)
    )

search(*, components: ComponentQuery | tuple[ComponentQuery, ...] | list[ComponentQuery] | None = None, required_components: tuple[ComponentQuery, ...] | list[ComponentQuery] | None = None, component_match: ComponentMatch = 'contains', system: tuple[ComponentQuery, ...] | list[ComponentQuery] | None = None, system_match: SystemMatch = 'exact', property_name: str | None = None, data_type: str | None = None, independent_variable: str | None = None, limit: int | None = None) -> tuple[DatasetMatch, ...]

Search and rank experimental datasets.

Parameters:

Name	Type	Description	Default
components	ComponentQuery | tuple[ComponentQuery, ...] | list[ComponentQuery] | None	One component or a list defining the component query. Queries match exact reported names, formulas, CAS numbers, InChI, or InChIKey, case-insensitively. Prefixes such as "formula:H2", "cas:1333-74-0", and "inchikey:..." restrict the identifier type. Friendly aliases are resolved across the complete collection before dataset matching.	None
required_components	tuple[ComponentQuery, ...] | list[ComponentQuery] | None	Mandatory subset of components when component_match is "within". Other listed components are optional, while unlisted components are rejected.	None
component_match	ComponentMatch	"exact" requires exactly components; "contains" requires every listed component and allows extras; "within" restricts systems to subsets of components and requires required_components.	'contains'
system	tuple[ComponentQuery, ...] | list[ComponentQuery] | None	Component identities describing a chemical system. By default the match is order-independent and exact.	None
system_match	SystemMatch	"exact" requires the complete system; "contains" allows additional components.	'exact'
property_name	str | None	Case-insensitive substring of the reported ThermoML property name.	None
data_type	str | None	Package classification such as "VLE", "volumetric", or "transport". The original ThermoML property group is also accepted.	None
independent_variable	str | None	Case-insensitive normalized substring of a reported ThermoML variable name, for example "pressure" or "temperature". Fixed constraints are retained as metadata but do not satisfy this filter.	None
limit	int | None	Maximum number of datasets returned. Ranking by the number of matching property observations occurs before truncation.	None

Returns:

Type	Description
tuple[DatasetMatch, ...]	Matches sorted by descending matching observation count, followed by deterministic publication and dataset identifiers.

Examples:

Search for the ten densest water/carbon-dioxide VLE datasets::

collection.search(
    system=("H2O", "CO2"), data_type="VLE", limit=10
)

Source code in src/thermoml_io/collection.py

def search(
    self,
    *,
    components: (
        ComponentQuery | tuple[ComponentQuery, ...] | list[ComponentQuery] | None
    ) = None,
    required_components: (tuple[ComponentQuery, ...] | list[ComponentQuery] | None) = None,
    component_match: ComponentMatch = "contains",
    system: tuple[ComponentQuery, ...] | list[ComponentQuery] | None = None,
    system_match: SystemMatch = "exact",
    property_name: str | None = None,
    data_type: str | None = None,
    independent_variable: str | None = None,
    limit: int | None = None,
) -> tuple[DatasetMatch, ...]:
    """Search and rank experimental datasets.

    Parameters
    ----------
    components:
        One component or a list defining the component query. Queries match
        exact reported names, formulas, CAS numbers, InChI, or InChIKey,
        case-insensitively. Prefixes such as ``"formula:H2"``,
        ``"cas:1333-74-0"``, and ``"inchikey:..."`` restrict the
        identifier type. Friendly aliases are resolved across the complete
        collection before dataset matching.
    required_components:
        Mandatory subset of ``components`` when ``component_match`` is
        ``"within"``. Other listed components are optional, while unlisted
        components are rejected.
    component_match:
        ``"exact"`` requires exactly ``components``; ``"contains"``
        requires every listed component and allows extras; ``"within"``
        restricts systems to subsets of ``components`` and requires
        ``required_components``.
    system:
        Component identities describing a chemical system. By default the
        match is order-independent and exact.
    system_match:
        ``"exact"`` requires the complete system; ``"contains"`` allows
        additional components.
    property_name:
        Case-insensitive substring of the reported ThermoML property name.
    data_type:
        Package classification such as ``"VLE"``, ``"volumetric"``, or
        ``"transport"``. The original ThermoML property group is also
        accepted.
    independent_variable:
        Case-insensitive normalized substring of a reported ThermoML
        variable name, for example ``"pressure"`` or ``"temperature"``.
        Fixed constraints are retained as metadata but do not satisfy this
        filter.
    limit:
        Maximum number of datasets returned. Ranking by the number of
        matching property observations occurs before truncation.

    Returns
    -------
    tuple[DatasetMatch, ...]
        Matches sorted by descending matching observation count, followed
        by deterministic publication and dataset identifiers.

    Examples
    --------
    Search for the ten densest water/carbon-dioxide VLE datasets::

        collection.search(
            system=("H2O", "CO2"), data_type="VLE", limit=10
        )
    """
    if components is not None and system is not None:
        raise ValueError("Specify either components or system, not both.")
    if component_match not in {"exact", "contains", "within"}:
        raise ValueError("component_match must be 'exact', 'contains', or 'within'.")
    if system_match not in {"exact", "contains"}:
        raise ValueError("system_match must be 'exact' or 'contains'.")
    if limit is not None and limit < 0:
        raise ValueError("limit must be non-negative or None.")
    component_queries = (
        (components,)
        if isinstance(components, str | ComponentIdentity)
        else tuple(components or ())
    )
    system_queries = tuple(system or ())
    required_queries = tuple(required_components or ())
    if required_queries and component_match != "within":
        raise ValueError("required_components is only valid with component_match='within'.")
    if component_match == "within" and not component_queries:
        raise ValueError("component_match='within' requires components.")
    resolved_components = self._component_index.resolve_many(component_queries)
    resolved_system = self._component_index.resolve_many(system_queries)
    resolved_required = self._component_index.resolve_many(required_queries)
    queries = resolved_components or resolved_system
    component_identifiers = {item.stable_identifier for item in resolved_components}
    if any(item.stable_identifier not in component_identifiers for item in resolved_required):
        raise ValueError("required_components must be a subset of components.")
    exact = (system is not None and system_match == "exact") or (
        components is not None and component_match == "exact"
    )
    normalized_property = property_name.casefold() if property_name else None
    normalized_type = normalize_term(data_type) if data_type else None
    normalized_variable = normalize_term(independent_variable) if independent_variable else None

    matches: list[DatasetMatch] = []
    for document in self.documents:
        for dataset in document.datasets:
            compounds = document.system_compounds(dataset)
            identities = tuple(
                ComponentIdentity.from_compound(compound) for compound in compounds
            )
            if component_match == "within" and components is not None:
                if not all(
                    any(query.matches(identity) for query in resolved_components)
                    for identity in identities
                ):
                    continue
                if not all(
                    any(query.matches(identity) for identity in identities)
                    for query in resolved_required
                ):
                    continue
            elif queries and not all(
                any(query.matches(identity) for identity in identities) for query in queries
            ):
                continue
            if exact and len(set(dataset.component_ids)) != len(queries):
                continue

            selected_variables = tuple(
                item
                for item in dataset.variables
                if normalized_variable and normalized_variable in normalize_term(item.name)
            )
            if independent_variable and not selected_variables:
                continue

            selected: list[PropertyDefinition] = []
            categories: list[str] = []
            for property_definition in dataset.properties:
                category = classify_property(property_definition, dataset)
                if normalized_property and normalized_property not in (
                    property_definition.name.casefold()
                ):
                    continue
                if normalized_type and normalized_type not in {
                    normalize_term(category),
                    normalize_term(property_definition.group),
                }:
                    continue
                selected.append(property_definition)
                categories.append(category)
            if (property_name or data_type) and not selected:
                continue
            selected_numbers = tuple(item.number for item in selected)
            if not property_name and not data_type:
                selected_numbers = tuple(item.number for item in dataset.properties)
                categories = [classify_property(item, dataset) for item in dataset.properties]
            selected_variable_numbers = tuple(item.number for item in selected_variables)
            count = 0
            for point in dataset.points:
                property_count = sum(
                    value.number in selected_numbers for value in point.property_values
                )
                if independent_variable:
                    variable_count = sum(
                        value.number in selected_variable_numbers
                        for value in point.variable_values
                    )
                    count += property_count * variable_count
                else:
                    count += property_count
            if count == 0:
                continue
            matches.append(
                DatasetMatch(
                    document=document,
                    dataset=dataset,
                    matching_property_numbers=selected_numbers,
                    observation_count=count,
                    data_types=tuple(dict.fromkeys(categories)),
                    matching_variable_numbers=selected_variable_numbers,
                )
            )
    matches.sort(
        key=lambda match: (
            -match.observation_count,
            match.document.citation.normalized_doi or match.document.provenance.sha256,
            match.dataset.number,
        )
    )
    if limit is not None:
        matches = matches[:limit]
    return tuple(matches)

all_matches() -> tuple[DatasetMatch, ...]

Return every non-empty dataset ranked by observation count.

Source code in src/thermoml_io/collection.py

def all_matches(self) -> tuple[DatasetMatch, ...]:
    """Return every non-empty dataset ranked by observation count."""
    return self.search()

thermoml_io.analysis

Descriptive metadata and coverage summaries for ThermoML collections.

RankedCount dataclass

One deterministic label/count pair in a collection ranking.

Source code in src/thermoml_io/analysis.py

@dataclass(frozen=True, slots=True)
class RankedCount:
    """One deterministic label/count pair in a collection ranking."""

    label: str
    count: int

CollectionSummary dataclass

Counts and ranked coverage of a ThermoML collection.

Source code in src/thermoml_io/analysis.py

@dataclass(frozen=True, slots=True)
class CollectionSummary:
    """Counts and ranked coverage of a ThermoML collection."""

    document_count: int
    dataset_count: int
    data_point_count: int
    observation_count: int
    reaction_dataset_count: int
    system_types: tuple[RankedCount, ...]
    dataset_system_types: tuple[RankedCount, ...]
    data_types: tuple[RankedCount, ...]
    property_groups: tuple[RankedCount, ...]
    properties: tuple[RankedCount, ...]
    systems: tuple[RankedCount, ...]
    components: tuple[RankedCount, ...]
    methods: tuple[RankedCount, ...]
    publications: tuple[RankedCount, ...]

    def top(self, field: str, limit: int = 10) -> tuple[RankedCount, ...]:
        """Return the first ``limit`` entries from one ranking field."""
        if limit < 0:
            raise ValueError("limit must be non-negative.")
        ranking = getattr(self, field)
        if not isinstance(ranking, tuple):
            raise ValueError(f"{field!r} is not a ranking field.")
        return ranking[:limit]

top(field: str, limit: int = 10) -> tuple[RankedCount, ...]

Return the first limit entries from one ranking field.

Source code in src/thermoml_io/analysis.py

def top(self, field: str, limit: int = 10) -> tuple[RankedCount, ...]:
    """Return the first ``limit`` entries from one ranking field."""
    if limit < 0:
        raise ValueError("limit must be non-negative.")
    ranking = getattr(self, field)
    if not isinstance(ranking, tuple):
        raise ValueError(f"{field!r} is not a ranking field.")
    return ranking[:limit]

summarize_documents(documents: Iterable[ThermoMLDocument]) -> CollectionSummary

Summarize a document stream without retaining the complete collection.

This is the scalable entry point for bulk archives. Every document is consumed exactly once, allowing callers to process millions of observations while retaining only aggregate counters.

Source code in src/thermoml_io/analysis.py

def summarize_documents(
    documents: Iterable[ThermoMLDocument],
) -> CollectionSummary:
    """Summarize a document stream without retaining the complete collection.

    This is the scalable entry point for bulk archives. Every document is
    consumed exactly once, allowing callers to process millions of observations
    while retaining only aggregate counters.
    """
    accumulator = _SummaryAccumulator.create()
    for document in documents:
        accumulator.add_document(document)
    return accumulator.finish()

summarize_collection(collection: ThermoMLCollection) -> CollectionSummary

Summarize all parsed experimental observations in collection.

Component and system counts are weighted by individual property observations. A dataset with 100 property values therefore contributes more than a dataset with 10 values, matching the ranking semantics of the search API.

Source code in src/thermoml_io/analysis.py

def summarize_collection(collection: ThermoMLCollection) -> CollectionSummary:
    """Summarize all parsed experimental observations in ``collection``.

    Component and system counts are weighted by individual property
    observations. A dataset with 100 property values therefore contributes
    more than a dataset with 10 values, matching the ranking semantics of the
    search API.
    """
    return summarize_documents(collection.documents)

thermoml_io.archive

Incremental analysis of local bulk ThermoML .tar/.tgz archives.

RankedDataset dataclass

Lightweight description of one dense experimental dataset.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class RankedDataset:
    """Lightweight description of one dense experimental dataset."""

    doi: str | None
    citation_title: str | None
    dataset_number: int
    system: str
    system_type: str
    observation_count: int
    data_types: tuple[str, ...]
    source_locator: str | None

ArchiveParseFailure dataclass

Explicit record of one archive member that could not be decoded.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class ArchiveParseFailure:
    """Explicit record of one archive member that could not be decoded."""

    member_name: str
    source_locator: str
    error_type: str
    message: str

ArchiveRecovery dataclass

One XML member recovered from its paired official NIST JSON member.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class ArchiveRecovery:
    """One XML member recovered from its paired official NIST JSON member."""

    xml_member_name: str
    json_member_name: str
    xml_sha256: str
    json_sha256: str
    error_type: str
    error_message: str
    lexical_numeric_representation_preserved: bool = False

ArchiveComponentIndex dataclass

Component identities discovered during one complete archive scan.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class ArchiveComponentIndex:
    """Component identities discovered during one complete archive scan."""

    archive_path: str
    xml_document_count: int
    parsed_document_count: int
    index: ComponentIndex
    failures: tuple[ArchiveParseFailure, ...]
    recoveries: tuple[ArchiveRecovery, ...]

    @property
    def identities(self) -> tuple[ComponentIdentity, ...]:
        """Return resolved archive-wide component identities."""
        return self.index.identities

    def resolve_component(self, query: ComponentQuery) -> ComponentIdentity:
        """Resolve one friendly or namespaced query against this snapshot."""
        return self.index.resolve(query)

identities: tuple[ComponentIdentity, ...] property

Return resolved archive-wide component identities.

resolve_component(query: ComponentQuery) -> ComponentIdentity

Resolve one friendly or namespaced query against this snapshot.

Source code in src/thermoml_io/archive.py

def resolve_component(self, query: ComponentQuery) -> ComponentIdentity:
    """Resolve one friendly or namespaced query against this snapshot."""
    return self.index.resolve(query)

ArchiveAnalysis dataclass

Streaming summary and dense-dataset ranking for a bulk archive.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class ArchiveAnalysis:
    """Streaming summary and dense-dataset ranking for a bulk archive."""

    archive_path: str
    xml_document_count: int
    parsed_document_count: int
    matched_document_count: int
    component_query: str | None
    resolved_component: ComponentIdentity | None
    serialized_prefilter: str | None
    summary: CollectionSummary
    top_datasets: tuple[RankedDataset, ...]
    failures: tuple[ArchiveParseFailure, ...]
    recoveries: tuple[ArchiveRecovery, ...]

CatalogEntry dataclass

One observed category/property/independent-variable combination.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class CatalogEntry:
    """One observed category/property/independent-variable combination."""

    data_category: str
    property_name: str
    independent_variable: str | None
    relationship_count: int
    dataset_count: int
    publication_count: int

ArchiveCatalog dataclass

Catalog of queryable property relationships in an archive snapshot.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class ArchiveCatalog:
    """Catalog of queryable property relationships in an archive snapshot."""

    archive_path: str
    xml_document_count: int
    parsed_document_count: int
    entries: tuple[CatalogEntry, ...]
    component_identities: tuple[ComponentIdentity, ...]
    failures: tuple[ArchiveParseFailure, ...]
    recoveries: tuple[ArchiveRecovery, ...]

    @property
    def component_index(self) -> ComponentIndex:
        """Return the component index collected during the catalog scan."""
        return ComponentIndex(self.component_identities)

    def resolve_component(self, query: ComponentQuery) -> ComponentIdentity:
        """Resolve one component without rescanning the archive."""
        return self.component_index.resolve(query)

    def categories(self) -> tuple[RankedCount, ...]:
        """Rank package data categories by property-variable relationships."""
        counts: Counter[str] = Counter()
        for entry in self.entries:
            counts[entry.data_category] += entry.relationship_count
        return tuple(
            RankedCount(label, count)
            for label, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
        )

    def properties(self, data_category: str) -> tuple[RankedCount, ...]:
        """List exact reported properties available within a category."""
        normalized = normalize_term(data_category)
        counts: Counter[str] = Counter()
        for entry in self.entries:
            if normalize_term(entry.data_category) == normalized:
                counts[entry.property_name] += entry.relationship_count
        return tuple(
            RankedCount(label, count)
            for label, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
        )

    def independent_variables(
        self, data_category: str, property_name: str
    ) -> tuple[RankedCount, ...]:
        """List independent variables for a category and property substring."""
        normalized_category = normalize_term(data_category)
        normalized_property = normalize_term(property_name)
        counts: Counter[str] = Counter()
        for entry in self.entries:
            variable = entry.independent_variable
            if (
                variable is not None
                and normalize_term(entry.data_category) == normalized_category
                and normalized_property in normalize_term(entry.property_name)
            ):
                counts[variable] += entry.relationship_count
        return tuple(
            RankedCount(label, count)
            for label, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
        )

component_index: ComponentIndex property

Return the component index collected during the catalog scan.

resolve_component(query: ComponentQuery) -> ComponentIdentity

Resolve one component without rescanning the archive.

Source code in src/thermoml_io/archive.py

def resolve_component(self, query: ComponentQuery) -> ComponentIdentity:
    """Resolve one component without rescanning the archive."""
    return self.component_index.resolve(query)

categories() -> tuple[RankedCount, ...]

Rank package data categories by property-variable relationships.

Source code in src/thermoml_io/archive.py

def categories(self) -> tuple[RankedCount, ...]:
    """Rank package data categories by property-variable relationships."""
    counts: Counter[str] = Counter()
    for entry in self.entries:
        counts[entry.data_category] += entry.relationship_count
    return tuple(
        RankedCount(label, count)
        for label, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
    )

properties(data_category: str) -> tuple[RankedCount, ...]

List exact reported properties available within a category.

Source code in src/thermoml_io/archive.py

def properties(self, data_category: str) -> tuple[RankedCount, ...]:
    """List exact reported properties available within a category."""
    normalized = normalize_term(data_category)
    counts: Counter[str] = Counter()
    for entry in self.entries:
        if normalize_term(entry.data_category) == normalized:
            counts[entry.property_name] += entry.relationship_count
    return tuple(
        RankedCount(label, count)
        for label, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
    )

independent_variables(data_category: str, property_name: str) -> tuple[RankedCount, ...]

List independent variables for a category and property substring.

Source code in src/thermoml_io/archive.py

def independent_variables(
    self, data_category: str, property_name: str
) -> tuple[RankedCount, ...]:
    """List independent variables for a category and property substring."""
    normalized_category = normalize_term(data_category)
    normalized_property = normalize_term(property_name)
    counts: Counter[str] = Counter()
    for entry in self.entries:
        variable = entry.independent_variable
        if (
            variable is not None
            and normalize_term(entry.data_category) == normalized_category
            and normalized_property in normalize_term(entry.property_name)
        ):
            counts[variable] += entry.relationship_count
    return tuple(
        RankedCount(label, count)
        for label, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
    )

PublicationRank dataclass

A publication ranked by returned property-variable relationships.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class PublicationRank:
    """A publication ranked by returned property-variable relationships."""

    publication_key: str
    doi: str | None
    title: str | None
    year: int | None
    authors_year: str
    citation_apa: str
    citation_bibtex: str
    relationship_count: int

ArchiveQueryResult dataclass

Property relationships and publication ranking from a complete scan.

Source code in src/thermoml_io/archive.py

@dataclass(frozen=True, slots=True)
class ArchiveQueryResult:
    """Property relationships and publication ranking from a complete scan."""

    archive_path: str
    components: tuple[str, ...]
    required_components: tuple[str, ...]
    resolved_components: tuple[ComponentIdentity, ...]
    resolved_required_components: tuple[ComponentIdentity, ...]
    component_match: ComponentMatch
    data_category: str | None
    property_name: str | None
    independent_variable: str
    xml_document_count: int
    parsed_document_count: int
    available_publication_count: int
    matched_dataset_count: int
    publications: tuple[PublicationRank, ...]
    table: ExperimentalTable
    failures: tuple[ArchiveParseFailure, ...]
    recoveries: tuple[ArchiveRecovery, ...]

    @property
    def analysis_table(self) -> ExperimentalTable:
        """Return one analysis-ready row per reported property observation."""
        return build_analysis_table(self.table)

    def write(
        self,
        path: str | Path,
        *,
        format: TableFormat | None = None,
        layout: Literal["analysis", "lossless"] = "analysis",
    ) -> Path:
        """Write an analysis-ready result, or explicitly request lossless layout.

        The default layout promotes physical quantities to ordinary columns
        named with their ThermoML-reported units. ``layout="lossless"`` writes
        the complete internal representation with structured JSON columns.
        """
        if layout == "analysis":
            selected = self.analysis_table
        elif layout == "lossless":
            selected = self.table
        else:
            raise ValueError("layout must be 'analysis' or 'lossless'.")
        return selected.write(path, format=format)

    def write_csv(self, path: str | Path) -> Path:
        """Write this query as an analysis-ready CSV file."""
        return self.write(path, format="csv")

    def write_json(self, path: str | Path) -> Path:
        """Write this query as an analysis-ready JSON table."""
        return self.write(path, format="json")

    def write_yaml(self, path: str | Path) -> Path:
        """Write this query as an analysis-ready YAML table."""
        return self.write(path, format="yaml")

    def write_parquet(self, path: str | Path) -> Path:
        """Write this query as an analysis-ready Parquet table."""
        return self.write(path, format="parquet")

    def write_lossless(self, path: str | Path, *, format: TableFormat | None = None) -> Path:
        """Write the complete provenance-oriented internal table explicitly."""
        return self.write(path, format=format, layout="lossless")

analysis_table: ExperimentalTable property

Return one analysis-ready row per reported property observation.

write(path: str | Path, *, format: TableFormat | None = None, layout: Literal['analysis', 'lossless'] = 'analysis') -> Path

Write an analysis-ready result, or explicitly request lossless layout.

The default layout promotes physical quantities to ordinary columns named with their ThermoML-reported units. layout="lossless" writes the complete internal representation with structured JSON columns.

Source code in src/thermoml_io/archive.py

def write(
    self,
    path: str | Path,
    *,
    format: TableFormat | None = None,
    layout: Literal["analysis", "lossless"] = "analysis",
) -> Path:
    """Write an analysis-ready result, or explicitly request lossless layout.

    The default layout promotes physical quantities to ordinary columns
    named with their ThermoML-reported units. ``layout="lossless"`` writes
    the complete internal representation with structured JSON columns.
    """
    if layout == "analysis":
        selected = self.analysis_table
    elif layout == "lossless":
        selected = self.table
    else:
        raise ValueError("layout must be 'analysis' or 'lossless'.")
    return selected.write(path, format=format)

write_csv(path: str | Path) -> Path

Write this query as an analysis-ready CSV file.

Source code in src/thermoml_io/archive.py

def write_csv(self, path: str | Path) -> Path:
    """Write this query as an analysis-ready CSV file."""
    return self.write(path, format="csv")

write_json(path: str | Path) -> Path

Write this query as an analysis-ready JSON table.

Source code in src/thermoml_io/archive.py

def write_json(self, path: str | Path) -> Path:
    """Write this query as an analysis-ready JSON table."""
    return self.write(path, format="json")

write_yaml(path: str | Path) -> Path

Write this query as an analysis-ready YAML table.

Source code in src/thermoml_io/archive.py

def write_yaml(self, path: str | Path) -> Path:
    """Write this query as an analysis-ready YAML table."""
    return self.write(path, format="yaml")

write_parquet(path: str | Path) -> Path

Write this query as an analysis-ready Parquet table.

Source code in src/thermoml_io/archive.py

def write_parquet(self, path: str | Path) -> Path:
    """Write this query as an analysis-ready Parquet table."""
    return self.write(path, format="parquet")

write_lossless(path: str | Path, *, format: TableFormat | None = None) -> Path

Write the complete provenance-oriented internal table explicitly.

Source code in src/thermoml_io/archive.py

def write_lossless(self, path: str | Path, *, format: TableFormat | None = None) -> Path:
    """Write the complete provenance-oriented internal table explicitly."""
    return self.write(path, format=format, layout="lossless")

iter_thermoml_archive(archive_path: str | Path, *, serialized_prefilter: str | bytes | None = None, json_fallback: JsonFallback = 'on_xml_error') -> Iterator[ThermoMLDocument]

Yield ThermoML documents from a local bulk archive.

Parameters:

Name	Type	Description	Default
archive_path	str | Path	Local tar-compatible archive. Members are read in archive order and are never extracted to the filesystem.	required
serialized_prefilter	str | bytes | None	Optional exact byte sequence that must occur in a serialized XML member before parsing. This is only a performance prefilter; callers must still apply semantic component/system matching to the parsed document.	None

Yields:

Type	Description
ThermoMLDocument	One provenance-labelled document at a time.

Notes

Bulk archive bytes remain subject to their source terms. This function does not persist, redistribute, or silently skip malformed matching documents.

Source code in src/thermoml_io/archive.py

def iter_thermoml_archive(
    archive_path: str | Path,
    *,
    serialized_prefilter: str | bytes | None = None,
    json_fallback: JsonFallback = "on_xml_error",
) -> Iterator[ThermoMLDocument]:
    """Yield ThermoML documents from a local bulk archive.

    Parameters
    ----------
    archive_path:
        Local tar-compatible archive. Members are read in archive order and are
        never extracted to the filesystem.
    serialized_prefilter:
        Optional exact byte sequence that must occur in a serialized XML member
        before parsing. This is only a performance prefilter; callers must still
        apply semantic component/system matching to the parsed document.

    Yields
    ------
    ThermoMLDocument
        One provenance-labelled document at a time.

    Notes
    -----
    Bulk archive bytes remain subject to their source terms. This function does
    not persist, redistribute, or silently skip malformed matching documents.
    """
    _validate_json_fallback(json_fallback)
    path = Path(archive_path)
    needle = (
        serialized_prefilter.encode("utf-8")
        if isinstance(serialized_prefilter, str)
        else serialized_prefilter
    )
    for member_name, raw in _archive_xml_members(path):
        if needle is not None and needle not in raw:
            continue
        try:
            document, _ = _parse_archive_document(
                path, member_name, raw, json_fallback=json_fallback
            )
            yield document
        except ThermoMLError as exc:
            raise ThermoMLArchiveError(
                f"Failed to parse archive member {member_name!r}: {exc}"
            ) from exc

analyze_thermoml_archive(archive_path: str | Path, *, component: ComponentQuery | None = None, component_index: ComponentIndex | ArchiveComponentIndex | None = None, serialized_prefilter: str | bytes | None = None, top_datasets: int = 10, on_error: Literal['raise', 'collect'] = 'raise', json_fallback: JsonFallback = 'on_xml_error') -> ArchiveAnalysis

Analyze an entire ThermoML archive with bounded aggregate memory.

component includes every pure or mixture dataset matching an archive-resolved identity. Friendly strings are resolved in a preliminary complete scan unless a reusable component_index is supplied. A :class:ComponentIdentity returned by an explicit resolver avoids that scan. serialized_prefilter can accelerate stable-identifier queries, but semantic matching remains the deciding filter. Dataset truncation occurs only after the complete archive has been scanned and ranked. on_error="collect" records every known ThermoML decoding failure in the result; the default strict mode raises at the first failure.

Source code in src/thermoml_io/archive.py

def analyze_thermoml_archive(
    archive_path: str | Path,
    *,
    component: ComponentQuery | None = None,
    component_index: ComponentIndex | ArchiveComponentIndex | None = None,
    serialized_prefilter: str | bytes | None = None,
    top_datasets: int = 10,
    on_error: Literal["raise", "collect"] = "raise",
    json_fallback: JsonFallback = "on_xml_error",
) -> ArchiveAnalysis:
    """Analyze an entire ThermoML archive with bounded aggregate memory.

    ``component`` includes every pure or mixture dataset matching an
    archive-resolved identity. Friendly strings are resolved in a preliminary
    complete scan unless a reusable ``component_index`` is supplied. A
    :class:`ComponentIdentity` returned by an explicit resolver avoids that
    scan. ``serialized_prefilter`` can accelerate stable-identifier queries,
    but semantic matching remains the deciding filter. Dataset truncation
    occurs only after the complete archive has been scanned and ranked.
    ``on_error="collect"`` records every known ThermoML decoding failure in the
    result; the default strict mode raises at the first failure.
    """
    if top_datasets < 0:
        raise ValueError("top_datasets must be non-negative.")
    if on_error not in {"raise", "collect"}:
        raise ValueError("on_error must be 'raise' or 'collect'.")
    _validate_json_fallback(json_fallback)
    path = Path(archive_path)
    resolved_component: ComponentIdentity | None = None
    if component is not None:
        if isinstance(component, ComponentIdentity):
            resolved_component = component
        else:
            reusable_index = (
                component_index.index
                if isinstance(component_index, ArchiveComponentIndex)
                else component_index
            )
            resolved_component = (
                reusable_index.resolve(component)
                if reusable_index is not None
                else explicit_component_identity(component)
            )
            if resolved_component is None:
                if reusable_index is None:
                    reusable_index = index_thermoml_archive(
                        path, on_error=on_error, json_fallback=json_fallback
                    ).index
                resolved_component = reusable_index.resolve(component)
    needle = (
        serialized_prefilter.encode("utf-8")
        if isinstance(serialized_prefilter, str)
        else serialized_prefilter
    )
    accumulator = _SummaryAccumulator.create()
    candidates: list[RankedDataset] = []
    xml_document_count = 0
    parsed_document_count = 0
    matched_document_count = 0
    failures: list[ArchiveParseFailure] = []
    recoveries: list[ArchiveRecovery] = []

    for member_name, raw in _archive_xml_members(path):
        xml_document_count += 1
        if needle is not None and needle not in raw:
            continue
        try:
            document, recovery = _parse_archive_document(
                path, member_name, raw, json_fallback=json_fallback
            )
        except ThermoMLError as exc:
            if on_error == "raise":
                raise ThermoMLArchiveError(
                    f"Failed to parse archive member {member_name!r}: {exc}"
                ) from exc
            failures.append(
                ArchiveParseFailure(
                    member_name=member_name,
                    source_locator=f"{path.resolve()}!{member_name}",
                    error_type=type(exc).__name__,
                    message=str(exc),
                )
            )
            continue
        if recovery is not None:
            recoveries.append(recovery)
        parsed_document_count += 1
        datasets = (
            document.datasets
            if resolved_component is None
            else _component_datasets(document, resolved_component)
        )
        if not datasets and component is not None:
            continue
        matched_document_count += 1
        accumulator.add_document(
            document,
            datasets,
            include_reactions=component is None,
        )
        candidates.extend(_ranked_dataset(document, dataset) for dataset in datasets)

    candidates.sort(
        key=lambda item: (
            -item.observation_count,
            item.doi or item.source_locator or "",
            item.dataset_number,
        )
    )
    selected = candidates[:top_datasets] if top_datasets else []
    return ArchiveAnalysis(
        archive_path=str(path),
        xml_document_count=xml_document_count,
        parsed_document_count=parsed_document_count,
        matched_document_count=matched_document_count,
        component_query=(component_query_label(component) if component is not None else None),
        resolved_component=resolved_component,
        serialized_prefilter=(
            serialized_prefilter.decode("utf-8")
            if isinstance(serialized_prefilter, bytes)
            else serialized_prefilter
        ),
        summary=accumulator.finish(),
        top_datasets=tuple(selected),
        failures=tuple(failures),
        recoveries=tuple(recoveries),
    )

index_thermoml_archive(archive_path: str | Path | None = None, *, on_error: Literal['raise', 'collect'] = 'collect', json_fallback: JsonFallback = 'on_xml_error') -> ArchiveComponentIndex

Build a reusable archive-wide index of component aliases.

Parameters:

Name	Type	Description	Default
archive_path	str | Path | None	Local archive path. The configured upstream snapshot is fetched when omitted.	None
on_error	Literal['raise', 'collect']	"raise" stops at the first malformed member; "collect" records failures in the returned index.	'collect'

Returns:

Type	Description
ArchiveComponentIndex	Detached identities connected through reported InChIKey, InChI, and CAS identifiers. No external chemical service is contacted.

Source code in src/thermoml_io/archive.py

def index_thermoml_archive(
    archive_path: str | Path | None = None,
    *,
    on_error: Literal["raise", "collect"] = "collect",
    json_fallback: JsonFallback = "on_xml_error",
) -> ArchiveComponentIndex:
    """Build a reusable archive-wide index of component aliases.

    Parameters
    ----------
    archive_path:
        Local archive path. The configured upstream snapshot is fetched when
        omitted.
    on_error:
        ``"raise"`` stops at the first malformed member; ``"collect"``
        records failures in the returned index.

    Returns
    -------
    ArchiveComponentIndex
        Detached identities connected through reported InChIKey, InChI, and
        CAS identifiers. No external chemical service is contacted.
    """
    if on_error not in {"raise", "collect"}:
        raise ValueError("on_error must be 'raise' or 'collect'.")
    _validate_json_fallback(json_fallback)
    path = _archive_path_or_fetch(archive_path)
    identities: list[ComponentIdentity] = []
    failures: list[ArchiveParseFailure] = []
    recoveries: list[ArchiveRecovery] = []
    xml_document_count = 0
    parsed_document_count = 0
    for member_name, raw in _archive_xml_members(path):
        xml_document_count += 1
        try:
            document, recovery = _parse_archive_document(
                path, member_name, raw, json_fallback=json_fallback
            )
        except ThermoMLError as exc:
            if on_error == "raise":
                raise ThermoMLArchiveError(
                    f"Failed to parse archive member {member_name!r}: {exc}"
                ) from exc
            failures.append(_parse_failure(path, member_name, exc))
            continue
        if recovery is not None:
            recoveries.append(recovery)
        parsed_document_count += 1
        identities.extend(
            ComponentIdentity.from_compound(compound) for compound in document.compounds
        )
    return ArchiveComponentIndex(
        archive_path=str(path),
        xml_document_count=xml_document_count,
        parsed_document_count=parsed_document_count,
        index=ComponentIndex.from_identities(identities),
        failures=tuple(failures),
        recoveries=tuple(recoveries),
    )

catalog_thermoml_archive(archive_path: str | Path | None = None, *, on_error: Literal['raise', 'collect'] = 'collect', json_fallback: JsonFallback = 'on_xml_error') -> ArchiveCatalog

Scan a complete snapshot and catalog queryable property relationships.

A relationship is one reported property value paired with one independent variable value in the same ThermoML NumValues record. Properties with no reported independent variable are retained with None as the independent-variable name.

Source code in src/thermoml_io/archive.py

def catalog_thermoml_archive(
    archive_path: str | Path | None = None,
    *,
    on_error: Literal["raise", "collect"] = "collect",
    json_fallback: JsonFallback = "on_xml_error",
) -> ArchiveCatalog:
    """Scan a complete snapshot and catalog queryable property relationships.

    A relationship is one reported property value paired with one independent
    variable value in the same ThermoML ``NumValues`` record. Properties with
    no reported independent variable are retained with ``None`` as the
    independent-variable name.
    """
    if on_error not in {"raise", "collect"}:
        raise ValueError("on_error must be 'raise' or 'collect'.")
    _validate_json_fallback(json_fallback)
    path = _archive_path_or_fetch(archive_path)
    relationships: Counter[tuple[str, str, str | None]] = Counter()
    dataset_counts: Counter[tuple[str, str, str | None]] = Counter()
    publication_keys: defaultdict[tuple[str, str, str | None], set[str]] = defaultdict(set)
    failures: list[ArchiveParseFailure] = []
    recoveries: list[ArchiveRecovery] = []
    component_identities: list[ComponentIdentity] = []
    xml_document_count = 0
    parsed_document_count = 0

    for member_name, raw in _archive_xml_members(path):
        xml_document_count += 1
        try:
            document, recovery = _parse_archive_document(
                path, member_name, raw, json_fallback=json_fallback
            )
        except ThermoMLError as exc:
            if on_error == "raise":
                raise ThermoMLArchiveError(
                    f"Failed to parse archive member {member_name!r}: {exc}"
                ) from exc
            failures.append(_parse_failure(path, member_name, exc))
            continue
        if recovery is not None:
            recoveries.append(recovery)
        parsed_document_count += 1
        component_identities.extend(
            ComponentIdentity.from_compound(compound) for compound in document.compounds
        )
        publication_key = document.citation.normalized_doi or document.provenance.sha256
        for dataset in document.datasets:
            properties = {item.number: item for item in dataset.properties}
            variables = {item.number: item for item in dataset.variables}
            dataset_entries: set[tuple[str, str, str | None]] = set()
            for point in dataset.points:
                point_variables = [
                    variables[value.number]
                    for value in point.variable_values
                    if value.number in variables
                ]
                for measured in point.property_values:
                    definition = properties.get(measured.number)
                    if definition is None:  # pragma: no cover - parser validates references
                        continue
                    category = classify_property(definition, dataset)
                    variable_names: tuple[str | None, ...] = (
                        tuple(item.name for item in point_variables) if point_variables else (None,)
                    )
                    for variable_name in variable_names:
                        key = (category, definition.name, variable_name)
                        relationships[key] += 1
                        dataset_entries.add(key)
            for key in dataset_entries:
                dataset_counts[key] += 1
                publication_keys[key].add(publication_key)

    entries = tuple(
        CatalogEntry(
            data_category=key[0],
            property_name=key[1],
            independent_variable=key[2],
            relationship_count=count,
            dataset_count=dataset_counts[key],
            publication_count=len(publication_keys[key]),
        )
        for key, count in sorted(
            relationships.items(),
            key=lambda item: (
                normalize_term(item[0][0]),
                normalize_term(item[0][1]),
                normalize_term(item[0][2] or ""),
            ),
        )
    )
    return ArchiveCatalog(
        archive_path=str(path),
        xml_document_count=xml_document_count,
        parsed_document_count=parsed_document_count,
        entries=entries,
        component_identities=ComponentIndex.from_identities(component_identities).identities,
        failures=tuple(failures),
        recoveries=tuple(recoveries),
    )

query_thermoml_archive(archive_path: str | Path | None = None, *, components: ComponentQuery | tuple[ComponentQuery, ...] | list[ComponentQuery], required_components: tuple[ComponentQuery, ...] | list[ComponentQuery] | None = None, component_match: ComponentMatch = 'contains', component_index: ComponentIndex | ArchiveComponentIndex | None = None, data_category: str | None = None, property_name: str | None = None, independent_variable: str, publication_limit: int | None = None, serialized_prefilters: tuple[str | bytes, ...] | list[str | bytes] = (), on_error: Literal['raise', 'collect'] = 'collect', json_fallback: JsonFallback = 'on_xml_error') -> ArchiveQueryResult

Return property-versus-condition rows from a complete archive scan.

Friendly strings are resolved against the complete archive before the data scan. Pass a reusable component_index from :func:index_thermoml_archive, or pass already resolved :class:ComponentIdentity objects, to avoid repeating the identity scan. Publications are ranked by the number of returned property-variable rows. publication_limit is applied only after every matching archive member has been scanned. The resulting table repeats full citation, provenance, system, method, phase, constraint, and uncertainty metadata on every row.

Source code in src/thermoml_io/archive.py

def query_thermoml_archive(
    archive_path: str | Path | None = None,
    *,
    components: (ComponentQuery | tuple[ComponentQuery, ...] | list[ComponentQuery]),
    required_components: (tuple[ComponentQuery, ...] | list[ComponentQuery] | None) = None,
    component_match: ComponentMatch = "contains",
    component_index: ComponentIndex | ArchiveComponentIndex | None = None,
    data_category: str | None = None,
    property_name: str | None = None,
    independent_variable: str,
    publication_limit: int | None = None,
    serialized_prefilters: tuple[str | bytes, ...] | list[str | bytes] = (),
    on_error: Literal["raise", "collect"] = "collect",
    json_fallback: JsonFallback = "on_xml_error",
) -> ArchiveQueryResult:
    """Return property-versus-condition rows from a complete archive scan.

    Friendly strings are resolved against the complete archive before the data
    scan. Pass a reusable ``component_index`` from
    :func:`index_thermoml_archive`, or pass already resolved
    :class:`ComponentIdentity` objects, to avoid repeating the identity scan.
    Publications are ranked by the number of returned property-variable rows.
    ``publication_limit`` is applied only after every matching archive member
    has been scanned. The resulting table repeats full citation, provenance,
    system, method, phase, constraint, and uncertainty metadata on every row.
    """
    if publication_limit is not None and publication_limit < 0:
        raise ValueError("publication_limit must be non-negative or None.")
    if on_error not in {"raise", "collect"}:
        raise ValueError("on_error must be 'raise' or 'collect'.")
    _validate_json_fallback(json_fallback)
    if component_match not in {"exact", "contains", "within"}:
        raise ValueError("component_match must be 'exact', 'contains', or 'within'.")
    path = _archive_path_or_fetch(archive_path)
    component_queries = (
        (components,) if isinstance(components, str | ComponentIdentity) else tuple(components)
    )
    required_queries = tuple(required_components or ())
    if not component_queries:
        raise ValueError("components must contain at least one component query.")
    if required_queries and component_match != "within":
        raise ValueError("required_components is only valid with component_match='within'.")
    reusable_index = (
        component_index.index
        if isinstance(component_index, ArchiveComponentIndex)
        else component_index
    )
    prepared_queries: tuple[ComponentQuery, ...]
    if reusable_index is not None:
        prepared_queries = (*component_queries, *required_queries)
    else:
        prepared_queries = tuple(
            explicit_component_identity(query) or query if isinstance(query, str) else query
            for query in (*component_queries, *required_queries)
        )
    prepared_components = prepared_queries[: len(component_queries)]
    prepared_required = prepared_queries[len(component_queries) :]
    needs_resolution = any(isinstance(query, str) for query in prepared_queries)
    if reusable_index is None and needs_resolution:
        reusable_index = index_thermoml_archive(
            path, on_error=on_error, json_fallback=json_fallback
        ).index
    resolver = reusable_index or ComponentIndex(())
    resolved_components = resolver.resolve_many(prepared_components)
    resolved_required = resolver.resolve_many(prepared_required)
    available = {item.stable_identifier for item in resolved_components}
    if any(item.stable_identifier not in available for item in resolved_required):
        raise ValueError("required_components must be a subset of components.")
    needles = tuple(
        item.encode("utf-8") if isinstance(item, str) else item for item in serialized_prefilters
    )
    failures: list[ArchiveParseFailure] = []
    recoveries: list[ArchiveRecovery] = []
    rows_by_publication: defaultdict[str, list[tuple[Cell, ...]]] = defaultdict(list)
    publications: dict[str, PublicationRank] = {}
    columns: tuple[str, ...] | None = None
    xml_document_count = 0
    parsed_document_count = 0
    matched_dataset_count = 0

    for member_name, raw in _archive_xml_members(path):
        xml_document_count += 1
        if needles and not all(needle in raw for needle in needles):
            continue
        try:
            document, recovery = _parse_archive_document(
                path, member_name, raw, json_fallback=json_fallback
            )
        except ThermoMLError as exc:
            if on_error == "raise":
                raise ThermoMLArchiveError(
                    f"Failed to parse archive member {member_name!r}: {exc}"
                ) from exc
            failures.append(_parse_failure(path, member_name, exc))
            continue
        if recovery is not None:
            recoveries.append(recovery)
        parsed_document_count += 1
        collection = ThermoMLCollection((document,))
        matches = collection.search(
            components=resolved_components,
            required_components=resolved_required,
            component_match=component_match,
            property_name=property_name,
            data_type=data_category,
            independent_variable=independent_variable,
        )
        if not matches:
            continue
        matched_dataset_count += len(matches)
        table = build_property_table(collection, matches=matches)
        columns = table.columns
        publication_key = document.citation.normalized_doi or document.provenance.sha256
        rows_by_publication[publication_key].extend(table.rows)
        formatted_citation = publication_citation(document.citation)
        publications[publication_key] = PublicationRank(
            publication_key=publication_key,
            doi=document.citation.normalized_doi,
            title=document.citation.title,
            year=document.citation.year,
            authors_year=formatted_citation.authors_year,
            citation_apa=formatted_citation.apa,
            citation_bibtex=formatted_citation.bibtex,
            relationship_count=len(rows_by_publication[publication_key]),
        )

    ranking = sorted(
        publications.values(),
        key=lambda item: (-item.relationship_count, item.publication_key),
    )
    selected = ranking if publication_limit is None else ranking[:publication_limit]
    empty = build_property_table(ThermoMLCollection(()), matches=())
    selected_rows = tuple(
        row for publication in selected for row in rows_by_publication[publication.publication_key]
    )
    return ArchiveQueryResult(
        archive_path=str(path),
        components=tuple(component_query_label(query) for query in component_queries),
        required_components=tuple(component_query_label(query) for query in required_queries),
        resolved_components=resolved_components,
        resolved_required_components=resolved_required,
        component_match=component_match,
        data_category=data_category,
        property_name=property_name,
        independent_variable=independent_variable,
        xml_document_count=xml_document_count,
        parsed_document_count=parsed_document_count,
        available_publication_count=len(ranking),
        matched_dataset_count=matched_dataset_count,
        publications=tuple(selected),
        table=ExperimentalTable(columns=columns or empty.columns, rows=selected_rows),
        failures=tuple(failures),
        recoveries=tuple(recoveries),
    )

thermoml_io.table

Loss-aware tabular views and CSV, JSON, YAML, and Parquet exporters.

ExperimentalTable dataclass

Rectangular long-form view of heterogeneous experimental observations.

Each row represents one property value. Variables, constraints, and the full uncertainty list are encoded as JSON text in scalar columns so the same schema can be exported consistently to CSV and Parquet.

Source code in src/thermoml_io/table.py

@dataclass(frozen=True, slots=True)
class ExperimentalTable:
    """Rectangular long-form view of heterogeneous experimental observations.

    Each row represents one property value. Variables, constraints, and the
    full uncertainty list are encoded as JSON text in scalar columns so the
    same schema can be exported consistently to CSV and Parquet.
    """

    columns: tuple[str, ...]
    rows: tuple[tuple[Cell, ...], ...]
    schema: str = EXPERIMENTAL_TABLE_SCHEMA

    @classmethod
    def concatenate(cls, *tables: ExperimentalTable) -> ExperimentalTable:
        """Concatenate compatible tables without dropping metadata columns."""
        if not tables:
            return cls(columns=(), rows=())
        columns = tables[0].columns
        schema = tables[0].schema
        if any(table.columns != columns or table.schema != schema for table in tables[1:]):
            raise ValueError("Cannot concatenate tables with different schemas.")
        return cls(
            columns=columns,
            rows=tuple(row for table in tables for row in table.rows),
            schema=schema,
        )

    def to_records(self) -> list[dict[str, Cell]]:
        """Return independent dictionaries suitable for dataframe creation."""
        return [dict(zip(self.columns, row, strict=True)) for row in self.rows]

    def to_pandas(self) -> Any:
        """Return a pandas DataFrame when the optional dependency is installed."""
        try:
            import pandas as pd  # type: ignore[import-untyped]
        except ImportError as exc:  # pragma: no cover - environment dependent
            raise OptionalDependencyError(
                "pandas is required for to_pandas(); install thermoml-io[pandas]."
            ) from exc
        return pd.DataFrame.from_records(self.to_records(), columns=self.columns)

    def write(self, path: str | Path, *, format: TableFormat | None = None) -> Path:
        """Write the table using a format inferred from the path by default."""
        output = Path(path)
        selected = format or output.suffix.lower().removeprefix(".")
        if selected == "yml":
            selected = "yaml"
        if selected not in {"csv", "json", "yaml", "parquet"}:
            raise ValueError(f"Unsupported table format {selected!r}.")
        output.parent.mkdir(parents=True, exist_ok=True)
        if selected == "csv":
            with output.open("w", encoding="utf-8", newline="") as stream:
                writer = csv.DictWriter(stream, fieldnames=self.columns)
                writer.writeheader()
                writer.writerows(self.to_records())
        elif selected == "json":
            payload = {
                "schema": self.schema,
                "columns": self.columns,
                "rows": self.to_records(),
            }
            output.write_text(
                json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
                encoding="utf-8",
            )
        elif selected == "yaml":
            try:
                import yaml  # type: ignore[import-untyped]
            except ImportError as exc:  # pragma: no cover - environment dependent
                raise OptionalDependencyError(
                    "PyYAML is required for YAML export; install thermoml-io[yaml]."
                ) from exc
            payload = {
                "schema": self.schema,
                "columns": list(self.columns),
                "rows": self.to_records(),
            }
            output.write_text(
                yaml.safe_dump(payload, allow_unicode=True, sort_keys=False),
                encoding="utf-8",
            )
        else:
            try:
                import pyarrow as pa
                import pyarrow.parquet as pq
            except ImportError as exc:  # pragma: no cover - environment dependent
                raise OptionalDependencyError(
                    "PyArrow is required for Parquet export; install thermoml-io[parquet]."
                ) from exc
            arrow_table = pa.Table.from_pylist(self.to_records())
            metadata = dict(arrow_table.schema.metadata or {})
            metadata[b"thermoml_io_schema"] = self.schema.removeprefix("thermoml-io.").encode(
                "utf-8"
            )
            arrow_table = arrow_table.replace_schema_metadata(metadata)
            pq.write_table(arrow_table, output)
        return output

concatenate(*tables: ExperimentalTable) -> ExperimentalTable classmethod

Concatenate compatible tables without dropping metadata columns.

Source code in src/thermoml_io/table.py

@classmethod
def concatenate(cls, *tables: ExperimentalTable) -> ExperimentalTable:
    """Concatenate compatible tables without dropping metadata columns."""
    if not tables:
        return cls(columns=(), rows=())
    columns = tables[0].columns
    schema = tables[0].schema
    if any(table.columns != columns or table.schema != schema for table in tables[1:]):
        raise ValueError("Cannot concatenate tables with different schemas.")
    return cls(
        columns=columns,
        rows=tuple(row for table in tables for row in table.rows),
        schema=schema,
    )

to_records() -> list[dict[str, Cell]]

Return independent dictionaries suitable for dataframe creation.

Source code in src/thermoml_io/table.py

def to_records(self) -> list[dict[str, Cell]]:
    """Return independent dictionaries suitable for dataframe creation."""
    return [dict(zip(self.columns, row, strict=True)) for row in self.rows]

to_pandas() -> Any

Return a pandas DataFrame when the optional dependency is installed.

Source code in src/thermoml_io/table.py

def to_pandas(self) -> Any:
    """Return a pandas DataFrame when the optional dependency is installed."""
    try:
        import pandas as pd  # type: ignore[import-untyped]
    except ImportError as exc:  # pragma: no cover - environment dependent
        raise OptionalDependencyError(
            "pandas is required for to_pandas(); install thermoml-io[pandas]."
        ) from exc
    return pd.DataFrame.from_records(self.to_records(), columns=self.columns)

write(path: str | Path, *, format: TableFormat | None = None) -> Path

Write the table using a format inferred from the path by default.

Source code in src/thermoml_io/table.py

def write(self, path: str | Path, *, format: TableFormat | None = None) -> Path:
    """Write the table using a format inferred from the path by default."""
    output = Path(path)
    selected = format or output.suffix.lower().removeprefix(".")
    if selected == "yml":
        selected = "yaml"
    if selected not in {"csv", "json", "yaml", "parquet"}:
        raise ValueError(f"Unsupported table format {selected!r}.")
    output.parent.mkdir(parents=True, exist_ok=True)
    if selected == "csv":
        with output.open("w", encoding="utf-8", newline="") as stream:
            writer = csv.DictWriter(stream, fieldnames=self.columns)
            writer.writeheader()
            writer.writerows(self.to_records())
    elif selected == "json":
        payload = {
            "schema": self.schema,
            "columns": self.columns,
            "rows": self.to_records(),
        }
        output.write_text(
            json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
    elif selected == "yaml":
        try:
            import yaml  # type: ignore[import-untyped]
        except ImportError as exc:  # pragma: no cover - environment dependent
            raise OptionalDependencyError(
                "PyYAML is required for YAML export; install thermoml-io[yaml]."
            ) from exc
        payload = {
            "schema": self.schema,
            "columns": list(self.columns),
            "rows": self.to_records(),
        }
        output.write_text(
            yaml.safe_dump(payload, allow_unicode=True, sort_keys=False),
            encoding="utf-8",
        )
    else:
        try:
            import pyarrow as pa
            import pyarrow.parquet as pq
        except ImportError as exc:  # pragma: no cover - environment dependent
            raise OptionalDependencyError(
                "PyArrow is required for Parquet export; install thermoml-io[parquet]."
            ) from exc
        arrow_table = pa.Table.from_pylist(self.to_records())
        metadata = dict(arrow_table.schema.metadata or {})
        metadata[b"thermoml_io_schema"] = self.schema.removeprefix("thermoml-io.").encode(
            "utf-8"
        )
        arrow_table = arrow_table.replace_schema_metadata(metadata)
        pq.write_table(arrow_table, output)
    return output

build_experimental_table(collection: ThermoMLCollection, *, matches: tuple[DatasetMatch, ...] | None = None) -> ExperimentalTable

Build a stable long-form table from all or selected datasets.

Source code in src/thermoml_io/table.py

def build_experimental_table(
    collection: ThermoMLCollection,
    *,
    matches: tuple[DatasetMatch, ...] | None = None,
) -> ExperimentalTable:
    """Build a stable long-form table from all or selected datasets."""
    selected_matches = matches if matches is not None else collection.all_matches()
    rows: list[tuple[Cell, ...]] = []
    for match in selected_matches:
        document = match.document
        dataset = match.dataset
        citation = publication_citation(document.citation)
        properties = {item.number: item for item in dataset.properties}
        variables = {item.number: item for item in dataset.variables}
        compounds = document.system_compounds(dataset)
        samples_json = json.dumps(
            [
                {
                    "component": compound.preferred_name,
                    "stable_identifier": compound.stable_identifier,
                    "common_names": compound.common_names,
                    "iupac_name": compound.iupac_name,
                    "cas_name": compound.cas_name,
                    "formula": compound.formula,
                    "standard_inchi": compound.standard_inchi,
                    "standard_inchi_key": compound.standard_inchi_key,
                    "cas_registry_number": compound.cas_registry_number,
                    "samples": [asdict(sample) for sample in compound.samples],
                }
                for compound in compounds
            ],
            default=_json_default,
            ensure_ascii=False,
            sort_keys=True,
        )
        constraints_json = json.dumps(
            [_constraint_record(item, document) for item in dataset.constraints],
            default=_json_default,
            ensure_ascii=False,
            sort_keys=True,
        )
        for point in dataset.points:
            variables_json = json.dumps(
                [
                    _quantity_record(variables[value.number], value, document)
                    for value in point.variable_values
                ],
                default=_json_default,
                ensure_ascii=False,
                sort_keys=True,
            )
            for measured in point.property_values:
                if measured.number not in match.matching_property_numbers:
                    continue
                definition = properties[measured.number]
                component = (
                    document.compound(definition.component_id).preferred_name
                    if definition.component_id is not None
                    else None
                )
                record: dict[str, Cell] = {
                    "source_locator": document.provenance.locator,
                    "source_sha256": document.provenance.sha256,
                    "source_retrieved_at": (
                        document.provenance.retrieved_at.isoformat()
                        if document.provenance.retrieved_at is not None
                        else None
                    ),
                    "source_media_type": document.provenance.media_type,
                    "source_related_xml_md5": document.provenance.related_xml_md5,
                    "source_recovery_json": (
                        json.dumps(
                            asdict(document.provenance.recovery),
                            ensure_ascii=False,
                            sort_keys=True,
                        )
                        if document.provenance.recovery is not None
                        else None
                    ),
                    "source_warnings_json": json.dumps(
                        document.warnings,
                        ensure_ascii=False,
                    ),
                    "thermoml_version": document.schema_version,
                    "doi": document.citation.normalized_doi,
                    "publication_year": document.citation.year,
                    "publication_date": document.citation.date,
                    "publication_document_type": document.citation.document_type,
                    "citation_title": document.citation.title,
                    "citation_authors": " | ".join(document.citation.authors),
                    "citation_authors_year": citation.authors_year,
                    "citation_apa": citation.apa,
                    "citation_bibtex": citation.bibtex,
                    "publication_name": document.citation.publication_name,
                    "publication_volume": document.citation.volume,
                    "publication_pages": document.citation.pages,
                    "citation_url": document.citation.url,
                    "trc_reference_id": document.citation.trc_reference_id,
                    "dataset_key": match.dataset_key,
                    "dataset_number": dataset.number,
                    "dataset_purpose": dataset.purpose,
                    "dataset_compiler": dataset.compiler,
                    "dataset_contributor": dataset.contributor,
                    "dataset_date_added": dataset.date_added,
                    "dataset_phases": " | ".join(dataset.phases),
                    "system_key": match.system_key,
                    "system_type": dataset.system_type,
                    "components": " | ".join(item.preferred_name for item in compounds),
                    "component_identifiers": " | ".join(
                        item.stable_identifier for item in compounds
                    ),
                    "samples_json": samples_json,
                    "data_type": classify_property(definition, dataset),
                    "property_number": definition.number,
                    "property_group": definition.group,
                    "property_name": definition.name,
                    "property_method": definition.method,
                    "property_phase": definition.phase,
                    "property_component": component,
                    "property_solvent_components": " | ".join(
                        document.compound(local_id).preferred_name
                        for local_id in definition.solvent_component_ids
                    ),
                    "property_presentation": definition.presentation,
                    "property_reference_phase": definition.reference_phase,
                    "property_standard_state": definition.standard_state,
                    "property_definition_uncertainties_json": json.dumps(
                        _uncertainty_records(definition.uncertainties),
                        default=_json_default,
                        ensure_ascii=False,
                        sort_keys=True,
                    ),
                    "property_repeatability_json": json.dumps(
                        [asdict(item) for item in definition.repeatability],
                        default=_json_default,
                        ensure_ascii=False,
                        sort_keys=True,
                    ),
                    "property_device_specifications_json": json.dumps(
                        [asdict(item) for item in definition.device_specifications],
                        default=_json_default,
                        ensure_ascii=False,
                        sort_keys=True,
                    ),
                    "value": measured.lexical_value,
                    "significant_digits": measured.significant_digits,
                    "uncertainties_json": json.dumps(
                        _uncertainty_records(measured.uncertainties),
                        default=_json_default,
                        ensure_ascii=False,
                        sort_keys=True,
                    ),
                    "value_repeatability_json": json.dumps(
                        [asdict(item) for item in measured.repeatability],
                        default=_json_default,
                        ensure_ascii=False,
                        sort_keys=True,
                    ),
                    "variables_json": variables_json,
                    "constraints_json": constraints_json,
                }
                rows.append(tuple(record[column] for column in _COLUMNS))
    return ExperimentalTable(columns=_COLUMNS, rows=tuple(rows))

build_property_table(collection: ThermoMLCollection, *, matches: tuple[DatasetMatch, ...]) -> ExperimentalTable

Build property-versus-independent-variable rows with full provenance.

matches must come from :meth:ThermoMLCollection.search with an independent_variable filter. One output row represents one reported property value paired with one matching variable value from the same ThermoML NumValues record. Fixed experimental conditions remain in constraints_json. complementary_conditions_json combines every other point variable with every fixed constraint, preserving their source so plots and regressions can distinguish isobars, isotherms, compositions, and other parameterizations. Every publication field from :func:build_experimental_table is retained.

Source code in src/thermoml_io/table.py

def build_property_table(
    collection: ThermoMLCollection,
    *,
    matches: tuple[DatasetMatch, ...],
) -> ExperimentalTable:
    """Build property-versus-independent-variable rows with full provenance.

    ``matches`` must come from :meth:`ThermoMLCollection.search` with an
    ``independent_variable`` filter. One output row represents one reported
    property value paired with one matching variable value from the same
    ThermoML ``NumValues`` record. Fixed experimental conditions remain in
    ``constraints_json``. ``complementary_conditions_json`` combines every
    other point variable with every fixed constraint, preserving their source
    so plots and regressions can distinguish isobars, isotherms, compositions,
    and other parameterizations. Every publication field from
    :func:`build_experimental_table` is retained.
    """
    if any(not match.matching_variable_numbers for match in matches):
        raise ValueError(
            "build_property_table requires matches selected with independent_variable."
        )
    selected_variables = {
        match.dataset_key: set(match.matching_variable_numbers) for match in matches
    }
    base = build_experimental_table(collection, matches=matches)
    rows: list[tuple[Cell, ...]] = []
    for record in base.to_records():
        dataset_key = record["dataset_key"]
        if not isinstance(dataset_key, str):  # pragma: no cover - stable schema
            continue
        variables_json = record["variables_json"]
        if not isinstance(variables_json, str):  # pragma: no cover - stable schema
            continue
        variables = json.loads(variables_json)
        for variable in variables:
            if variable["number"] not in selected_variables[dataset_key]:
                continue
            constraints_json = record["constraints_json"]
            if not isinstance(constraints_json, str):  # pragma: no cover - stable schema
                continue
            complementary_conditions = [
                {"source": "variable", **item}
                for item in variables
                if item["number"] != variable["number"]
            ]
            complementary_conditions.extend(
                {"source": "constraint", **item} for item in json.loads(constraints_json)
            )
            relation = {
                **record,
                "independent_variable_number": variable["number"],
                "independent_variable_name": variable["name"],
                "independent_variable_phase": variable["phase"],
                "independent_variable_component": variable["component"],
                "independent_variable_solvent_components": " | ".join(
                    variable["solvent_components"]
                ),
                "independent_variable_value": variable["value"],
                "independent_variable_significant_digits": variable["significant_digits"],
                "independent_variable_uncertainties_json": json.dumps(
                    variable["uncertainties"], ensure_ascii=False, sort_keys=True
                ),
                "independent_variable_definition_uncertainties_json": json.dumps(
                    variable["definition_uncertainties"],
                    ensure_ascii=False,
                    sort_keys=True,
                ),
                "independent_variable_repeatability_json": json.dumps(
                    variable["value_repeatability"],
                    ensure_ascii=False,
                    sort_keys=True,
                ),
                "independent_variable_definition_repeatability_json": json.dumps(
                    variable["definition_repeatability"],
                    ensure_ascii=False,
                    sort_keys=True,
                ),
                "independent_variable_device_specifications_json": json.dumps(
                    variable["device_specifications"],
                    ensure_ascii=False,
                    sort_keys=True,
                ),
                "complementary_conditions_json": json.dumps(
                    complementary_conditions,
                    ensure_ascii=False,
                    sort_keys=True,
                ),
            }
            rows.append(tuple(relation[column] for column in _PROPERTY_RELATION_COLUMNS))
    return ExperimentalTable(columns=_PROPERTY_RELATION_COLUMNS, rows=tuple(rows))

build_analysis_table(table: ExperimentalTable) -> ExperimentalTable

Pivot a property-condition table into an analysis-ready wide table.

Physical quantities become ordinary columns named exactly as reported by ThermoML, including their units (for example Temperature, K and Viscosity, Pa*s). Each row remains one reported property observation. DOI, authors/year, system, phases, method, category, and dataset identity are repeated as scalar columns. Less frequently used metrological and provenance details are retained as compact JSON in metadata.

If the same reported quantity name has multiple semantic meanings within one observation, its columns are explicitly qualified by source, phase, or component. Differences that occur only between rows remain in metadata so the main physical columns stay compact. Duplicate indistinguishable conditions in one observation raise rather than silently overwriting a value.

Parameters:

Name	Type	Description	Default
table	ExperimentalTable	Lossless table returned by :func:build_property_table.	required

Returns:

Type	Description
ExperimentalTable	Analysis-ready table with schema thermoml-io.analysis-table.v1.

Raises:

Type	Description
ValueError	If table is not a property-condition table or contains ambiguous duplicate conditions that cannot be represented safely in one row.

Source code in src/thermoml_io/table.py

def build_analysis_table(table: ExperimentalTable) -> ExperimentalTable:
    """Pivot a property-condition table into an analysis-ready wide table.

    Physical quantities become ordinary columns named exactly as reported by
    ThermoML, including their units (for example ``Temperature, K`` and
    ``Viscosity, Pa*s``). Each row remains one reported property observation.
    DOI, authors/year, system, phases, method, category, and dataset identity
    are repeated as scalar columns. Less frequently used metrological and
    provenance details are retained as compact JSON in ``metadata``.

    If the same reported quantity name has multiple semantic meanings within
    one observation, its columns are explicitly qualified by source, phase, or
    component. Differences that occur only between rows remain in metadata so
    the main physical columns stay compact. Duplicate indistinguishable
    conditions in one observation raise rather than silently overwriting a
    value.

    Parameters
    ----------
    table:
        Lossless table returned by :func:`build_property_table`.

    Returns
    -------
    ExperimentalTable
        Analysis-ready table with schema ``thermoml-io.analysis-table.v1``.

    Raises
    ------
    ValueError
        If ``table`` is not a property-condition table or contains ambiguous
        duplicate conditions that cannot be represented safely in one row.
    """
    missing = set(_PROPERTY_RELATION_COLUMNS).difference(table.columns)
    if missing:
        raise ValueError(
            "build_analysis_table requires a property-condition table; "
            f"missing columns: {', '.join(sorted(missing))}."
        )

    prepared: list[tuple[dict[str, Cell], list[dict[str, Any]]]] = []
    condition_signatures: dict[str, set[tuple[str, str, str]]] = {}
    condition_names_ambiguous_within_row: set[str] = set()
    property_signatures: dict[str, set[tuple[str, str]]] = {}
    for record in table.to_records():
        independent_name = record["independent_variable_name"]
        if not isinstance(independent_name, str):
            raise ValueError("Independent-variable names must be strings.")
        independent = {
            "role": "independent",
            "source": "variable",
            "number": record["independent_variable_number"],
            "name": independent_name,
            "phase": record["independent_variable_phase"],
            "component": record["independent_variable_component"],
            "solvent_components": record["independent_variable_solvent_components"],
            "value": record["independent_variable_value"],
            "significant_digits": record["independent_variable_significant_digits"],
            "uncertainties": _decoded_list(record, "independent_variable_uncertainties_json"),
            "definition_uncertainties": _decoded_list(
                record, "independent_variable_definition_uncertainties_json"
            ),
            "repeatability": _decoded_list(record, "independent_variable_repeatability_json"),
            "definition_repeatability": _decoded_list(
                record, "independent_variable_definition_repeatability_json"
            ),
            "device_specifications": _decoded_list(
                record, "independent_variable_device_specifications_json"
            ),
        }
        complementary = _decoded_list(record, "complementary_conditions_json")
        for item in complementary:
            item["role"] = "complementary"
        conditions = [independent, *complementary]
        row_condition_signatures: dict[str, set[tuple[str, str, str]]] = {}
        for condition in conditions:
            name = condition.get("name")
            if not isinstance(name, str):
                raise ValueError("Condition names must be strings.")
            signature = _condition_signature(condition)
            condition_signatures.setdefault(name, set()).add(signature)
            row_condition_signatures.setdefault(name, set()).add(signature)
        condition_names_ambiguous_within_row.update(
            name for name, signatures in row_condition_signatures.items() if len(signatures) > 1
        )
        property_name = record["property_name"]
        if not isinstance(property_name, str):
            raise ValueError("Property names must be strings.")
        property_signatures.setdefault(property_name, set()).add(_property_signature(record))
        prepared.append((record, conditions))

    condition_signatures = {
        name: (
            signatures if name in condition_names_ambiguous_within_row else {next(iter(signatures))}
        )
        for name, signatures in condition_signatures.items()
    }
    property_signatures = {
        name: {next(iter(signatures))} for name, signatures in property_signatures.items()
    }
    condition_columns = sorted(
        {
            _condition_label(name, signature, condition_signatures)
            for name, signatures in condition_signatures.items()
            for signature in signatures
        },
        key=_condition_sort_key,
    )
    property_columns = sorted(
        {
            _property_label(name, signature, property_signatures)
            for name, signatures in property_signatures.items()
            for signature in signatures
        },
        key=str.casefold,
    )
    columns = (*condition_columns, *property_columns, *_ANALYSIS_METADATA_COLUMNS)
    rows: list[tuple[Cell, ...]] = []
    for record, conditions in prepared:
        output: dict[str, Cell] = dict.fromkeys(columns)
        for condition in conditions:
            name = str(condition["name"])
            label = _condition_label(name, _condition_signature(condition), condition_signatures)
            if output[label] is not None:
                raise ValueError(
                    "Cannot represent duplicate indistinguishable condition "
                    f"{label!r} in one analysis row."
                )
            output[label] = condition.get("value")
        property_name = str(record["property_name"])
        property_label = _property_label(
            property_name, _property_signature(record), property_signatures
        )
        output[property_label] = record["value"]
        components = str(record["components"] or "")
        output.update(
            {
                "DOI": record["doi"],
                "Authors/Year": record["citation_authors_year"],
                "System": " + ".join(components.split(" | ")),
                "System Type": record["system_type"],
                "Phases": record["dataset_phases"],
                "Method": record["property_method"],
                "Data Category": record["data_type"],
                "Dataset": record["dataset_key"],
                "metadata": _analysis_metadata(record, conditions),
            }
        )
        rows.append(tuple(output[column] for column in columns))
    return ExperimentalTable(
        columns=columns,
        rows=tuple(rows),
        schema=ANALYSIS_TABLE_SCHEMA,
    )

thermoml_io.upstream

Versioned upstream discovery and verified ThermoML archive downloads.

ArchiveSource dataclass

Immutable description of one checksum-pinned upstream archive.

Source code in src/thermoml_io/upstream.py

@dataclass(frozen=True, slots=True)
class ArchiveSource:
    """Immutable description of one checksum-pinned upstream archive."""

    source_name: str
    metadata_url: str
    record_id: str
    record_version: str
    record_modified: str
    doi: str
    filename: str
    download_url: str
    media_type: str
    size_bytes: int
    sha256: str
    snapshot_date: str | None
    description: str

CordraSnapshot dataclass

Deterministic identity census of the live NIST ThermoML Cordra API.

Source code in src/thermoml_io/upstream.py

@dataclass(frozen=True, slots=True)
class CordraSnapshot:
    """Deterministic identity census of the live NIST ThermoML Cordra API."""

    api_url: str
    query: str
    object_type: str
    object_count: int
    identifiers_sha256: str
    first_identifier: str
    last_identifier: str

get_archive_source(source: str = 'nist') -> ArchiveSource

Load a checksum-pinned archive source shipped with the package.

Source code in src/thermoml_io/upstream.py

def get_archive_source(source: str = "nist") -> ArchiveSource:
    """Load a checksum-pinned archive source shipped with the package."""
    try:
        registry = json.loads(_registry_path().read_text(encoding="utf-8"))
        if registry["schema_version"] != 1:
            raise ThermoMLSourceError("Unsupported archive-source registry schema.")
        value = registry["sources"][source]
    except KeyError as exc:
        raise ThermoMLSourceError(f"Unknown ThermoML archive source {source!r}.") from exc
    except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
        raise ThermoMLSourceError(f"Invalid archive-source registry: {exc}") from exc
    return _source_from_mapping(source, value)

get_cordra_snapshot() -> CordraSnapshot

Load the packaged identity census for the live NIST Cordra API.

Source code in src/thermoml_io/upstream.py

def get_cordra_snapshot() -> CordraSnapshot:
    """Load the packaged identity census for the live NIST Cordra API."""
    try:
        registry = json.loads(_cordra_path().read_text(encoding="utf-8"))
        if registry["schema_version"] != 1:
            raise ThermoMLSourceError("Unsupported Cordra registry schema.")
        value = cast(dict[str, Any], registry["snapshot"])
        snapshot = CordraSnapshot(
            api_url=str(value["api_url"]),
            query=str(value["query"]),
            object_type=str(value["object_type"]),
            object_count=int(value["object_count"]),
            identifiers_sha256=str(value["identifiers_sha256"]).casefold(),
            first_identifier=str(value["first_identifier"]),
            last_identifier=str(value["last_identifier"]),
        )
    except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
        raise ThermoMLSourceError(f"Invalid Cordra snapshot registry: {exc}") from exc
    return _validate_cordra_snapshot(snapshot)

discover_archive_source(source: str = 'nist', *, timeout: float = 30.0) -> ArchiveSource

Discover the newest tar-compatible ThermoML snapshot in NIST NERDm.

Source code in src/thermoml_io/upstream.py

def discover_archive_source(source: str = "nist", *, timeout: float = 30.0) -> ArchiveSource:
    """Discover the newest tar-compatible ThermoML snapshot in NIST NERDm."""
    configured = get_archive_source(source)
    metadata = _download_json(configured.metadata_url, timeout=timeout)
    candidates: list[tuple[str, dict[str, Any]]] = []
    for raw_component in metadata.get("components", []):
        if not isinstance(raw_component, dict):
            continue
        component = cast(dict[str, Any], raw_component)
        filename = component.get("filepath")
        if not isinstance(filename, str) or not filename.endswith(_ARCHIVE_SUFFIXES):
            continue
        match = _SNAPSHOT_DATE.search(filename)
        candidates.append((match.group(1) if match else "", component))
    if not candidates:
        raise ThermoMLSourceError("NIST metadata contains no tar-compatible archive.")
    _, selected = max(candidates, key=lambda item: (item[0], str(item[1]["filepath"])))
    checksum = selected.get("checksum")
    if not isinstance(checksum, dict):
        raise ThermoMLSourceError("Selected NIST archive has no checksum metadata.")
    algorithm = checksum.get("algorithm")
    tag = algorithm.get("tag") if isinstance(algorithm, dict) else None
    if str(tag).casefold() != "sha256":
        raise ThermoMLSourceError("Selected NIST archive has no SHA-256 checksum.")
    filename = str(selected["filepath"])
    match = _SNAPSHOT_DATE.search(filename)
    return _validate_source(
        ArchiveSource(
            source_name=source,
            metadata_url=configured.metadata_url,
            record_id=str(metadata.get("ediid", configured.record_id)),
            record_version=str(metadata.get("version", "")),
            record_modified=str(metadata.get("modified", "")),
            doi=str(metadata.get("doi", configured.doi)),
            filename=filename,
            download_url=str(selected["downloadURL"]),
            media_type=str(selected.get("mediaType", "application/octet-stream")),
            size_bytes=int(selected["size"]),
            sha256=str(checksum["hash"]).casefold(),
            snapshot_date=match.group(1) if match else None,
            description=str(selected.get("description", "")).strip(),
        )
    )

discover_cordra_snapshot(*, timeout: float = 180.0) -> CordraSnapshot

Census all live Cordra ThermoML IDs independently of the bulk archive.

The Cordra API intentionally exposes metadata and data-point counts, not the numerical observations. This census detects additions or removals even when the checksum-pinned NERDm .tgz snapshot has not changed.

Source code in src/thermoml_io/upstream.py

def discover_cordra_snapshot(*, timeout: float = 180.0) -> CordraSnapshot:
    """Census all live Cordra ThermoML IDs independently of the bulk archive.

    The Cordra API intentionally exposes metadata and data-point counts, not
    the numerical observations. This census detects additions or removals even
    when the checksum-pinned NERDm ``.tgz`` snapshot has not changed.
    """
    configured = get_cordra_snapshot()
    parameters = urlencode({"query": configured.query})
    payload = _download_json(f"{configured.api_url}?{parameters}&ids", timeout=timeout)
    results = payload.get("results")
    reported_size = payload.get("size")
    if not isinstance(results, list) or any(not isinstance(item, str) for item in results):
        raise ThermoMLSourceError("Cordra ID response must contain a string results list.")
    identifiers = sorted(cast(list[str], results))
    if len(set(identifiers)) != len(identifiers):
        raise ThermoMLSourceError("Cordra ID response contains duplicate identifiers.")
    if not isinstance(reported_size, int) or reported_size != len(identifiers):
        raise ThermoMLSourceError(
            "Cordra ID response is incomplete: reported size does not match results."
        )
    if not identifiers:
        raise ThermoMLSourceError("Cordra ID response contains no identifiers.")
    digest = hashlib.sha256(("\n".join(identifiers) + "\n").encode("utf-8")).hexdigest()
    return _validate_cordra_snapshot(
        CordraSnapshot(
            api_url=configured.api_url,
            query=configured.query,
            object_type=configured.object_type,
            object_count=len(identifiers),
            identifiers_sha256=digest,
            first_identifier=identifiers[0],
            last_identifier=identifiers[-1],
        )
    )

default_cache_dir() -> Path

Return the platform-neutral user cache directory for archive bytes.

Source code in src/thermoml_io/upstream.py

def default_cache_dir() -> Path:
    """Return the platform-neutral user cache directory for archive bytes."""
    explicit = os.environ.get("THERMOML_IO_CACHE")
    if explicit:
        return Path(explicit).expanduser()
    xdg = os.environ.get("XDG_CACHE_HOME")
    if xdg:
        return Path(xdg).expanduser() / "thermoml-io"
    return Path.home() / ".cache" / "thermoml-io"

fetch_thermoml_archive(*, cache_dir: str | Path | None = None, source: str = 'nist', force: bool = False, timeout: float = 120.0) -> Path

Return a locally cached, size- and SHA-256-verified archive.

No URL, filename, or checksum is required from the caller. A temporary sibling file is downloaded and verified before atomically replacing an invalid or explicitly refreshed cache entry.

Source code in src/thermoml_io/upstream.py

def fetch_thermoml_archive(
    *,
    cache_dir: str | Path | None = None,
    source: str = "nist",
    force: bool = False,
    timeout: float = 120.0,
) -> Path:
    """Return a locally cached, size- and SHA-256-verified archive.

    No URL, filename, or checksum is required from the caller. A temporary
    sibling file is downloaded and verified before atomically replacing an
    invalid or explicitly refreshed cache entry.
    """
    selected = get_archive_source(source)
    directory = Path(cache_dir) if cache_dir is not None else default_cache_dir()
    directory.mkdir(parents=True, exist_ok=True)
    destination = directory / selected.filename
    if not force and _verified(destination, selected):
        return destination
    temporary: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w+b", prefix=f".{selected.filename}.", dir=directory, delete=False
        ) as stream:
            temporary = Path(stream.name)
            digest, size = _stream_archive(selected, cast(BinaryIO, stream), timeout=timeout)
        if size != selected.size_bytes or digest != selected.sha256:
            raise ThermoMLDownloadError(
                "Downloaded archive failed the registered size or SHA-256 check."
            )
        temporary.replace(destination)
    finally:
        if temporary is not None and temporary.exists():
            temporary.unlink()
    return destination

archive_source_record(source: ArchiveSource) -> dict[str, Any]

Return the deterministic registry representation used by maintenance CI.

Source code in src/thermoml_io/upstream.py

def archive_source_record(source: ArchiveSource) -> dict[str, Any]:
    """Return the deterministic registry representation used by maintenance CI."""
    value = asdict(source)
    source_name = str(value.pop("source_name"))
    archive_keys = (
        "filename",
        "download_url",
        "media_type",
        "size_bytes",
        "sha256",
        "snapshot_date",
        "description",
    )
    archive = {key: value.pop(key) for key in archive_keys}
    return {"schema_version": 1, "sources": {source_name: {**value, "archive": archive}}}

cordra_snapshot_record(snapshot: CordraSnapshot) -> dict[str, Any]

Return the deterministic registry representation used by monthly CI.

Source code in src/thermoml_io/upstream.py

def cordra_snapshot_record(snapshot: CordraSnapshot) -> dict[str, Any]:
    """Return the deterministic registry representation used by monthly CI."""
    return {"schema_version": 1, "snapshot": asdict(snapshot)}

thermoml_io.conformance

Metadata-only registry of external ThermoML conformance material.

ConformanceSource dataclass

External specification or example corpus, never an experimental source.

Source code in src/thermoml_io/conformance.py

@dataclass(frozen=True, slots=True)
class ConformanceSource:
    """External specification or example corpus, never an experimental source."""

    source_id: str
    title: str
    organization: str
    project_url: str
    publication_doi: str
    artifact_url: str
    media_type: str
    size_bytes: int
    sha256: str
    thermoml_version: str
    use_case_count: int
    purpose: str
    included_by_default: bool
    experimental_query_eligible: bool

list_conformance_sources() -> tuple[ConformanceSource, ...]

List registered external examples without downloading or redistributing them.

Source code in src/thermoml_io/conformance.py

def list_conformance_sources() -> tuple[ConformanceSource, ...]:
    """List registered external examples without downloading or redistributing them."""
    try:
        registry = json.loads(_registry_path().read_text(encoding="utf-8"))
        if registry["schema_version"] != 1:
            raise ThermoMLSourceError("Unsupported conformance registry schema.")
        mappings = cast(dict[str, dict[str, object]], registry["sources"])
        sources = tuple(
            _validate(
                ConformanceSource(
                    source_id=source_id,
                    title=str(value["title"]),
                    organization=str(value["organization"]),
                    project_url=str(value["project_url"]),
                    publication_doi=str(value["publication_doi"]),
                    artifact_url=str(value["artifact_url"]),
                    media_type=str(value["media_type"]),
                    size_bytes=_integer(value["size_bytes"], field="size_bytes"),
                    sha256=str(value["sha256"]),
                    thermoml_version=str(value["thermoml_version"]),
                    use_case_count=_integer(value["use_case_count"], field="use_case_count"),
                    purpose=str(value["purpose"]),
                    included_by_default=_boolean(
                        value["included_by_default"], field="included_by_default"
                    ),
                    experimental_query_eligible=_boolean(
                        value["experimental_query_eligible"],
                        field="experimental_query_eligible",
                    ),
                )
            )
            for source_id, value in mappings.items()
        )
    except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
        raise ThermoMLSourceError(f"Invalid conformance-source registry: {exc}") from exc
    return sources

get_conformance_source(source_id: str) -> ConformanceSource

Return one registered conformance source by stable package identifier.

Source code in src/thermoml_io/conformance.py

def get_conformance_source(source_id: str) -> ConformanceSource:
    """Return one registered conformance source by stable package identifier."""
    for source in list_conformance_sources():
        if source.source_id == source_id:
            return source
    raise ThermoMLSourceError(f"Unknown ThermoML conformance source {source_id!r}.")