Compare commits
18 Commits
52802ae6c5
...
v0.0.3
| Author | SHA1 | Date | |
|---|---|---|---|
| 1fab228272 | |||
| fc82ec0eb9 | |||
| de1774f60e | |||
| 8c145ef2ff | |||
| 0ecd86bd64 | |||
| 555c781aed | |||
| 604cb61870 | |||
| 69e370f964 | |||
| 6eb7eeea40 | |||
| 18e6fb3d14 | |||
| fe614d026a | |||
| 870a55042e | |||
| 2848fdcf54 | |||
| d0f641f2b5 | |||
| e8eba43647 | |||
| 0d601f623c | |||
| 66735a9eba | |||
| 7525c9f653 |
@@ -29,6 +29,11 @@ def setup_command_line_arguments():
|
||||
help="RevPi PiControl object",
|
||||
)
|
||||
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:
|
||||
@@ -39,6 +44,9 @@ def main() -> int:
|
||||
if obj == "picontrol":
|
||||
rc = cli_picontrol.main()
|
||||
|
||||
elif obj == "config":
|
||||
rc = cli_config.main()
|
||||
|
||||
else:
|
||||
log.error(f"Unknown object: {obj}")
|
||||
rc = 1
|
||||
|
||||
93
src/revpi_middleware/cli_commands/cli_config.py
Normal file
93
src/revpi_middleware/cli_commands/cli_config.py
Normal file
@@ -0,0 +1,93 @@
|
||||
# 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
|
||||
@@ -17,6 +17,18 @@ class BusType(Enum):
|
||||
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(
|
||||
method: str,
|
||||
*args,
|
||||
|
||||
@@ -10,6 +10,7 @@ from pydbus import SessionBus, SystemBus
|
||||
|
||||
from . import REVPI_DBUS_NAME
|
||||
from .process_image import InterfacePiControl
|
||||
from .system_config import InterfaceRevpiConfig
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
@@ -41,6 +42,7 @@ class BusProvider(Thread):
|
||||
# ("Subdir2/Whatever", Example())
|
||||
lst_interfaces = [
|
||||
InterfacePiControl(self.picontrol_device, self.config_rsc),
|
||||
InterfaceRevpiConfig(),
|
||||
]
|
||||
|
||||
try:
|
||||
|
||||
@@ -8,10 +8,12 @@ from logging import getLogger
|
||||
from .revpi_config import (
|
||||
ConfigActions,
|
||||
configure_avahi_daemon,
|
||||
configure_bluetooth,
|
||||
configure_con_can,
|
||||
configure_dphys_swapfile,
|
||||
configure_external_antenna,
|
||||
configure_gui,
|
||||
configure_wlan,
|
||||
simple_systemd,
|
||||
)
|
||||
from ..dbus_helper import DbusInterface
|
||||
@@ -91,8 +93,8 @@ AVAILABLE_FEATURES = {
|
||||
simple_systemd, ["noderedrevpinodes-server.service"]
|
||||
),
|
||||
"revpipyload": FeatureFunction(simple_systemd, ["revpipyload.service"]),
|
||||
"bluetooth": False,
|
||||
"ieee80211": False,
|
||||
"bluetooth": FeatureFunction(configure_bluetooth, []),
|
||||
"wlan": FeatureFunction(configure_wlan, []),
|
||||
"avahi": FeatureFunction(configure_avahi_daemon, []),
|
||||
"external-antenna": FeatureFunction(configure_external_antenna, []),
|
||||
}
|
||||
|
||||
@@ -6,10 +6,11 @@ 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
|
||||
from typing import List
|
||||
from os.path import exists, join
|
||||
from typing import List, Optional
|
||||
|
||||
from pydbus import SystemBus
|
||||
|
||||
@@ -19,6 +20,10 @@ 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):
|
||||
UNKNOWN = 0
|
||||
@@ -40,9 +45,9 @@ class RevPiConfig:
|
||||
|
||||
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 = ""
|
||||
@@ -76,11 +81,19 @@ class RevPiConfig:
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# 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 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 ConBridge
|
||||
could_have_con_bridge = self._cm_type in (ComputeModuleTypes.CM3, ComputeModuleTypes.CM4S)
|
||||
@@ -88,6 +101,10 @@ 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:
|
||||
return self._wlan_class_path
|
||||
|
||||
@property
|
||||
def cm_type(self) -> ComputeModuleTypes:
|
||||
return self._cm_type
|
||||
@@ -97,8 +114,8 @@ class RevPiConfig:
|
||||
return self._revpi_with_con_bridge
|
||||
|
||||
@property
|
||||
def with_wifi(self) -> bool:
|
||||
return self._cm_with_wifi
|
||||
def with_wlan(self) -> bool:
|
||||
return bool(self._wlan_class_path)
|
||||
|
||||
|
||||
class ConfigTxt:
|
||||
@@ -106,7 +123,7 @@ class ConfigTxt:
|
||||
|
||||
def __init__(self):
|
||||
self._config_txt_path = ""
|
||||
for path in ("/boot/firmware/config.txt", "/boot/config.txt"):
|
||||
for path in CONFIG_TXT_LOCATIONS:
|
||||
if exists(path):
|
||||
self._config_txt_path = path
|
||||
break
|
||||
@@ -196,6 +213,70 @@ 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")
|
||||
|
||||
@@ -214,21 +295,21 @@ def configure_dphys_swapfile(action: ConfigActions):
|
||||
def configure_external_antenna(action: ConfigActions):
|
||||
revpi = RevPiConfig()
|
||||
if action is ConfigActions.AVAILABLE:
|
||||
return revpi.with_wifi
|
||||
return revpi.with_wlan
|
||||
|
||||
config_txt = ConfigTxt()
|
||||
|
||||
if action is ConfigActions.ENABLE and revpi.with_wifi:
|
||||
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_wifi:
|
||||
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_wifi and "ant2" in config_txt.get_values("dtparam")
|
||||
return revpi.with_wlan and "ant2" in config_txt.get_values("dtparam")
|
||||
|
||||
else:
|
||||
raise ValueError(f"action {action} not supported")
|
||||
@@ -258,27 +339,32 @@ def configure_gui(action: ConfigActions):
|
||||
raise ValueError(f"action {action} not supported")
|
||||
|
||||
|
||||
def configure_con_can(action: ConfigActions):
|
||||
def configure_wlan(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:
|
||||
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")
|
||||
|
||||
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:
|
||||
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.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.AVAILABLE:
|
||||
return revpi.with_wlan
|
||||
|
||||
elif action is ConfigActions.STATUS:
|
||||
return revpi.with_con_bridge and dt_overlay in config_txt.get_values("dtparam")
|
||||
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")
|
||||
@@ -286,6 +372,16 @@ def configure_con_can(action: ConfigActions):
|
||||
return None
|
||||
|
||||
|
||||
def get_rfkill_index(device_class_path: str) -> Optional[int]:
|
||||
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):
|
||||
bus = SystemBus()
|
||||
systemd_manager = bus.get(".systemd1")
|
||||
@@ -328,7 +424,10 @@ if __name__ == "__main__":
|
||||
print("Model:", rc.model)
|
||||
print("Serial: ", rc.serial)
|
||||
print("CM Type: ", rc.cm_type.name)
|
||||
print("With wifi: ", rc.with_wifi)
|
||||
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 con-bridge:", rc.with_con_bridge)
|
||||
|
||||
config_txt = ConfigTxt()
|
||||
|
||||
Reference in New Issue
Block a user