# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.
"""Portable declarations for plugin-provided example workspaces."""
from __future__ import annotations
import dataclasses
import sys
from collections.abc import Iterator, Mapping, Sequence
from contextlib import contextmanager
from importlib import resources
from pathlib import Path
from types import MappingProxyType
from typing import TYPE_CHECKING
from sigima.objects import ImageObj, SignalObj
from datalab.plugins.resources import (
LOCAL_ID_PATTERN,
resolve_package_resource,
split_package_resource,
)
if TYPE_CHECKING:
if sys.version_info >= (3, 11):
from importlib.resources.abc import Traversable
else:
from importlib.abc import Traversable
__all__ = ["PluginExample", "PluginExampleData"]
_RESOURCE_LABEL = "Plugin example resource"
[docs]
@dataclasses.dataclass(frozen=True)
class PluginExampleData:
"""Scientific objects and per-recipe defaults generated by a plugin.
Args:
objects: generated signals and images
parameter_values: recipe ID to parameter values suited to these objects
"""
objects: Sequence[SignalObj | ImageObj]
parameter_values: Mapping[str, Mapping[str, object]] = dataclasses.field(
default_factory=dict
)
def __post_init__(self) -> None:
"""Freeze and validate generated example data."""
objects = tuple(self.objects)
if not objects or not all(
isinstance(obj, (SignalObj, ImageObj)) for obj in objects
):
raise TypeError("Plugin example data must contain signal or image objects")
if not isinstance(self.parameter_values, Mapping):
raise TypeError("Plugin example parameters must map recipe IDs to values")
frozen: dict[str, Mapping[str, object]] = {}
for recipe_id, values in self.parameter_values.items():
if not isinstance(recipe_id, str) or recipe_id.count(":") != 1:
raise ValueError("Plugin example parameters must use recipe IDs")
if not isinstance(values, Mapping) or not all(
isinstance(key, str) for key in values
):
raise TypeError(
"Plugin example parameters must map string keys to values"
)
frozen[recipe_id] = MappingProxyType(dict(values))
object.__setattr__(self, "objects", objects)
object.__setattr__(self, "parameter_values", MappingProxyType(frozen))
[docs]
def values_for(self, recipe_id: str) -> Mapping[str, object]:
"""Return the parameter values suited to one recipe (empty if none)."""
return self.parameter_values.get(recipe_id, MappingProxyType({}))
[docs]
@dataclasses.dataclass(frozen=True)
class PluginExample:
"""Packaged or generated example, and the recipes it was designed for.
An example without recipe is presented as a plain dataset.
"""
id: str
title: str
resource: str | None = None
description: str = ""
recipe_ids: Sequence[str] = ()
expected_checks: Sequence[str] = ()
def __post_init__(self) -> None:
"""Validate identity, package resource, and optional workflow metadata."""
if not isinstance(self.id, str) or not LOCAL_ID_PATTERN.fullmatch(self.id):
raise ValueError(
"Plugin example ID must contain lowercase letters, digits, '.', "
"'_' or '-'"
)
if not isinstance(self.title, str) or not self.title.strip():
raise ValueError("Plugin example title must be a non-empty string")
if not isinstance(self.description, str):
raise TypeError("Plugin example description must be a string")
if self.resource is not None:
split_package_resource(self.resource, _RESOURCE_LABEL)
if isinstance(self.recipe_ids, (str, bytes)) or not isinstance(
self.recipe_ids, Sequence
):
raise TypeError("Plugin example recipe IDs must be a sequence")
recipe_ids = tuple(self.recipe_ids)
if not all(
isinstance(recipe_id, str) and recipe_id.count(":") == 1
for recipe_id in recipe_ids
):
raise ValueError("Plugin example recipe IDs must be namespaced")
if len(recipe_ids) != len(set(recipe_ids)):
raise ValueError("Plugin example recipe IDs must be unique")
object.__setattr__(self, "recipe_ids", recipe_ids)
if isinstance(self.expected_checks, (str, bytes)) or not isinstance(
self.expected_checks, Sequence
):
raise TypeError("Plugin example expected checks must be a sequence")
checks = tuple(self.expected_checks)
if not all(isinstance(check, str) and check.strip() for check in checks):
raise ValueError("Plugin example expected checks must be non-empty strings")
if len(checks) != len(set(checks)):
raise ValueError("Plugin example expected checks must be unique")
object.__setattr__(self, "expected_checks", checks)
@property
def package(self) -> str:
"""Return the importable package containing the example resource."""
if self.resource is None:
raise ValueError("Generated plugin example has no package resource")
return self.resource.split(":", maxsplit=1)[0]
@property
def resource_path(self) -> str:
"""Return the POSIX-style path relative to the resource package."""
if self.resource is None:
raise ValueError("Generated plugin example has no package resource")
return self.resource.split(":", maxsplit=1)[1]
[docs]
def resolve(self) -> Traversable:
"""Resolve the declared resource without requiring a filesystem path."""
if self.resource is None:
raise ValueError("Generated plugin example has no package resource")
return resolve_package_resource(self.resource, _RESOURCE_LABEL)
[docs]
@contextmanager
def as_file(self) -> Iterator[Path]:
"""Materialize the resource for APIs that require a filesystem path."""
with resources.as_file(self.resolve()) as filename:
yield Path(filename)
