Reference

systemctl2mqtt package.

ServiceActiveType = Literal['active', 'inactive', 'failed'] module-attribute

Systemctl service active status

ServiceEventStateType = Literal['on', 'off'] module-attribute

Service event state

ServiceEventStatusType = Literal['running', 'exited', 'failed'] module-attribute

Service event Systemctl status

ServiceLoadType = Literal['loaded', 'not-found', 'error'] module-attribute

Systemctl service load status

Systemctl2MqttConfigException

Bases: Systemctl2MqttException

Config exception occurred.

Systemctl2MqttConnectionException

Bases: Systemctl2MqttException

Connection exception occurred.

Systemctl2MqttEventsException

Bases: Systemctl2MqttException

Events processing exception occurred.

Systemctl2MqttException

Bases: Exception

General processing exception occurred.

Systemctl2MqttStatsException

Bases: Systemctl2MqttException

Stats processing exception occurred.

Systemctl2Mqtt

systemctl2mqtt class.

Attributes:

Name	Type	Description
version	str	The version of systemctl2mqtt
cfg	Systemctl2MqttConfig	The config for systemctl2mqtt
b_stats	bool	Activate the stats
b_events	bool	Activate the events
systemctl_events	Queue[dict[str, str]]	Queue with systemctl events
systemctl_stats	Queue[list[str]]	Queue with systemctl stats
known_event_services	dict[str, ServiceEvent]	The dict with the known service events
known_stat_services	dict[str, dict[int, ServiceStatsRef]]	The dict with the known service stats references
last_stat_services	dict[str, ServiceStats | dict[str, Any]]	The dict with the last service stats
mqtt	Client	The mqtt client
systemctl_events_t	Thread	The thread to collect events from systemctl
systemctl_stats_t	Thread	The thread to collect stats from systemctl
systemctl_version	str	The systemctl version
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
stats_topic	str	Topic template for stats
events_topic	str	Topic template for an events
do_not_exit	bool	Prevent exit from within Systemctl2mqtt, when handled outside

Initialize the Systemctl2mqtt.

Parameters:

Name	Type	Description	Default
cfg	Systemctl2MqttConfig	The configuration object for Systemctl2mqtt	required
do_not_exit	bool	Prevent exit from within Systemctl2mqtt, when handled outside	False

Source code in systemctl2mqtt/systemctl2mqtt.py

def __init__(self, cfg: Systemctl2MqttConfig, do_not_exit: bool = False):
    """Initialize the Systemctl2mqtt.

    Parameters
    ----------
    cfg
        The configuration object for Systemctl2mqtt
    do_not_exit
        Prevent exit from within Systemctl2mqtt, when handled outside

    """

    self.cfg = cfg
    self.do_not_exit = do_not_exit

    self.discovery_binary_sensor_topic = f"{cfg['homeassistant_prefix']}/binary_sensor/{cfg['mqtt_topic_prefix']}/{cfg['systemctl2mqtt_hostname']}_{{}}/config"
    self.discovery_sensor_topic = f"{cfg['homeassistant_prefix']}/sensor/{cfg['mqtt_topic_prefix']}/{cfg['systemctl2mqtt_hostname']}_{{}}/config"
    self.status_topic = (
        f"{cfg['mqtt_topic_prefix']}/{cfg['systemctl2mqtt_hostname']}/status"
    )
    self.version_topic = (
        f"{cfg['mqtt_topic_prefix']}/{cfg['systemctl2mqtt_hostname']}/version"
    )
    self.stats_topic = (
        f"{cfg['mqtt_topic_prefix']}/{cfg['systemctl2mqtt_hostname']}/{{}}/stats"
    )
    self.events_topic = (
        f"{cfg['mqtt_topic_prefix']}/{cfg['systemctl2mqtt_hostname']}/{{}}/events"
    )

    if self.cfg["enable_events"]:
        self.b_events = True
    if self.cfg["enable_stats"]:
        self.b_stats = True

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

    try:
        self.systemctl_version = self._get_systemctl_version()
    except FileNotFoundError as ex:
        raise Systemctl2MqttConfigException(
            "Could not get systemctl version"
        ) from ex

    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)

    main_logger.info("Events enabled: %d", self.b_events)
    main_logger.info("Stats enabled: %d", self.b_stats)

    try:
        # Setup MQTT
        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"],
        )
        self.mqtt.username_pw_set(
            username=self.cfg["mqtt_user"], password=self.cfg["mqtt_password"]
        )
        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 Systemctl2MqttConnectionException from ex

    # Register services
    self._reload_services()

    started = False
    try:
        if self.b_events:
            logging.info("Starting Events thread")
            self._start_readline_events_thread()
            started = True
    except Exception as ex:
        main_logger.exception("Error while trying to start events thread.")
        main_logger.debug(ex)
        raise Systemctl2MqttConfigException from ex

    try:
        if self.b_stats:
            started = True
            logging.info("Starting Stats thread")
            self._start_readline_stats_thread()
    except Exception as ex:
        main_logger.exception("Error while trying to start stats thread.")
        main_logger.debug(ex)
        raise Systemctl2MqttConfigException from ex

    if started is False:
        logging.critical("Nothing started, check your config!")
        sys.exit(1)

__del__

Destroy the class.

Source code in systemctl2mqtt/systemctl2mqtt.py

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

loop

Start the loop.

Raises:

Type	Description
Systemctl2MqttEventsException	If anything goes wrong in the processing of the events
Systemctl2MqttStatsException	If anything goes wrong in the processing of the stats
Systemctl2MqttException	If anything goes wrong outside of the known exceptions

