iqrfpy.async_response

  1"""Async response module.
  2
  3This module contains Asynchronous DPA response class.
  4"""
  5
  6from typing import List, Optional, Union
  7from iqrfpy.enums.peripherals import Peripheral
  8from iqrfpy.enums.commands import Command
  9from iqrfpy.enums.message_types import GenericMessages
 10from iqrfpy.iresponse import IResponseGetterMixin
 11from iqrfpy.utils.common import Common
 12import iqrfpy.utils.dpa as dpa_constants
 13from iqrfpy.utils.dpa import ResponsePacketMembers, ResponseCodes
 14from iqrfpy.utils.validators import DpaValidator
 15
 16__all__ = ['AsyncResponse']
 17
 18
 19class AsyncResponse(IResponseGetterMixin):
 20    """Asynchronous response class.
 21
 22    This class is used to capture and resolve asynchronous responses,
 23    as this is a generic response class, the data is only sent as string of bytes in hexadecimal
 24    representation separated by dots.
 25    """
 26
 27    def __init__(self, nadr: int, pnum: Peripheral, pcmd: Command, hwpid: int = dpa_constants.HWPID_MAX,
 28                 rcode: int = 0x80, dpa_value: int = 0, pdata: Union[List[int], None] = None,
 29                 msgid: Optional[str] = None, result: Optional[dict] = None):
 30        """AsyncResponse constructor.
 31
 32        Args:
 33            nadr (int): Device address.
 34            pnum (Peripheral): Peripheral.
 35            pcmd (Command): Peripheral command.
 36            hwpid (int, optional): Hardware profile ID. Defaults to 65535, this value ignores HWPID check.
 37            rcode (int, optional): Response code. Defaults to 128.
 38            dpa_value (int, optional): DPA value. Defaults to 0.
 39            pdata (List[int], optional): DPA response data. Defaults to None.
 40            msgid (str, optional): Message ID. Defaults to None.
 41            result (dict, optional): JSON response data. Defaults to None.
 42        """
 43        super().__init__(
 44            nadr=nadr,
 45            pnum=pnum,
 46            pcmd=pcmd,
 47            m_type=GenericMessages.RAW,
 48            hwpid=hwpid,
 49            rcode=rcode,
 50            dpa_value=dpa_value,
 51            pdata=pdata,
 52            msgid=msgid,
 53            result=result
 54        )
 55
 56    @classmethod
 57    def from_dpa(cls, dpa: bytes) -> 'AsyncResponse':
 58        """Asynchronous response DPA factory method.
 59
 60        Parses DPA data and constructs :obj:`AsyncResponse` object.
 61
 62        Args:
 63            dpa (bytes): DPA response bytes.
 64
 65        Returns:
 66            :obj:`AsyncResponse`: Asynchronous response message object.
 67        """
 68        DpaValidator.base_response_length(dpa=dpa)
 69        nadr = dpa[ResponsePacketMembers.NADR]
 70        hwpid = Common.hwpid_from_dpa(dpa[ResponsePacketMembers.HWPID_HI], dpa[ResponsePacketMembers.HWPID_LO])
 71        pnum = Common.pnum_from_dpa(dpa[ResponsePacketMembers.PNUM])
 72        pcmd = Common.request_pcmd_from_dpa(pnum, dpa[ResponsePacketMembers.PCMD])
 73        rcode = dpa[ResponsePacketMembers.RCODE]
 74        dpa_value = dpa[ResponsePacketMembers.DPA_VALUE]
 75        result = None
 76        if rcode == ResponseCodes.ASYNC_RESPONSE:
 77            if len(dpa) > 8:
 78                result = {'rData': list(dpa)}
 79        return cls(nadr=nadr, pnum=pnum, pcmd=pcmd, hwpid=hwpid, rcode=rcode, dpa_value=dpa_value,
 80                   pdata=list(dpa), result=result)
 81
 82    @classmethod
 83    def from_json(cls, json: dict) -> 'AsyncResponse':
 84        """Asynchronous response JSON factory method.
 85
 86        Parses JSON API response and constructs :obj:`AsyncResponse` object.
 87
 88        Args:
 89            json (dict): JSON API response.
 90
 91        Returns:
 92            :obj:`AsyncResponse`: Asynchronous response message object.
 93        """
 94        msgid = Common.msgid_from_json(json)
 95        result = json['data']['rsp']
 96        packet = result['rData'].replace('.', '')
 97        pdata = bytes.fromhex(packet)
 98        ldata = Common.hex_string_to_list(packet)
 99        nadr = ldata[ResponsePacketMembers.NADR]
100        hwpid = Common.hwpid_from_dpa(ldata[ResponsePacketMembers.HWPID_HI], ldata[ResponsePacketMembers.HWPID_LO])
101        pnum = Common.pnum_from_dpa(ldata[ResponsePacketMembers.PNUM])
102        pcmd = Common.request_pcmd_from_dpa(pnum, ldata[ResponsePacketMembers.PCMD])
103        rcode = ldata[ResponsePacketMembers.RCODE]
104        dpa_value = ldata[ResponsePacketMembers.DPA_VALUE]
105        return cls(nadr=nadr, pnum=pnum, pcmd=pcmd, hwpid=hwpid, rcode=rcode, dpa_value=dpa_value,
106                   pdata=list(pdata), msgid=msgid, result=result)