"""I2PControl RPC method implementations.""" from __future__ import annotations class I2PControlMethods: """RPC method implementations. Each method takes a *params* dict and returns a result dict. Uses a mock/stub router context for now. """ def __init__(self, router_context: dict | None = None) -> None: self._context: dict = router_context or {} self._config: dict[str, str] = {} # ----- router methods --------------------------------------------------- def router_info(self, params: dict) -> dict: """Return router info: version, status, uptime, etc.""" return { "version": "0.1.0", "status": self._context.get("status", "running"), "uptime": self._context.get("uptime", 0), "network_status": self._context.get("network_status", "ok"), } def router_status(self, params: dict) -> dict: """Return router status state.""" return {"status": self._context.get("status", "running")} # ----- tunnel methods --------------------------------------------------- def tunnel_list(self, params: dict) -> dict: """Return list of active tunnels.""" return {"tunnels": self._context.get("tunnels", [])} # ----- peer methods ----------------------------------------------------- def peer_list(self, params: dict) -> dict: """Return list of known peers, with optional limit.""" limit = params.get("limit", 50) peers = self._context.get("peers", []) return {"peers": peers[:limit]} # ----- netdb methods ---------------------------------------------------- def netdb_stats(self, params: dict) -> dict: """Return network database statistics.""" return { "total_routers": self._context.get("total_routers", 0), "total_leasesets": self._context.get("total_leasesets", 0), } # ----- bandwidth -------------------------------------------------------- def bw_status(self, params: dict) -> dict: """Return bandwidth status.""" return { "inbound_bps": self._context.get("inbound_bps", 0), "outbound_bps": self._context.get("outbound_bps", 0), } # ----- config ----------------------------------------------------------- def config_get(self, params: dict) -> dict: """Get a configuration value by key.""" key = params.get("key", "") return {"key": key, "value": self._config.get(key)} def config_set(self, params: dict) -> dict: """Set a configuration value.""" key = params.get("key", "") value = params.get("value", "") self._config[key] = value return {"key": key, "value": value, "success": True} # ----- dispatch --------------------------------------------------------- def dispatch(self, method: str, params: dict) -> dict: """Dispatch a method call by name. Dot-separated method names (e.g. ``router.info``) are converted to underscores (``router_info``) for lookup. Raises ``ValueError`` for unknown methods. """ handler_name = method.replace(".", "_") handler = getattr(self, handler_name, None) if handler is None or handler_name.startswith("_") or handler_name == "dispatch": raise ValueError(f"Unknown method: {method}") return handler(params)