Skip to content

ThingDescriptionService

hololinked.server.zmq.services.ThingDescriptionService

Generates Thing Descriptions for Things served over ZMQ.

This object would be a service in layered architecture.

Source code in repo/hololinked/hololinked/server/zmq/services.py 
 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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
class ThingDescriptionService:
    """
    Generates Thing Descriptions for `Thing`s served over ZMQ.

    This object would be a service in layered architecture.
    """

    def __init__(self, server: ZMQServer, logger: structlog.stdlib.BoundLogger) -> None:
        """
        Initialize the Thing Description service.

        Parameters
        ----------
        server: ZMQServer
            the server whose sockets the forms address - a Thing Description is mostly forms, and a
            form is an address on one of them, which is why only this server can build one
        logger: structlog.stdlib.BoundLogger
            The logger to use for logging messages
        """
        self.server = server
        self.logger = logger.bind(layer="service", impl=self.__class__.__name__)

    def socket_addresses_for(self, protocol: str) -> tuple[str, str]:
        """
        The request and event socket addresses a form should carry for one transport.

        Parameters
        ----------
        protocol: str
            the transport to address - `INPROC`, `IPC` or `TCP`

        Returns
        -------
        tuple[str, str]
            the REQ-REP address and the PUB-SUB address, in that order

        Raises
        ------
        RuntimeError
            if the server does not serve that transport
        ValueError
            if the protocol is not one of `INPROC`, `IPC` or `TCP`
        """
        server = self.server
        if protocol.lower() == "inproc":
            if server.inproc_server is None or server.inproc_event_publisher is None:
                raise RuntimeError(
                    "This server cannot generate TD for INPROC protocol, consider using thing model directly."
                )
            return server.inproc_server.socket_address, server.inproc_event_publisher.socket_address
        if protocol.lower() == "ipc":
            if server.ipc_server is None or server.ipc_event_publisher is None:
                raise RuntimeError(
                    "This server cannot generate TD for IPC protocol, consider using thing model directly."
                )
            return server.ipc_server.socket_address, server.ipc_event_publisher.socket_address
        if protocol.lower() == "tcp":
            if server.tcp_server is None or server.tcp_event_publisher is None:
                raise RuntimeError(
                    "This server cannot generate TD for TCP protocol, consider using thing model directly."
                )
            # a socket bound to every interface is not an address a client can dial
            hostname = socket.gethostname()
            req_rep_socket_address = server.tcp_server.socket_address.replace(
                "*", hostname
            ).replace(
                "0.0.0.0",
                hostname,
            )  # SAST(id='hololinked.server.zmq.services.ThingDescriptionService.socket_addresses_for.req_rep_socket_address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
            pub_sub_socket_address = server.tcp_event_publisher.socket_address.replace(
                "*", hostname
            ).replace(
                "0.0.0.0",
                hostname,
            )  # SAST(id='hololinked.server.zmq.services.ThingDescriptionService.socket_addresses_for.pub_sub_socket_address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
            return req_rep_socket_address, pub_sub_socket_address
        raise ValueError(f"Unsupported protocol '{protocol}' for ZMQ.")

    async def generate(
        self,
        thing: Thing,
        protocol: str,
        ignore_errors: bool = False,
        skip_names: list[str] = [],
    ) -> dict[str, Any]:
        """
        Generate the Thing Description for one `Thing`, with ZMQ forms for one transport.

        Parameters
        ----------
        thing: Thing
            The `Thing` whose description is generated
        protocol: str
            The protocol for which to generate the TD - `INPROC`, `IPC` or `TCP`
        ignore_errors: bool
            Whether to ignore errors while generating the TD. Default is False.
        skip_names: list[str]
            List of property, action or event names to skip while generating the TD.

        Returns
        -------
        dict[str, Any]
            The Thing Description in JSON format.

        Raises
        ------
        RuntimeError
            if the server does not serve the requested protocol
        ValueError
            if the protocol is not one of `INPROC`, `IPC` or `TCP`
        """
        TM = thing.get_thing_model(ignore_errors=ignore_errors, skip_names=skip_names).json()  # type: dict[str, Any]
        TD = copy.deepcopy(TM)
        req_rep_socket_address, pub_sub_socket_address = self.socket_addresses_for(protocol)

        self.add_properties(TD, TM, thing, req_rep_socket_address, pub_sub_socket_address, ignore_errors)
        self.add_actions(TD, TM, thing, req_rep_socket_address, ignore_errors)
        self.add_events(TD, TM, thing, pub_sub_socket_address, ignore_errors)

        return TD

    def content_type_for(self, thing: Thing, name: str) -> str:
        """
        The content type one objekt's payloads are declared with.

        Parameters
        ----------
        thing: Thing
            the `Thing` the objekt belongs to
        name: str
            name of the property, action or event

        Returns
        -------
        str
            the registered content type, or the one its serializer declares
        """
        content_type = Serializers.get_content_type_for_object(thing.id, thing.__class__.__name__, name)
        if not content_type:
            content_type = Serializers.for_object(thing.id, thing.__class__.__name__, name).content_type
        return content_type

    def add_properties(
        self,
        TD: dict[str, Any],
        TM: dict[str, Any],
        thing: Thing,
        req_rep_socket_address: str,
        pub_sub_socket_address: str,
        ignore_errors: bool,
    ) -> None:
        """
        Add properties to the TD with ZMQ forms, one per operation the property supports.

        Parameters
        ----------
        TD: dict[str, Any]
            The Thing Description to modify in place
        TM: dict[str, Any]
            The Thing Model the description is built from
        thing: Thing
            The `Thing` being described
        req_rep_socket_address: str
            address reads and writes are addressed to
        pub_sub_socket_address: str
            address change events are addressed to
        ignore_errors: bool
            Whether to warn and carry on rather than raise
        """
        for name in TM.get("properties", []):
            try:
                affordance = PropertyAffordance.from_TD(name, TM)
                if not TD["properties"][name].get("forms", None):
                    TD["properties"][name]["forms"] = []
                content_type = self.content_type_for(thing, name)

                form = Form()
                form.href = req_rep_socket_address
                form.op = Operations.readproperty
                form.contentType = content_type
                TD["properties"][name]["forms"].append(form.json())

                if not affordance.readOnly:
                    form = Form()
                    form.href = req_rep_socket_address
                    form.op = Operations.writeproperty
                    form.contentType = content_type
                    TD["properties"][name]["forms"].append(form.json())

                if affordance.observable:
                    form = Form()
                    form.href = pub_sub_socket_address
                    form.op = Operations.observeproperty
                    form.contentType = content_type
                    TD["properties"][name]["forms"].append(form.json())
            except Exception as ex:
                if not ignore_errors:
                    raise ex from None
                thing.logger.warning(
                    "error while generating TD forms for property",
                    name=name,
                    error=str(ex),
                )

    def add_actions(
        self,
        TD: dict[str, Any],
        TM: dict[str, Any],
        thing: Thing,
        req_rep_socket_address: str,
        ignore_errors: bool,
    ) -> None:
        """
        Add actions to the TD with ZMQ forms.

        Parameters
        ----------
        TD: dict[str, Any]
            The Thing Description to modify in place
        TM: dict[str, Any]
            The Thing Model the description is built from
        thing: Thing
            The `Thing` being described
        req_rep_socket_address: str
            address invocations are addressed to
        ignore_errors: bool
            Whether to warn and carry on rather than raise
        """
        for name in TM.get("actions", []):
            try:
                ActionAffordance.from_TD(name, TM)
                if not TD["actions"][name].get("forms", None):
                    TD["actions"][name]["forms"] = []

                form = Form()
                form.href = req_rep_socket_address
                form.op = Operations.invokeaction
                form.contentType = self.content_type_for(thing, name)
                TD["actions"][name]["forms"].append(form.json())
            except Exception as ex:
                if not ignore_errors:
                    raise ex from None
                thing.logger.warning(
                    "error while generating TD forms for action",
                    name=name,
                    error=str(ex),
                )

    def add_events(
        self,
        TD: dict[str, Any],
        TM: dict[str, Any],
        thing: Thing,
        pub_sub_socket_address: str,
        ignore_errors: bool,
    ) -> None:
        """
        Add events to the TD with ZMQ forms.

        Parameters
        ----------
        TD: dict[str, Any]
            The Thing Description to modify in place
        TM: dict[str, Any]
            The Thing Model the description is built from
        thing: Thing
            The `Thing` being described
        pub_sub_socket_address: str
            address subscriptions are addressed to
        ignore_errors: bool
            Whether to warn and carry on rather than raise
        """
        for name in TM.get("events", []):
            try:
                EventAffordance.from_TD(name, TM)
                if not TD["events"][name].get("forms", None):
                    TD["events"][name]["forms"] = []

                form = Form()
                form.href = pub_sub_socket_address
                form.op = Operations.subscribeevent
                form.contentType = self.content_type_for(thing, name)
                TD["events"][name]["forms"].append(form.json())
            except Exception as ex:
                if not ignore_errors:
                    raise ex from None
                thing.logger.warning(
                    "error while generating TD forms for event",
                    name=name,
                    error=str(ex),
                )

