Reference¶

Constants

linux2mqtt package.

`SensorType = Literal['sensor', 'binary_sensor']` `module-attribute` ¶

Sensor type for discovery

`StatusType = Literal['online', 'offline']` `module-attribute` ¶

Metric status

`Linux2MqttConfigException` ¶

Bases: Linux2MqttException

Bad config exception occurred.

`Linux2MqttConnectionException` ¶

Bases: Linux2MqttException

Connection processing exception occurred.

`Linux2MqttException` ¶

Bases: Exception

General processing exception occurred.

`Linux2MqttMetricsException` ¶

Bases: Linux2MqttException

Metrics processing exception occurred.

`Linux2Mqtt` ¶

linux2mqtt class.

Attributes:

Name	Type	Description
`version`	`str`	The version of linux2mqtt
`cfg`	`Linux2MqttConfig`	The config for linux2mqtt
`discovery_binary_sensor_topic`	`str`	Topic template for a binary sensor
`discovery_sensor_topic`	`str`	Topic template for a nary sensor
`status_topic`	`str`	Topic template for a status value
`version_topic`	`str`	Topic template for a version value
`state_topic`	`str`	Topic template for a state dict
`availability_topic`	`str`	Topic template for a availability value
`deferred_metrics_queue`	`Queue[BaseMetric]`	Queue with metrics to publish available through lazy gathering
`do_not_exit`	`bool`	Prevent exit from within linux2mqtt, when handled outside

Initialize the linux2mqtt.

Parameters:

Name	Type	Description	Default
`cfg`	`Linux2MqttConfig`	The config for the linux2mqtt	required
`do_not_exit`	`bool`	Prevent exit from within linux2mqtt, when handled outside	`True`

Raises:

Type	Description
`Linux2MqttConfigException`	Bad config

Source code in linux2mqtt/linux2mqtt.py

def __init__(
    self,
    cfg: Linux2MqttConfig,
    do_not_exit: bool = True,
):
    """Initialize the linux2mqtt.

    Parameters
    ----------
    cfg
        The config for the linux2mqtt
    do_not_exit
        Prevent exit from within linux2mqtt, when handled outside

    Raises
    ------
    Linux2MqttConfigException
        Bad config

    """

    self.cfg = cfg
    self.do_not_exit = do_not_exit
    self.metrics = []
    self.connected = False

    system_name_sanitized = sanitize(self.cfg["linux2mqtt_hostname"])

    self.discovery_binary_sensor_topic = f"{self.cfg['homeassistant_prefix']}/binary_sensor/{self.cfg['mqtt_topic_prefix']}/{system_name_sanitized}_{{}}/config"
    self.discovery_sensor_topic = f"{self.cfg['homeassistant_prefix']}/sensor/{self.cfg['mqtt_topic_prefix']}/{system_name_sanitized}_{{}}/config"
    self.availability_topic = (
        f"{self.cfg['mqtt_topic_prefix']}/{system_name_sanitized}/{{}}/availability"
    )
    self.state_topic = (
        f"{self.cfg['mqtt_topic_prefix']}/{system_name_sanitized}/{{}}/state"
    )
    self.status_topic = (
        f"{self.cfg['mqtt_topic_prefix']}/{system_name_sanitized}/status"
    )
    self.version_topic = (
        f"{self.cfg['mqtt_topic_prefix']}/{system_name_sanitized}/version"
    )

    main_logger.setLevel(self.cfg["log_level"].upper())

    if not self.do_not_exit:
        main_logger.info("Register signal handlers for SIGINT and SIGTERM")
        signal.signal(signal.SIGTERM, self._signal_handler)
        signal.signal(signal.SIGINT, self._signal_handler)

    if self.cfg["interval"] < MIN_INTERVAL:
        raise Linux2MqttConfigException(
            "linux2mqtt could not start due to bad config"
        ) from ValueError(
            f"The interval for the mqtt update must at least {MIN_INTERVAL}s"
        )
    elif self.cfg["interval"] > MAX_INTERVAL:
        raise Linux2MqttConfigException(
            "linux2mqtt could not start due to bad config"
        ) from ValueError(
            f"The interval for the mqtt update must at most {MAX_INTERVAL}s"
        )

