iqrfpy.itransport

  1"""Transport abstract class.
  2
  3Serves as an abstract for communication channels.
  4"""
  5from abc import ABC, abstractmethod
  6from typing import Callable, Optional
  7from iqrfpy.irequest import IRequest
  8from iqrfpy.iresponse import IResponse
  9from iqrfpy.confirmation import Confirmation
 10
 11
 12class ITransport(ABC):
 13    """Abstract class providing interface for communication channels."""
 14
 15    @abstractmethod
 16    def initialize(self) -> None:
 17        """Initialize transport and create a connection if applicable.
 18
 19        Raises:
 20            NotImplementedError: Abstract method not implemented.
 21        """
 22        raise NotImplementedError("Abstract method not implemented.")
 23
 24    @abstractmethod
 25    def terminate(self, force: bool = False) -> None:
 26        """Terminates transport.
 27
 28        Args:
 29            force (bool): Force terminate transport.
 30
 31        Raises:
 32            NotImplementedError: Abstract method not implemented.
 33        """
 34        raise NotImplementedError("Abstract method not implemented.")
 35
 36    @abstractmethod
 37    def send(self, request: IRequest) -> None:
 38        """Serialize passed request to format acceptable by the communication channel and send request.
 39
 40        Args:
 41            request (IRequest): Request message to send.
 42
 43        Raises:
 44            NotImplementedError: Abstract method not implemented.
 45        """
 46        raise NotImplementedError("Abstract method not implemented.")
 47
 48    @abstractmethod
 49    def send_and_receive(self, request: IRequest, timeout: Optional[float] = None) -> IResponse:
 50        """Serialize request to format acceptable by the communication channel, send request and receive response synchronously.
 51
 52        Args:
 53            request (IRequest): Request message to send.
 54            timeout (float, optional): Response timeout.
 55
 56        Returns:
 57            :obj:`IResponse`: Response message object.
 58
 59        Raises:
 60            NotImplementedError: Abstract method not implemented.
 61        """
 62        raise NotImplementedError("Abstract method not implemented.")
 63
 64    @abstractmethod
 65    def receive(self, timeout: Optional[float] = None) -> IResponse:
 66        """Receive and return response synchronously.
 67
 68        Args:
 69            timeout (float, optional): Response timeout.
 70
 71        Returns:
 72            :obj:`IResponse`: Response message object.
 73
 74        Raises:
 75            NotImplementedError: Abstract method not implemented.
 76        """
 77        raise NotImplementedError("Abstract method not implemented.")
 78
 79    @abstractmethod
 80    def confirmation(self) -> Confirmation:
 81        """Receive and return confirmation synchronously.
 82
 83        Returns:
 84            :obj:`Confirmation`: Confirmation message object.
 85
 86        Raises:
 87            NotImplementedError: Abstract method not implemented.
 88        """
 89        raise NotImplementedError("Abstract method not implemented.")
 90
 91    @abstractmethod
 92    def set_receive_callback(self, callback: Callable[[IResponse], None]) -> None:
 93        """Set callback to handle asynchronously received messages.
 94
 95        Args:
 96            callback (Callable[[IResponse], None]): Function to call once a message has been received and successfully
 97                                                    deserialized.
 98
 99        Raises:
100            NotImplementedError: Abstract method not implemented.
101        """
102        raise NotImplementedError("Abstract method not implemented.")