http_device

frog.hardware.http_device

Provides a device which communicates via HTTP requests.

Attributes

Classes

HTTPDevice(timeout=DEFAULT_HTTP_TIMEOUT)

Bases: AbstractDevice

A device which communicates via HTTP requests.

Create a new HTTPDevice.

Parameters:

Name	Type	Description	Default
timeout	float	The maximum time in seconds to wait for a response from the server	DEFAULT_HTTP_TIMEOUT

Source code in frog/hardware/http_device.py

def __init__(self, timeout: float = DEFAULT_HTTP_TIMEOUT) -> None:
    """Create a new HTTPDevice.

    Args:
        timeout: The maximum time in seconds to wait for a response from the server
    """
    if timeout <= 0:
        raise ValueError("Timeout must be greater than zero")

    self._timeout = timeout
    self._manager = QNetworkAccessManager()

Functions

handle_response(response) abstractmethod

The default callback for successful HTTP responses.

Source code in frog/hardware/http_device.py

@abstractmethod
def handle_response(self, response: str) -> None:
    """The default callback for successful HTTP responses."""

make_request(url, callback=None)

Make a new HTTP request in the background.

Parameters:

Name	Type	Description	Default
url	str	The URL to connect to	required
callback	Callable[[str], Any] | None	A function to be called when a successful response is received	None

Source code in frog/hardware/http_device.py

def make_request(
    self, url: str, callback: Callable[[str], Any] | None = None
) -> None:
    """Make a new HTTP request in the background.

    Args:
        url: The URL to connect to
        callback: A function to be called when a successful response is received
    """
    if not callback:
        callback = self.handle_response

    request = QNetworkRequest(url)
    request.setTransferTimeout(round(1000 * self._timeout))
    reply = self._manager.get(request)
    reply.finished.connect(
        partial(self.pubsub_errors(self._on_reply_received), reply, callback)
    )