`connect` ¶

Initialize the linux2mqtt.

Raises:

Type	Description
`Linux2MqttConnectionException`	If anything with the mqtt connection goes wrong

Source code in linux2mqtt/linux2mqtt.py

def connect(self) -> None:
    """Initialize the linux2mqtt.

    Raises
    ------
    Linux2MqttConnectionException
        If anything with the mqtt connection goes wrong

    """
    try:
        self.mqtt = paho.mqtt.client.Client(
            callback_api_version=paho.mqtt.client.CallbackAPIVersion.VERSION2,  # type: ignore[attr-defined, call-arg]
            client_id=self.cfg["mqtt_client_id"],
        )
        if self.cfg["mqtt_user"] or self.cfg["mqtt_password"]:
            self.mqtt.username_pw_set(
                self.cfg["mqtt_user"], self.cfg["mqtt_password"]
            )
        self.mqtt.on_connect = self._on_connect
        self.mqtt.will_set(
            self.status_topic,
            "offline",
            qos=self.cfg["mqtt_qos"],
            retain=True,
        )
        self.mqtt.connect(
            self.cfg["mqtt_host"], self.cfg["mqtt_port"], self.cfg["mqtt_timeout"]
        )
        self.mqtt.loop_start()
        self._mqtt_send(self.status_topic, "online", retain=True)
        self._mqtt_send(self.version_topic, self.version, retain=True)
    except paho.mqtt.client.WebsocketConnectionError as ex:
        main_logger.exception("Error while trying to connect to MQTT broker.")
        main_logger.debug(ex)
        raise Linux2MqttConnectionException from ex

`del` ¶

Destroy the class.

Source code in linux2mqtt/linux2mqtt.py

def __del__(self) -> None:
    """Destroy the class."""
    self._cleanup()

`add_metric` ¶

Add metric to linux2mqtt.

Parameters:

Name	Type	Description	Default
`metric`	`BaseMetric`	The metric to add	required

Source code in linux2mqtt/linux2mqtt.py

def add_metric(self, metric: BaseMetric) -> None:
    """Add metric to linux2mqtt.

    Parameters
    ----------
    metric
        The metric to add

    """
    self.metrics.append(metric)

`loop_busy` ¶

Monitor the metrics and handle the update interval for each metric.

When not connected, it waits for it until the process is exited or a connection is established.

Parameters:

Name	Type	Description	Default
`raise_known_exceptions`	`bool`	Should any known processing exception be raised or ignored	`False`

Raises:

Type	Description
`Linux2MqttConnectionException`	If anything with the mqtt connection goes wrong

Source code in linux2mqtt/linux2mqtt.py

def loop_busy(self, raise_known_exceptions: bool = False) -> None:
    """Monitor the metrics and handle the update interval for each metric.

    When not connected, it waits for it until the process is exited or a connection is established.

    Parameters
    ----------
    raise_known_exceptions
        Should any known processing exception be raised or ignored

    Raises
    ------
    Linux2MqttConnectionException
        If anything with the mqtt connection goes wrong

    """
    while not self.connected:
        main_logger.debug("Waiting for connection.")
        time.sleep(1)

    self._create_discovery_topics()
    while True:
        try:
            x = 0
            while x < self.cfg["interval"]:
                # Check the queue for deferred results one/sec
                time.sleep(1)
                self._check_queue()
                x += 1
            for metric in self.metrics:
                is_deferred = metric.poll(result_queue=self.deferred_metrics_queue)
                if not is_deferred:
                    self._publish_metric(metric)
        except Linux2MqttConnectionException as ex:
            if raise_known_exceptions:
                raise ex  # noqa: TRY201
            else:
                main_logger.warning(
                    "Do not raise due to raise_known_exceptions=False: %s", str(ex)
                )

`BaseMetric` ¶

Base metric class.

Attributes:

