iqrfpy.irequest

  1"""Request message class abstraction."""
  2
  3from abc import ABC, abstractmethod
  4from typing import List, Optional, Union
  5from uuid import uuid4
  6import iqrfpy.utils.dpa as dpa_constants
  7from iqrfpy.enums.commands import Command
  8from iqrfpy.enums.message_types import MessageType
  9from iqrfpy.enums.peripherals import Peripheral
 10from iqrfpy.exceptions import RequestNadrInvalidError, RequestPnumInvalidError, RequestPcmdInvalidError, \
 11    RequestHwpidInvalidError, RequestParameterInvalidValueError
 12
 13__all__ = ['IRequest']
 14
 15
 16class IRequest(ABC):
 17    """This abstract class serves as an interface for embedded peripherals and standards request messages.
 18
 19    The class provides getters and setters for shared request properties in the form of the @property decorator.
 20    Shared properties such as device address, peripheral, peripheral command and hwpid are validated here.
 21    The class also provides generic DPA and JSON serialization methods for derived classes to amend.
 22    While the abstract class cannot be instantiated, it's methods can be called from a derived class using super.
 23
 24    Network address (nadr) specifies target device: 0 (coordinator), 1-239 (node), 252 (local device),
 25    254 (temporary address) and 255 (broadcast), other values are reserved. Values in range 0-255.
 26
 27    Peripheral number (pnum) specifies peripheral to use: 0-31 are served for embedded peripherals and
 28    64-127 are reserved for IQRF standard peripherals. Peripheral number can be passed as a member of one
 29    of the peripheral enum class members (see iqrfpy.enums.peripherals) or plain integer. Values in range 0-255.
 30
 31    Peripheral command (pcmd) specifies a command to execute, each peripheral implements different commands.
 32    See DPA framework document for specific command numbers. Peripheral command can be passed as a member of one
 33    of the peripheral command enum class members (see iqrfpy.enums.commands) or plain integer.
 34    Peripheral number and peripheral command are pre-defined for each command message this library implements,
 35    however they will be required for user-defined messages. Values in range 0-255, but note that only values
 36    0-127 are used for request commands, as response command value is equal to request command value plus 128 (0x80).
 37
 38    Hardware profile ID (hwpid) specifies a unique device (by its functionality), this argument can be used to filter
 39    which devices execute a commend specified by request message. Only devices with matching hwpid will execute
 40    a command, unless the value 65535 (0xFFFF) is used, which ignores hwpid checking. Values in range 0-65535.
 41
 42    Packet data (pdata) can be required or optional data sent with the request message, depending on the command.
 43    On a general level, packet data is treated as list of unsigned 1B integers. This means that when defining a user
 44    command and a request message, request data should be serialized into a list of previously mentioned
 45    format and stored in the pdata member.
 46
 47    For more detailed information, see the IQRF DPA Framework documentation.
 48
 49    Message type (m_type) is an optional argument, however, it is required when using JSON API as message type
 50    is equivalent to a combination of peripheral number and peripheral command (see iqrfpy.enums.message_types).
 51    For example, `OSMessages.READ` (iqrfEmbedOs_Read) is equivalent to `EmbedPeripherals.OS` (2) and
 52    `OSRequestCommands.READ` (0).
 53
 54    Request parameters (params) is an optional dictionary argument, that is used when constructing a JSON API request.
 55    When creating a user message, use the params member to propagate your request parameters into the final
 56    JSON request object.
 57
 58    DPA response time (dpa_rsp_time) is an optional argument specifying DPA request timeout, not time between reception
 59    of a request and delivery of response. Note that handling of this argument is the responsibility of a transport
 60    and system facilitating communication between the library and devices of the IQRF network.
 61
 62    Device processing time (dev_process_time) is an optional argument used to specify time allotted to the device
 63    executing a task beyond simple data processing. For example, a device may need to measure some quantity
 64    for certain amount of time for the measured value to be valid. Note that handling of this argument
 65    is the responsibility of a transport and system facilitating communication between the library and devices
 66    of the IQRF network.
 67
 68    Message ID (msgid) is an optional argument used only by the JSON API at this time. It is used to pair
 69    JSON API requests and responses as the message type argument only allows to pair requests of the same type,
 70    but not pairing a specific request and response, given the asynchronous nature of the JSON API.
 71    """
 72
 73    __slots__ = '_nadr', '_pnum', '_pcmd', '_mtype', '_hwpid', '_pdata', '_msgid', '_params', '_dpa_rsp_time', \
 74        '_dev_process_time'
 75
 76    def __init__(self, nadr: int, pnum: Union[Peripheral, int], pcmd: Union[Command, int],
 77                 hwpid: int = dpa_constants.HWPID_MAX, pdata: Optional[List[int]] = None,
 78                 m_type: Optional[MessageType] = None, msgid: Optional[str] = None,
 79                 params: Optional[dict] = None, dpa_rsp_time: Optional[float] = None,
 80                 dev_process_time: Optional[float] = None):
 81        """IRequest constructor.
 82
 83        Args:
 84            nadr (int): Network address.
 85            pnum (Union[Peripheral, int]): Peripheral number.
 86            pcmd (Union[Command, int]): Peripheral command.
 87            hwpid (int, optional): Hardware profile ID. Defaults to 65535 (Ignore HWPID check).
 88            pdata (List[int], optional): DPA request data. Defaults to None.
 89            m_type (MessageType, optional): JSON API message type. Defaults to None.
 90            msgid (str, optional): JSON API message ID. Defaults to None. If the parameter is not specified, a random
 91                UUIDv4 string is generated and used.
 92            params (dict, optional): JSON API request data. Defaults to None.
 93            dpa_rsp_time (float, optional): DPA request timeout in seconds. Defaults to None.
 94            dev_process_time (float, optional): Device processing time. Defaults to None.
 95        """
 96        self._nadr = nadr
 97        self._pnum = pnum
 98        self._pcmd = pcmd
 99        self._hwpid = hwpid
