Classes:

• DistributionModel –

  Model object for probability distributions

Functions:

• distrib_from_sampled_quantity –

  Extract a distribution from a :SampledQuantity node through :from-distribution path.

• sample_from_distrib –

  Sample from a distribution model based on its type.

DistributionModel

DistributionModel(distrib_id, graph)

Bases: ModelBase

Model object for probability distributions

Attributes:

• distrib_type (URIRef) –

  the type of distribution to be handled

Parameters:

• distrib_id (URIRef) –

  URI of the distribution in the graph

• graph (Graph) –

  RDF graph for loading attributes

Source code in src/rdf_utils/models/distribution.py

def __init__(self, distrib_id: URIRef, graph: Graph) -> None:
    super().__init__(node_id=distrib_id, graph=graph)

    if URI_DISTRIB_TYPE_UNIFORM_ROT in self.types:
        self.distrib_type = URI_DISTRIB_TYPE_UNIFORM_ROT
    elif URI_DISTRIB_TYPE_UNIFORM in self.types:
        self.distrib_type = URI_DISTRIB_TYPE_UNIFORM
        self._load_uniform_distrib_attrs(graph=graph)
    elif URI_DISTRIB_TYPE_NORMAL in self.types:
        self.distrib_type = URI_DISTRIB_TYPE_NORMAL
        self._load_normal_distrib_attrs(graph=graph)
    else:
        raise RuntimeError(f"Distrib '{self.id}' has unhandled types: {self.types}")

distrib_from_sampled_quantity

distrib_from_sampled_quantity(quantity_id, graph)

Extract a distribution from a :SampledQuantity node through :from-distribution path.

Parameters:

• quantity_id (URIRef) –

  URI of the :SampledQuantity node

• graph (Graph) –

  RDF graph to look for distribution nodes and attributes

Returns:

• DistributionModel –

  distribution model object

Source code in src/rdf_utils/models/distribution.py

def distrib_from_sampled_quantity(quantity_id: URIRef, graph: Graph) -> DistributionModel:
    """Extract a distribution from a :SampledQuantity node through :from-distribution path.

    Parameters:
        quantity_id: URI of the :SampledQuantity node
        graph: RDF graph to look for distribution nodes and attributes

    Returns:
        distribution model object
    """
    distrib_id = graph.value(subject=quantity_id, predicate=URI_DISTRIB_PRED_FROM_DISTRIB)
    assert isinstance(
        distrib_id, URIRef
    ), f"Node '{quantity_id}' does not link to a distribution node: {distrib_id}"
    return DistributionModel(distrib_id=distrib_id, graph=graph)

sample_from_distrib

sample_from_distrib(distrib, size=None)

Sample from a distribution model based on its type.

Parameters:

• distrib (DistributionModel) –

  distribution model

• size (Optional[int | tuple[int, ...]], default: None ) –

  Size of the sample, which matches size argument in numpy.random calls. Will be ignored for random rotations at the moment. For uniform and normal distribs, tuple size should have last dimension matching the distrib's dimension.

Returns:

• Any –

  distribution sample with dimension matching given size

Source code in src/rdf_utils/models/distribution.py

def sample_from_distrib(
    distrib: DistributionModel, size: Optional[int | tuple[int, ...]] = None
) -> Any:
    """Sample from a distribution model based on its type.

    Parameters:
        distrib: distribution model
        size: Size of the sample, which matches size argument in numpy.random calls.
              Will be ignored for random rotations at the moment. For uniform and normal distribs,
              tuple size should have last dimension matching the distrib's dimension.

    Returns:
        distribution sample with dimension matching given size
    """
    if URI_DISTRIB_TYPE_UNIFORM_ROT in distrib.types:
        try:
            from scipy.spatial.transform import Rotation
        except ImportError:
            raise RuntimeError("to sample random rotations, 'scipy' must be installed")

        return Rotation.random()

    if URI_DISTRIB_TYPE_UNIFORM in distrib.types:
        lower_bounds = distrib.get_attr(key=URI_DISTRIB_PRED_LOWER)
        upper_bounds = distrib.get_attr(key=URI_DISTRIB_PRED_UPPER)
        assert isinstance(lower_bounds, list) and isinstance(
            upper_bounds, list
        ), f"Uniform distrib '{distrib.id}' does not have valid lower & upper bounds"
        return np.random.uniform(lower_bounds, upper_bounds, size=size)

    if URI_DISTRIB_TYPE_NORMAL in distrib.types:
        dim = distrib.get_attr(key=URI_DISTRIB_PRED_DIM)
        assert (
            isinstance(dim, int) and dim > 0
        ), f"Normal distrib '{distrib.id}' does not have valid dimension: {dim}"

        mean = distrib.get_attr(key=URI_DISTRIB_PRED_MEAN)
        assert (
            isinstance(mean, list) and len(mean) == dim
        ), f"Normal distrib '{distrib.id}' does not have valid mean: {mean}"

        if dim == 1:
            std = distrib.get_attr(key=URI_DISTRIB_PRED_STD)
            assert isinstance(
                std, float
            ), f"Normal distrib '{distrib.id}' does not have valid standard deviation: {std}"
            return np.random.normal(loc=mean[0], scale=std, size=size)

        # multivariate normal
        cov = distrib.get_attr(key=URI_DISTRIB_PRED_COV)
        assert isinstance(
            cov, np.ndarray
        ), f"Normal distrib '{distrib.id}' does not have valid covariance: {cov}"
        return np.random.multivariate_normal(mean=mean, cov=cov, size=size)

    raise RuntimeError(f"Distrib '{distrib.id}' has unhandled types: {distrib.types}")