Name	Type	Description
`_name`	`str`	the name of the metric
`unit_of_measurement`	`str \| None`	The unit of the metric
`device_class`	`str \| None`	The device_class of the metric
`icon`	`str \| None`	The icon of the metric
`state_field`	`str`	The field for the state in the data dict of .polled_result
`ha_sensor_typ`		The sensor type of the metric
`polled_result`	`dict[str, str \| int \| float \| None] \| None`	The dict with the polled result data for the state and attributes

Initialize base class.

Source code in linux2mqtt/metrics.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize base class."""
    self.polled_result = None

`name` `property` ¶

Return the name of the metric.

Returns:

Type	Description
`str`	The metrics name

`name_sanitized` `property` ¶

Return the sanitized name of the metric.

Returns:

Type	Description
`str`	The metrics sanitized name

`get_discovery` ¶

Get the discovery topic config data.

Parameters:

Name	Type	Description	Default
`state_topic`	`str`	The state topic where to find the data for state and attributes	required
`availability_topic`	`str`	The availability topic for the entry	required
`device_definition`	`LinuxDeviceEntry`	The device entry fro the homeassistant config	required

Returns:

Type	Description
`LinuxEntry`	The homeassistant config entry

Source code in linux2mqtt/metrics.py

def get_discovery(
    self,
    state_topic: str,
    availability_topic: str,
    device_definition: LinuxDeviceEntry,
) -> LinuxEntry:
    """Get the discovery topic config data.

    Parameters
    ----------
    state_topic
        The state topic where to find the data for state and attributes
    availability_topic
        The availability topic for the entry
    device_definition
        The device entry fro the homeassistant config

    Returns
    -------
    LinuxEntry
        The homeassistant config entry

    """
    return LinuxEntry(
        {
            "name": self.name,
            "unique_id": f"{device_definition['identifiers']}_{self.name_sanitized}",
            "availability_topic": availability_topic.format(self.name_sanitized),
            "payload_available": "online",
            "payload_not_available": "offline",
            "state_topic": state_topic.format(self.name_sanitized),
            "value_template": f"{{{{ value_json.{self.state_field} if value_json is not undefined and value_json.{self.state_field} is not undefined else None }}}}",
            "unit_of_measurement": self.unit_of_measurement,
            "icon": self.icon,
            "device_class": self.device_class,
            "payload_on": "on",
            "payload_off": "off",
            "device": device_definition,
            "json_attributes_topic": state_topic.format(self.name_sanitized),
            "qos": 1,
        }
    )

`poll` ¶

Poll new data for the metric. Can happened instantly or lazily (separate thread for example).

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[Any]`	The queue where to post new data if data is gathered lazily	required

Returns:

Type	Description
`bool`	False if data is readily available, True if data is gathered lazily

Raises:

Type	Description
`Linux2MqttException`	General exception.

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[Any]) -> bool:
    """Poll new data for the metric. Can happened instantly or lazily (separate thread for example).

    Parameters
    ----------
    result_queue
        The queue where to post new data if data is gathered lazily

    Returns
    -------
    bool
        False if data is readily available, True if data is gathered lazily

    Raises
    ------
    Linux2MqttException
        General exception.

    """
    raise Linux2MqttException from NotImplementedError

`BaseMetricThread` ¶

Bases: Thread

Base metric thread.

Attributes:

Name	Type	Description
`result_queue`	`Queue[BaseMetric]`	The queue to put the metric into once data is gathered
`metric`	`BaseMetric`	The metric to gather data for
`interval`	`int`	The interval to gather data over

`CPUMetrics` ¶

Bases: BaseMetric

CPU metric.

Attributes:

Name	Type	Description
`interval`	`int`	The interval to gather cpu data over

Initialize the cpu metric.

Raises:

Type	Description
`Linux2MqttConfigException`	Bad config

Source code in linux2mqtt/metrics.py

def __init__(self, interval: int):
    """Initialize the cpu metric.

    Raises
    ------
    Linux2MqttConfigException
        Bad config

    """
    super().__init__()
    self.interval = interval

    if interval < MIN_CPU_INTERVAL:
        raise Linux2MqttConfigException(
            "cpu metric could not start due to bad config"
        ) from ValueError(
            f"The interval for the cpu must at least {MIN_CPU_INTERVAL}s"
        )
    elif interval > MAX_CPU_INTERVAL:
        raise Linux2MqttConfigException(
            "cpu metric could not start due to bad config"
        ) from ValueError(
            f"The interval for the cpu must at most {MAX_CPU_INTERVAL}s"
        )

`poll` ¶

Poll new data for the cpu metric.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[BaseMetric]`	The queue where to post new data once gathered	required

Returns:

Type	Description
`bool = False`	True as the data is gathered lazily

Raises:

Type	Description
`Linux2MqttException`	General exception

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[BaseMetric]) -> bool:
    """Poll new data for the cpu metric.

    Parameters
    ----------
    result_queue
        The queue where to post new data once gathered

    Returns
    -------
    bool = False
        True as the data is gathered lazily

    Raises
    ------
    Linux2MqttException
        General exception


    """
    try:
        assert result_queue
    except ReferenceError as ex:
        raise Linux2MqttException(
            "Cannot start cpu metric due to missing result_queue"
        ) from ex
    self.result_queue = result_queue
    th = CPUMetricThread(
        result_queue=result_queue, metric=self, interval=self.interval
    )
    th.daemon = True
    th.start()
    return True  # Expect a deferred result

`CPUMetricThread` ¶

Bases: BaseMetricThread

CPU metric thread.

Initialize the cpu thread.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[BaseMetric]`	The queue to put the metric into once the data is gathered	required
`metric`	`BaseMetric`	The cpu metric to gather data for	required
`interval`	`int`	The interval to gather data over	required

Source code in linux2mqtt/metrics.py

def __init__(
    self, result_queue: Queue[BaseMetric], metric: BaseMetric, interval: int
):
    """Initialize the cpu thread.

    Parameters
    ----------
    result_queue
        The queue to put the metric into once the data is gathered
    metric
        The cpu metric to gather data for
    interval
        The interval to gather data over

    """
    threading.Thread.__init__(self)
    self.result_queue = result_queue
    self.metric = metric
    self.interval = interval

`run` ¶

Run the cpu thread. Once data is gathered, it is put into the queue and the thread exits.

Raises:

Type	Description
`Linux2MqttMetricsException`	cpu information could not be gathered or prepared for publishing

Source code in linux2mqtt/metrics.py

def run(self) -> None:
    """Run the cpu thread. Once data is gathered, it is put into the queue and the thread exits.

    Raises
    ------
    Linux2MqttMetricsException
        cpu information could not be gathered or prepared for publishing

    """
    try:
        cpu_times = psutil.cpu_times_percent(interval=self.interval, percpu=False)
        self.metric.polled_result = {
            **jsons.dump(cpu_times),  # type: ignore[unused-ignore]
            "used": 100.0 - cpu_times.idle,
        }
        self.result_queue.put(self.metric)
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish cpu data"
        ) from ex

`DiskUsageMetrics` ¶

Bases: BaseMetric

Disk usage metrics.

Attributes:

Name	Type	Description
`_name_template`		The template to create the name using the mountpoint as value
`mountpoint`	`str`	The mountpoint to check the metric

Initialize the disk usage metric.

Source code in linux2mqtt/metrics.py

def __init__(self, mountpoint: str):
    """Initialize the disk usage metric."""
    super().__init__()
    self.mountpoint = mountpoint
    self._name = self._name_template.format(mountpoint)

`poll` ¶

Poll new data for the virtual memory metric.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[Self]`	(Unused)	required

Returns:

Type	Description
`bool`	True as the data is readily available

Raises:

Type	Description
`Linux2MqttMetricsException`	virtual memory information could not be gathered or prepared for publishing

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[Self]) -> bool:
    """Poll new data for the virtual memory metric.

    Parameters
    ----------
    result_queue
        (Unused)

    Returns
    -------
    bool
        True as the data is readily available

    Raises
    ------
    Linux2MqttMetricsException
        virtual memory information could not be gathered or prepared for publishing

    """
    try:
        disk = psutil.disk_usage(self.mountpoint)
        self.polled_result = {
            "total": float(disk.total) / 1_000_000_000,
            "used": float(disk.used) / 1_000_000_000,
            "free": float(disk.free) / 1_000_000_000,
            "percent": disk.percent,
        }
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish disk usage data"
        ) from ex
    else:
        return False

`FanSpeedMetrics` ¶

Bases: BaseMetric

Fan speed metric.

Initialize the fan speed metric.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device	required
`fan`	`str`	The fan	required

Raises:

Type	Description
`Linux2MqttConfigException`	Bad config

Source code in linux2mqtt/metrics.py

def __init__(self, device: str, fan: str):
    """Initialize the fan speed metric.

    Parameters
    ----------
    device
        The device
    fan
        The fan

    Raises
    ------
    Linux2MqttConfigException
        Bad config

    """
    super().__init__()
    self._device = device
    self._fan = fan
    self._name = self._name_template.format(device, fan)

`poll` ¶

Poll new data for the thermal zone metric.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[Self]`	(Unused)	required

Returns:

Type	Description
`bool`	True as the data is readily available

Raises:

Type	Description
`Linux2MqttMetricsException`	Fan speed information could not be gathered or prepared for publishing

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[Self]) -> bool:
    """Poll new data for the thermal zone metric.

    Parameters
    ----------
    result_queue
        (Unused)

    Returns
    -------
    bool
        True as the data is readily available

    Raises
    ------
    Linux2MqttMetricsException
        Fan speed information could not be gathered or prepared for publishing

    """
    try:
        st = psutil.sensors_fans()  # type: ignore[attr-defined]
        fan = next(
            (item for item in st.get(self._device, []) if item.label == self._fan),
            None,
        )
        assert fan
        self.polled_result = {
            "label": fan.label,
            "current": fan.current,
            "unit": "rpm",
        }
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish fan speed data"
        ) from ex
    else:
        return False

`NetworkMetrics` ¶

Bases: BaseMetric

Network metric thread.

Attributes:

Name	Type	Description
`_name_template`		The template to create the name using the nic as value
`interval`	`int`	The interval to gather cpu data over
`nic`	`str`	The network interface to gather data for.

Initialize the network metric.

Parameters:

Name	Type	Description	Default
`nic`	`str`	The network interface	required
`interval`	`int`	The interval to gather data over	required

Raises:

Type	Description
`Linux2MqttConfigException`	Bad config

Source code in linux2mqtt/metrics.py

def __init__(self, nic: str, interval: int):
    """Initialize the network metric.

    Parameters
    ----------
    nic
        The network interface
    interval
        The interval to gather data over

    Raises
    ------
    Linux2MqttConfigException
        Bad config

    """
    super().__init__()
    self.interval = interval
    self.nic = nic
    self._name = self._name_template.format(nic)

    if interval < MIN_NET_INTERVAL:
        raise ValueError(
            f"The interval for the network {nic} must at least {MIN_NET_INTERVAL}s"
        )
    elif interval > MAX_NET_INTERVAL:
        raise ValueError(
            f"The interval for the network {nic} must at most {MAX_NET_INTERVAL}s"
        )

`poll` ¶

Poll new data for the network metric.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[BaseMetric]`	The queue where to post new data once gathered	required

Returns:

Type	Description
`bool = False`	True as the data is gathered lazily

Raises:

Type	Description
`Linux2MqttException`	General exception

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[BaseMetric]) -> bool:
    """Poll new data for the network metric.

    Parameters
    ----------
    result_queue
        The queue where to post new data once gathered

    Returns
    -------
    bool = False
        True as the data is gathered lazily

    Raises
    ------
    Linux2MqttException
        General exception

    """
    try:
        assert result_queue
    except ReferenceError as e:
        raise Linux2MqttException(
            "Cannot start network metric due to missing result_queue"
        ) from e
    self.result_queue = result_queue
    th = NetworkMetricThread(
        result_queue=result_queue, metric=self, interval=self.interval, nic=self.nic
    )
    th.daemon = True
    th.start()
    return True  # Expect a deferred result

`NetworkMetricThread` ¶

Bases: BaseMetricThread

Network metric thread.

Attributes:

Name	Type	Description
`nic`		The network interface to gather data for.

Initialize the cpu thread.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[BaseMetric]`	The queue to put the metric into once the data is gathered	required
`metric`	`BaseMetric`	The cpu metric to gather data for	required
`interval`	`int`	The interval to gather data over	required
`nic`	`str`	The network interface	required

Source code in linux2mqtt/metrics.py

def __init__(
    self,
    result_queue: Queue[BaseMetric],
    metric: BaseMetric,
    interval: int,
    nic: str,
):
    """Initialize the cpu thread.

    Parameters
    ----------
    result_queue: Queue[BaseMetric]
        The queue to put the metric into once the data is gathered
    metric
        The cpu metric to gather data for
    interval
        The interval to gather data over
    nic
        The network interface

    """
    threading.Thread.__init__(self)
    self.result_queue = result_queue
    self.metric = metric
    self.interval = interval
    self.nic = nic

`run` ¶

Run the cpu thread. Once data is gathered, it is put into the queue and the thread exits.

Raises:

Type	Description
`Linux2MqttMetricsException`	network information could not be gathered or prepared for publishing

Source code in linux2mqtt/metrics.py

def run(self) -> None:
    """Run the cpu thread. Once data is gathered, it is put into the queue and the thread exits.

    Raises
    ------
    Linux2MqttMetricsException
        network information could not be gathered or prepared for publishing

    """
    try:
        x = 0
        interval = self.interval
        tx_bytes = []
        rx_bytes = []
        prev_tx = 0
        prev_rx = 0
        base_tx = 0
        base_rx = 0
        while x < interval:
            nics = psutil.net_io_counters(pernic=True)
            if self.nic in nics:
                tx = nics[self.nic].bytes_sent
                rx = nics[self.nic].bytes_recv
                if tx < prev_tx:
                    # TX counter rollover
                    base_tx += prev_tx
                if rx < prev_rx:
                    # RX counter rollover
                    base_rx += prev_rx
                tx_bytes.append(base_tx + tx)
                rx_bytes.append(base_rx + rx)
                prev_tx = tx
                prev_rx = rx
            time.sleep(1)
            x += 1

        if self.nic in nics:
            tx_rate_bytes_sec = np.average(np.diff(np.array(tx_bytes)))
            tx_rate = tx_rate_bytes_sec / 125.0  # bytes/sec to kilobits/sec
            rx_rate_bytes_sec = np.average(np.diff(np.array(rx_bytes)))
            rx_rate = rx_rate_bytes_sec / 125.0  # bytes/sec to kilobits/sec

            self.metric.polled_result = {
                "total_rate": int(tx_rate + rx_rate),
                "tx_rate": int(tx_rate),
                "rx_rate": int(rx_rate),
            }
            self.result_queue.put(self.metric)
        else:
            metric_logger.warning("Network %s not available", self.nic)
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish network data"
        ) from ex

`TempMetrics` ¶

Bases: BaseMetric

Thermal zones metric.

Initialize the thermal zone metric.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device	required
`thermal_zone`	`str`	The thermal zone	required

Raises:

Type	Description
`Linux2MqttConfigException`	Bad config

Source code in linux2mqtt/metrics.py

def __init__(self, device: str, thermal_zone: str):
    """Initialize the thermal zone metric.

    Parameters
    ----------
    device
        The device
    thermal_zone
        The thermal zone

    Raises
    ------
    Linux2MqttConfigException
        Bad config

    """
    super().__init__()
    self._device = device
    self._thermal_zone = thermal_zone
    self._name = self._name_template.format(device, thermal_zone)

`poll` ¶

Poll new data for the thermal zone metric.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[Self]`	(Unused)	required

Returns:

Type	Description
`bool`	True as the data is readily available

Raises:

Type	Description
`Linux2MqttMetricsException`	Thermal zone information could not be gathered or prepared for publishing

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[Self]) -> bool:
    """Poll new data for the thermal zone metric.

    Parameters
    ----------
    result_queue
        (Unused)

    Returns
    -------
    bool
        True as the data is readily available

    Raises
    ------
    Linux2MqttMetricsException
        Thermal zone information could not be gathered or prepared for publishing

    """
    try:
        st = psutil.sensors_temperatures()  # type: ignore[attr-defined]
        thermal_zone = next(
            (
                item
                for item in st.get(self._device, [])
                if item.label == self._thermal_zone
            ),
            None,
        )
        assert thermal_zone
        self.polled_result = {
            "label": thermal_zone.label,
            "current": thermal_zone.current,
            "high": thermal_zone.high,
            "critical": thermal_zone.critical,
        }
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish thermal zone data"
        ) from ex
    else:
        return False

`VirtualMemoryMetrics` ¶

Bases: BaseMetric

Virtual memory metric.

Source code in linux2mqtt/metrics.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize base class."""
    self.polled_result = None