Attributes

server instance-attribute 

server = server

logger instance-attribute 

logger = bind(layer='service', impl=__name__)

Functions

__init__ 

__init__(server: ZMQServer, logger: BoundLogger) -> None

Initialize the Thing Description service.

Parameters:

Name Type Description Default
server
ZMQServer

the server whose sockets the forms address - a Thing Description is mostly forms, and a form is an address on one of them, which is why only this server can build one

 required
logger
BoundLogger

The logger to use for logging messages

 required
Source code in repo/hololinked/hololinked/server/zmq/services.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, server: ZMQServer, logger: structlog.stdlib.BoundLogger) -> None:
    """
    Initialize the Thing Description service.

    Parameters
    ----------
    server: ZMQServer
        the server whose sockets the forms address - a Thing Description is mostly forms, and a
        form is an address on one of them, which is why only this server can build one
    logger: structlog.stdlib.BoundLogger
        The logger to use for logging messages
    """
    self.server = server
    self.logger = logger.bind(layer="service", impl=self.__class__.__name__)

generate async 

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

Generate the Thing Description for one Thing, with ZMQ forms for one transport.

Parameters:

Name Type Description Default
thing
Thing

The Thing whose description is generated

 required
protocol
str

The protocol for which to generate the TD - INPROC, IPC or TCP

 required
