ragraph.generic

Generic purpose classes and methods such as metadata models.

Metadata includes the kinds, labels, weights and annotations that we assign to our data objects. Standardizing this enables a more predictable experience.

Annotations

Annotations(*args, **kwargs)

Bases: Mapping

Miscellaneous properties mapping such as tool-specific metadata.

Source code in ragraph/generic.py

def __init__(self, *args, **kwargs):
    # The internal storage dictionary.
    self._data: Dict[str, Any] = dict()

    # Update the mapping with args and kwargs.
    self.update(*args, **kwargs)

Bound

Bound(
    value: Union[int, float],
    inclusive: bool = True,
    report: Optional[str] = None,
)

Numerical lower or upper bound for a value. Use with comparison operators.

Parameters:

Name	Type	Description	Default
value	Union[int, float]	Numerical bound value.	required
inclusive	bool	Whether the bound is inclusive.	True
report	Optional[str]	Whether to report an "error", "warn" or nothing (None) on bound violations.	None

Source code in ragraph/generic.py

def __init__(
    self,
    value: Union[int, float],
    inclusive: bool = True,
    report: Optional["str"] = None,
):
    self.value = value
    self.inclusive = inclusive
    self.report = report

as_dict

as_dict() -> Dict[str, Any]

Serializable dictionary representation.

Source code in ragraph/generic.py

def as_dict(self) -> Dict[str, Any]:
    """Serializable dictionary representation."""
    return dict(value=self.value, inclusive=self.inclusive, report=self.report)

ContinuousDomain

ContinuousDomain(
    lower: Optional[Bound] = None,
    upper: Optional[Bound] = None,
)

Numerical domain for a value. Use with "in" operator.

Source code in ragraph/generic.py

def __init__(
    self,
    lower: Optional[Bound] = None,
    upper: Optional[Bound] = None,
):
    self.lower = lower
    self.upper = upper

as_dict

as_dict() -> Dict[str, Any]

Serializable dictionary representation.

Source code in ragraph/generic.py

def as_dict(self) -> Dict[str, Any]:
    """Serializable dictionary representation."""
    return dict(
        lower=None if self.lower is None else self.lower.as_dict(),
        upper=None if self.upper is None else self.upper.as_dict(),
    )

Convention

Bases: StrEnum

Convention for matrix conversion to follow. Either:

• Inputs in rows, feedback above diagonal (IR_FAD)
• Inputs in columns, feedback below diagonal (IC_FBD)

IC_FBD class-attribute instance-attribute

IC_FBD = auto()

Inputs in columns, feedback below diagonal.

IR_FAD class-attribute instance-attribute

IR_FAD = auto()

Inputs in rows, feedback above diagonal.

Mapping

Mapping(*args, **kwargs)

A dictionary like object that with property-based access fields.

It's possible to include allowed keys, default values, and optional validators for certain keys/properties of a derived class.

from ragraph.generic import Mapping, field
def check_int(value):
    assert value == 1

class MyMap(Mapping):
    _protected = True
    _defaults = dict(myfield=1)
    _validators = dict(myfield=check_int)
    @field
    def myfield(self) -> int:
        '''My field's docstring'''

m = MyMap(myfield=3)
assert m.myfield == 3,            "This should return our set value."
assert m.myfield == m['myfield'], "A mapping works like a dictionary."
del(m.myfield)                    # This deletes our override.
m.validate()                      # Checks whether myfield == 1.

Source code in ragraph/generic.py

def __init__(self, *args, **kwargs):
    # The internal storage dictionary.
    self._data: Dict[str, Any] = dict()

    # Update the mapping with args and kwargs.
    self.update(*args, **kwargs)

as_dict

as_dict() -> Dict[str, Any]

Return a copy as a dictionary with all sub-mappings as dictionaries, too.

Source code in ragraph/generic.py

def as_dict(self) -> Dict[str, Any]:
    """Return a copy as a dictionary with all sub-mappings as dictionaries, too."""
    m = self._defaults.copy()
    m.update(self._data)
    for key, value in m.items():
        if isinstance(value, Mapping):
            m[key] = value.as_dict()
    return deepcopy(m)

get

get(key: str, fallback: Any = None) -> Any

Get an attribute with a fallback value if it isn't found.

Source code in ragraph/generic.py

def get(self, key: str, fallback: Any = None) -> Any:
    """Get an attribute with a fallback value if it isn't found."""
    return getattr(self, key, fallback)

items

items()

Get defaults and overrides as a (key, value) dict_items iterator.

Source code in ragraph/generic.py

def items(self):
    """Get defaults and overrides as a (key, value) dict_items iterator."""
    m = deepcopy(self._defaults)
    m.update(self._data)
    return m.items()

keys

keys() -> Set[str]

The keys in this mapping. (defaults and overrides).