`poll` ¶

Poll new data for the virtual memory metric.

Parameters:

Name	Type	Description	Default
`result_queue`	`Queue[Self]`	(Unused)	required

Returns:

Type	Description
`bool`	True as the data is readily available

Raises:

Type	Description
`Linux2MqttMetricsException`	virtual memory information could not be gathered or prepared for publishing

Source code in linux2mqtt/metrics.py

def poll(self, result_queue: Queue[Self]) -> bool:
    """Poll new data for the virtual memory metric.

    Parameters
    ----------
    result_queue
        (Unused)

    Returns
    -------
    bool
        True as the data is readily available

    Raises
    ------
    Linux2MqttMetricsException
        virtual memory information could not be gathered or prepared for publishing

    """
    try:
        vm = psutil.virtual_memory()
        self.polled_result = {
            "total": float(vm.total) / 1_000_000,
            "available": float(vm.available) / 1_000_000,
            "percent": vm.percent,
            "used": float(vm.used) / 1_000_000,
            "free": float(vm.free) / 1_000_000,
            "active": float(vm.active) / 1_000_000,
            "inactive": float(vm.inactive) / 1_000_000,
        }
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish virtual memory data"
        ) from ex
    else:
        return False

`Linux2MqttConfig` ¶

Bases: TypedDict