100        self._pdata = pdata
101        self._mtype = m_type
102        self._msgid = msgid if msgid is not None else str(uuid4())
103        self._params = params if params is not None else {}
104        self._dpa_rsp_time = dpa_rsp_time
105        self._dev_process_time = dev_process_time
106        self._validate_base(nadr, pnum, pcmd, hwpid, dpa_rsp_time, dev_process_time)
107
108    def _validate_base(self, nadr: int, pnum: Union[Peripheral, int], pcmd: Union[Command, int], hwpid: int,
109                       dpa_rsp_time: Optional[float], dev_process_time: Optional[float]) -> None:
110        """Shared members validation method.
111
112        Validates shared request members to ensure the data can fit into the DPA packet and JSON request
113        object and do not violate rules for expected format of these properties.
114
115        Args:
116            nadr (int): Network address.
117            pnum (Union[Peripheral, int]): Peripheral number.
118            pcmd (Union[Command, int]): Peripheral command.
119            hwpid (int, optional): Hardware profile ID. Defaults to 65535 (Ignore HWPID check).
120            dpa_rsp_time (float, optional): DPA request timeout in seconds. Defaults to None.
121            dev_process_time (float, optional): Device processing time. Defaults to None.
122        """
123        self._validate_nadr(nadr)
124        self._validate_pnum(pnum)
125        self._validate_pcmd(pcmd)
126        self._validate_hwpid(hwpid)
127        self._validate_dpa_rsp_time(dpa_rsp_time)
128        self._validate_dev_process_time(dev_process_time)
129
130    @staticmethod
131    def _validate_nadr(nadr: int):
132        """Validates network address parameter.
133
134        Args:
135            nadr (int): Network address value.
136
137        Raises:
138            RequestNadrInvalidError: If nadr is less than 0 or greater than 255.
139        """
140        if not dpa_constants.BYTE_MIN <= nadr <= dpa_constants.BYTE_MAX:
141            raise RequestNadrInvalidError('NADR should be between 0 and 255.')
142
143    @property
144    def nadr(self) -> int:
145        """:obj:`int`: Network address.
146
147        Getter and setter.
148        """
149        return self._nadr
150
151    @nadr.setter
152    def nadr(self, value: int):
153        self._validate_nadr(value)
154        self._nadr = value
155
156    @staticmethod
157    def _validate_pnum(pnum: Union[Peripheral, int]):
158        """Validates peripheral number parameter.
159
160        Args:
161            pnum (Union[Peripheral, int]): Peripheral number value.
162
163        Raises:
164            RequestPnumInvalidError: If pnum is less than 0 or greater than 255.
165        """
166        if not dpa_constants.BYTE_MIN <= pnum <= dpa_constants.BYTE_MAX:
167            raise RequestPnumInvalidError('PNUM should be between 0 and 255.')
168
169    @property
170    def pnum(self) -> Union[Peripheral, int]:
171        """:obj:`iqrfpy.enums.peripherals.Peripheral` of :obj:`int`: Peripheral number.
172
173        Getter and setter.
174        """
175        return self._pnum
176
177    @pnum.setter
178    def pnum(self, value: Union[Peripheral, int]):
179        self._validate_pnum(pnum=value)
180        self._pnum = value
181
182    @staticmethod
183    def _validate_pcmd(pcmd: Union[Command, int]):
184        """Validates peripheral command parameter.
185
186        Args:
187            pcmd (Union[Command, int]): Peripheral command value.
188
189        Raises:
190            RequestPcmdInvalidError: If pcmd is less than 0 or greater than 255.
191        """
192        if not dpa_constants.BYTE_MIN <= pcmd <= dpa_constants.BYTE_MAX:
193            raise RequestPcmdInvalidError('PCMD should be between 0 and 255.')
194
195    @property
196    def pcmd(self) -> Union[Command, int]:
197        """:obj:`iqrfpy.enums.commands.Command` or :obj:`int`: Peripheral command.
198
199        Getter and setter.
200        """
201        return self._pcmd
202
203    @pcmd.setter
204    def pcmd(self, value: Union[Command, int]):
205        self._validate_pcmd(pcmd=value)
206        self._pcmd = value
207
208    @staticmethod
209    def _validate_hwpid(hwpid: int):
210        """Validates Hardware profile ID parameter.
211
212        Args:
213            hwpid (int): Hardware profile ID value.
214
215        Raises:
216            RequestHwpidInvalidError: If hwpid is less than 0 or greater than 65535.
217        """
218        if not dpa_constants.HWPID_MIN <= hwpid <= dpa_constants.HWPID_MAX:
219            raise RequestHwpidInvalidError('HWPID should be between 0 and 65535.')
220
221    @property
222    def hwpid(self) -> int:
223        """:obj:`int`: Hardware profile ID.
224
225        Getter and setter.
226        """
227        return self._hwpid
228
229    @staticmethod
230    def _validate_dpa_rsp_time(dpa_rsp_time: Optional[float] = None):
231        """Validates dpa timeout parameter.
232
233        Args:
234            dpa_rsp_time (float, optional): DPA timeout value in seconds. Defaults to None.
235
236        Raises:
237            RequestParameterInvalidValueError: If dpa_rsp_time is not None and is less than 0.
238        """
239        if dpa_rsp_time is None:
240            return
241        if dpa_rsp_time < 0:
242            raise RequestParameterInvalidValueError('DPA response time should a positive integer.')
243
244    @property
245    def dpa_rsp_time(self) -> Optional[float]:
246        """:obj:`float` or :obj:`None`: DPA timeout.
247
248        Getter and setter.
249        """
250        return self._dpa_rsp_time
251
252    @dpa_rsp_time.setter
253    def dpa_rsp_time(self, value: Optional[float] = None):
254        self._validate_dpa_rsp_time(value)
255        self._dpa_rsp_time = value
256
257    @staticmethod
258    def _validate_dev_process_time(dev_process_time: Optional[float] = None):
259        """Validates device processing time parameter.
260
261        Args:
262            dev_process_time (float, optional): Device processing time value. Defaults to None.
263
264        Raises:
265            RequestParameterInvalidValueError: If dev_process_time is not None and is less than 0.
266        """
267        if dev_process_time is None:
268            return
269        if dev_process_time < 0:
270            raise RequestParameterInvalidValueError('Device processing time should be a positive integer.')
271
272    @property
273    def dev_process_time(self) -> Optional[float]:
274        """:obj:`float` or :obj:`None`: Device processing time.
275
276        Getter and setter.
277        """
278        return self._dev_process_time
279
280    @dev_process_time.setter
281    def dev_process_time(self, value: Optional[float] = None):
282        self._validate_dev_process_time(value)
283        self._dev_process_time = value
284
285    @property
286    def msgid(self) -> str:
287        """:obj:`str`: Message ID.
288
289        Getter and setter.
290        """
291        return self._msgid
292
293    @msgid.setter
294    def msgid(self, value: str):
295        self._msgid = value
296
297    @property
298    def mtype(self) -> MessageType:
299        """:obj:`MessageType` Message type.
300
301        Getter only.
302        """
303        return self._mtype
304
305    @abstractmethod
306    def to_dpa(self, mutable: bool = False) -> Union[bytes, bytearray]:
307        """DPA request serialization method.
308
309        Serializes a request to DPA request packet format. Each request contains the DPA request header.
310        If pdata is specified, it is appended to the DPA request header. The request can be serialized
311        into both mutable and immutable (default) byte formats.
312
313        Args:
314            mutable (bool, optional): Serialize into mutable byte representation of DPA request packet.
315                Defaults to False.
316
317        Returns:
318            :obj:`bytes`: Immutable byte representation of DPA request packet.\n
319            :obj:`bytearray`: Mutable byte representation of DPA request packet (if argument mutable is True).
320        """
321        raise NotImplementedError('Abstract method not implemented.')
322
323    @abstractmethod
324    def to_json(self) -> dict:
325        """JSON API request serialization method.
326
327        Serializes a request to JSON API request object. Each request contains message type and data.
328        Data carries message ID for pairing requests with responses, and `req` object that contains network address,
329        hwpid and optional request parameters (equivalent to DPA packet data) if `params` is not None.
330        If DPA timeout is specified, the value is multiplied by 1000 as JSON API accepts timeout in milliseconds.
331
332        Returns:
333            :obj:`dict`: JSON API request object.
334        """
335        raise NotImplementedError('Abstract method not implemented')