Skip to content

hololinked.metadata.td.tm.ThingModel

Bases: WoTSchema, Metadata

Thing Model as per W3C WoT Thing Description v1.1.

Source code in repo/hololinked/hololinked/metadata/td/tm.py
class ThingModel(WoTSchema, Metadata):
    """
    Thing Model as per W3C WoT Thing Description v1.1.

    - [Specification](https://www.w3.org/TR/wot-thing-description11/)
    - [UML Diagram](https://docs.hololinked.dev/UML/PDF/ThingModel.pdf)
    """

    context: list[str | dict[str, str]] = Field(["https://www.w3.org/2022/wot/td/v1.1"], alias="@context")
    type: Optional[str | list[str]] = None
    id: Optional[str] = None
    title: Optional[str] = None
    description: Optional[str] = None
    version: Optional[VersionInfo] = None
    created: Optional[str] = None
    modified: Optional[str] = None
    support: Optional[str] = None
    base: Optional[str] = None
    properties: dict[str, DataSchema] = Field(default_factory=dict)
    actions: dict[str, ActionAffordance] = Field(default_factory=dict)
    events: dict[str, EventAffordance] = Field(default_factory=dict)

    model_config = ConfigDict(extra="allow")

    def __init__(
        self,
        thing: Thing | None = None,
        allow_loose_schema: Optional[bool] = False,
        ignore_errors: bool = False,
        skip_names: Optional[list[str]] = [],
    ) -> None:
        WoTSchema.__init__(self)
        Metadata.__init__(
            self,
            thing=thing,
            ignore_errors=ignore_errors,
            skip_names=skip_names,
        )
        self.allow_loose_schema = allow_loose_schema

    def generate(self) -> ThingModel:
        """
        Populate the thing model.

        Returns
        -------
        ThingModel

        Raises
        ------
        ValueError
            If the thing instance is not attached to the thing model. Usually, when API first approach is used,
            the thing instance could not yet be instantiated. This method is not to be used at that time.
        """
        if self.thing is None:
            raise ValueError("Thing Model instance not attached to a Thing instance to generate metadata.")
        self.id = self.thing.id
        self.title = self.thing.__class__.__name__
        self.context = ["https://www.w3.org/2022/wot/td/v1.1"]
        # default value of context is not being picked up although we only use exclude_unset=True
        if self.thing.__doc__:
            self.description = WoTSchema.format_doc(self.thing.__doc__)
        self.properties = dict()
        self.actions = dict()
        self.events = dict()
        self.add_interaction_affordances()
        return self

    def produce(self) -> Thing:
        """Produce a Thing instance from the Thing Model, not implemented yet."""
        raise NotImplementedError("This will be implemented in a future release for an API first approach")

    # not the best code and logic, but works for now
    skip_properties: list[str] = ["expose", "thing_description", "GUI", "object_info"]
    """list of property names to skip when generating the TD"""

    skip_actions: list[str] = [
        Thing._add_property.name,
        Thing._get_properties.name,
        Thing._get_properties_in_db.name,
        Thing._set_properties.name,
        "get_postman_collection",
        "get_our_thing_model",
    ]
    """list of action names to skip when generating the TD"""

    skip_events: list[str] = []
    """list of event names to skip when generating the TD"""

    def add_interaction_affordances(self) -> None:
        """
        Add interaction affordances to the thing model.

        Raises
        ------
        ValueError
            If the thing instance is not attached to the thing model. Usually, when API first approach is used,
            the thing instance could not yet be instantiated. This method is not to be used at that time.
        """
        if self.thing is None:
            raise ValueError("Thing Model instance not attached to a Thing instance to generate metadata.")
        affordances: list[
            tuple[str, ItemsView[str, Any], type[PropertyAffordance | ActionAffordance | EventAffordance], list[str]]
        ] = [
            ("properties", self.thing.properties.remote_objects.items(), PropertyAffordance, self.skip_properties),
            ("actions", self.thing.actions.descriptors.items(), ActionAffordance, self.skip_actions),
            ("events", self.thing.events.plain.items(), EventAffordance, self.skip_events),
        ]
        for affordance, items, affordance_cls, skip_list in affordances:
            for name, obj in items:
                if name in skip_list or name in self.skip_names:
                    continue
                if name == "state" and affordance == "properties" and not hasattr(self.thing, "state_machine"):
                    continue
                try:
                    affordance_dict = getattr(self, affordance)
                    affordance_dict[name] = affordance_cls.from_descriptor(obj, self.thing)
                except Exception as ex:
                    if not self.ignore_errors:
                        raise ex from None
                    if hasattr(self.thing, "logger"):
                        self.thing.logger.error(f"Error while generating schema for {name} - {ex}")

    def model_dump(self, **kwargs) -> dict[str, Any]:
        """
        Return the JSON representation.

        Returns
        -------
        dict[str, Any]
            JSON representation
        """

        def dump_value(value):
            nonlocal kwargs
            if hasattr(value, "model_dump"):
                return value.model_dump(**kwargs)
            elif isinstance(value, dict):
                return {k: dump_value(v) for k, v in value.items()}
            elif isinstance(value, list):
                return [dump_value(v) for v in value]
            elif isinstance(value, tuple):
                return tuple(dump_value(v) for v in value)
            else:
                return value

        result = {}
        for field in self.__class__.model_fields:
            if field in self.skip_keys:
                continue
            if not hasattr(self, field) or getattr(self, field) is None:
                continue
            if field in [
                "instance",
                "skip_keys",
                "skip_properties",
                "skip_actions",
                "skip_events",
                "skip_names",
                "ignore_errors",
                "allow_loose_schema",
            ]:
                continue
            value = getattr(self, field)
            result[field] = dump_value(value)
        return result