A config object.

Attributes:

Name	Type	Description
`log_level`	`str`	Log verbosity
`homeassistant_prefix`	`str`	MQTT discovery topic prefix
`linux2mqtt_hostname`	`str`	A descriptive name for the system being monitored
`mqtt_client_id`	`str`	Client Id for MQTT broker client
`mqtt_user`	`str`	Username for MQTT broker authentication
`mqtt_password`	`str`	Password for MQTT broker authentication
`mqtt_host`	`str`	Hostname or IP address of the MQTT broker
`mqtt_port`	`int`	Port or IP address of the MQTT broker
`mqtt_timeout`	`int`	Timeout for MQTT messages
`mqtt_topic_prefix`	`str`	MQTT topic prefix
`mqtt_qos`	`int`	QOS for standard MQTT messages
`interval`	`int`	Publish metrics to MQTT broker every n seconds

`LinuxDeviceEntry` ¶

Bases: TypedDict

A linux device entry object for discovery in home assistant.

Attributes:

Name	Type	Description
`identifiers`	`str`	A unique str to identify the device in home assistant
`name`	`str`	The name of the device to display in home assistant
`model`	`str`	The model of the device as additional info

`LinuxEntry` ¶

Bases: TypedDict

A linux entry object for discovery in home assistant.

Attributes:

