em27_sensors

frog.hardware.plugins.sensors.em27_sensors

This module provides an interface to the EM27 monitor.

This is used to scrape the PSF27Sensor data table off the server.

Attributes

Classes

EM27Error

Bases: Exception

Indicates than an error occurred while parsing the webpage.

EM27Sensors(host=EM27_HOST, poll_interval=EM27_SENSORS_POLL_INTERVAL, timeout=DEFAULT_EM27_HTTP_TIMEOUT)

Bases: EM27SensorsBase

An interface for EM27 sensors on the real device.

Create a new EM27Sensors.

Parameters:

Name	Type	Description	Default
host	str	The IP address or hostname of the EM27 device	EM27_HOST
poll_interval	float	How often to poll the sensors (seconds)	EM27_SENSORS_POLL_INTERVAL
timeout	float	The maximum time in seconds to wait for a response from the server	DEFAULT_EM27_HTTP_TIMEOUT

Source code in frog/hardware/plugins/sensors/em27_sensors.py

def __init__(
    self,
    host: str = EM27_HOST,
    poll_interval: float = EM27_SENSORS_POLL_INTERVAL,
    timeout: float = DEFAULT_EM27_HTTP_TIMEOUT,
) -> None:
    """Create a new EM27Sensors.

    Args:
        host: The IP address or hostname of the EM27 device
        poll_interval: How often to poll the sensors (seconds)
        timeout: The maximum time in seconds to wait for a response from the server
    """
    super().__init__(EM27_SENSORS_URL.format(host=host), poll_interval, timeout)

Functions

EM27SensorsBase(url, poll_interval=float('nan'), timeout=DEFAULT_HTTP_TIMEOUT)

Bases: HTTPDevice, SensorsBase

An interface for monitoring EM27 properties.

Create a new EM27 property monitor.

Parameters:

Name	Type	Description	Default
url	str	Web address of the automation units diagnostics page.	required
poll_interval	float	How often to poll the device (seconds)	float('nan')
timeout	float	The maximum time in seconds to wait for a response from the server	DEFAULT_HTTP_TIMEOUT

Source code in frog/hardware/plugins/sensors/em27_sensors.py

def __init__(
    self,
    url: str,
    poll_interval: float = float("nan"),
    timeout: float = DEFAULT_HTTP_TIMEOUT,
) -> None:
    """Create a new EM27 property monitor.

    Args:
        url: Web address of the automation units diagnostics page.
        poll_interval: How often to poll the device (seconds)
        timeout: The maximum time in seconds to wait for a response from the server
    """
    self._url: str = url
    self._connected = False

    HTTPDevice.__init__(self, timeout)
    SensorsBase.__init__(self, poll_interval)

    self.request_readings()

Functions

handle_response(response)

Process received sensor data.

Source code in frog/hardware/plugins/sensors/em27_sensors.py

def handle_response(self, response: str):
    """Process received sensor data."""
    readings = get_em27_sensor_data(response)

    if not self._connected:
        self._connected = True
        self.signal_is_opened()
        self.start_polling()

    self.send_readings_message(readings)

request_readings()

Request the EM27 property data from the web server.

The HTTP request is made on a background thread.

Source code in frog/hardware/plugins/sensors/em27_sensors.py

def request_readings(self) -> None:
    """Request the EM27 property data from the web server.

    The HTTP request is made on a background thread.
    """
    self.make_request(self._url)

Functions

get_em27_sensor_data(content)

Search for the PSF27Sensor table and store the data.

Parameters:

Name	Type	Description	Default
content	str	HTML content in which to search for PSF27Sensor table	required

Returns:

Name	Type	Description
data_table	list[SensorReading]	a list of sensor properties and their values

Source code in frog/hardware/plugins/sensors/em27_sensors.py

def get_em27_sensor_data(content: str) -> list[SensorReading]:
    """Search for the PSF27Sensor table and store the data.

    Args:
        content: HTML content in which to search for PSF27Sensor table

    Returns:
        data_table: a list of sensor properties and their values
    """
    table_header = (
        "<TR><TH>No</TH><TH>Name</TH><TH>Description</TH>"
        + "<TH>Status</TH><TH>Value</TH><TH>Meas. Unit</TH></TR>"
    )
    table_start = content.find(table_header)
    if table_start == -1:
        raise EM27Error("PSF27Sensor table not found")

    table_end = table_start + content[table_start:].find("</TABLE>")
    table = content[table_start:table_end].splitlines()
    data_table = []
    for row in range(1, len(table)):
        data_table.append(
            SensorReading(
                table[row].split("<TD>")[2].rstrip("</TD>"),
                Decimal(table[row].split("<TD>")[5].strip("</TD>")),
                table[row].split("<TD>")[6].rstrip("</TD></TR"),
            )
        )

    return data_table