Attributes

skip_actions class-attribute instance-attribute

skip_actions: list[str] = [name, name, name, name, 'get_postman_collection', 'get_our_thing_model']

list of action names to skip when generating the TD

skip_events class-attribute instance-attribute

skip_events: list[str] = []

list of event names to skip when generating the TD

skip_properties class-attribute instance-attribute

skip_properties: list[str] = ['expose', 'thing_description', 'GUI', 'object_info']

list of property names to skip when generating the TD

Functions

add_interaction_affordances

add_interaction_affordances() -> None

Add interaction affordances to the thing model.

Raises:

Type Description
ValueError

If the thing instance is not attached to the thing model. Usually, when API first approach is used, the thing instance could not yet be instantiated. This method is not to be used at that time.

Source code in repo/hololinked/hololinked/metadata/td/tm.py
def add_interaction_affordances(self) -> None:
    """
    Add interaction affordances to the thing model.

    Raises
    ------
    ValueError
        If the thing instance is not attached to the thing model. Usually, when API first approach is used,
        the thing instance could not yet be instantiated. This method is not to be used at that time.
    """
    if self.thing is None:
        raise ValueError("Thing Model instance not attached to a Thing instance to generate metadata.")
    affordances: list[
        tuple[str, ItemsView[str, Any], type[PropertyAffordance | ActionAffordance | EventAffordance], list[str]]
    ] = [
        ("properties", self.thing.properties.remote_objects.items(), PropertyAffordance, self.skip_properties),
        ("actions", self.thing.actions.descriptors.items(), ActionAffordance, self.skip_actions),
        ("events", self.thing.events.plain.items(), EventAffordance, self.skip_events),
    ]
    for affordance, items, affordance_cls, skip_list in affordances:
        for name, obj in items:
            if name in skip_list or name in self.skip_names:
                continue
            if name == "state" and affordance == "properties" and not hasattr(self.thing, "state_machine"):
                continue
            try:
                affordance_dict = getattr(self, affordance)
                affordance_dict[name] = affordance_cls.from_descriptor(obj, self.thing)
            except Exception as ex:
                if not self.ignore_errors:
                    raise ex from None
                if hasattr(self.thing, "logger"):
                    self.thing.logger.error(f"Error while generating schema for {name} - {ex}")

generate

generate() -> ThingModel

Populate the thing model.

Returns:

Type Description
ThingModel

Raises:

Type Description
ValueError

If the thing instance is not attached to the thing model. Usually, when API first approach is used, the thing instance could not yet be instantiated. This method is not to be used at that time.

Source code in repo/hololinked/hololinked/metadata/td/tm.py
def generate(self) -> ThingModel:
    """
    Populate the thing model.

    Returns
    -------
    ThingModel

    Raises
    ------
    ValueError
        If the thing instance is not attached to the thing model. Usually, when API first approach is used,
        the thing instance could not yet be instantiated. This method is not to be used at that time.
    """
    if self.thing is None:
        raise ValueError("Thing Model instance not attached to a Thing instance to generate metadata.")
    self.id = self.thing.id
    self.title = self.thing.__class__.__name__
    self.context = ["https://www.w3.org/2022/wot/td/v1.1"]
    # default value of context is not being picked up although we only use exclude_unset=True
    if self.thing.__doc__:
        self.description = WoTSchema.format_doc(self.thing.__doc__)
    self.properties = dict()
    self.actions = dict()
    self.events = dict()
    self.add_interaction_affordances()
    return self

model_dump

model_dump(**kwargs) -> dict[str, Any]

Return the JSON representation.

Returns:

Type Description
dict[str, Any]

JSON representation

Source code in repo/hololinked/hololinked/metadata/td/tm.py
def model_dump(self, **kwargs) -> dict[str, Any]:
    """
    Return the JSON representation.

    Returns
    -------
    dict[str, Any]
        JSON representation
    """

    def dump_value(value):
        nonlocal kwargs
        if hasattr(value, "model_dump"):
            return value.model_dump(**kwargs)
        elif isinstance(value, dict):
            return {k: dump_value(v) for k, v in value.items()}
        elif isinstance(value, list):
            return [dump_value(v) for v in value]
        elif isinstance(value, tuple):
            return tuple(dump_value(v) for v in value)
        else:
            return value

    result = {}
    for field in self.__class__.model_fields:
        if field in self.skip_keys:
            continue
        if not hasattr(self, field) or getattr(self, field) is None:
            continue
        if field in [
            "instance",
            "skip_keys",
            "skip_properties",
            "skip_actions",
            "skip_events",
            "skip_names",
            "ignore_errors",
            "allow_loose_schema",
        ]:
            continue
        value = getattr(self, field)
        result[field] = dump_value(value)
    return result

produce

produce() -> Thing

Produce a Thing instance from the Thing Model, not implemented yet.

Source code in repo/hololinked/hololinked/metadata/td/tm.py
def produce(self) -> Thing:
    """Produce a Thing instance from the Thing Model, not implemented yet."""
    raise NotImplementedError("This will be implemented in a future release for an API first approach")