ignore_errors
bool

Whether to ignore errors while generating the TD. Default is False.

 False
skip_names
list[str]

List of property, action or event names to skip while generating the TD.

 []

Returns:

Type Description
dict[str, Any]

The Thing Description in JSON format.

Raises:

Type Description
RuntimeError

if the server does not serve the requested protocol
ValueError

if the protocol is not one of INPROC, IPC or TCP
Source code in repo/hololinked/hololinked/server/zmq/services.py 
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
async def generate(
    self,
    thing: Thing,
    protocol: str,
    ignore_errors: bool = False,
    skip_names: list[str] = [],
) -> dict[str, Any]:
    """
    Generate the Thing Description for one `Thing`, with ZMQ forms for one transport.

    Parameters
    ----------
    thing: Thing
        The `Thing` whose description is generated
    protocol: str
        The protocol for which to generate the TD - `INPROC`, `IPC` or `TCP`
    ignore_errors: bool
        Whether to ignore errors while generating the TD. Default is False.
    skip_names: list[str]
        List of property, action or event names to skip while generating the TD.

    Returns
    -------
    dict[str, Any]
        The Thing Description in JSON format.

    Raises
    ------
    RuntimeError
        if the server does not serve the requested protocol
    ValueError
        if the protocol is not one of `INPROC`, `IPC` or `TCP`
    """
    TM = thing.get_thing_model(ignore_errors=ignore_errors, skip_names=skip_names).json()  # type: dict[str, Any]
    TD = copy.deepcopy(TM)
    req_rep_socket_address, pub_sub_socket_address = self.socket_addresses_for(protocol)

    self.add_properties(TD, TM, thing, req_rep_socket_address, pub_sub_socket_address, ignore_errors)
    self.add_actions(TD, TM, thing, req_rep_socket_address, ignore_errors)
    self.add_events(TD, TM, thing, pub_sub_socket_address, ignore_errors)

    return TD

socket_addresses_for 

socket_addresses_for(protocol: str) -> tuple[str, str]

The request and event socket addresses a form should carry for one transport.

Parameters:

Name Type Description Default
protocol
str

the transport to address - INPROC, IPC or TCP

 required

Returns:

Type Description
tuple[str, str]

the REQ-REP address and the PUB-SUB address, in that order

Raises:

Type Description
RuntimeError

if the server does not serve that transport
ValueError