Name	Type	Description
`name`	`str`	The name of the sensor to display in home assistant
`unique_id`	`str`	The unique id of the sensor in home assistant
`icon`	`str \| None`	The icon of the sensor to display
`availability_topic`	`str`	The topic to check the availability of the sensor
`payload_available`	`str`	The payload of availability_topic of the sensor when available
`payload_unavailable`		The payload of availability_topic of the sensor when unavailable
`state_topic`	`str`	The topic containing all information for the state of the sensor
`value_template`	`str`	The jinja2 template to extract the state value from the state_topic for the sensor
`unit_of_measurement`	`str \| None`	The unit of measurement of the sensor
`payload_on`	`str`	When a binary sensor: The value of extracted state of the sensor to be considered 'on'
`payload_off`	`str`	When a binary sensor: The value of extracted state of the sensor to be considered 'off'
`device`	`LinuxDeviceEntry`	The device the sensor is attributed to
`device_class`	`str \| None`	The device class of the sensor
`state_topic`	`str`	The topic containing all information for the attributes of the sensor
`qos`	`int`	The QOS of the discovery message

`clean_for_discovery` ¶

Cleanup a typed dict for home assistant discovery, which is quite picky and does not like empty of None values.

Parameters:

Name	Type	Description	Default
`val`	`LinuxEntry`	The TypedDict to cleanup	required

