Compare commits
2 Commits
v0.0.3
...
debian/0.0
| Author | SHA1 | Date | |
|---|---|---|---|
| f23a91bd4f | |||
| 28d4ce0116 |
27
debian/changelog
vendored
Normal file
27
debian/changelog
vendored
Normal file
@@ -0,0 +1,27 @@
|
||||
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
|
||||
36
debian/control
vendored
Normal file
36
debian/control
vendored
Normal file
@@ -0,0 +1,36 @@
|
||||
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:
|
||||
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:
|
||||
${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
Normal file
27
debian/copyright
vendored
Normal file
@@ -0,0 +1,27 @@
|
||||
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
Normal file
7
debian/gbp.conf
vendored
Normal file
@@ -0,0 +1,7 @@
|
||||
[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
|
||||
4
debian/revpi-middleware.install
vendored
Normal file
4
debian/revpi-middleware.install
vendored
Normal file
@@ -0,0 +1,4 @@
|
||||
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/
|
||||
2
debian/revpi-middleware.links
vendored
Normal file
2
debian/revpi-middleware.links
vendored
Normal file
@@ -0,0 +1,2 @@
|
||||
/usr/share/revpi-middleware/revpi-middleware /usr/sbin/revpi-middleware
|
||||
/usr/share/revpi-middleware/revpicli /usr/bin/revpicli
|
||||
7
debian/rules
vendored
Executable file
7
debian/rules
vendored
Executable file
@@ -0,0 +1,7 @@
|
||||
#!/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
|
||||
1
debian/source/format
vendored
Normal file
1
debian/source/format
vendored
Normal file
@@ -0,0 +1 @@
|
||||
3.0 (quilt)
|
||||
@@ -5,11 +5,12 @@
|
||||
This module provides the foundation for the RevPi middleware CLI commands
|
||||
and argument parsing setup.
|
||||
"""
|
||||
from argparse import ArgumentParser
|
||||
from logging import getLogger
|
||||
|
||||
from . import cli_config, cli_picontrol
|
||||
from revpi_middleware.cli_commands import cli_picontrol
|
||||
from revpi_middleware.proginit import StdLogOutput
|
||||
from .. import proginit as pi
|
||||
from ..proginit import StdLogOutput
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
@@ -29,11 +30,6 @@ 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:
|
||||
@@ -44,9 +40,6 @@ 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
|
||||
|
||||
@@ -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
|
||||
@@ -4,7 +4,7 @@
|
||||
from argparse import ArgumentParser
|
||||
from logging import getLogger
|
||||
|
||||
from .dbus_helper import BusType, await_signal, simple_call
|
||||
from .dbus_helper import await_signal, simple_call
|
||||
from .. import proginit as pi
|
||||
from ..dbus_middleware1 import extend_interface
|
||||
|
||||
@@ -36,21 +36,12 @@ def add_subparsers(parent_parser: ArgumentParser):
|
||||
|
||||
def method_reset():
|
||||
log.debug("D-Bus call of method ResetDriver")
|
||||
simple_call(
|
||||
"ResetDriver",
|
||||
interface=extend_interface("PiControl"),
|
||||
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
|
||||
)
|
||||
simple_call("ResetDriver", interface=extend_interface("picontrol"))
|
||||
log.info("ResetDriver called via D-Bus")
|
||||
|
||||
|
||||
def method_await_reset(timeout: int = 0):
|
||||
detected_signal = await_signal(
|
||||
"NotifyDriverReset",
|
||||
timeout,
|
||||
extend_interface("PiControl"),
|
||||
bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
|
||||
)
|
||||
def method_await_reset(timout: int = 0):
|
||||
detected_signal = await_signal("NotifyDriverReset", timout, extend_interface("picontrol"))
|
||||
if detected_signal:
|
||||
log.info("ResetDriver signal received")
|
||||
else:
|
||||
|
||||
@@ -1,113 +1,50 @@
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""D-Bus helper functions for cli commands."""
|
||||
from enum import Enum
|
||||
from threading import Thread
|
||||
from time import sleep
|
||||
|
||||
from gi.repository import GLib
|
||||
from pydbus import SessionBus, SystemBus
|
||||
from pydbus import SystemBus
|
||||
|
||||
from ..dbus_middleware1 import REVPI_DBUS_BASE_PATH
|
||||
from ..dbus_middleware1 import REVPI_DBUS_NAME
|
||||
|
||||
|
||||
class BusType(Enum):
|
||||
SESSION = "session"
|
||||
SYSTEM = "system"
|
||||
PICONTROL_INTERFACE = "com.revolutionpi.middleware1.picontrol"
|
||||
RESET_DRIVER_METHOD = "ResetDriver"
|
||||
|
||||
|
||||
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,
|
||||
interface: str,
|
||||
object_path=REVPI_DBUS_BASE_PATH,
|
||||
bus_type=BusType.SYSTEM,
|
||||
):
|
||||
def simple_call(method: str, *args, interface: str, object_path=REVPI_DBUS_BASE_PATH):
|
||||
"""
|
||||
Performs a call to a D-Bus method on a specified interface and object.
|
||||
|
||||
This function uses the D-Bus messaging system to dynamically call a method
|
||||
of a specified interface, using the given object path and bus type. It
|
||||
provides a way to interact with D-Bus interfaces, using either a system or
|
||||
session bus, and returns the result of executing the specified method.
|
||||
Executes a method on a specific D-Bus object interface within the RevPi system. This function
|
||||
connects to the system bus, retrieves the desired interface and object path, and invokes
|
||||
the specified method with provided arguments.
|
||||
|
||||
Parameters:
|
||||
method: str
|
||||
The name of the method to invoke on the D-Bus interface.
|
||||
*args:
|
||||
Additional positional arguments to pass to the specified D-Bus method.
|
||||
The name of the method to be invoked on the targeted interface.
|
||||
*args: tuple
|
||||
Positional arguments to be passed to the method being invoked.
|
||||
interface: str
|
||||
The name of the D-Bus interface containing the method.
|
||||
object_path:
|
||||
The path of the D-Bus object on which the interface is defined. Defaults
|
||||
to REVPI_DBUS_BASE_PATH.
|
||||
bus_type: BusType
|
||||
Specifies whether to use the system or session bus. Defaults to BusType.SYSTEM.
|
||||
The name of the D-Bus interface providing the required functionality.
|
||||
object_path: str, optional
|
||||
The D-Bus object path of the RevPi interface. Defaults to REVPI_DBUS_BASE_PATH.
|
||||
|
||||
Returns:
|
||||
The value returned by the D-Bus method.
|
||||
|
||||
Raises:
|
||||
Any errors raised from the D-Bus call will propagate to the caller.
|
||||
Any
|
||||
The result of the method invocation on the targeted D-Bus interface.
|
||||
"""
|
||||
bus = SessionBus() if bus_type is BusType.SESSION else SystemBus()
|
||||
bus = SystemBus()
|
||||
revpi = bus.get(REVPI_DBUS_NAME, object_path)
|
||||
iface = revpi[interface]
|
||||
return getattr(iface, method)(*args)
|
||||
|
||||
|
||||
def await_signal(
|
||||
signal_name: str,
|
||||
timeout: int,
|
||||
interface: str,
|
||||
object_path=REVPI_DBUS_BASE_PATH,
|
||||
bus_type=BusType.SYSTEM,
|
||||
):
|
||||
"""
|
||||
Waits for a specific signal and returns whether the signal was detected.
|
||||
|
||||
This function connects to a D-Bus interface and waits for a specific signal
|
||||
to be emitted. If the signal is not received within the specified timeout
|
||||
period, the function will return False. If the signal is detected within
|
||||
the timeout, the function will return True. It can connect to either the
|
||||
system bus or the session bus, depending on the provided `bus_type`.
|
||||
|
||||
Parameters:
|
||||
signal_name: str
|
||||
The name of the signal to be awaited.
|
||||
timeout: int
|
||||
The maximum time to wait for the signal, in seconds. A value of 0 or
|
||||
less means that there is no timeout.
|
||||
interface: str
|
||||
The name of the D-Bus interface to listen on.
|
||||
object_path
|
||||
The D-Bus object path where the interface resides. Defaults to
|
||||
REVPI_DBUS_BASE_PATH.
|
||||
bus_type
|
||||
The type of D-Bus to connect to. Can be either BusType.SYSTEM or
|
||||
BusType.SESSION. Defaults to BusType.SYSTEM.
|
||||
|
||||
Returns:
|
||||
bool
|
||||
True if the signal was detected within the timeout period, False
|
||||
otherwise.
|
||||
"""
|
||||
def await_signal(signal_name: str, timeout: int, interface: str, object_path=REVPI_DBUS_BASE_PATH):
|
||||
detected_signal = False
|
||||
timeout = int(timeout)
|
||||
loop = GLib.MainLoop()
|
||||
th_sleep = Thread()
|
||||
|
||||
def th_timeout():
|
||||
sleep(timeout)
|
||||
@@ -118,7 +55,7 @@ def await_signal(
|
||||
detected_signal = True
|
||||
loop.quit()
|
||||
|
||||
bus = SessionBus() if bus_type is BusType.SESSION else SystemBus()
|
||||
bus = SystemBus()
|
||||
revpi = bus.get(REVPI_DBUS_NAME, object_path)
|
||||
iface = revpi[interface]
|
||||
|
||||
|
||||
@@ -42,7 +42,7 @@ class MiddlewareDaemon:
|
||||
if self.bus_provider and self.bus_provider.is_alive():
|
||||
return
|
||||
|
||||
self.bus_provider = BusProvider(use_system_bus=not pi.pargs.use_session_bus)
|
||||
self.bus_provider = BusProvider()
|
||||
self.bus_provider.start()
|
||||
|
||||
log.debug("leave MiddlewareDaemon.dbus_start")
|
||||
|
||||
@@ -2,7 +2,27 @@
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""D-Bus middleware version 1 of revpi_middleware."""
|
||||
from .dbus_helper import REVPI_DBUS_BASE_PATH, REVPI_DBUS_NAME
|
||||
from .dbus_helper import extend_interface
|
||||
from ..__about__ import __author__, __copyright__, __license__, __version__
|
||||
|
||||
from .bus_provider import BusProvider
|
||||
REVPI_DBUS_NAME = "com.revolutionpi.middleware1"
|
||||
REVPI_DBUS_BASE_PATH = "/com/revolutionpi/middleware1"
|
||||
|
||||
|
||||
def extend_interface(*args) -> str:
|
||||
"""
|
||||
Extends an interface name by appending additional segments to a pre-defined base name.
|
||||
|
||||
This function takes multiple arguments, concatenates them with a predefined base
|
||||
interface name, and returns the resulting string, effectively constructing an
|
||||
extended interface name.
|
||||
|
||||
Args:
|
||||
*args: str
|
||||
Components to be appended to the base interface name.
|
||||
|
||||
Returns:
|
||||
str
|
||||
Fully constructed interface name by joining the base interface name with
|
||||
the provided segments.
|
||||
"""
|
||||
return ".".join([REVPI_DBUS_NAME, *args])
|
||||
|
||||
@@ -6,71 +6,35 @@ from logging import getLogger
|
||||
from threading import Thread
|
||||
|
||||
from gi.repository import GLib
|
||||
from pydbus import SessionBus, SystemBus
|
||||
from pydbus import SystemBus
|
||||
|
||||
from . import REVPI_DBUS_NAME
|
||||
from .process_image import InterfacePiControl
|
||||
from .system_config import InterfaceRevpiConfig
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
|
||||
class BusProvider(Thread):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
picontrol_device="/dev/piControl0",
|
||||
config_rsc="/etc/revpi/config.rsc",
|
||||
use_system_bus=True,
|
||||
):
|
||||
def __init__(self):
|
||||
log.debug("enter BusProvider.__init__")
|
||||
super().__init__()
|
||||
|
||||
self._bus = SystemBus() if use_system_bus else SessionBus()
|
||||
self._bus = SystemBus()
|
||||
self._loop = GLib.MainLoop()
|
||||
|
||||
self.picontrol_device = picontrol_device
|
||||
self.config_rsc = config_rsc
|
||||
|
||||
def run(self):
|
||||
log.debug("enter BusProvider.run")
|
||||
|
||||
# The 2nd, 3rd, ... arguments can be objects or tuples of a path and an object
|
||||
# Example(),
|
||||
# ("Subdir1", Example()),
|
||||
# ("Subdir2", Example()),
|
||||
# ("Subdir2/Whatever", Example())
|
||||
lst_interfaces = [
|
||||
InterfacePiControl(self.picontrol_device, self.config_rsc),
|
||||
InterfaceRevpiConfig(),
|
||||
]
|
||||
|
||||
try:
|
||||
self._bus.publish(
|
||||
REVPI_DBUS_NAME,
|
||||
*lst_interfaces,
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(f"can not publish dbus {REVPI_DBUS_NAME}: {e}")
|
||||
|
||||
try:
|
||||
self._loop.run()
|
||||
except Exception as e:
|
||||
log.error(f"can not run dbus mainloop: {e}")
|
||||
|
||||
# Clean up all interfaces
|
||||
for interface in lst_interfaces:
|
||||
if type(interface) is tuple:
|
||||
_, interface = interface
|
||||
interface.cleanup()
|
||||
self._bus.publish(
|
||||
REVPI_DBUS_NAME,
|
||||
InterfacePiControl(),
|
||||
)
|
||||
|
||||
self._loop.run()
|
||||
log.debug("leave BusProvider.run")
|
||||
|
||||
def stop(self):
|
||||
log.debug("enter BusProvider.stop")
|
||||
self._loop.quit()
|
||||
log.debug("leave BusProvider.stop")
|
||||
|
||||
@property
|
||||
def running(self):
|
||||
return self._loop.is_running()
|
||||
|
||||
@@ -1,78 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""Helper for dbus."""
|
||||
|
||||
from logging import getLogger
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
REVPI_DBUS_NAME = "com.revolutionpi.middleware1"
|
||||
REVPI_DBUS_BASE_PATH = "/com/revolutionpi/middleware1"
|
||||
|
||||
|
||||
class DbusInterface:
|
||||
|
||||
def cleanup(self):
|
||||
"""
|
||||
Represents a method responsible for performing cleanup operations. This method is executed to properly
|
||||
release resources, close connections, or perform other necessary finalization tasks.
|
||||
|
||||
This method does not take any arguments or return a value.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
def extend_interface(*args) -> str:
|
||||
"""
|
||||
Extends an interface name by appending additional segments to a pre-defined base name.
|
||||
|
||||
This function takes multiple arguments, concatenates them with a predefined base
|
||||
interface name, and returns the resulting string, effectively constructing an
|
||||
extended interface name.
|
||||
|
||||
Args:
|
||||
*args: str
|
||||
Components to be appended to the base interface name.
|
||||
|
||||
Returns:
|
||||
str
|
||||
Fully constructed interface name by joining the base interface name with
|
||||
the provided segments.
|
||||
"""
|
||||
return ".".join([REVPI_DBUS_NAME, *args])
|
||||
|
||||
|
||||
def grep(pattern, filename):
|
||||
"""
|
||||
Searches for lines in a file that contain a given pattern and returns them as a list.
|
||||
|
||||
The function reads lines from the specified file and checks whether each line
|
||||
contains the provided pattern. It returns a list of lines that match the
|
||||
pattern. If the file is not found, an empty list is returned. Any other
|
||||
exceptions during the file reading process are caught and logged.
|
||||
|
||||
Args:
|
||||
pattern (str): The substring to search for within the file's lines.
|
||||
filename (str): The path to the file that will be searched.
|
||||
|
||||
Returns:
|
||||
list[str]: A list containing lines that include the provided pattern, with
|
||||
leading and trailing spaces removed.
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: This error is caught if the file specified is not
|
||||
found.
|
||||
Exception: Any unforeseen exception during file operations is caught and
|
||||
logged.
|
||||
"""
|
||||
try:
|
||||
with open(filename, "r") as file:
|
||||
# Gibt alle Zeilen zurück, die das Muster enthalten
|
||||
matching_lines = [line.strip() for line in file if pattern in line]
|
||||
return matching_lines
|
||||
except FileNotFoundError:
|
||||
return []
|
||||
except Exception as e:
|
||||
log.error(f"Error reading file: {e}")
|
||||
return []
|
||||
@@ -1,5 +1,5 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-FileCopyrightText: 2020-2023 Sven Sager
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""
|
||||
Helper for the process image.
|
||||
@@ -7,8 +7,8 @@ Helper for the process image.
|
||||
The ResetDriverWatchdog class is a copy of revpipyload project module "watchdogs"
|
||||
https://github.com/naruxde/revpipyload/blob/b51c2b617a57cc7d96fd67e1da9f090a0624eacb/src/revpipyload/watchdogs.py
|
||||
"""
|
||||
|
||||
import os
|
||||
from ctypes import c_int
|
||||
from fcntl import ioctl
|
||||
from logging import getLogger
|
||||
from threading import Thread
|
||||
@@ -19,9 +19,9 @@ log = getLogger(__name__)
|
||||
class ResetDriverWatchdog(Thread):
|
||||
"""Watchdog to catch the reset_driver action."""
|
||||
|
||||
def __init__(self, picontrol_device: str):
|
||||
def __init__(self, pi_control_device="/dev/piControl0"):
|
||||
super(ResetDriverWatchdog, self).__init__()
|
||||
self.procimg = picontrol_device
|
||||
self.procimg = pi_control_device
|
||||
self.daemon = True
|
||||
self._calls = []
|
||||
self._exit = False
|
||||
@@ -35,7 +35,7 @@ class ResetDriverWatchdog(Thread):
|
||||
"""
|
||||
Mainloop of watchdog for reset_driver.
|
||||
|
||||
If the thread cannot open the process image or the IOCTL is not
|
||||
If the thread can not open the process image or the IOCTL is not
|
||||
implemented (wheezy), the thread function will stop. The trigger
|
||||
property will always return True.
|
||||
"""
|
||||
@@ -51,7 +51,7 @@ class ResetDriverWatchdog(Thread):
|
||||
)
|
||||
return
|
||||
|
||||
# The ioctl will return 2 bytes (c-type int)
|
||||
# The ioctl will return 2 byte (c-type int)
|
||||
byte_buff = bytearray(2)
|
||||
while not self._exit:
|
||||
try:
|
||||
@@ -73,7 +73,7 @@ class ResetDriverWatchdog(Thread):
|
||||
def register_call(self, function):
|
||||
"""Register a function, if watchdog triggers."""
|
||||
if not callable(function):
|
||||
raise ValueError("Function is not callable.")
|
||||
return ValueError("Function is not callable.")
|
||||
if function not in self._calls:
|
||||
self._calls.append(function)
|
||||
|
||||
@@ -89,7 +89,7 @@ class ResetDriverWatchdog(Thread):
|
||||
log.debug("leave ResetDriverWatchdog.stop()")
|
||||
|
||||
def unregister_call(self, function=None):
|
||||
"""Remove a function from the watchdog trigger."""
|
||||
"""Remove a function call on watchdog trigger."""
|
||||
if function is None:
|
||||
self._calls.clear()
|
||||
elif function in self._calls:
|
||||
@@ -101,24 +101,3 @@ class ResetDriverWatchdog(Thread):
|
||||
rc = self._triggered
|
||||
self._triggered = False
|
||||
return rc
|
||||
|
||||
|
||||
class PiControlIoctl:
|
||||
IOCTL_RESET_DRIVER = 19212
|
||||
|
||||
def __init__(self, picontrol_device: str):
|
||||
self.picontrol_device = picontrol_device
|
||||
|
||||
def ioctl(self, request, arg=0):
|
||||
if type(arg) is c_int:
|
||||
arg = arg.value
|
||||
|
||||
_fd = os.open(self.picontrol_device, os.O_WRONLY)
|
||||
return_value = ioctl(_fd, 19212, arg)
|
||||
os.close(_fd)
|
||||
|
||||
return return_value
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self.picontrol_device
|
||||
@@ -2,20 +2,21 @@
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""D-Bus interfaces for piControl."""
|
||||
import os
|
||||
from fcntl import ioctl
|
||||
from logging import getLogger
|
||||
|
||||
from pydbus.generic import signal
|
||||
|
||||
from .process_image_helper import PiControlIoctl, ResetDriverWatchdog
|
||||
from ..dbus_helper import DbusInterface
|
||||
from ..interface_helper import ResetDriverWatchdog
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
|
||||
class InterfacePiControl(DbusInterface):
|
||||
class InterfacePiControl:
|
||||
"""
|
||||
<node>
|
||||
<interface name='com.revolutionpi.middleware1.PiControl'>
|
||||
<interface name='com.revolutionpi.middleware1.picontrol'>
|
||||
<signal name="NotifyDriverReset">
|
||||
</signal>
|
||||
<method name='ResetDriver'>
|
||||
@@ -26,16 +27,12 @@ class InterfacePiControl(DbusInterface):
|
||||
|
||||
NotifyDriverReset = signal()
|
||||
|
||||
def __init__(self, picontrol_device: str, config_rsc: str):
|
||||
self.picontrol_device = picontrol_device
|
||||
self.config_rsc = config_rsc
|
||||
def __init__(self):
|
||||
self.pi_control = "/dev/piControl0"
|
||||
|
||||
self.wd_reset_driver = ResetDriverWatchdog(self.picontrol_device)
|
||||
self.wd_reset_driver = ResetDriverWatchdog(self.pi_control)
|
||||
self.wd_reset_driver.register_call(self.notify_reset_driver)
|
||||
|
||||
def cleanup(self):
|
||||
self.wd_reset_driver.stop()
|
||||
|
||||
def notify_reset_driver(self):
|
||||
self.NotifyDriverReset()
|
||||
|
||||
@@ -43,9 +40,21 @@ class InterfacePiControl(DbusInterface):
|
||||
log.debug("enter InterfacePiControl.ResetDriver")
|
||||
|
||||
try:
|
||||
picontrol_ioctl = PiControlIoctl(self.picontrol_device)
|
||||
picontrol_ioctl.ioctl(PiControlIoctl.IOCTL_RESET_DRIVER)
|
||||
fd = os.open(self.pi_control, os.O_WRONLY)
|
||||
except Exception as e:
|
||||
log.warning(f"could not open ${self.pi_control} to reset driver")
|
||||
raise e
|
||||
|
||||
execption = None
|
||||
try:
|
||||
# KB_RESET _IO('K', 12 ) // reset the piControl driver including the config file
|
||||
ioctl(fd, 19212)
|
||||
log.info("reset piControl driver")
|
||||
except Exception as e:
|
||||
log.warning(f"could not reset piControl driver: ${e}")
|
||||
raise e
|
||||
execption = e
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
if execption:
|
||||
raise execption
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""D-Bus interfaces for system configuration."""
|
||||
from .interface_config import InterfaceRevpiConfig
|
||||
@@ -1,100 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""D-Bus interfaces for hardware configuration."""
|
||||
from collections import namedtuple
|
||||
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
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
FeatureFunction = namedtuple("FeatureFunction", ["function", "args"])
|
||||
|
||||
|
||||
class InterfaceRevpiConfig(DbusInterface):
|
||||
"""
|
||||
<node>
|
||||
<interface name="com.revolutionpi.middleware1.RevpiConfig">
|
||||
<method name="Disable">
|
||||
<arg name="feature" type="s" direction="in"/>
|
||||
</method>
|
||||
<method name="Enable">
|
||||
<arg name="feature" type="s" direction="in"/>
|
||||
</method>
|
||||
<method name="GetStatus">
|
||||
<arg name="feature" type="s" direction="in"/>
|
||||
<arg name="status" type="b" direction="out"/>
|
||||
</method>
|
||||
<method name="GetAvailability">
|
||||
<arg name="feature" type="s" direction="in"/>
|
||||
<arg name="available" type="b" direction="out"/>
|
||||
</method>
|
||||
<property name="available_features" type="as" access="read"/>
|
||||
</interface>
|
||||
</node>
|
||||
"""
|
||||
|
||||
def Disable(self, feature: str) -> None:
|
||||
"""Disable the feature."""
|
||||
feature_function = get_feature(feature)
|
||||
feature_function.function(ConfigActions.DISABLE, *feature_function.args)
|
||||
|
||||
def Enable(self, feature: str) -> None:
|
||||
"""Enable the feature."""
|
||||
feature_function = get_feature(feature)
|
||||
feature_function.function(ConfigActions.ENABLE, *feature_function.args)
|
||||
|
||||
def GetStatus(self, feature: str) -> bool:
|
||||
"""Get feature status."""
|
||||
feature_function = get_feature(feature)
|
||||
return feature_function.function(ConfigActions.STATUS, *feature_function.args)
|
||||
|
||||
def GetAvailability(self, feature: str) -> bool:
|
||||
"""Get feature availability on the RevPi."""
|
||||
feature_function = get_feature(feature)
|
||||
return feature_function.function(ConfigActions.AVAILABLE, *feature_function.args)
|
||||
|
||||
@property
|
||||
def available_features(self) -> list[str]:
|
||||
return list(AVAILABLE_FEATURES.keys())
|
||||
|
||||
|
||||
def get_feature(feature: str) -> FeatureFunction:
|
||||
if feature not in AVAILABLE_FEATURES:
|
||||
raise ValueError(f"feature {feature} does not exist")
|
||||
feature_function = AVAILABLE_FEATURES[feature]
|
||||
if not feature_function:
|
||||
raise NotImplementedError(f"feature {feature} is not implemented")
|
||||
return feature_function
|
||||
|
||||
|
||||
AVAILABLE_FEATURES = {
|
||||
"gui": FeatureFunction(configure_gui, []),
|
||||
"revpi-con-can": FeatureFunction(configure_con_can, []),
|
||||
"dphys-swapfile": FeatureFunction(configure_dphys_swapfile, []),
|
||||
"pimodbus-master": FeatureFunction(simple_systemd, ["pimodbus-master.service"]),
|
||||
"pimodbus-slave": FeatureFunction(simple_systemd, ["pimodbus-slave.service"]),
|
||||
"systemd-timesyncd": FeatureFunction(simple_systemd, ["systemd-timesyncd.service"]),
|
||||
"ssh": FeatureFunction(simple_systemd, ["ssh.service"]),
|
||||
"nodered": FeatureFunction(simple_systemd, ["nodered.service"]),
|
||||
"noderedrevpinodes-server": FeatureFunction(
|
||||
simple_systemd, ["noderedrevpinodes-server.service"]
|
||||
),
|
||||
"revpipyload": FeatureFunction(simple_systemd, ["revpipyload.service"]),
|
||||
"bluetooth": FeatureFunction(configure_bluetooth, []),
|
||||
"wlan": FeatureFunction(configure_wlan, []),
|
||||
"avahi": FeatureFunction(configure_avahi_daemon, []),
|
||||
"external-antenna": FeatureFunction(configure_external_antenna, []),
|
||||
}
|
||||
@@ -1,434 +0,0 @@
|
||||
# -*- 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
|
||||
|
||||
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):
|
||||
UNKNOWN = 0
|
||||
CM1 = 6
|
||||
CM3 = 10
|
||||
CM4 = 20
|
||||
CM4S = 21
|
||||
CM5 = 24
|
||||
|
||||
|
||||
class ConfigActions(Enum):
|
||||
ENABLE = "enable"
|
||||
DISABLE = "disable"
|
||||
STATUS = "status"
|
||||
AVAILABLE = "available"
|
||||
|
||||
|
||||
class RevPiConfig:
|
||||
|
||||
def __init__(self):
|
||||
self._cm_type = ComputeModuleTypes.UNKNOWN
|
||||
|
||||
self._revpi_with_con_bridge = False
|
||||
self._wlan_class_path = ""
|
||||
|
||||
self.serial = ""
|
||||
self.model = ""
|
||||
|
||||
self._init_device_info()
|
||||
|
||||
def _init_device_info(self):
|
||||
dc_cpuinfo = {}
|
||||
|
||||
# Extract CPU information
|
||||
with open("/proc/cpuinfo", "r") as f:
|
||||
line = "\n"
|
||||
while line:
|
||||
line = f.readline()
|
||||
if line.startswith(("Revision", "Serial", "Model")):
|
||||
key, value = line.split(":", 1)
|
||||
key = key.strip().lower()
|
||||
value = value.strip()
|
||||
dc_cpuinfo[key] = value
|
||||
|
||||
self.model = dc_cpuinfo.get("model", "")
|
||||
self.serial = dc_cpuinfo.get("serial", "")
|
||||
|
||||
# Detect Compute Module type
|
||||
revision = dc_cpuinfo.get("revision", "")
|
||||
if revision:
|
||||
revision = int(revision, 16)
|
||||
mask = 4080 # 0xFF0 in dezimal
|
||||
try:
|
||||
self._cm_type = ComputeModuleTypes((revision & mask) >> 4)
|
||||
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 ConBridge
|
||||
could_have_con_bridge = self._cm_type in (ComputeModuleTypes.CM3, ComputeModuleTypes.CM4S)
|
||||
if could_have_con_bridge:
|
||||
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
|
||||
|
||||
@property
|
||||
def with_con_bridge(self) -> bool:
|
||||
return self._revpi_with_con_bridge
|
||||
|
||||
@property
|
||||
def with_wlan(self) -> bool:
|
||||
return bool(self._wlan_class_path)
|
||||
|
||||
|
||||
class ConfigTxt:
|
||||
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:
|
||||
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]:
|
||||
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):
|
||||
with open(self._config_txt_path, "r") as f:
|
||||
self._config_txt_lines = f.readlines()
|
||||
|
||||
def save_config(self):
|
||||
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):
|
||||
# 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:
|
||||
return self._clear_name_values("dtoverlay", dtoverlays)
|
||||
|
||||
def clear_dtparams(self, dtparams: str or list) -> int:
|
||||
return self._clear_name_values("dtparam", dtparams)
|
||||
|
||||
def get_values(self, var_name: str) -> list:
|
||||
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:
|
||||
return self._config_txt_path
|
||||
|
||||
|
||||
def configure_avahi_daemon(action: ConfigActions):
|
||||
return_value = simple_systemd(action, "avahi-daemon.service")
|
||||
|
||||
# Post actions for avahi-daemon
|
||||
if action in (ConfigActions.ENABLE, ConfigActions.DISABLE):
|
||||
# Apply the enable/disable action to the avahi socket AFTER the service
|
||||
# unit, because a connected socket could interrupt stop
|
||||
simple_systemd(action, "avahi-daemon.socket")
|
||||
|
||||
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")
|
||||
|
||||
# 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_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_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]:
|
||||
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")
|
||||
|
||||
if action is ConfigActions.ENABLE:
|
||||
systemd_manager.UnmaskUnitFiles([unit], False)
|
||||
systemd_manager.EnableUnitFiles([unit], False, False)
|
||||
systemd_manager.StartUnit(unit, "replace")
|
||||
|
||||
elif action is ConfigActions.DISABLE:
|
||||
systemd_manager.StopUnit(unit, "replace")
|
||||
systemd_manager.DisableUnitFiles([unit], False)
|
||||
|
||||
elif action is ConfigActions.STATUS:
|
||||
try:
|
||||
unit_path = systemd_manager.LoadUnit(unit)
|
||||
properties = bus.get(".systemd1", unit_path)
|
||||
except Exception:
|
||||
log.warning(f"could not get systemd unit {unit}")
|
||||
return False
|
||||
|
||||
return properties.UnitFileState == "enabled" and properties.ActiveState == "active"
|
||||
|
||||
elif action is ConfigActions.AVAILABLE:
|
||||
try:
|
||||
unit_path = systemd_manager.LoadUnit(unit)
|
||||
properties = bus.get(".systemd1", unit_path)
|
||||
except Exception:
|
||||
log.warning(f"could not get systemd unit {unit}")
|
||||
return False
|
||||
|
||||
return properties.LoadState != "not-found"
|
||||
|
||||
else:
|
||||
raise ValueError(f"action {action} not supported")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
rc = RevPiConfig()
|
||||
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 con-bridge:", rc.with_con_bridge)
|
||||
|
||||
config_txt = ConfigTxt()
|
||||
print("Config file: ", config_txt.config_txt_path)
|
||||
@@ -9,7 +9,7 @@ __version__ = "1.4.0"
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from argparse import ArgumentParser, Namespace, SUPPRESS
|
||||
from argparse import ArgumentParser, Namespace
|
||||
from configparser import ConfigParser
|
||||
from enum import Enum
|
||||
from os import R_OK, W_OK, access, environ, getpid, remove
|
||||
@@ -262,16 +262,6 @@ parser = ArgumentParser(
|
||||
prog=programname,
|
||||
description="Program description",
|
||||
)
|
||||
|
||||
# Use session bus of D-Bus for local testing and development proposes (hidden)
|
||||
parser.add_argument(
|
||||
"--use-session-bus",
|
||||
dest="use_session_bus",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help=SUPPRESS,
|
||||
)
|
||||
|
||||
parser.add_argument("--version", action="version", version=f"%(prog)s {program_version}")
|
||||
parser.add_argument(
|
||||
"-f",
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
@@ -1,7 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
from os import environ
|
||||
|
||||
# D-BUS needs a DISPLAY variable to work with the session bus
|
||||
environ["DISPLAY"] = ":0"
|
||||
@@ -1,36 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
from time import sleep
|
||||
|
||||
from tests.dbus_middleware1.fake_devices import PiControlDeviceMockup
|
||||
|
||||
|
||||
class TestBusProvider(PiControlDeviceMockup):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
# Do not import things on top of the module. Some classes or functions need to be mocked up first.
|
||||
from revpi_middleware.dbus_middleware1 import BusProvider
|
||||
|
||||
# Prepare the bus provider and start it
|
||||
self.bp = BusProvider(
|
||||
self.picontrol.name,
|
||||
use_system_bus=False,
|
||||
)
|
||||
self.bp.start()
|
||||
|
||||
# Wait 5 seconds until the bus provider has started the main loop
|
||||
counter = 50
|
||||
while not self.bp.running and counter > 0:
|
||||
counter -= 1
|
||||
sleep(0.1)
|
||||
|
||||
def tearDown(self):
|
||||
self.bp.stop()
|
||||
self.bp.join(10.0)
|
||||
if self.bp.is_alive():
|
||||
raise RuntimeError("Bus provider thread is still running")
|
||||
|
||||
super().tearDown()
|
||||
@@ -1,120 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
from ctypes import c_int
|
||||
from queue import Empty, Queue
|
||||
from tempfile import NamedTemporaryFile
|
||||
from threading import Event, Thread
|
||||
from unittest import TestCase
|
||||
|
||||
IOCTL_QUEUE = Queue()
|
||||
RESET_DRIVER_EVENT = Event()
|
||||
|
||||
|
||||
class FakePiControlDevice:
|
||||
IOCTL_RESET_DRIVER = 19212
|
||||
|
||||
def __init__(self, picontrol_device: str):
|
||||
self._fh = NamedTemporaryFile("wb+", 0, prefix="fake_device_")
|
||||
self.name = self._fh.name
|
||||
|
||||
def __del__(self):
|
||||
self._fh.close()
|
||||
|
||||
def reset_process_image(self):
|
||||
self._fh.write(b"\x00" * 4096)
|
||||
self._fh.seek(0)
|
||||
|
||||
def ioctl(self, request, arg=0) -> int:
|
||||
if type(arg) is c_int:
|
||||
arg = arg.value
|
||||
|
||||
if request == self.IOCTL_RESET_DRIVER:
|
||||
pass
|
||||
else:
|
||||
raise NotImplementedError(f"Unknown IOCTL request: {request}")
|
||||
|
||||
IOCTL_QUEUE.put_nowait((request, arg))
|
||||
return arg
|
||||
|
||||
def close(self):
|
||||
self._fh.close()
|
||||
|
||||
def read(self, size):
|
||||
return self._fh.read(size)
|
||||
|
||||
def seek(self, offset, whence):
|
||||
self._fh.seek(offset, whence)
|
||||
|
||||
def write(self, buffer):
|
||||
return self._fh.write(buffer)
|
||||
|
||||
|
||||
class FakeResetDriverWatchdog(Thread):
|
||||
|
||||
def __init__(self, picontrol_device: str):
|
||||
super().__init__()
|
||||
self.daemon = True
|
||||
self._calls = []
|
||||
self._exit = False
|
||||
self.not_implemented = True
|
||||
self._triggered = False
|
||||
self.start()
|
||||
|
||||
def run(self):
|
||||
while not self._exit:
|
||||
if RESET_DRIVER_EVENT.wait(0.1):
|
||||
RESET_DRIVER_EVENT.clear()
|
||||
self._triggered = True
|
||||
for func in self._calls:
|
||||
func()
|
||||
|
||||
def register_call(self, function):
|
||||
"""Register a function, if watchdog triggers."""
|
||||
if not callable(function):
|
||||
raise ValueError("Function is not callable.")
|
||||
if function not in self._calls:
|
||||
self._calls.append(function)
|
||||
|
||||
def stop(self):
|
||||
"""Stop watchdog for reset_driver."""
|
||||
self._exit = True
|
||||
|
||||
def unregister_call(self, function=None):
|
||||
"""Remove a function from the watchdog trigger."""
|
||||
if function is None:
|
||||
self._calls.clear()
|
||||
elif function in self._calls:
|
||||
self._calls.remove(function)
|
||||
|
||||
@property
|
||||
def triggered(self):
|
||||
"""Will return True one time after watchdog was triggered."""
|
||||
rc = self._triggered
|
||||
self._triggered = False
|
||||
return rc
|
||||
|
||||
|
||||
class PiControlDeviceMockup(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
# Empty the queue
|
||||
while True:
|
||||
try:
|
||||
IOCTL_QUEUE.get_nowait()
|
||||
except Empty:
|
||||
break
|
||||
|
||||
# Replace classes with mockup classes
|
||||
import revpi_middleware.dbus_middleware1.process_image.interface_picontrol as test_helpers
|
||||
test_helpers.PiControlIoctl = FakePiControlDevice
|
||||
test_helpers.ResetDriverWatchdog = FakeResetDriverWatchdog
|
||||
|
||||
# Create a fake picontrol0 device
|
||||
self.picontrol = FakePiControlDevice(picontrol_device="/dev/fake_device_0")
|
||||
|
||||
def tearDown(self):
|
||||
self.picontrol.close()
|
||||
super().tearDown()
|
||||
@@ -1,3 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
@@ -1,45 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
from threading import Thread
|
||||
from time import sleep
|
||||
|
||||
from revpi_middleware.cli_commands.dbus_helper import BusType, await_signal, simple_call
|
||||
from revpi_middleware.dbus_middleware1 import extend_interface
|
||||
from tests.dbus_middleware1.bus_provider import TestBusProvider
|
||||
from tests.dbus_middleware1.fake_devices import IOCTL_QUEUE, RESET_DRIVER_EVENT
|
||||
|
||||
|
||||
class TestObjectPicontrol(TestBusProvider):
|
||||
|
||||
def test_is_active(self):
|
||||
self.assertTrue(self.bp.running)
|
||||
|
||||
def test_reset_driver(self):
|
||||
simple_call(
|
||||
"ResetDriver",
|
||||
interface=extend_interface("PiControl"),
|
||||
bus_type=BusType.SESSION,
|
||||
)
|
||||
ioctl_call = IOCTL_QUEUE.get(timeout=2.0)
|
||||
self.assertEqual((19212, 0), ioctl_call)
|
||||
|
||||
def test_notify_reset_driver(self):
|
||||
timeout = 5
|
||||
|
||||
def target_call_reset_driver():
|
||||
sleep(1.0)
|
||||
RESET_DRIVER_EVENT.set()
|
||||
|
||||
th_wait_for_reset = Thread(target=target_call_reset_driver, daemon=True)
|
||||
th_wait_for_reset.start()
|
||||
|
||||
result = await_signal(
|
||||
"NotifyDriverReset",
|
||||
timeout,
|
||||
extend_interface("PiControl"),
|
||||
bus_type=BusType.SESSION,
|
||||
)
|
||||
self.assertTrue(result)
|
||||
|
||||
th_wait_for_reset.join(timeout=timeout)
|
||||
Reference in New Issue
Block a user