if the protocol is not one of INPROC, IPC or TCP
Source code in repo/hololinked/hololinked/server/zmq/services.py 
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
def socket_addresses_for(self, protocol: str) -> tuple[str, str]:
    """
    The request and event socket addresses a form should carry for one transport.

    Parameters
    ----------
    protocol: str
        the transport to address - `INPROC`, `IPC` or `TCP`

    Returns
    -------
    tuple[str, str]
        the REQ-REP address and the PUB-SUB address, in that order

    Raises
    ------
    RuntimeError
        if the server does not serve that transport
    ValueError
        if the protocol is not one of `INPROC`, `IPC` or `TCP`
    """
    server = self.server
    if protocol.lower() == "inproc":
        if server.inproc_server is None or server.inproc_event_publisher is None:
            raise RuntimeError(
                "This server cannot generate TD for INPROC protocol, consider using thing model directly."
            )
        return server.inproc_server.socket_address, server.inproc_event_publisher.socket_address
    if protocol.lower() == "ipc":
        if server.ipc_server is None or server.ipc_event_publisher is None:
            raise RuntimeError(
                "This server cannot generate TD for IPC protocol, consider using thing model directly."
            )
        return server.ipc_server.socket_address, server.ipc_event_publisher.socket_address
    if protocol.lower() == "tcp":
        if server.tcp_server is None or server.tcp_event_publisher is None:
            raise RuntimeError(
                "This server cannot generate TD for TCP protocol, consider using thing model directly."
            )
        # a socket bound to every interface is not an address a client can dial
        hostname = socket.gethostname()
        req_rep_socket_address = server.tcp_server.socket_address.replace(
            "*", hostname
        ).replace(
            "0.0.0.0",
            hostname,
        )  # SAST(id='hololinked.server.zmq.services.ThingDescriptionService.socket_addresses_for.req_rep_socket_address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
        pub_sub_socket_address = server.tcp_event_publisher.socket_address.replace(
            "*", hostname
        ).replace(
            "0.0.0.0",
            hostname,
        )  # SAST(id='hololinked.server.zmq.services.ThingDescriptionService.socket_addresses_for.pub_sub_socket_address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
        return req_rep_socket_address, pub_sub_socket_address
    raise ValueError(f"Unsupported protocol '{protocol}' for ZMQ.")

content_type_for 

content_type_for(thing: Thing, name: str) -> str

The content type one objekt's payloads are declared with.

Parameters:

Name Type Description Default
thing
Thing

the Thing the objekt belongs to

 required
name
str

name of the property, action or event

 required

Returns:

Type Description
str

the registered content type, or the one its serializer declares
Source code in repo/hololinked/hololinked/server/zmq/services.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def content_type_for(self, thing: Thing, name: str) -> str:
    """
    The content type one objekt's payloads are declared with.

    Parameters
    ----------
    thing: Thing
        the `Thing` the objekt belongs to
    name: str
        name of the property, action or event

    Returns
    -------
    str
        the registered content type, or the one its serializer declares
    """
    content_type = Serializers.get_content_type_for_object(thing.id, thing.__class__.__name__, name)
    if not content_type:
        content_type = Serializers.for_object(thing.id, thing.__class__.__name__, name).content_type
    return content_type

add_properties 

add_properties(TD: dict[str, Any], TM: dict[str, Any], thing: Thing, req_rep_socket_address: str, pub_sub_socket_address: str, ignore_errors: bool) -> None

Add properties to the TD with ZMQ forms, one per operation the property supports.

Parameters:

Name Type Description Default
TD
dict[str, Any]

The Thing Description to modify in place

 required
TM
dict[str, Any]

The Thing Model the description is built from

 required
thing
Thing

The Thing being described

 required
req_rep_socket_address
str

address reads and writes are addressed to

 required
pub_sub_socket_address
str

address change events are addressed to

 required
ignore_errors
bool

Whether to warn and carry on rather than raise

 required
Source code in repo/hololinked/hololinked/server/zmq/services.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def add_properties(
    self,
    TD: dict[str, Any],
    TM: dict[str, Any],
    thing: Thing,
    req_rep_socket_address: str,
    pub_sub_socket_address: str,
    ignore_errors: bool,
) -> None:
    """
    Add properties to the TD with ZMQ forms, one per operation the property supports.

    Parameters
    ----------
    TD: dict[str, Any]
        The Thing Description to modify in place
    TM: dict[str, Any]
        The Thing Model the description is built from
    thing: Thing
        The `Thing` being described
    req_rep_socket_address: str
        address reads and writes are addressed to
    pub_sub_socket_address: str
        address change events are addressed to
    ignore_errors: bool
        Whether to warn and carry on rather than raise
    """
    for name in TM.get("properties", []):
        try:
            affordance = PropertyAffordance.from_TD(name, TM)
            if not TD["properties"][name].get("forms", None):
                TD["properties"][name]["forms"] = []
            content_type = self.content_type_for(thing, name)

            form = Form()
            form.href = req_rep_socket_address
            form.op = Operations.readproperty
            form.contentType = content_type
            TD["properties"][name]["forms"].append(form.json())

            if not affordance.readOnly:
                form = Form()
                form.href = req_rep_socket_address
                form.op = Operations.writeproperty
                form.contentType = content_type
                TD["properties"][name]["forms"].append(form.json())

            if affordance.observable:
                form = Form()
                form.href = pub_sub_socket_address
                form.op = Operations.observeproperty
                form.contentType = content_type
                TD["properties"][name]["forms"].append(form.json())
        except Exception as ex:
            if not ignore_errors:
                raise ex from None
            thing.logger.warning(
                "error while generating TD forms for property",
                name=name,
                error=str(ex),
            )

