|
|
|
|
@@ -1,16 +1,10 @@
|
|
|
|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
|
|
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
|
|
|
import re
|
|
|
|
|
import shutil
|
|
|
|
|
import subprocess
|
|
|
|
|
from collections import namedtuple
|
|
|
|
|
from enum import Enum, IntEnum
|
|
|
|
|
from glob import glob
|
|
|
|
|
from logging import getLogger
|
|
|
|
|
from os import X_OK, access
|
|
|
|
|
from os.path import exists, join
|
|
|
|
|
from typing import List, Optional
|
|
|
|
|
|
|
|
|
|
from pydbus import SystemBus
|
|
|
|
|
|
|
|
|
|
@@ -18,29 +12,8 @@ from ..dbus_helper import grep
|
|
|
|
|
|
|
|
|
|
log = getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
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")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
CM1 = 6
|
|
|
|
|
CM3 = 10
|
|
|
|
|
@@ -50,13 +23,6 @@ class ComputeModuleTypes(IntEnum):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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"
|
|
|
|
|
DISABLE = "disable"
|
|
|
|
|
STATUS = "status"
|
|
|
|
|
@@ -64,26 +30,12 @@ class ConfigActions(Enum):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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):
|
|
|
|
|
self._cm_type = ComputeModuleTypes.UNKNOWN
|
|
|
|
|
self._cm_with_wifi = False
|
|
|
|
|
|
|
|
|
|
self._revpi_with_con_bridge = False
|
|
|
|
|
self._wlan_class_path = ""
|
|
|
|
|
|
|
|
|
|
self.serial = ""
|
|
|
|
|
self.model = ""
|
|
|
|
|
@@ -91,29 +43,6 @@ class RevPiConfig:
|
|
|
|
|
self._init_device_info()
|
|
|
|
|
|
|
|
|
|
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 = {}
|
|
|
|
|
|
|
|
|
|
# Extract CPU information
|
|
|
|
|
@@ -140,19 +69,11 @@ class RevPiConfig:
|
|
|
|
|
except ValueError:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
# Detect WLAN on CM module
|
|
|
|
|
could_have_wlan = self._cm_type in (ComputeModuleTypes.CM4, ComputeModuleTypes.CM5)
|
|
|
|
|
if could_have_wlan:
|
|
|
|
|
wlan_interface = join(LINUX_WLAN_CLASS_PATH, "phy0")
|
|
|
|
|
if grep("DRIVER=brcmfmac", join(wlan_interface, "device", "uevent")):
|
|
|
|
|
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 WiFi
|
|
|
|
|
could_have_wifi = self._cm_type in (ComputeModuleTypes.CM4, ComputeModuleTypes.CM5)
|
|
|
|
|
if could_have_wifi:
|
|
|
|
|
lst_grep = grep("DRIVER=brcmfmac", "/sys/class/ieee80211/phy0/device/uevent")
|
|
|
|
|
self._cm_with_wifi = len(lst_grep) > 0 and self._cm_type in (ComputeModuleTypes)
|
|
|
|
|
|
|
|
|
|
# Detect ConBridge
|
|
|
|
|
could_have_con_bridge = self._cm_type in (ComputeModuleTypes.CM3, ComputeModuleTypes.CM4S)
|
|
|
|
|
@@ -160,275 +81,17 @@ class RevPiConfig:
|
|
|
|
|
lst_grep = grep("kunbus,revpi-connect", "/proc/device-tree/compatible")
|
|
|
|
|
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
|
|
|
|
|
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
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
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
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def with_wlan(self) -> bool:
|
|
|
|
|
"""
|
|
|
|
|
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:
|
|
|
|
|
"""
|
|
|
|
|
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*$")
|
|
|
|
|
|
|
|
|
|
def __init__(self):
|
|
|
|
|
self._config_txt_path = ""
|
|
|
|
|
for path in CONFIG_TXT_LOCATIONS:
|
|
|
|
|
if exists(path):
|
|
|
|
|
self._config_txt_path = path
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
if not self._config_txt_path:
|
|
|
|
|
raise FileNotFoundError("no config.txt found")
|
|
|
|
|
|
|
|
|
|
self._config_txt_lines = []
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
if type(values) is str:
|
|
|
|
|
values = [values]
|
|
|
|
|
|
|
|
|
|
for config_var in self._get_all_name_values():
|
|
|
|
|
if config_var.name == name and config_var.value in values:
|
|
|
|
|
self._config_txt_lines.pop(config_var.line_index)
|
|
|
|
|
counter += 1
|
|
|
|
|
|
|
|
|
|
return counter
|
|
|
|
|
|
|
|
|
|
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:
|
|
|
|
|
self.reload_config()
|
|
|
|
|
|
|
|
|
|
lst_return = []
|
|
|
|
|
|
|
|
|
|
for i in range(len(self._config_txt_lines)):
|
|
|
|
|
match = self.re_name_value.match(self._config_txt_lines[i])
|
|
|
|
|
if match:
|
|
|
|
|
lst_return.append(ConfigVariable(match.group("name"), match.group("value"), i))
|
|
|
|
|
|
|
|
|
|
return lst_return
|
|
|
|
|
|
|
|
|
|
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:
|
|
|
|
|
self._config_txt_lines = f.readlines()
|
|
|
|
|
|
|
|
|
|
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:
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
tmp_path = f"{self._config_txt_path}.tmp"
|
|
|
|
|
with open(tmp_path, "w") as f:
|
|
|
|
|
f.writelines(self._config_txt_lines)
|
|
|
|
|
shutil.move(tmp_path, self._config_txt_path)
|
|
|
|
|
|
|
|
|
|
self._config_txt_lines.clear()
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
for config_var in self._get_all_name_values():
|
|
|
|
|
if config_var.name == name and config_var.value == value:
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
self._config_txt_lines.append(f"{name}={value}\n")
|
|
|
|
|
|
|
|
|
|
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)
|
|
|
|
|
|
|
|
|
|
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)
|
|
|
|
|
|
|
|
|
|
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 = []
|
|
|
|
|
|
|
|
|
|
for config_var in self._get_all_name_values():
|
|
|
|
|
if config_var.name == var_name:
|
|
|
|
|
var_values.append(config_var.value)
|
|
|
|
|
|
|
|
|
|
return var_values
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
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
|
|
|
|
|
def with_wifi(self) -> bool:
|
|
|
|
|
return self._cm_with_wifi
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def configure_avahi_daemon(action: ConfigActions):
|
|
|
|
|
@@ -443,70 +106,6 @@ def configure_avahi_daemon(action: ConfigActions):
|
|
|
|
|
return return_value
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def configure_bluetooth(action: ConfigActions):
|
|
|
|
|
hci_device = join(LINUX_BT_CLASS_PATH, "hci0")
|
|
|
|
|
bt_rfkill_index = get_rfkill_index(hci_device)
|
|
|
|
|
|
|
|
|
|
# If the bluetooth device is not present, the device should have been
|
|
|
|
|
# brought up by revpi-bluetooth's udev rules or vendor magic (devices
|
|
|
|
|
# based on CM4 and newer). Nothing we can do here, so treat the interface
|
|
|
|
|
# as disabled.
|
|
|
|
|
|
|
|
|
|
if action is ConfigActions.ENABLE:
|
|
|
|
|
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:
|
|
|
|
|
with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "w") as f:
|
|
|
|
|
f.write("1")
|
|
|
|
|
|
|
|
|
|
elif action is ConfigActions.STATUS:
|
|
|
|
|
if bt_rfkill_index is None:
|
|
|
|
|
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:
|
|
|
|
|
raise ValueError(f"action {action} not supported")
|
|
|
|
|
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def configure_con_can(action: ConfigActions):
|
|
|
|
|
revpi = RevPiConfig()
|
|
|
|
|
if action is ConfigActions.AVAILABLE:
|
|
|
|
|
return revpi.with_con_bridge
|
|
|
|
|
|
|
|
|
|
dt_overlay = "revpi-con-can"
|
|
|
|
|
config_txt = ConfigTxt()
|
|
|
|
|
|
|
|
|
|
if action is ConfigActions.ENABLE and revpi.with_con_bridge:
|
|
|
|
|
config_txt.clear_dtoverlays([dt_overlay])
|
|
|
|
|
config_txt.add_name_value("dtoverlay", dt_overlay)
|
|
|
|
|
config_txt.save_config()
|
|
|
|
|
subprocess.call(["/usr/bin/dtoverlay", dt_overlay])
|
|
|
|
|
|
|
|
|
|
elif action is ConfigActions.DISABLE and revpi.with_con_bridge:
|
|
|
|
|
config_txt.clear_dtoverlays([dt_overlay])
|
|
|
|
|
config_txt.save_config()
|
|
|
|
|
subprocess.call(["/usr/bin/dtoverlay", "-r", dt_overlay])
|
|
|
|
|
|
|
|
|
|
elif action is ConfigActions.STATUS:
|
|
|
|
|
return revpi.with_con_bridge and dt_overlay in config_txt.get_values("dtparam")
|
|
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
raise ValueError(f"action {action} not supported")
|
|
|
|
|
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def configure_dphys_swapfile(action: ConfigActions):
|
|
|
|
|
return_value = simple_systemd(action, "dphys-swapfile.service")
|
|
|
|
|
|
|
|
|
|
@@ -522,31 +121,6 @@ def configure_dphys_swapfile(action: ConfigActions):
|
|
|
|
|
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)
|
|
|
|
|
|
|
|
|
|
@@ -554,11 +128,7 @@ def configure_gui(action: ConfigActions):
|
|
|
|
|
return gui_available
|
|
|
|
|
|
|
|
|
|
bus = SystemBus()
|
|
|
|
|
systemd = bus.get(
|
|
|
|
|
"org.freedesktop.systemd1",
|
|
|
|
|
"/org/freedesktop/systemd1",
|
|
|
|
|
)
|
|
|
|
|
systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
|
|
|
|
|
systemd_manager = bus.get(".systemd1")
|
|
|
|
|
|
|
|
|
|
if action is ConfigActions.ENABLE:
|
|
|
|
|
systemd_manager.SetDefaultTarget("graphical.target", True)
|
|
|
|
|
@@ -573,96 +143,9 @@ def configure_gui(action: ConfigActions):
|
|
|
|
|
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):
|
|
|
|
|
"""
|
|
|
|
|
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()
|
|
|
|
|
systemd = bus.get(
|
|
|
|
|
"org.freedesktop.systemd1",
|
|
|
|
|
"/org/freedesktop/systemd1",
|
|
|
|
|
)
|
|
|
|
|
systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
|
|
|
|
|
systemd_manager = bus.get(".systemd1")
|
|
|
|
|
|
|
|
|
|
if action is ConfigActions.ENABLE:
|
|
|
|
|
systemd_manager.UnmaskUnitFiles([unit], False)
|
|
|
|
|
@@ -676,7 +159,7 @@ def simple_systemd(action: ConfigActions, unit: str):
|
|
|
|
|
elif action is ConfigActions.STATUS:
|
|
|
|
|
try:
|
|
|
|
|
unit_path = systemd_manager.LoadUnit(unit)
|
|
|
|
|
properties = bus.get("org.freedesktop.systemd1", unit_path)
|
|
|
|
|
properties = bus.get(".systemd1", unit_path)
|
|
|
|
|
except Exception:
|
|
|
|
|
log.warning(f"could not get systemd unit {unit}")
|
|
|
|
|
return False
|
|
|
|
|
@@ -686,7 +169,7 @@ def simple_systemd(action: ConfigActions, unit: str):
|
|
|
|
|
elif action is ConfigActions.AVAILABLE:
|
|
|
|
|
try:
|
|
|
|
|
unit_path = systemd_manager.LoadUnit(unit)
|
|
|
|
|
properties = bus.get("org.freedesktop.systemd1", unit_path)
|
|
|
|
|
properties = bus.get(".systemd1", unit_path)
|
|
|
|
|
except Exception:
|
|
|
|
|
log.warning(f"could not get systemd unit {unit}")
|
|
|
|
|
return False
|
|
|
|
|
@@ -702,11 +185,5 @@ if __name__ == "__main__":
|
|
|
|
|
print("Model:", rc.model)
|
|
|
|
|
print("Serial: ", rc.serial)
|
|
|
|
|
print("CM Type: ", rc.cm_type.name)
|
|
|
|
|
print("With WLAN: ", rc.with_wlan)
|
|
|
|
|
if rc.with_wlan:
|
|
|
|
|
print(" class path: ", rc.class_path_wlan)
|
|
|
|
|
print(" rfkill index: ", get_rfkill_index(rc.class_path_wlan))
|
|
|
|
|
print("With wifi: ", rc.with_wifi)
|
|
|
|
|
print("With con-bridge:", rc.with_con_bridge)
|
|
|
|
|
|
|
|
|
|
config_txt = ConfigTxt()
|
|
|
|
|
print("Config file: ", config_txt.config_txt_path)
|
|
|
|
|
|