Skip to content

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 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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 
302
303
304
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 
364
365
366
367
368
369
370
371
372
373
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 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
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:
            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)
                )
        x = 0
        while x < self.cfg["interval"]:
            # Check the queue for deferred results one/sec
            time.sleep(1)
            self._check_queue()
            x += 1

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 | list[str] | list[int] | None] | None

The dict with the polled result data for the state and attributes

Initialize base class.

Source code in linux2mqtt/metrics.py 
65
66
67
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 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
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 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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 
366
367
368
369
370
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 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
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 
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
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 
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
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

NetConnectionMetrics

Bases: BaseMetric

Network connections metric.

Extract local IPs for evaluation during poll.

Parameters:

Name Type Description Default
interval int

The interval to gather data over

 required
Source code in linux2mqtt/metrics.py 
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
def __init__(self, interval: int) -> None:
    """Extract local IPs for evaluation during poll.

    Parameters
    ----------
    interval
        The interval to gather data over

    """
    super().__init__()
    interface_addrs = psutil.net_if_addrs()
    self.ips = set()
    self.interval = interval

    for snicaddrs in interface_addrs.values():
        for snicaddr in snicaddrs:
            if snicaddr.family.value in (2, 10):
                self.ips.add(snicaddr.address)

poll

Poll new data for the network connection metrics.

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

network connection information could not be gathered or prepared for publishing
Source code in linux2mqtt/metrics.py 
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
def poll(self, result_queue: Queue[Self]) -> bool:
    """Poll new data for the network connection metrics.

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

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

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

    """
    try:
        st = psutil.net_connections()
        listening_ports = {
            x.laddr.port
            for x in st
            if x.status == "LISTEN"
            and isinstance(x.laddr, addr)
            and x.laddr.ip in ("0.0.0.0", "::")
        }

        self.polled_result = {
            "count": len([x for x in st if x.status == "ESTABLISHED"]),
            "ipv4": len(
                [
                    x
                    for x in st
                    if x.family.value == 2
                    and x.status == "ESTABLISHED"
                    and isinstance(x.laddr, addr)
                    and not x.laddr.ip.startswith("127.")
                ]
            ),
            "ipv6": len(
                [
                    x
                    for x in st
                    if x.family.value == 10
                    and x.status == "ESTABLISHED"
                    and isinstance(x.laddr, addr)
                    and x.laddr.ip != "::1"
                ]
            ),
            "listening_ports": list(listening_ports),
            "outbound": [
                f"{x.raddr.ip}:{x.raddr.port}"
                for x in st
                if x.status == "ESTABLISHED"
                and isinstance(x.laddr, addr)
                and isinstance(x.raddr, addr)
                and x.laddr.ip in self.ips
                and x.raddr.ip not in ("::1", "127.0.0.1")
            ],
            "inbound": [
                f"{x.raddr.ip}:{x.raddr.port} -> {x.laddr.ip}:{x.laddr.port}"
                for x in st
                if x.status == "ESTABLISHED"
                and isinstance(x.laddr, addr)
                and isinstance(x.raddr, addr)
                and x.laddr.port in listening_ports
            ],
        }
    except Exception as ex:
        raise Linux2MqttMetricsException(
            "Could not gather and publish net connections"
        ) 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 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
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 
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
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 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
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 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
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 
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
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 
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
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 
65
66
67
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 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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())