Skip to content

ThingDescriptionService

hololinked.server.mqtt.services.ThingDescriptionService

Generates MQTT Thing Descriptions for Things from Thing Models.

Source code in repo/hololinked/hololinked/server/mqtt/services.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class ThingDescriptionService:
    """Generates MQTT Thing Descriptions for `Thing`s from Thing Models."""

    def __init__(
        self,
        hostname: str,
        port: int,
        logger: structlog.stdlib.BoundLogger,
        thing: Thing,
        ssl: bool = True,
    ) -> None:
        """
        Initialize the Thing Description service.

        Parameters
        ----------
        hostname: str
            The MQTT broker hostname, to fill in the TD forms
        port: int
            The MQTT broker port, to fill in the TD forms
        logger: structlog.stdlib.BoundLogger
            The logger to use for logging messages
        thing: Thing
            The `Thing` whose description is generated
        ssl: bool
            Whether the broker is using SSL or not
        """
        self.hostname = hostname
        self.port = port
        self.logger = logger.bind(impl=self.__class__.__name__)
        self.thing = thing  # type: Thing
        self.ssl = ssl

    async def generate(
        self,
        ignore_errors: bool = False,
        skip_names: list[str] = [],
    ) -> dict[str, Any]:
        """
        Generates the Thing Description for the specified `Thing`, adds observable properties and events.

        Parameters
        ----------
        ignore_errors: bool
            Whether to ignore errors when generating the model and when adding properties/events to the TD
        skip_names: list[str]
            List of property/event names to skip when adding to the TD

        Returns
        -------
        dict[str, Any]
            The generated MQTT Thing Description
        """
        thing_model = self.thing.get_thing_model(ignore_errors=ignore_errors, skip_names=skip_names).json()
        TD = copy.deepcopy(thing_model)
        # remove actions as they dont push events
        TD.pop("actions", None)

        self.add_properties(TD, thing_model, ignore_errors, skip_names)
        self.add_events(TD, thing_model, ignore_errors, skip_names)

        return TD

    def add_properties(
        self,
        TD: dict[str, Any],
        thing_model: dict[str, Any],
        ignore_errors: bool = False,
        skip_names: list[str] = [],
    ) -> None:
        """
        Adds observable properties to the Thing Description with MQTT forms.

        Parameters
        ----------
        TD: dict[str, Any]
            The seed Thing Description to modify in place. This method does not have a return value, therefore
            just supply the TD dict and it will be modified. Non-observable properties will be removed.
        thing_model: dict[str, Any]
            The Thing Model the description is built from
        ignore_errors: bool
            Whether to ignore errors when adding properties to the TD
        skip_names: list[str]
            List of property names to skip when adding to the TD
        """
        for name in thing_model.get("properties", {}).keys():
            if name in skip_names:
                continue
            try:
                affordance = PropertyAffordance.from_TD(name, thing_model)
                if not affordance.observable:
                    TD["properties"].pop(name)
                    continue
                TD["properties"][name]["forms"] = []
                form = Form()
                form.op = Operations.observeproperty
                form.contentType = Serializers.for_object(TD["id"], self.thing.__class__.__name__, name).content_type
                form.href = f"mqtt{'s' if self.ssl else ''}://{self.hostname}:{self.port}"
                form.mqv_topic = f"{TD['id']}/{name}"
                TD["properties"][name]["forms"].append(form.json())
            except Exception as ex:
                if ignore_errors:
                    self.logger.warning(f"Could not add property {name} to MQTT TD: {ex}")
                    continue
                raise ex from None

    def add_events(
        self,
        TD: dict[str, Any],
        thing_model: dict[str, Any],
        ignore_errors: bool = False,
        skip_names: list[str] = [],
    ) -> None:
        """
        Adds events to the Thing Description with MQTT forms.

        Parameters
        ----------
        TD: dict[str, Any]
            The seed Thing Description to modify in place. This method does not have a return value, therefore
            just supply the TD dict and it will be modified.
        thing_model: dict[str, Any]
            The Thing Model the description is built from
        ignore_errors: bool
            Whether to ignore errors when adding events to the TD
        skip_names: list[str]
            List of event names to skip when adding to the TD
        """
        # repurpose event
        for name in thing_model.get("events", {}).keys():
            if name in skip_names:
                continue
            try:
                EventAffordance.from_TD(name, thing_model)
                TD["events"][name]["forms"] = []
                form = Form()
                form.op = Operations.subscribeevent
                form.contentType = Serializers.for_object(TD["id"], self.thing.__class__.__name__, name).content_type
                form.href = f"mqtt{'s' if self.ssl else ''}://{self.hostname}:{self.port}"
                form.mqv_topic = f"{TD['id']}/{name}"
                TD["events"][name]["forms"].append(form.json())
            except Exception as ex:
                if ignore_errors:
                    self.logger.warning(f"Could not add event {name} to MQTT TD: {ex}")
                    continue
                raise ex from None