add_actions 

add_actions(TD: dict[str, Any], TM: dict[str, Any], thing: Thing, req_rep_socket_address: str, ignore_errors: bool) -> None

Add actions to the TD with ZMQ forms.

Parameters:

Name Type Description Default
TD
dict[str, Any]

The Thing Description to modify in place

 required
TM
dict[str, Any]

The Thing Model the description is built from

 required
thing
Thing

The Thing being described

 required
req_rep_socket_address
str

address invocations are addressed to

 required
ignore_errors
bool

Whether to warn and carry on rather than raise

 required
Source code in repo/hololinked/hololinked/server/zmq/services.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def add_actions(
    self,
    TD: dict[str, Any],
    TM: dict[str, Any],
    thing: Thing,
    req_rep_socket_address: str,
    ignore_errors: bool,
) -> None:
    """
    Add actions to the TD with ZMQ forms.

    Parameters
    ----------
    TD: dict[str, Any]
        The Thing Description to modify in place
    TM: dict[str, Any]
        The Thing Model the description is built from
    thing: Thing
        The `Thing` being described
    req_rep_socket_address: str
        address invocations are addressed to
    ignore_errors: bool
        Whether to warn and carry on rather than raise
    """
    for name in TM.get("actions", []):
        try:
            ActionAffordance.from_TD(name, TM)
            if not TD["actions"][name].get("forms", None):
                TD["actions"][name]["forms"] = []

            form = Form()
            form.href = req_rep_socket_address
            form.op = Operations.invokeaction
            form.contentType = self.content_type_for(thing, name)
            TD["actions"][name]["forms"].append(form.json())
        except Exception as ex:
            if not ignore_errors:
                raise ex from None
            thing.logger.warning(
                "error while generating TD forms for action",
                name=name,
                error=str(ex),
            )

add_events 

add_events(TD: dict[str, Any], TM: dict[str, Any], thing: Thing, pub_sub_socket_address: str, ignore_errors: bool) -> None

Add events to the TD with ZMQ forms.

Parameters:

Name Type Description Default
TD
dict[str, Any]

The Thing Description to modify in place

 required
TM
dict[str, Any]

The Thing Model the description is built from

 required
thing
Thing

The Thing being described

 required
pub_sub_socket_address
str

address subscriptions are addressed to

 required
ignore_errors
bool

Whether to warn and carry on rather than raise

 required
Source code in repo/hololinked/hololinked/server/zmq/services.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def add_events(
    self,
    TD: dict[str, Any],
    TM: dict[str, Any],
    thing: Thing,
    pub_sub_socket_address: str,
    ignore_errors: bool,
) -> None:
    """
    Add events to the TD with ZMQ forms.

    Parameters
    ----------
    TD: dict[str, Any]
        The Thing Description to modify in place
    TM: dict[str, Any]
        The Thing Model the description is built from
    thing: Thing
        The `Thing` being described
    pub_sub_socket_address: str
        address subscriptions are addressed to
    ignore_errors: bool
        Whether to warn and carry on rather than raise
    """
    for name in TM.get("events", []):
        try:
            EventAffordance.from_TD(name, TM)
            if not TD["events"][name].get("forms", None):
                TD["events"][name]["forms"] = []

            form = Form()
            form.href = pub_sub_socket_address
            form.op = Operations.subscribeevent
            form.contentType = self.content_type_for(thing, name)
            TD["events"][name]["forms"].append(form.json())
        except Exception as ex:
            if not ignore_errors:
                raise ex from None
            thing.logger.warning(
                "error while generating TD forms for event",
                name=name,
                error=str(ex),
            )