Source code in systemctl2mqtt/systemctl2mqtt.py

def loop(self) -> None:
    """Start the loop.

    Raises
    ------
    Systemctl2MqttEventsException
        If anything goes wrong in the processing of the events
    Systemctl2MqttStatsException
        If anything goes wrong in the processing of the stats
    Systemctl2MqttException
        If anything goes wrong outside of the known exceptions

    """

    self._remove_destroyed_services()

    self._handle_events_queue()

    self._handle_stats_queue()

    try:
        if self.b_events and not self.systemctl_events_t.is_alive():
            main_logger.warning("Restarting events thread")
            self._start_readline_events_thread()
    except Exception as e:
        main_logger.exception("Error while trying to restart events thread.")
        main_logger.debug(e)
        raise Systemctl2MqttConfigException from e

    try:
        if self.b_stats and not self.systemctl_stats_t.is_alive():
            main_logger.warning("Restarting stats thread")
            self._start_readline_stats_thread()
    except Exception as e:
        main_logger.exception("Error while trying to restart stats thread.")
        main_logger.debug(e)
        raise Systemctl2MqttConfigException from e

loop_busy

Start the loop (blocking).

Parameters:

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

Raises:

Type	Description
Systemctl2MqttEventsException	If anything goes wrong in the processing of the events
Systemctl2MqttStatsException	If anything goes wrong in the processing of the stats
Systemctl2MqttException	If anything goes wrong outside of the known exceptions

Source code in systemctl2mqtt/systemctl2mqtt.py

def loop_busy(self, raise_known_exceptions: bool = False) -> None:
    """Start the loop (blocking).

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

    Raises
    ------
    Systemctl2MqttEventsException
        If anything goes wrong in the processing of the events
    Systemctl2MqttStatsException
        If anything goes wrong in the processing of the stats
    Systemctl2MqttException
        If anything goes wrong outside of the known exceptions

    """

    while True:
        try:
            self.loop()
        except Systemctl2MqttEventsException 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)
                )
        except Systemctl2MqttStatsException 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)
                )

        # Calculate next iteration between (~0.2s and 0.001s)
        sleep_time = 0.001 + 0.2 / MAX_QUEUE_SIZE * (
            MAX_QUEUE_SIZE
            - max(self.systemctl_events.qsize(), self.systemctl_stats.qsize())
        )
        main_logger.debug("Sleep for %.5fs until next iteration", sleep_time)
        sleep(sleep_time)

PIDStats

Bases: TypedDict

A PID stats object which is part of the services stats.

Attributes:

Name	Type	Description
pid	int	The pid of the Service
memory	float	Used memory in MB
cpu	float	The cpu usage by the Service in cpu-% (ex.: a Systemctl with 4 cores has 400% cpu available)

ServiceDeviceEntry

Bases: TypedDict

A Service 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

ServiceEntry

Bases: TypedDict

A Service 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 | None	When a binary sensor: The value of extracted state of the sensor to be considered 'on'
payload_off	str | None	When a binary sensor: The value of extracted state of the sensor to be considered 'off'
device	ServiceDeviceEntry	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

ServiceEvent

Bases: TypedDict

A Service event object to send to an mqtt topic.

Attributes:

Name	Type	Description
name	str	The name of the Service
description	str	The description of the Service
pid	int	The pid of the Service
cpids	list[int]	The child pids of the Service
status	ServiceEventStatusType	The Systemctl status the Service is in
state	ServiceEventStateType	The state of the Service

ServiceStats

Bases: TypedDict

A Service stats object to send to an mqtt topic.

Attributes:

Name	Type	Description
name	str	The name of the Service
host	str	The Systemctl host
memory	float	Used memory in MB
cpu	float	The cpu usage by the Service in cpu-% (ex.: a Systemctl with 4 cores has 400% cpu available)
pid_stats	dict[int, PIDStats]	The stats for all pids

ServiceStatsRef

Bases: TypedDict

A Service stats ref object compare between current and past stats.

Attributes:

Name	Type	Description
last	datetime	When the last stat rotation happened

Systemctl2MqttConfig

Bases: TypedDict

A config object.

Attributes:

Name	Type	Description
log_level	str	Log verbosity
homeassistant_prefix	str	MQTT discovery topic prefix
systemctl2mqtt_hostname	str	A descriptive name for the Systemctl 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
destroyed_service_ttl	int	How long, in seconds, before destroyed services are removed from Home Assistant. Services won't be removed if the service is restarted before the TTL expires.
service_whitelist	list[str]	Whitelist the services to monitor, if empty, everything is monitored. The entries are either match as literal strings or as regex.
service_blacklist	list[str]	Blacklist the services to monitor, takes priority over whitelist. The entries are either match as literal strings or as regex.
enable_events	bool	Flag to enable event monitoring
enable_stats	bool	Flag to enable stat monitoring
stats_record_seconds	int	Interval every how many seconds the stats are published via MQTT

SystemctlService

Bases: TypedDict

A systemctl service definition.

Attributes:

Name	Type	Description
unit	str	The name of the service
load	ServiceLoadType	Is the service loaded
active	ServiceActiveType	High level status of the service
sub	ServiceEventStatusType	Detailed state of the service
description	str	Description of the service

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	ServiceEntry	The TypedDict to cleanup	required

Returns:

Type	Description
dict	The cleaned dict

Source code in systemctl2mqtt/helpers.py

def clean_for_discovery(
    val: ServiceEntry,
) -> 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, "")
    }