Attributes

thing instance-attribute 

thing = thing

hostname instance-attribute 

hostname = hostname

port instance-attribute 

port = port

ssl instance-attribute 

ssl = ssl

logger instance-attribute 

logger = bind(impl=__name__)

Functions

__init__ 

__init__(hostname: str, port: int, logger: BoundLogger, thing: Thing, ssl: bool = True) -> None

Initialize the Thing Description service.

Parameters:

Name Type Description Default
hostname
str

The MQTT broker hostname, to fill in the TD forms

 required
port
int

The MQTT broker port, to fill in the TD forms

 required
logger
BoundLogger

The logger to use for logging messages

 required
thing
Thing

The Thing whose description is generated

 required
ssl
bool

Whether the broker is using SSL or not

 True
Source code in repo/hololinked/hololinked/server/mqtt/services.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    hostname: str,
    port: int,
    logger: structlog.stdlib.BoundLogger,
    thing: Thing,
    ssl: bool = True,
) -> None:
    """
    Initialize the Thing Description service.

    Parameters
    ----------
    hostname: str
        The MQTT broker hostname, to fill in the TD forms
    port: int
        The MQTT broker port, to fill in the TD forms
    logger: structlog.stdlib.BoundLogger
        The logger to use for logging messages
    thing: Thing
        The `Thing` whose description is generated
    ssl: bool
        Whether the broker is using SSL or not
    """
    self.hostname = hostname
    self.port = port
    self.logger = logger.bind(impl=self.__class__.__name__)
    self.thing = thing  # type: Thing
    self.ssl = ssl

generate async 

generate(ignore_errors: bool = False, skip_names: list[str] = []) -> dict[str, Any]

Generates the Thing Description for the specified Thing, adds observable properties and events.

Parameters:

Name Type Description Default
ignore_errors
bool

Whether to ignore errors when generating the model and when adding properties/events to the TD

 False
skip_names
list[str]

List of property/event names to skip when adding to the TD

 []

Returns:

Type Description
dict[str, Any]

The generated MQTT Thing Description
Source code in repo/hololinked/hololinked/server/mqtt/services.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
async def generate(
    self,
    ignore_errors: bool = False,
    skip_names: list[str] = [],
) -> dict[str, Any]:
    """
    Generates the Thing Description for the specified `Thing`, adds observable properties and events.

    Parameters
    ----------
    ignore_errors: bool
        Whether to ignore errors when generating the model and when adding properties/events to the TD
    skip_names: list[str]
        List of property/event names to skip when adding to the TD

    Returns
    -------
    dict[str, Any]
        The generated MQTT Thing Description
    """
    thing_model = self.thing.get_thing_model(ignore_errors=ignore_errors, skip_names=skip_names).json()
    TD = copy.deepcopy(thing_model)
    # remove actions as they dont push events
    TD.pop("actions", None)

    self.add_properties(TD, thing_model, ignore_errors, skip_names)
    self.add_events(TD, thing_model, ignore_errors, skip_names)

    return TD

add_properties 

add_properties(TD: dict[str, Any], thing_model: dict[str, Any], ignore_errors: bool = False, skip_names: list[str] = []) -> None

Adds observable properties to the Thing Description with MQTT forms.