Returns:

Type	Description
`dict`	The cleaned dict

Source code in linux2mqtt/helpers.py

def clean_for_discovery(val: LinuxEntry) -> dict[str, str | int | float | object]:
    """Cleanup a typed dict for home assistant discovery, which is quite picky and does not like empty of None values.

    Parameters
    ----------
    val
        The TypedDict to cleanup

    Returns
    -------
    dict
        The cleaned dict

    """

    return {
        k: v
        for k, v in dict(val).items()
        if isinstance(v, str | int | float | object) and v not in (None, "")
    }

`sanitize` ¶

Sanitize a value for unique_id usage.

Parameters:

Name	Type	Description	Default
`val`	`str`	The string to sanitize	required

Returns:

Type	Description
`str`	The sanitized value

Source code in linux2mqtt/helpers.py

def sanitize(val: str) -> str:
    """Sanitize a value for unique_id usage.

    Parameters
    ----------
    val
        The string to sanitize

    Returns
    -------
    str
        The sanitized value

    """
    return re.sub(r"[^a-zA-Z0-9_-]", "_", val.lower())

Reference¶

SensorType = Literal['sensor', 'binary_sensor'] module-attribute ¶

StatusType = Literal['online', 'offline'] module-attribute ¶

Linux2MqttConfigException ¶