Source code in ragraph/generic.py

def keys(self) -> Set[str]:
    """The keys in this mapping. (defaults and overrides)."""
    return set(self._defaults.keys()).union(set(self._data.keys()))

update

update(*args: Dict[str, Any], **kwargs: Any) -> None

Update multiple keys at once.

Parameters:

Name	Type	Description	Default
args	Dict[str, Any]	Dictionaries of key value pairs to update the set data with.	()
kwargs	Any	Keyword arguments to update the set data with.	{}

Source code in ragraph/generic.py

def update(self, *args: Dict[str, Any], **kwargs: Any) -> None:
    """Update multiple keys at once.

    Arguments:
        args: Dictionaries of key value pairs to update the set data with.
        kwargs: Keyword arguments to update the set data with.
    """
    for arg in args:
        try:
            for key, value in arg.items():
                setattr(self, key, value)
        except AttributeError:
            raise TypeError(f"Expected a dictionary or a Mapping. Found a {type(arg)}.")
    for key, value in kwargs.items():
        setattr(self, key, value)

validate

validate() -> None

Check whether the current data passes validation.

Source code in ragraph/generic.py

def validate(self) -> None:
    """Check whether the current data passes validation."""
    # Get the complete mapping (default overridden with set values).
    m = deepcopy(self._defaults)
    m.update(self._data)

    # Per key validation.
    for key, value in m.items():
        try:
            # Validate submappings.
            if isinstance(value, Mapping):
                value.validate()

            # Validate using validator if available.
            validator = self._validators.get(key)
            if validator is not None:
                validator(value)
        except Exception:
            raise MappingValidationError(
                f"Validation of '{key}' failed: '{value}' did not pass validation."
            )

    # Post-validation.
    try:
        self._post_validation()
    except Exception:
        raise MappingValidationError("Post-validation did not succeed.")

Metadata

Metadata(
    name: Optional[str] = None,
    kind: Optional[str] = None,
    labels: Optional[List[str]] = None,
    weights: Optional[Dict[str, Union[int, float]]] = None,
    annotations: Optional[
        Union[Annotations, Dict[str, Any]]
    ] = None,
    uuid: Optional[Union[str, UUID]] = None,
)

Metadata for graph elements.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Instance name. Set to a copy of the UUID if None provided.	None
kind	Optional[str]	Kind or main category of this instance.	None
labels	Optional[List[str]]	Labels categorizing this instance.	None
weights	Optional[Dict[str, Union[int, float]]]	Dictionary of weights attached to this instance.	None
annotations	Optional[Union[Annotations, Dict[str, Any]]]	Miscellaneous properties of this instance.	None
uuid	Optional[Union[str, UUID]]	UUID of this instance, generated when None provided.	None

Source code in ragraph/generic.py

def __init__(
    self,
    name: Optional[str] = None,
    kind: Optional[str] = None,
    labels: Optional[List[str]] = None,
    weights: Optional[Dict[str, Union[int, float]]] = None,
    annotations: Optional[Union[Annotations, Dict[str, Any]]] = None,
    uuid: Optional[Union[str, _uuid.UUID]] = None,
):
    self._uuid: _uuid.UUID
    self._name: str
    self._kind: str
    self._labels: List[str]
    self._weights: Dict[str, Union[int, float]]
    self._annotations: Annotations

    setattr(self, "uuid", uuid)
    setattr(self, "name", name)
    setattr(self, "kind", kind)
    setattr(self, "labels", labels)
    setattr(self, "weights", weights)
    setattr(self, "annotations", annotations)

annotations property writable

annotations: Annotations

Annotations of this instance.

Defaults to an empty Annotations instance.

kind property writable

kind: str

Kind or main category of this instance.

labels property writable

labels: List[str]

Labels categorizing this instance.

name property writable

name: str

Instance name. Given a UUID if None was provided.

uuid property writable

uuid: UUID

Instance UUID.

weight property

weight: Union[int, float]

Cumulative weight of this instance (read-only).

Returns the sum of self.weights.

weights property writable

weights: Dict[str, Union[int, float]]

Dictionary of weights attached to this instance.

MetadataFilter

MetadataFilter(
    uuids: Optional[
        Union[Iterable[str], Iterable[UUID]]
    ] = None,
    names: Optional[Iterable[str]] = None,
    kinds: Optional[Iterable[str]] = None,
    labels: Optional[Iterable[str]] = None,
    weights: Optional[Iterable[str]] = None,
    weight_domains: Optional[
        Dict[str, ContinuousDomain]
    ] = None,
    annotations: Optional[Iterable[str]] = None,
)

Metadata filtering options.

Parameters:

Name	Type	Description	Default
uuids	Optional[Union[Iterable[str], Iterable[UUID]]]	Filter by UUIDs.	None
names	Optional[Iterable[str]]	Filter by names.	None
kinds	Optional[Iterable[str]]	Filter by kinds.	None
labels	Optional[Iterable[str]]	Filter by labels. Items should match at least one.	None
weights	Optional[Iterable[str]]	Filter by weight labels. Items should match at least one.	None
weight_domains	Optional[Dict[str, ContinuousDomain]]	Filter items by weight domains. (upper/lower bound)	None
annotations	Optional[Iterable[str]]	Filter by annotation keys. Items should match at least one.	None

Source code in ragraph/generic.py

def __init__(
    self,
    uuids: Optional[Union[Iterable[str], Iterable[_uuid.UUID]]] = None,
    names: Optional[Iterable[str]] = None,
    kinds: Optional[Iterable[str]] = None,
    labels: Optional[Iterable[str]] = None,
    weights: Optional[Iterable[str]] = None,
    weight_domains: Optional[Dict[str, ContinuousDomain]] = None,
    annotations: Optional[Iterable[str]] = None,
):
    if uuids:
        self.uuids: Set[_uuid.UUID] = {
            i if isinstance(i, _uuid.UUID) else _uuid.UUID(i) for i in uuids
        }
    if names:
        self.names: Set[str] = set(names)
    if kinds:
        self.kinds: Set[str] = set(kinds)
    if labels:
        self.labels: Set[str] = set(labels)
    if weights:
        self.weights: Set[str] = set(weights)
    if weight_domains:
        self.weight_domains: Dict[str, ContinuousDomain] = weight_domains
    if annotations:
        self.annotations: Set[str] = set(annotations)

check_annotations

check_annotations(item: Metadata) -> bool

Check if item satisfies annotation keys filter.

Source code in ragraph/generic.py

def check_annotations(self, item: Metadata) -> bool:
    """Check if item satisfies annotation keys filter."""
    return any(self.annotations.intersection(item.annotations.keys()))

check_kinds

check_kinds(item: Metadata) -> bool

Check if item satisfies kinds filter.

Source code in ragraph/generic.py

def check_kinds(self, item: Metadata) -> bool:
    """Check if item satisfies kinds filter."""
    return item.kind in self.kinds

check_labels

check_labels(item: Metadata) -> bool

Check if item satisfies labels filter.

Source code in ragraph/generic.py

def check_labels(self, item: Metadata) -> bool:
    """Check if item satisfies labels filter."""
    return any(self.labels.intersection(item.labels))

check_names

check_names(item: Metadata) -> bool

Check if item satisfies names filter.

Source code in ragraph/generic.py

def check_names(self, item: Metadata) -> bool:
    """Check if item satisfies names filter."""
    return item.name in self.names

check_uuids

check_uuids(item: Metadata) -> bool

Check if item satisfies UUID filter.

Source code in ragraph/generic.py

def check_uuids(self, item: Metadata) -> bool:
    """Check if item satisfies UUID filter."""
    return item.uuid in self.uuids

check_weight_domains

check_weight_domains(item: Metadata) -> bool

Check if item satisfies weight domains filter.

Source code in ragraph/generic.py

def check_weight_domains(self, item: Metadata) -> bool:
    """Check if item satisfies weight domains filter."""
    return all(
        (key in item.weights and item.weights[key] in domain)
        for key, domain in self.weight_domains.items()
    )

check_weights

check_weights(item: Metadata) -> bool

Check if item satisfies weight keys filter.

Source code in ragraph/generic.py

def check_weights(self, item: Metadata) -> bool:
    """Check if item satisfies weight keys filter."""
    return any(self.weights.intersection(item.weights.keys()))

filter

filter(
    data: Iterable[Metadata], as_list: bool = True
) -> Union[List[Metadata], Generator[Metadata, None, None]]

Filter data using the set metadata filters.

Source code in ragraph/generic.py

def filter(
    self, data: Iterable[Metadata], as_list: bool = True
) -> Union[List[Metadata], Generator[Metadata, None, None]]:
    """Filter data using the set metadata filters."""

    def _generator():
        checks = self.get_checks()
        if checks:
            for item in data:
                if all(check(item) for check in checks):
                    yield item
        else:
            yield from data

    generator = _generator()
    return list(generator) if as_list else generator

get_checks

get_checks() -> List[Callable[[Metadata], bool]]

Active filter methods.

Source code in ragraph/generic.py

def get_checks(self) -> List[Callable[[Metadata], bool]]:
    """Active filter methods."""
    return [
        getattr(self, f"check_{filter}")
        for filter in self._filters
        if getattr(self, filter, False)
    ]

MetadataOptions

MetadataOptions(
    objects: Iterable[Metadata],
    skip_uuids: bool = False,
    skip_names: bool = False,
)

Seen values in an iterable of Metadata instances.

