plaid.storage.reader¶

def to_dict(
    self,
    dataset: Any,
    idx: int,
    features: Optional[list[str]] = None,
    indexers: Optional[dict[str, Any]] = None,
) -> dict[float, dict[str, Any]]:
    """Convert a dataset sample to dictionary format.

    Args:
        dataset: The dataset object containing the sample.
        idx: Index of the sample to convert.
        features: Optional list of feature names to include from the variable fields.
            If None, all variable features available for the backend are included.
        indexers: Optional mapping ``feature_path -> indexer`` used to extract only
            selected indices inside variable features. Indexing semantics are
            backend-dependent and ignored for non-requested features.

    Returns:
        dict: Sample data in dictionary format.

    Raises:
        ValueError: If called with CGNS backend.
    """
    if self.backend_spec.to_var_sample_dict is None:  # pragma: no cover
        raise ValueError(
            f"Converter.to_dict not available for {self.backend} backend"
        )

    if features:
        features = update_features_for_CGNS_compatibility(
            features,
            self.constant_features,
            self.variable_features,
        )
        req_var_feat = [f for f in features if f in self.variable_features]
    else:
        req_var_feat = None

    if indexers is not None:
        unknown = set(indexers.keys()) - self.variable_features
        if unknown:
            raise KeyError(
                f"Indexers contain unknown variable features: {sorted(unknown)}"
            )
        if req_var_feat is not None:
            not_requested = set(indexers.keys()) - set(req_var_feat)
            if not_requested:
                raise KeyError(
                    "Indexers contain features not present in requested variable "
                    f"features: {sorted(not_requested)}"
                )

    var_sample_dict = self.backend_spec.to_var_sample_dict(
        dataset, idx, features=req_var_feat, indexers=indexers
    )
    return to_sample_dict(var_sample_dict, self.flat_cst, self.cgns_types, features)

def to_plaid(
    self,
    dataset: Any,
    idx: int,
    features: Optional[list[str]] = None,
    indexers: Optional[dict[str, Any]] = None,
) -> Sample:
    """Convert a dataset sample to PLAID Sample object.

    Args:
        dataset: The dataset object containing the sample.
        idx: Index of the sample to convert.
        features: Optional list of feature names to include from the variable fields.
            If None, all variable features available for the backend are included.
            Features are retreated based on self.constant_features and self.variable_features to satisfy the CGNS conventions.
        indexers: Optional mapping ``feature_path -> indexer`` used to extract only
            selected indices inside variable features.

    Returns:
        Sample: A PLAID Sample object.
    """
    # Note: we deliberately do NOT call
    # ``update_features_for_CGNS_compatibility`` here. ``to_dict`` runs it
    # once for non-CGNS backends, and the CGNS branch ignores ``features``
    # entirely. Calling the helper twice used to break feature filtering
    # because its missing-key check only validates the *input* list while
    # the helper itself appends auxiliary paths (parent FlowSolution,
    # ``GridLocation``, ``Base_times``, ``ZoneType``, ...) that may not
    # be declared in ``constant_features`` / ``variable_features``. On
    # the second call those additions look "missing" and the helper
    # raises ``KeyError("Missing features in dataset/converter: ...")``.
    if self.backend != "cgns":
        sample_dict = self.to_dict(dataset, idx, features, indexers=indexers)
        return to_plaid_sample(sample_dict, self.cgns_types)
    else:
        return dataset[idx]

def sample_to_dict(self, sample: Any) -> dict[float, dict[str, Any]]:
    """Convert a PLAID Sample to dictionary format.

    Args:
        sample: The PLAID Sample object to convert.

    Returns:
        dict: Sample data in dictionary format.

    Raises:
        ValueError: If called with CGNS backend.
    """
    if self.backend_spec.sample_to_var_sample_dict is None:  # pragma: no cover
        raise ValueError(
            f"Converter.sample_to_var_sample_dict not available for {self.backend} backend"
        )
    var_sample_dict = self.backend_spec.sample_to_var_sample_dict(sample)
    return to_sample_dict(var_sample_dict, self.flat_cst, self.cgns_types)

def sample_to_plaid(self, sample: Any) -> Sample:
    """Convert a sample to PLAID format (identity function for most backends).

    Args:
        sample: The sample object to convert.

    Returns:
        Sample: A PLAID Sample object.
    """
    if self.backend != "cgns":
        sample_dict = self.sample_to_dict(sample)
        return to_plaid_sample(sample_dict, self.cgns_types)
    else:
        return sample

def plaid_to_dict(self, plaid_sample: Sample) -> dict[str, Any]:
    """Convert a PLAID Sample to dictionary format for storage.

    Args:
        plaid_sample: The PLAID Sample object to convert.

    Returns:
        dict: Sample data in dictionary format suitable for storage.
    """
    return plaid_to_sample_dict(
        plaid_sample,
        self.variable_features,
        self.constant_features,
    )

def __repr__(self) -> str:
    """String representation of the Converter.

    Returns:
        str: String representation including the backend.
    """
    return f"Converter(backend={self.backend})"