Linux2MqttConnectionException ¶

Linux2MqttException ¶

Linux2MqttMetricsException ¶

Linux2Mqtt ¶

connect ¶

__del__ ¶

add_metric ¶

loop_busy ¶

BaseMetric ¶

name property ¶

name_sanitized property ¶

get_discovery ¶

poll ¶

BaseMetricThread ¶

CPUMetrics ¶

poll ¶

CPUMetricThread ¶

run ¶

DiskUsageMetrics ¶

poll ¶

FanSpeedMetrics ¶

poll ¶

NetworkMetrics ¶

poll ¶

NetworkMetricThread ¶

run ¶

TempMetrics ¶

poll ¶

VirtualMemoryMetrics ¶

poll ¶

Linux2MqttConfig ¶

LinuxDeviceEntry ¶

LinuxEntry ¶

clean_for_discovery ¶

sanitize ¶

`SensorType = Literal['sensor', 'binary_sensor']` `module-attribute` ¶

`StatusType = Literal['online', 'offline']` `module-attribute` ¶

`Linux2MqttConfigException` ¶

`Linux2MqttConnectionException` ¶

`Linux2MqttException` ¶

`Linux2MqttMetricsException` ¶

`Linux2Mqtt` ¶

`connect` ¶

`del` ¶

`add_metric` ¶

`loop_busy` ¶

`BaseMetric` ¶

`name` `property` ¶

`name_sanitized` `property` ¶

`get_discovery` ¶

`poll` ¶

`BaseMetricThread` ¶

`CPUMetrics` ¶

`poll` ¶

`CPUMetricThread` ¶

`run` ¶

`DiskUsageMetrics` ¶

`poll` ¶

`FanSpeedMetrics` ¶

`poll` ¶

`NetworkMetrics` ¶

`poll` ¶

`NetworkMetricThread` ¶

`run` ¶

`TempMetrics` ¶

`poll` ¶

`VirtualMemoryMetrics` ¶

`poll` ¶

`Linux2MqttConfig` ¶

`LinuxDeviceEntry` ¶

`LinuxEntry` ¶

`clean_for_discovery` ¶

`sanitize` ¶