Parameters:

Name	Type	Description	Default
objects	Iterable[Metadata]	Objects derivated of the Metadata class.	required
skip_names	bool	Whether to skip the names field.	False

Source code in ragraph/generic.py

def __init__(
    self,
    objects: Iterable[Metadata],
    skip_uuids: bool = False,
    skip_names: bool = False,
):
    uuids: Optional[Set[_uuid.UUID]] = None if skip_uuids else set()
    names: Optional[Set[str]] = None if skip_names else set()

    kinds: Set[str] = set()
    labels: Set[str] = set()
    weights: Set[str] = set()
    weight_domains: Dict[str, ContinuousDomain] = defaultdict(ContinuousDomain)
    annotations: Set[str] = set()

    for obj in objects:
        if uuids is not None:
            uuids.add(obj.uuid)
        if names is not None:
            names.add(obj.name)
        kinds.add(obj.kind)
        labels.update(obj.labels)
        weights.update(obj.weights.keys())
        annotations.update(obj.annotations.keys())

        for k, v in obj.weights.items():
            domain = weight_domains[k]
            if domain.lower is None or v < domain.lower:
                domain.lower = Bound(value=v, inclusive=True, report=None)
            if domain.upper is None or v > domain.upper:
                domain.upper = Bound(value=v, inclusive=True, report=None)

    self.uuids = uuids
    self.names = names
    self.kinds = kinds
    self.labels = labels
    self.weights = weights
    self.weight_domains = weight_domains
    self.annotations = annotations

as_dict

as_dict() -> Dict[str, Any]

Serializable dictionary representation.

Source code in ragraph/generic.py

def as_dict(self) -> Dict[str, Any]:
    """Serializable dictionary representation."""
    result: Dict[str, Any] = dict(
        kinds=sorted(self.kinds),
        labels=sorted(self.labels),
        weights=sorted(self.weights),
        weight_domains={k: v.as_dict() for k, v in self.weight_domains.items()},
        annotations=sorted(self.annotations),
    )
    if self.uuids is not None:
        result["uuids"] = sorted(self.uuids, key=lambda x: str(x))
    if self.names is not None:
        result["names"] = sorted(self.names)

    return result

field

field(fget: Callable)

A Mapping field is a property that utilizes the Mapping's data.

Use it like the @property decorator and leave the function body blank (or pass):

By inspecting the wrapped method's name we derive the property's key.

Getting data from the mapping is done by retrieving the key from the set values. If that key is not found, an attempt is made on the defaults. An error is thrown when the key is neither to be found in the data or defaults.

Setting data checks for nested Mapping keys to update those and otherwise defaults to simply storing the value. Updating a nested Mapping field only updates the keys that are provided.

Deleting data pops the key from the set values, but does not remove any defaults.

Note

The method contents of a Mapping field are ignored.

Source code in ragraph/generic.py

def field(fget: Callable):
    """A [`Mapping`][ragraph.generic.Mapping] `field` is a property that utilizes the
    [`Mapping`][ragraph.generic.Mapping]'s data.

    Use it like the ``@property`` decorator and leave the function body blank (or pass):

    By inspecting the wrapped method's name we derive the property's key.

    Getting data from the mapping is done by retrieving the key from the set values. If that key is
    not found, an attempt is made on the defaults. An error is thrown when the key is neither to be
    found in the data or defaults.

    Setting data checks for nested [`Mapping`][ragraph.generic.Mapping] keys to update those and
    otherwise defaults to simply storing the value. Updating a nested
    [`Mapping`][ragraph.generic.Mapping] field only updates the keys that are provided.

    Deleting data pops the key from the set values, but does not remove any defaults.

    Note:
        The method contents of a [`Mapping`][ragraph.generic.Mapping] `field` are ignored.
    """
    key = fget.__name__

    def getter(self):
        """Get a mapping value."""
        if key in self._data:
            return self._data[key]

        if key in self._defaults:
            return self._defaults[key]

        raise AttributeError(f"Mapping does not contain a (default) value for key '{key}'.")

    def setter(self, value: Any):
        """Set a mapping value."""
        # Setting the mapping to None means resetting it to default.
        if value is None:
            self._data.pop(key, None)
            return

        # Handle nested mapping keys.
        if self._is_mapping(key):
            if key in self._data:
                # Mapping already set, update it.
                self._data[key].update(value)
            else:
                # Create a copy of the default and update that.
                self._data[key] = deepcopy(self._defaults.get(key))
                self._data[key].update(value)
            return

        # Otherwise just store the value.
        else:
            self._data[key] = value

    def deleter(self):
        """Delete a mapping value (e.g. unset it)."""
        self._data.pop(key, None)

    return property(fget=getter, fset=setter, fdel=deleter, doc=fget.__doc__)