8 Commits

Author SHA1 Message Date
1c52859fb4 feat(dbus): Add support for configuring 'revpi-con-can' feature
Introduce the `configure_con_can` function to manage enabling,
disabling, status checking, and availability of the 'revpi-con-can'
feature. Update the `AVAILABLE_FEATURES` dictionary to integrate
'revpi-con-can' as a configurable feature.
2025-04-21 08:03:02 +02:00
f9f50d7167 feat(dbus): Add support for configuring the external antenna
Introduce the `configure_external_antenna` function to manage external
antenna settings, including enable, disable, and status checks. Update
the feature configuration in `interface_config` to include this
functionality. This enhances WiFi-related configuration options.
2025-04-21 08:03:02 +02:00
e21a61232d feat(dbus): Add ConfigTxt class for managing config.txt file
Introduced the ConfigTxt class to handle parsing, editing, and saving of
config.txt files. This includes methods for adding, clearing, and
retrieving configuration values, as well as handling dtoverlays and
dtparams. Enhanced system configuration capabilities by providing
structured support for config file operations.
2025-04-21 08:03:02 +02:00
8d8d3bbae4 feat(dbus): Add dphys-swapfile configuration functionality
Implemented `configure_dphys_swapfile` to manage the dphys-swapfile
service, including automatic swapfile removal when the service is
disabled. Updated feature registry to support dphys-swapfile
configuration.
2025-04-20 16:32:57 +02:00
1ae661a5e9 feat(dbus): Add avahi-daemon configuration to system services
Introduce a `configure_avahi_daemon` function to manage the avahi-daemon
service and socket with proper post actions. Update the interface
configuration to enable avahi management using the new function.
2025-04-20 16:16:51 +02:00
ffdb27049e feat(dbus): Remove 'var-log.mount' feature
The 'var-log.mount' feature was dropped and has been removed from
the AVAILABLE_FEATURES dictionary.
2025-04-20 16:02:39 +02:00
38f495e864 refactor(dbus): Move system configuration methods to revpi_config.py
Centralized `ConfigActions`, `configure_gui`, and `simple_systemd` into
`revpi_config.py` to eliminate duplication and improve maintainability.
Updated `interface_config.py` to import these methods, ensuring
consistent functionality across modules.
2025-04-20 15:43:09 +02:00
85dc6a7e56 feat(dbus): Add RevPiConfig class for device information handling
This commit introduces a new `RevPiConfig` class to manage RevPi device
configuration details, such as model, serial, compute module type, WiFi,
and ConBridge detection. It parses CPU info from `/proc/cpuinfo` and
leverages helper methods for hardware-specific checks, enhancing the
middleware's ability to identify and manage hardware features.
2025-04-20 15:39:02 +02:00
15 changed files with 68 additions and 740 deletions

View File

@@ -1,4 +1,4 @@
<!-- /usr/share/dbus-1/system.d/com.revolutionpi.middleware1.conf --> <!-- /etc/dbus-1/system.d/revpi-middleware.conf -->
<busconfig> <busconfig>
<!-- Allow full access to root as the bus owner --> <!-- Allow full access to root as the bus owner -->
<policy user="root"> <policy user="root">
@@ -12,7 +12,7 @@
<allow send_destination="com.revolutionpi.middleware1" <allow send_destination="com.revolutionpi.middleware1"
send_interface="org.freedesktop.DBus.Introspectable"/> send_interface="org.freedesktop.DBus.Introspectable"/>
<allow send_destination="com.revolutionpi.middleware1" <allow send_destination="com.revolutionpi.middleware1"
send_interface="com.revolutionpi.middleware1.PiControl"/> send_interface="com.revolutionpi.middleware1.picontrol"/>
</policy> </policy>
<!-- Standard-Policy --> <!-- Standard-Policy -->

88
debian/changelog vendored
View File