Parameters:

Name Type Description Default
TD
dict[str, Any]

The seed Thing Description to modify in place. This method does not have a return value, therefore just supply the TD dict and it will be modified. Non-observable properties will be removed.

 required
thing_model
dict[str, Any]

The Thing Model the description is built from

 required
ignore_errors
bool

Whether to ignore errors when adding properties to the TD

 False
skip_names
list[str]

List of property names to skip when adding to the TD

 []
Source code in repo/hololinked/hololinked/server/mqtt/services.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def add_properties(
    self,
    TD: dict[str, Any],
    thing_model: dict[str, Any],
    ignore_errors: bool = False,
    skip_names: list[str] = [],
) -> None:
    """
    Adds observable properties to the Thing Description with MQTT forms.

    Parameters
    ----------
    TD: dict[str, Any]
        The seed Thing Description to modify in place. This method does not have a return value, therefore
        just supply the TD dict and it will be modified. Non-observable properties will be removed.
    thing_model: dict[str, Any]
        The Thing Model the description is built from
    ignore_errors: bool
        Whether to ignore errors when adding properties to the TD
    skip_names: list[str]
        List of property names to skip when adding to the TD
    """
    for name in thing_model.get("properties", {}).keys():
        if name in skip_names:
            continue
        try:
            affordance = PropertyAffordance.from_TD(name, thing_model)
            if not affordance.observable:
                TD["properties"].pop(name)
                continue
            TD["properties"][name]["forms"] = []
            form = Form()
            form.op = Operations.observeproperty
            form.contentType = Serializers.for_object(TD["id"], self.thing.__class__.__name__, name).content_type
            form.href = f"mqtt{'s' if self.ssl else ''}://{self.hostname}:{self.port}"
            form.mqv_topic = f"{TD['id']}/{name}"
            TD["properties"][name]["forms"].append(form.json())
        except Exception as ex:
            if ignore_errors:
                self.logger.warning(f"Could not add property {name} to MQTT TD: {ex}")
                continue
            raise ex from None

add_events 

add_events(TD: dict[str, Any], thing_model: dict[str, Any], ignore_errors: bool = False, skip_names: list[str] = []) -> None

Adds events to the Thing Description with MQTT forms.

Parameters:

Name Type Description Default
TD
dict[str, Any]

The seed Thing Description to modify in place. This method does not have a return value, therefore just supply the TD dict and it will be modified.

 required
thing_model
dict[str, Any]

The Thing Model the description is built from

 required
ignore_errors
bool

Whether to ignore errors when adding events to the TD

 False
skip_names
list[str]

List of event names to skip when adding to the TD

 []
Source code in repo/hololinked/hololinked/server/mqtt/services.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def add_events(
    self,
    TD: dict[str, Any],
    thing_model: dict[str, Any],
    ignore_errors: bool = False,
    skip_names: list[str] = [],
) -> None:
    """
    Adds events to the Thing Description with MQTT forms.

    Parameters
    ----------
    TD: dict[str, Any]
        The seed Thing Description to modify in place. This method does not have a return value, therefore
        just supply the TD dict and it will be modified.
    thing_model: dict[str, Any]
        The Thing Model the description is built from
    ignore_errors: bool
        Whether to ignore errors when adding events to the TD
    skip_names: list[str]
        List of event names to skip when adding to the TD
    """
    # repurpose event
    for name in thing_model.get("events", {}).keys():
        if name in skip_names:
            continue
        try:
            EventAffordance.from_TD(name, thing_model)
            TD["events"][name]["forms"] = []
            form = Form()
            form.op = Operations.subscribeevent
            form.contentType = Serializers.for_object(TD["id"], self.thing.__class__.__name__, name).content_type
            form.href = f"mqtt{'s' if self.ssl else ''}://{self.hostname}:{self.port}"
            form.mqv_topic = f"{TD['id']}/{name}"
            TD["events"][name]["forms"].append(form.json())
        except Exception as ex:
            if ignore_errors:
                self.logger.warning(f"Could not add event {name} to MQTT TD: {ex}")
                continue
            raise ex from None