diff --git a/libervia/cli/cmd_serverless.py b/libervia/cli/cmd_serverless.py new file mode 100644 index 00000000..92ea1906 --- /dev/null +++ b/libervia/cli/cmd_serverless.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 + +# Libervia CLI +# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program 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 Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""CLI commands for serverless.""" + +from rich.table import Table + +from libervia.backend.core.i18n import _ +from libervia.backend.plugins.plugin_misc_serverless.models import ( + DiscoverNodes, + Node, + ServerlessInfo, +) +from libervia.cli.constants import Const as C + +from . import base + +__commands__ = ["Serverless"] + + +class ServerlessInfos(base.CommandBase): + """Show serverless device info.""" + + def __init__(self, host) -> None: + extra_outputs = {"default": self.default_output} + super().__init__( + host, + "info", + help=_("show serverless device information"), + use_output=C.OUTPUT_MODEL, + extra_outputs=extra_outputs, + use_verbose=True, + ) + + def add_parser_options(self) -> None: + pass + + def default_output(self, info: ServerlessInfo) -> None: + self.console.print(f" Device ID: {info.device_id}") + self.console.print(f" mDNS name: {info.service_name}") + self.console.print(f" Peers found: {info.peer_count}") + + async def start(self) -> None: + try: + data_s = await self.host.bridge.serverless_infos_get() + info = ServerlessInfo.model_validate_json(data_s) + except Exception as e: + self.disp(f"Failed to get serverless info: {e}", error=True) + self.host.quit(C.EXIT_BRIDGE_ERRBACK) + return + + await self.output(info) + self.host.quit() + + +class ServerlessDiscover(base.CommandBase): + """Discover peers on the local network.""" + + def __init__(self, host) -> None: + extra_outputs = {"default": self.default_output} + super().__init__( + host, + "discover", + help=_("discover peers on the local network"), + use_output=C.OUTPUT_MODEL, + extra_outputs=extra_outputs, + use_verbose=True, + ) + + def add_parser_options(self) -> None: + self.parser.add_argument( + "-f", + "--follow", + action="store_true", + help=_("keep monitoring for new/lost peers"), + ) + + def _render_table(self, nodes: list[Node], title: str) -> Table: + table = Table(title=title) + table.add_column("👤 " + _("Name")) + table.add_column("🪪 " + _("Device ID")) + table.add_column("🖧 " + _("Addresses")) + if self.verbosity: + table.add_column("📋 " + _("Metadata")) + for node in nodes: + row = [node.name, node.device_id, ", ".join(node.addrs)] + if self.verbosity: + row.append( + ", ".join(f"{k}={v}" for k, v in node.metadata.items()) + if node.metadata + else "" + ) + table.add_row(*row) + return table + + def default_output(self, nodes: DiscoverNodes) -> None: + if not nodes: + self.console.print(_("No peers discovered.")) + return + self.console.print(self._render_table(list(nodes), _("Discovered peers"))) + + async def _on_peer_found(self, node_json: str) -> None: + node = Node.model_validate_json(node_json) + self._peers[node.device_id] = node + self.console.print("") + self.console.print(self._render_table([node], _("[green]New peer found[/green]"))) + + async def _on_peer_lost(self, device_id: str) -> None: + node = self._peers.pop(device_id, None) + self.console.print("") + if node is not None: + self.console.print(self._render_table([node], _("[red]Peer lost[/red]"))) + else: + self.console.print( + _("[red]Peer lost:[/red] {device_id}").format(device_id=device_id) + ) + + async def start(self) -> None: + try: + nodes_s = await self.host.bridge.serverless_discover("{}") + nodes = DiscoverNodes.model_validate_json(nodes_s) + except Exception as e: + self.disp(f"Failed to discover peers: {e}", error=True) + self.host.quit(C.EXIT_BRIDGE_ERRBACK) + return + + await self.output(nodes) + + if not self.args.follow: + self.host.quit() + return + + # Follow mode: subscribe to signals and keep running + self._peers: dict[str, Node] = {node.device_id: node for node in nodes} + self.host.bridge.register_signal( + "serverless_peer_found", self._on_peer_found, "plugin" + ) + self.host.bridge.register_signal( + "serverless_peer_lost", self._on_peer_lost, "plugin" + ) + + +class Serverless(base.CommandBase): + """Serverless mode commands.""" + + subcommands = ( + ServerlessInfos, + ServerlessDiscover, + ) + + def __init__(self, host) -> None: + super().__init__( + host, + "serverless", + use_profile=False, + help=_("serverless mode commands"), + aliases=["sl"], + )