@@ -1,88 +0,0 @@
revpi-middleware (0.0.4-1+deb12+1) bookworm; urgency=medium
* fix(dbus): Update DBus policy file path and interface name
* fix(dbus): Update systemd interface and path handling
* doc(revpiconfig): Add detailed docstrings to ConfigTxt methods
* doc(revpiconfig): Add docstrings to `RevPiConfig` class and methods
* doc(revpiconfig): Add docstrings to enums in `revpi_config.py`
* doc(revpiconfig): Docstrings for `get_rfkill_index` and `simple_systemd`
-- Sven Sager <s.sager@kunbus.com> Tue, 22 Apr 2025 11:08:12 +0200
revpi-middleware (0.0.3-1+deb12+1) bookworm; urgency=medium
* refactor: Update interface name from 'picontrol' to 'PiControl'
* feat(dbus): Add `grep` function to search for patterns in a file
* fix(cli): Change absolute imports to relative imports
* feat(dbus): Add D-Bus interface for system configuration in middleware
* feat(dbus): Add GUI configuration handling to interface_config.py
* feat(revpiconfig): Add RevPiConfig class for device information handling
* refactor(dbus): Move system configuration methods to revpi_config.py
* feat(dbus): Remove 'var-log.mount' feature
* feat(dbus): Add avahi-daemon configuration to system services
* feat(dbus): Add dphys-swapfile configuration functionality
* feat(revpiconfig): Add ConfigTxt class for managing config.txt file
* feat(dbus): Add support for configuring the external antenna
* feat(dbus): Add support for configuring 'revpi-con-can' feature
* feat(revpiconfig): Enhance Wi-Fi detection and add rfkill index support
* feat(dbus): Add Wi-Fi configuration support to the system config
* refactor(revpiconfig: Change Wi-Fi detection and rfkill index logic
* feat(dbus): Add Bluetooth configuration functionality
* feat(dbus): Add InterfaceRevpiConfig to DBus interfaces list
* feat(cli): Add `get_properties` helper function for DBus interactions
* feat(cli): Add CLI command for configuring Revpi features
* feat(cli): Add CLI support for RevPi configuration object (revpi-config)
* feat(revpiconfig): Replace rfkill subprocess calls with sysfs writes
* refactor: Rename WiFi to WLAN for consistent terminology
* feat(deb): Add dependencies for `dtoverlay` command
-- Sven Sager <s.sager@kunbus.com> Mon, 21 Apr 2025 13:44:18 +0200
revpi-middleware (0.0.2-1+deb12+1) bookworm; urgency=medium
* refactor(dbus): Move D-Bus helper functions to a dedicated file
* refactor(dbus): Move ResetDriverWatchdog to process_image_helper.py
* refactor(dbus): Parameterize `picontrol_device` and `config_rsc`
* feat: Add session bus option for local testing and development
* feat(dbus): Add import for BusProvider in dbus_middleware1 module
* feat(dbus): Add `running` property to `BusProvider`
* refactor(cli): D-Bus helpers support session and system bus types
* fix(dbus): Add error handling for DBus publishing and main loop
* refactor(dbus): piControl driver reset with PiControlIoctl class
* test(dbus): Add unit test framework for dbus_middleware1 module
* test(dbus): Add unit tests for PiControl D-Bus interface
* refactor(dbus): Fix typo and remove unused thread instance
* refactor(dbus): D-Bus interface management with cleanup support.
* test(dbus): Add support for testing driver reset notification
* feat(deb): Add dbus for testing to build dependencies
* fix(deb): Skip tests because of missing SystemBus in build container
-- Sven Sager <s.sager@kunbus.com> Sat, 19 Apr 2025 16:34:20 +0200
revpi-middleware (0.0.1-1+deb12+1) bookworm; urgency=medium
* docs: Start git project with python git-ignore and Readme
* docs: Use 'reuse' for SPDX Headers and Licenses
* feat: Add python base project files
* feat: Add proginit application basic module
* feat: Add the data directory for additional data files for the project
* test: Add tests directory with a dummy test
* build: Add all necessary files for the build system
* feat: Add dummy main application script
* feat: Add systemd file and data to integrate the app as a daemon
* chore: Update proginit to 1.4.0
* feat(dbus): Add ResetDriverWatchdog helper as global dbus helper
* feat(process_image): Add D-Bus interface for piControl driver
* feat(dbus): Add initial D-Bus middleware implementation
* feat(dbus): Add `extend_interface` function for dynamic interface naming
* feat(dbus): Add DBus policy configuration for revpi-middleware
* feat: Add MiddlewareDaemon implementation to revpi-middleware
* feat: Add daemon mode and signal handling to main application
* feat(cli): Add D-Bus helper functions for CLI commands.
* feat(cli): Add await_signal function to handle D-Bus signals
* feat(cli): Add `await-reset` to wait for piControl reset signal
* chore(build): Update requirements for this project
* feat(cli): Add new CLI tool entry point for `revpictl`
* feat(deb): Start packaging branch
-- Sven Sager <s.sager@kunbus.com> Fri, 18 Apr 2025 19:02:20 +0200

39
debian/control vendored
View File

@@ -1,39 +0,0 @@
Source: revpi-middleware
Section: python
Priority: optional
Maintainer: KUNBUS GmbH <support@kunbus.com>
Rules-Requires-Root: no
Homepage: https://revolutionpi.com/
Vcs-Browser: https://gitlab.com/revolutionpi/revpi-middleware
Vcs-Git: https://gitlab.com/revolutionpi/revpi-middleware.git -b debian/bookworm
Build-Depends:
dbus,
dbus-x11,
debhelper-compat (= 13),
dh-python,
python3-all,
python3-gi (>= 3.42.2),
python3-pydbus (>= 0.6.0),
python3-setuptools,
Standards-Version: 4.6.2
Package: revpi-middleware
Architecture: all
Pre-Depends: ${misc:Pre-Depends}
Depends:
raspi-utils-dt | libraspberrypi-bin,
${python3:Depends},
${misc:Depends}
Description: Revolution Pi middleware with D-Bus interface
The Revolution Pi middleware provides a robust communication interface for
Revolution Pi industrial computers. It enables seamless integration between
hardware components and applications through a D-Bus interface. The middleware
serves as a bridge for data exchange, device configuration, and system
monitoring.
.
Key features:
* Hardware abstraction layer for Revolution Pi I/O modules
* Real-time data processing and event handling
* Simplified API for accessing Revolution Pi hardware features
* Extensive configuration options for industrial automation tasks
* Built-in monitoring and diagnostic capabilities

27
debian/copyright vendored
View File

@@ -1,27 +0,0 @@
Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
Source: https://gitlab.com/revolutionpi/opcua-revpi-server
Files: *
Copyright: 2025 KUNBUS GmbH
License: GPL-2+
Files: debian/*
Copyright: 2025 KUNBUS GmbH
License: GPL-2+
License: GPL-2+
This package is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
.
This package is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>
.
On Debian systems, the complete text of the GNU General
Public License version 2 can be found in "/usr/share/common-licenses/GPL-2".

7
debian/gbp.conf vendored
View File

@@ -1,7 +0,0 @@
[DEFAULT]
upstream-branch = main
upstream-tag = v%(version)s
debian-branch=debian/bookworm
debian-tag = debian/%(version)s
debian-tag-msg = %(pkg)s Debian release %(version)s
pristine-tar = True

View File

@@ -1,4 +0,0 @@
data/dbus-policy/com.revolutionpi.middleware1.conf /usr/share/dbus-1/system.d
data/etc/default/revpi-middleware /etc/default/
data/etc/revpi-middleware/revpi-middleware.conf /etc/revpi-middleware/
data/systemd/before_253/revpi-middleware.service /lib/systemd/system/

View File

@@ -1,2 +0,0 @@
/usr/share/revpi-middleware/revpi-middleware /usr/sbin/revpi-middleware
/usr/share/revpi-middleware/revpicli /usr/bin/revpicli

12
debian/rules vendored
View File

@@ -1,12 +0,0 @@
#!/usr/bin/make -f
export PYBUILD_NAME=revpi-middleware
export PYBUILD_INSTALL_ARGS=--install-lib=/usr/share/$(PYBUILD_NAME)/ --install-scripts=/usr/share/$(PYBUILD_NAME)/
%:
dh $@ --with python3 --buildsystem=pybuild
override_dh_auto_test:
# Currently, the tests have to be skipped, because no SystemBus is
# available in the Docker container.
@echo "Skipped tests"

View File

@@ -1 +0,0 @@
3.0 (quilt)

View File

@@ -29,11 +29,6 @@ def setup_command_line_arguments():
help="RevPi PiControl object", help="RevPi PiControl object",
) )
cli_picontrol.add_subparsers(obj_picontrol) cli_picontrol.add_subparsers(obj_picontrol)
obj_config = rpictl_obj.add_parser(
"config",
help="RevPi configuration object (revpi-config)",
)
cli_config.add_subparsers(obj_config)
def main() -> int: def main() -> int:
@@ -44,9 +39,6 @@ def main() -> int:
if obj == "picontrol": if obj == "picontrol":
rc = cli_picontrol.main() rc = cli_picontrol.main()
elif obj == "config":
rc = cli_config.main()
else: else:
log.error(f"Unknown object: {obj}") log.error(f"Unknown object: {obj}")
rc = 1 rc = 1

View File

@@ -1,93 +0,0 @@
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
# SPDX-License-Identifier: GPL-2.0-or-later
"""Command-Line for the picontrol object of CLI."""
from argparse import ArgumentParser
from logging import getLogger
from .dbus_helper import BusType, get_properties, simple_call
from .. import proginit as pi
from ..dbus_middleware1 import extend_interface
log = getLogger(__name__)
def add_subparsers(parent_parser: ArgumentParser):
parent_parser.add_argument(
"action",
choices=["enable", "disable", "status", "available", "list-features"],
help="Action to be executed: enable, disable, status or available. "
"To get all available features, use 'list-features'.",
)
parent_parser.add_argument(
"feature",
nargs="?",
default="",
help="Name of the feature to configer. To list all features use 'list-features' as action.",
)
def main() -> int:
action = pi.pargs.action
dbus_value = False
try:
if action == "list-features":
dbus_value = get_properties(
"available_features",
interface=extend_interface("RevpiConfig"),
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
)
for feature in dbus_value:
print(feature)
return 0
# For the following actions, a feature name is required
if pi.pargs.feature == "":
raise Exception("Feature name is required")
if action == "enable":
simple_call(
"Enable",
pi.pargs.feature,
interface=extend_interface("RevpiConfig"),
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
)
elif action == "disable":
simple_call(
"Disable",
pi.pargs.feature,
interface=extend_interface("RevpiConfig"),
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
)
elif action == "status":
dbus_value = simple_call(
"GetStatus",
pi.pargs.feature,
interface=extend_interface("RevpiConfig"),
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
)
elif action == "available":
dbus_value = simple_call(
"GetAvailability",
pi.pargs.feature,
interface=extend_interface("RevpiConfig"),
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
)
else:
raise Exception(f"Unknown action: {action}")
except Exception as e:
log.error(f"Error: {e}")
return 1
log.debug(
f"D-Bus call of method {action} for feature {pi.pargs.feature} returned: {dbus_value}"
)
print(int(dbus_value))
return 0

View File

@@ -17,18 +17,6 @@ class BusType(Enum):
SYSTEM = "system" SYSTEM = "system"
def get_properties(
property_name: str,
interface: str,
object_path=REVPI_DBUS_BASE_PATH,
bus_type=BusType.SYSTEM,
):
bus = SessionBus() if bus_type is BusType.SESSION else SystemBus()
revpi = bus.get(REVPI_DBUS_NAME, object_path)
iface = revpi[interface]
return getattr(iface, property_name)
def simple_call( def simple_call(
method: str, method: str,
*args, *args,

View File

@@ -10,7 +10,6 @@ from pydbus import SessionBus, SystemBus
from . import REVPI_DBUS_NAME from . import REVPI_DBUS_NAME
from .process_image import InterfacePiControl from .process_image import InterfacePiControl
from .system_config import InterfaceRevpiConfig
log = getLogger(__name__) log = getLogger(__name__)
@@ -42,7 +41,6 @@ class BusProvider(Thread):
# ("Subdir2/Whatever", Example()) # ("Subdir2/Whatever", Example())
lst_interfaces = [ lst_interfaces = [
InterfacePiControl(self.picontrol_device, self.config_rsc), InterfacePiControl(self.picontrol_device, self.config_rsc),
InterfaceRevpiConfig(),
] ]
try: try:

View File

@@ -8,12 +8,10 @@ from logging import getLogger
from .revpi_config import ( from .revpi_config import (
ConfigActions, ConfigActions,
configure_avahi_daemon, configure_avahi_daemon,
configure_bluetooth,
configure_con_can, configure_con_can,
configure_dphys_swapfile, configure_dphys_swapfile,
configure_external_antenna, configure_external_antenna,
configure_gui, configure_gui,
configure_wlan,
simple_systemd, simple_systemd,
) )
from ..dbus_helper import DbusInterface from ..dbus_helper import DbusInterface
@@ -93,8 +91,8 @@ AVAILABLE_FEATURES = {
simple_systemd, ["noderedrevpinodes-server.service"] simple_systemd, ["noderedrevpinodes-server.service"]
), ),
"revpipyload": FeatureFunction(simple_systemd, ["revpipyload.service"]), "revpipyload": FeatureFunction(simple_systemd, ["revpipyload.service"]),
"bluetooth": FeatureFunction(configure_bluetooth, []), "bluetooth": False,
"wlan": FeatureFunction(configure_wlan, []), "ieee80211": False,
"avahi": FeatureFunction(configure_avahi_daemon, []), "avahi": FeatureFunction(configure_avahi_daemon, []),
"external-antenna": FeatureFunction(configure_external_antenna, []), "external-antenna": FeatureFunction(configure_external_antenna, []),
} }

View File

@@ -6,11 +6,10 @@ import shutil
import subprocess import subprocess
from collections import namedtuple from collections import namedtuple
from enum import Enum, IntEnum from enum import Enum, IntEnum
from glob import glob
from logging import getLogger from logging import getLogger
from os import X_OK, access from os import X_OK, access
from os.path import exists, join from os.path import exists
from typing import List, Optional from typing import List
from pydbus import SystemBus from pydbus import SystemBus
@@ -20,27 +19,10 @@ log = getLogger(__name__)
ConfigVariable = namedtuple("ConfigVariable", ["name", "value", "line_index"]) ConfigVariable = namedtuple("ConfigVariable", ["name", "value", "line_index"])
LINUX_BT_CLASS_PATH = "/sys/class/bluetooth"
LINUX_WLAN_CLASS_PATH = "/sys/class/ieee80211"
CONFIG_TXT_LOCATIONS = ("/boot/firmware/config.txt", "/boot/config.txt") CONFIG_TXT_LOCATIONS = ("/boot/firmware/config.txt", "/boot/config.txt")
class ComputeModuleTypes(IntEnum): class ComputeModuleTypes(IntEnum):
"""
Enumeration class to represent compute module types.
This class is an enumeration that defines various types of compute
modules and assigns them associated integer values for identifying
different module types.
Attributes:
UNKNOWN (int): Represents an unknown or undefined compute module type.
CM1 (int): Represents a Compute Module 1.
CM3 (int): Represents a Compute Module 3.
CM4 (int): Represents a Compute Module 4.
CM4S (int): Represents a Compute Module 4S.
CM5 (int): Represents a Compute Module 5.
"""
UNKNOWN = 0 UNKNOWN = 0
CM1 = 6 CM1 = 6
CM3 = 10 CM3 = 10
@@ -50,13 +32,6 @@ class ComputeModuleTypes(IntEnum):
class ConfigActions(Enum): class ConfigActions(Enum):
"""
Enumeration class for defining configuration actions.
This enumeration provides predefined constants for common configuration
actions. It can be used to ensure consistency when working with or defining
such actions in a system.
"""
ENABLE = "enable" ENABLE = "enable"
DISABLE = "disable" DISABLE = "disable"
STATUS = "status" STATUS = "status"
@@ -64,26 +39,12 @@ class ConfigActions(Enum):
class RevPiConfig: class RevPiConfig:
"""
Represents the configuration and hardware details of a Revolution Pi system.
This class provides methods and properties to initialize and fetch
information related to the Revolution Pi device, such as model, serial
number, compute module type, WLAN capability, and the presence of a
connection bridge. The class works by parsing system-level files (e.g.,
`/proc/cpuinfo`) and using this data to identify hardware characteristics
and features.
Attributes:
serial (str): The serial number of the Revolution Pi device.
model (str): The model name of the Revolution Pi device.
"""
def __init__(self): def __init__(self):
self._cm_type = ComputeModuleTypes.UNKNOWN self._cm_type = ComputeModuleTypes.UNKNOWN
self._cm_with_wifi = False
self._revpi_with_con_bridge = False self._revpi_with_con_bridge = False
self._wlan_class_path = ""
self.serial = "" self.serial = ""
self.model = "" self.model = ""
@@ -91,29 +52,6 @@ class RevPiConfig:
self._init_device_info() self._init_device_info()
def _init_device_info(self): def _init_device_info(self):
"""
Initialize and retrieve detailed hardware information, including CPU details,
device type, WLAN interface, and connectivity features.
This method gathers information from system files and other sources to
initialize device-specific attributes such as model, serial number,
compute module type, and optional features like integrated WLAN
or ConBridge support. It performs checks specific to the detected
module type to accurately populate necessary device details.
Attributes
----------
model : str
The model of the CPU based on information from /proc/cpuinfo.
serial : str
The serial number extracted from /proc/cpuinfo.
_cm_type : ComputeModuleTypes, optional
The type of the compute module derived from the hardware revision value.
_wlan_class_path : str, optional
Filesystem path to the detected WLAN interface, if any.
_revpi_with_con_bridge : bool
Indicates whether the device supports the ConBridge feature.
"""
dc_cpuinfo = {} dc_cpuinfo = {}
# Extract CPU information # Extract CPU information
@@ -140,19 +78,11 @@ class RevPiConfig:
except ValueError: except ValueError:
pass pass
# Detect WLAN on CM module # Detect WiFi on CM module
could_have_wlan = self._cm_type in (ComputeModuleTypes.CM4, ComputeModuleTypes.CM5) could_have_wifi = self._cm_type in (ComputeModuleTypes.CM4, ComputeModuleTypes.CM5)
if could_have_wlan: if could_have_wifi:
wlan_interface = join(LINUX_WLAN_CLASS_PATH, "phy0") lst_grep = grep("DRIVER=brcmfmac", "/sys/class/ieee80211/phy0/device/uevent")
if grep("DRIVER=brcmfmac", join(wlan_interface, "device", "uevent")): self._cm_with_wifi = len(lst_grep) > 0 and self._cm_type in (ComputeModuleTypes)
self._wlan_class_path = wlan_interface
# If no build in WLAN on the CM, detect third party WLAN on RevPi Flat
if not self._wlan_class_path and grep("revpi-flat", "/proc/device-tree/compatible"):
lst_wlan_interfaces = glob("/sys/class/ieee80211/*")
for wlan_interface in lst_wlan_interfaces:
if grep("DRIVER=mwifiex_sdio", join(wlan_interface, "device", "uevent")):
self._wlan_class_path = wlan_interface
# Detect ConBridge # Detect ConBridge
could_have_con_bridge = self._cm_type in (ComputeModuleTypes.CM3, ComputeModuleTypes.CM4S) could_have_con_bridge = self._cm_type in (ComputeModuleTypes.CM3, ComputeModuleTypes.CM4S)
@@ -160,77 +90,20 @@ class RevPiConfig:
lst_grep = grep("kunbus,revpi-connect", "/proc/device-tree/compatible") lst_grep = grep("kunbus,revpi-connect", "/proc/device-tree/compatible")
self._revpi_with_con_bridge = len(lst_grep) > 0 self._revpi_with_con_bridge = len(lst_grep) > 0
@property
def class_path_wlan(self) -> str:
"""
Provides access to the WLAN class path.
This property retrieves the stored WLAN class path, allowing the user to access it when
needed.
Returns:
str: The WLAN class path.
"""
return self._wlan_class_path
@property @property
def cm_type(self) -> ComputeModuleTypes: def cm_type(self) -> ComputeModuleTypes:
"""
Gets the type of the compute module.
The property provides access to the type of the compute
module used. The type is represented as an instance of
the `ComputeModuleTypes` class.
Returns
-------
ComputeModuleTypes
The type of the compute module.
"""
return self._cm_type return self._cm_type
@property @property
def with_con_bridge(self) -> bool: def with_con_bridge(self) -> bool:
"""
Indicates if the device is configured with a connection bridge.
This property checks the internal status and determines whether the device setup
includes a connection bridge functionality. It is read-only.
Returns:
bool: True if the connection bridge is configured, False otherwise.
"""
return self._revpi_with_con_bridge return self._revpi_with_con_bridge
@property @property
def with_wlan(self) -> bool: def with_wifi(self) -> bool:
""" return self._cm_with_wifi
Checks if WLAN is available.
This property evaluates whether WLAN is enabled or available by checking
the presence or value of the internal attribute `_wlan_class_path`.
Returns:
bool: True if WLAN is available, False otherwise.
"""
return bool(self._wlan_class_path)
class ConfigTxt: class ConfigTxt:
"""
Configuration file handler for managing 'config.txt'.
This class provides an interface to read, modify, save, and reload
Raspbian's configuration file `config.txt`. It includes functionalities
to manipulate specific parameters within the configuration and supports
managing dtoverlay and dtparam entries. The primary aim of this class
is to abstract file operations and make modifications user-friendly.
Attributes:
_config_txt_path (str): The path to the configuration file `config.txt`.
_config_txt_lines (list[str]): Contains all lines of the configuration
file as a list of strings, where each string represents a line.
"""
re_name_value = re.compile(r"^\s*(?!#)(?P<name>[^=\s].+?)\s*=\s*(?P<value>\S+)\s*$") re_name_value = re.compile(r"^\s*(?!#)(?P<name>[^=\s].+?)\s*=\s*(?P<value>\S+)\s*$")
def __init__(self): def __init__(self):
@@ -246,24 +119,6 @@ class ConfigTxt:
self._config_txt_lines = [] self._config_txt_lines = []
def _clear_name_values(self, name: str, values: str or list) -> int: def _clear_name_values(self, name: str, values: str or list) -> int:
"""
Removes all occurrences of specified name-value pairs from the configuration.
This method searches for all name-value pairs in the configuration and
removes those that match the given name and value(s). It returns the
number of occurrences removed.
Arguments:
name: str
The name of the configuration variable to search for.
values: str or list
The value or list of values to match the configuration variable
against.
Returns:
int: The number of name-value pairs removed from the configuration.
"""
counter = 0 counter = 0
if type(values) is str: if type(values) is str:
values = [values] values = [values]
@@ -276,21 +131,6 @@ class ConfigTxt:
return counter return counter
def _get_all_name_values(self) -> List[ConfigVariable]: def _get_all_name_values(self) -> List[ConfigVariable]:
"""
Retrieves all name-value pairs from the configuration text lines.
This method parses the configuration text lines to extract all name-value
pairs. If the configuration text lines are not loaded, it reloads the
configuration before processing. Each extracted name-value pair is added to a
list as a ConfigVariable object, which also holds the index of the match in
the text lines. The method returns the compiled list of these ConfigVariable
objects.
Returns:
List[ConfigVariable]: A list of ConfigVariable objects representing the
name-value pairs found in the configuration text lines, along with their
corresponding indexes.
"""
if not self._config_txt_lines: if not self._config_txt_lines:
self.reload_config() self.reload_config()
@@ -304,28 +144,10 @@ class ConfigTxt:
return lst_return return lst_return
def reload_config(self): def reload_config(self):
"""
Reloads the configuration file and updates the list of configuration lines.
This method reads the content of the configuration file specified by the
attribute `_config_txt_path` and updates `_config_txt_lines` with the file
contents as a list of strings, where each string represents a line.
Returns:
None
"""
with open(self._config_txt_path, "r") as f: with open(self._config_txt_path, "r") as f:
self._config_txt_lines = f.readlines() self._config_txt_lines = f.readlines()
def save_config(self): def save_config(self):
"""
Saves the current configuration to a file. The method ensures atomicity by first writing
to a temporary file and then moving it to the desired path. After the configuration is
saved, the internal list of configuration lines is cleared.
Raises:
OSError: If there is an issue writing to or moving the file.
"""
if not self._config_txt_lines: if not self._config_txt_lines:
return return
@@ -337,20 +159,6 @@ class ConfigTxt:
self._config_txt_lines.clear() self._config_txt_lines.clear()
def add_name_value(self, name: str, value: str): def add_name_value(self, name: str, value: str):
"""
Adds a name-value pair to the configuration if it does not already exist.
This method checks if the given name-value pair is already present in
the configuration. If it is not present, the pair is appended to the
configuration text lines.
Parameters:
name (str): The name to be added to the configuration.
value (str): The value corresponding to the name to be added.
Returns:
None
"""
# Check weather name and value already exists # Check weather name and value already exists
for config_var in self._get_all_name_values(): for config_var in self._get_all_name_values():
if config_var.name == name and config_var.value == value: if config_var.name == name and config_var.value == value:
@@ -359,55 +167,12 @@ class ConfigTxt:
self._config_txt_lines.append(f"{name}={value}\n") self._config_txt_lines.append(f"{name}={value}\n")
def clear_dtoverlays(self, dtoverlays: str or list) -> int: def clear_dtoverlays(self, dtoverlays: str or list) -> int:
"""
Clears the specified device tree overlays. This method removes one or more
device tree overlays by clearing their corresponding name-value pairs.
Args:
dtoverlays (str or list): A device tree overlay name as a string, or a
list of such overlay names to be cleared.
Returns:
int: The number of device tree overlay name-value pairs successfully
cleared.
"""
return self._clear_name_values("dtoverlay", dtoverlays) return self._clear_name_values("dtoverlay", dtoverlays)
def clear_dtparams(self, dtparams: str or list) -> int: def clear_dtparams(self, dtparams: str or list) -> int:
"""
Clears the specified device tree parameters.
This method removes the given device tree parameters by utilizing
the underlying `_clear_name_values` function with a predefined
parameter type.
Parameters:
dtparams: str or list
A string or list of strings specifying the device tree
parameters to remove.
Returns:
int
The number of parameters cleared.
"""
return self._clear_name_values("dtparam", dtparams) return self._clear_name_values("dtparam", dtparams)
def get_values(self, var_name: str) -> list: def get_values(self, var_name: str) -> list:
"""
Get all values associated with a given variable name.
This method retrieves a list of values corresponding to the specified
variable name by iterating through a collection of configuration
variables. Each configuration variable is checked for a matching name,
and its value is appended to the resulting list if a match is found.
Parameters:
var_name (str): The name of the variable for which values are to
be retrieved.
Returns:
list: A list of values associated with the specified variable name.
"""
var_values = [] var_values = []
for config_var in self._get_all_name_values(): for config_var in self._get_all_name_values():
@@ -418,16 +183,6 @@ class ConfigTxt:
@property @property
def config_txt_path(self) -> str: def config_txt_path(self) -> str:
"""
Get the file path for the configuration text file.
This property provides access to the private attribute `_config_txt_path` which
stores the file path to the configuration text file.
Returns:
str
The file path to the configuration text file.
"""
return self._config_txt_path return self._config_txt_path
@@ -443,35 +198,39 @@ def configure_avahi_daemon(action: ConfigActions):
return return_value return return_value
def configure_bluetooth(action: ConfigActions): def configure_dphys_swapfile(action: ConfigActions):
hci_device = join(LINUX_BT_CLASS_PATH, "hci0") return_value = simple_systemd(action, "dphys-swapfile.service")
bt_rfkill_index = get_rfkill_index(hci_device)
# If the bluetooth device is not present, the device should have been # Post actions for dphys-swapfile
# brought up by revpi-bluetooth's udev rules or vendor magic (devices if action is ConfigActions.DISABLE:
# based on CM4 and newer). Nothing we can do here, so treat the interface # Remove swapfile afer disabling the service unit
# as disabled. subprocess.call(
["/sbin/dphys-swapfile", "uninstall"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
if action is ConfigActions.ENABLE: return return_value
if bt_rfkill_index is not None:
with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "w") as f:
f.write("0")
elif action is ConfigActions.DISABLE:
if bt_rfkill_index is not None: def configure_external_antenna(action: ConfigActions):
with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "w") as f: revpi = RevPiConfig()
f.write("1") if action is ConfigActions.AVAILABLE:
return revpi.with_wifi
config_txt = ConfigTxt()
if action is ConfigActions.ENABLE and revpi.with_wifi:
config_txt.clear_dtparams(["ant1", "ant2"])
config_txt.add_name_value("dtparam", "ant2")
config_txt.save_config()
elif action is ConfigActions.DISABLE and revpi.with_wifi:
config_txt.clear_dtparams(["ant1", "ant2"])
config_txt.save_config()
elif action is ConfigActions.STATUS: elif action is ConfigActions.STATUS:
if bt_rfkill_index is None: return revpi.with_wifi and "ant2" in config_txt.get_values("dtparam")
return False
with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "r") as f:
buffer = f.read().strip()
return buffer == "0"
elif action is ConfigActions.AVAILABLE:
return bt_rfkill_index is not None
else: else:
raise ValueError(f"action {action} not supported") raise ValueError(f"action {action} not supported")
@@ -479,6 +238,28 @@ def configure_bluetooth(action: ConfigActions):
return None return None
def configure_gui(action: ConfigActions):
gui_available = access("/usr/bin/startx", X_OK)
if action is ConfigActions.AVAILABLE:
return gui_available
bus = SystemBus()
systemd_manager = bus.get(".systemd1")
if action is ConfigActions.ENABLE:
systemd_manager.SetDefaultTarget("graphical.target", True)
elif action is ConfigActions.DISABLE:
systemd_manager.SetDefaultTarget("multi-user.target", True)
elif action is ConfigActions.STATUS:
return systemd_manager.GetDefaultTarget() == "graphical.target"
else:
raise ValueError(f"action {action} not supported")
def configure_con_can(action: ConfigActions): def configure_con_can(action: ConfigActions):
revpi = RevPiConfig() revpi = RevPiConfig()
if action is ConfigActions.AVAILABLE: if action is ConfigActions.AVAILABLE:
@@ -507,162 +288,9 @@ def configure_con_can(action: ConfigActions):
return None return None
def configure_dphys_swapfile(action: ConfigActions):
return_value = simple_systemd(action, "dphys-swapfile.service")
# Post actions for dphys-swapfile
if action is ConfigActions.DISABLE:
# Remove swapfile afer disabling the service unit
subprocess.call(
["/sbin/dphys-swapfile", "uninstall"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return return_value
def configure_external_antenna(action: ConfigActions):
revpi = RevPiConfig()
if action is ConfigActions.AVAILABLE:
return revpi.with_wlan
config_txt = ConfigTxt()
if action is ConfigActions.ENABLE and revpi.with_wlan:
config_txt.clear_dtparams(["ant1", "ant2"])
config_txt.add_name_value("dtparam", "ant2")
config_txt.save_config()
elif action is ConfigActions.DISABLE and revpi.with_wlan:
config_txt.clear_dtparams(["ant1", "ant2"])
config_txt.save_config()
elif action is ConfigActions.STATUS:
return revpi.with_wlan and "ant2" in config_txt.get_values("dtparam")
else:
raise ValueError(f"action {action} not supported")
return None
def configure_gui(action: ConfigActions):
gui_available = access("/usr/bin/startx", X_OK)
if action is ConfigActions.AVAILABLE:
return gui_available
bus = SystemBus()
systemd = bus.get(
"org.freedesktop.systemd1",
"/org/freedesktop/systemd1",
)
systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
if action is ConfigActions.ENABLE:
systemd_manager.SetDefaultTarget("graphical.target", True)
elif action is ConfigActions.DISABLE:
systemd_manager.SetDefaultTarget("multi-user.target", True)
elif action is ConfigActions.STATUS:
return systemd_manager.GetDefaultTarget() == "graphical.target"
else:
raise ValueError(f"action {action} not supported")
def configure_wlan(action: ConfigActions):
revpi = RevPiConfig()
if action is ConfigActions.ENABLE:
if revpi.with_wlan:
wlan_rfkill_index = get_rfkill_index(revpi.class_path_wlan)
with open(f"/sys/class/rfkill/rfkill{wlan_rfkill_index}/soft", "w") as f:
f.write("0")
elif action is ConfigActions.DISABLE:
if revpi.with_wlan:
wlan_rfkill_index = get_rfkill_index(revpi.class_path_wlan)
with open(f"/sys/class/rfkill/rfkill{wlan_rfkill_index}/soft", "w") as f:
f.write("1")
elif action is ConfigActions.AVAILABLE:
return revpi.with_wlan
elif action is ConfigActions.STATUS:
if not revpi.with_wlan:
return False
wlan_rfkill_index = get_rfkill_index(revpi.class_path_wlan)
with open(f"/sys/class/rfkill/rfkill{wlan_rfkill_index}/soft", "r") as f:
buffer = f.read().strip()
return buffer == "0"
else:
raise ValueError(f"action {action} not supported")
return None
def get_rfkill_index(device_class_path: str) -> Optional[int]:
"""
Get the rfkill index for a device under a specific device class path.
This function searches for and extracts the rfkill index associated with
devices located under the given device class path. It uses a regular
expression to identify and parse the rfkill index from the paths
of matching rfkill device files.
Parameters:
device_class_path: str
The path to the device class directory where rfkill entries
are located.
Returns:
Optional[int]:
The index of the rfkill device if found, otherwise None.
"""
re_rfkill_index = re.compile(r"^/.+/rfkill(?P<index>\d+)$")
for rfkill_path in glob(join(device_class_path, "rfkill*")):
match_index = re_rfkill_index.match(rfkill_path)
if match_index:
return int(match_index.group("index"))
return None
def simple_systemd(action: ConfigActions, unit: str): def simple_systemd(action: ConfigActions, unit: str):
"""
Performs specified actions on systemd units.
This function allows interaction with systemd units for various operations
such as enabling, disabling, checking the status, and verifying availability.
It communicates with the systemd manager via the SystemBus and handles units
based on the action specified.
Parameters:
action (ConfigActions): Specifies the action to be performed on the
systemd unit. Supported actions include ENABLE,
DISABLE, STATUS, and AVAILABLE.
unit (str): The name of the systemd unit on which the action is to be
performed.
Returns:
bool: For STATUS and AVAILABLE actions, returns True if the corresponding
criteria are met (e.g., enabled and active for STATUS, or not found
for AVAILABLE). Otherwise, returns False.
Raises:
ValueError: If the specified action is not supported.
"""
bus = SystemBus() bus = SystemBus()
systemd = bus.get( systemd_manager = bus.get(".systemd1")
"org.freedesktop.systemd1",
"/org/freedesktop/systemd1",
)
systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
if action is ConfigActions.ENABLE: if action is ConfigActions.ENABLE:
systemd_manager.UnmaskUnitFiles([unit], False) systemd_manager.UnmaskUnitFiles([unit], False)
@@ -676,7 +304,7 @@ def simple_systemd(action: ConfigActions, unit: str):
elif action is ConfigActions.STATUS: elif action is ConfigActions.STATUS:
try: try:
unit_path = systemd_manager.LoadUnit(unit) unit_path = systemd_manager.LoadUnit(unit)
properties = bus.get("org.freedesktop.systemd1", unit_path) properties = bus.get(".systemd1", unit_path)
except Exception: except Exception:
log.warning(f"could not get systemd unit {unit}") log.warning(f"could not get systemd unit {unit}")
return False return False
@@ -686,7 +314,7 @@ def simple_systemd(action: ConfigActions, unit: str):
elif action is ConfigActions.AVAILABLE: elif action is ConfigActions.AVAILABLE:
try: try:
unit_path = systemd_manager.LoadUnit(unit) unit_path = systemd_manager.LoadUnit(unit)
properties = bus.get("org.freedesktop.systemd1", unit_path) properties = bus.get(".systemd1", unit_path)
except Exception: except Exception:
log.warning(f"could not get systemd unit {unit}") log.warning(f"could not get systemd unit {unit}")
return False return False
@@ -702,10 +330,7 @@ if __name__ == "__main__":
print("Model:", rc.model) print("Model:", rc.model)
print("Serial: ", rc.serial) print("Serial: ", rc.serial)
print("CM Type: ", rc.cm_type.name) print("CM Type: ", rc.cm_type.name)
print("With WLAN: ", rc.with_wlan) print("With wifi: ", rc.with_wifi)
if rc.with_wlan:
print(" class path: ", rc.class_path_wlan)
print(" rfkill index: ", get_rfkill_index(rc.class_path_wlan))
print("With con-bridge:", rc.with_con_bridge) print("With con-bridge:", rc.with_con_bridge)
config_txt = ConfigTxt() config_txt = ConfigTxt()