feat(cli): Add D-Bus helper functions for CLI commands.
This introduces `simple_call`, a utility function to interact with D-Bus interfaces within the RevPi system. It facilitates method invocation on specific interfaces and paths, improving modularity and code reuse in middleware operations.
This commit is contained in:
5
src/revpi_middleware/cli_commands/__init__.py
Normal file
5
src/revpi_middleware/cli_commands/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""CLI commands to control revpi_middleware."""
|
||||
from ..__about__ import __author__, __copyright__, __license__, __version__
|
||||
53
src/revpi_middleware/cli_commands/cli_base.py
Normal file
53
src/revpi_middleware/cli_commands/cli_base.py
Normal file
@@ -0,0 +1,53 @@
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""Command-line interface base module for RevPi middleware daemon.
|
||||
|
||||
This module provides the foundation for the RevPi middleware CLI commands
|
||||
and argument parsing setup.
|
||||
"""
|
||||
from argparse import ArgumentParser
|
||||
from logging import getLogger
|
||||
|
||||
from revpi_middleware.cli_commands import cli_picontrol
|
||||
from revpi_middleware.proginit import StdLogOutput
|
||||
from .. import proginit as pi
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
|
||||
def setup_command_line_arguments():
|
||||
pi.parser.prog = "revpictl"
|
||||
|
||||
# Add all objects here
|
||||
rpictl_obj = pi.parser.add_subparsers(
|
||||
dest="obj",
|
||||
title="object",
|
||||
help="Available RevPi objects",
|
||||
)
|
||||
|
||||
obj_picontrol = rpictl_obj.add_parser(
|
||||
"picontrol",
|
||||
help="RevPi PiControl object",
|
||||
)
|
||||
cli_picontrol.add_subparsers(obj_picontrol)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
setup_command_line_arguments()
|
||||
pi.init_app(StdLogOutput.STDERR)
|
||||
|
||||
obj = pi.pargs.obj
|
||||
if obj == "picontrol":
|
||||
rc = cli_picontrol.main()
|
||||
|
||||
else:
|
||||
log.error(f"Unknown object: {obj}")
|
||||
rc = 1
|
||||
|
||||
pi.cleanup()
|
||||
|
||||
return rc
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
exit(main())
|
||||
46
src/revpi_middleware/cli_commands/cli_picontrol.py
Normal file
46
src/revpi_middleware/cli_commands/cli_picontrol.py
Normal file
@@ -0,0 +1,46 @@
|
||||
# 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 simple_call
|
||||
from .. import proginit as pi
|
||||
from ..dbus_middleware1 import extend_interface
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
|
||||
def add_subparsers(parent_parser: ArgumentParser):
|
||||
methods = parent_parser.add_subparsers(
|
||||
dest="methods",
|
||||
title="methods",
|
||||
help="Available RevPi PiControl methods",
|
||||
)
|
||||
methods.add_parser(
|
||||
"reset",
|
||||
help="Reset the piControl driver",
|
||||
)
|
||||
|
||||
|
||||
def method_reset():
|
||||
log.debug("D-Bus call of method ResetDriver")
|
||||
simple_call("ResetDriver", interface=extend_interface("picontrol"))
|
||||
log.info("ResetDriver called via D-Bus")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
method = pi.pargs.methods
|
||||
try:
|
||||
|
||||
if method == "reset":
|
||||
method_reset()
|
||||
|
||||
else:
|
||||
raise Exception(f"Unknown method: {method}")
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
36
src/revpi_middleware/cli_commands/dbus_helper.py
Normal file
36
src/revpi_middleware/cli_commands/dbus_helper.py
Normal file
@@ -0,0 +1,36 @@
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
"""D-Bus helper functions for cli commands."""
|
||||
from pydbus import SystemBus
|
||||
|
||||
from ..dbus_middleware1 import REVPI_DBUS_BASE_PATH
|
||||
from ..dbus_middleware1 import REVPI_DBUS_NAME
|
||||
|
||||
PICONTROL_INTERFACE = "com.revolutionpi.middleware1.picontrol"
|
||||
RESET_DRIVER_METHOD = "ResetDriver"
|
||||
|
||||
|
||||
def simple_call(method: str, *args, interface: str, object_path=REVPI_DBUS_BASE_PATH):
|
||||
"""
|
||||
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 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 providing the required functionality.
|
||||
object_path: str, optional
|
||||
The D-Bus object path of the RevPi interface. Defaults to REVPI_DBUS_BASE_PATH.
|
||||
|
||||
Returns:
|
||||
Any
|
||||
The result of the method invocation on the targeted D-Bus interface.
|
||||
"""
|
||||
bus = SystemBus()
|
||||
revpi = bus.get(REVPI_DBUS_NAME, object_path)
|
||||
iface = revpi[interface]
|
||||
return getattr(iface, method)(*args)
|
||||
Reference in New Issue
Block a user