from __future__ import annotations
import asyncio
import functools
import logging
from collections.abc import Callable, Coroutine
from typing import TYPE_CHECKING, Any, overload
from pyobs.events import Event, LogEvent, ModuleClosedEvent, registered_events
from pyobs.interfaces import Interface
from pyobs.interfaces.interface import get_registered_interface
from pyobs.utils.enums import ModuleState
from .commlogging import CommLoggingHandler
from .proxy import Proxy, ProxyType, _ProxyContext
if TYPE_CHECKING:
from pyobs.modules import Module
StateCallback = Callable[[Any], None]
PresenceCallback = Callable[["ModuleState", str], None]
EventHandler = Callable[[Event, str], Coroutine[Any, Any, bool]]
log = logging.getLogger(__name__)
class Comm:
"""Base class for all Comm modules in pyobs."""
__module__ = "pyobs.comm"
def __init__(self) -> None:
"""Creates a comm module."""
self._proxies: dict[str, Proxy] = {}
self._state_subscriptions: dict[str, list[tuple[type[Interface], StateCallback]]] = {}
self._presence_subscriptions: dict[str, list[PresenceCallback]] = {}
self._module: Module | None = None
self._log_queue: asyncio.Queue[LogEvent] = asyncio.Queue()
self._logging_task: asyncio.Task[Any] | None = None
self._event_handlers: dict[type[Event], list[EventHandler]] = {}
self._event_handler_tasks: dict[tuple[type[Event], EventHandler], set[asyncio.Task[Any]]] = {}
self._events_sent: set[type[Event]] = set()
self._events_subscribed: set[type[Event]] = set()
self._closing = asyncio.Event()
self._published_state: set[type[Interface]] = set()
@property
def has_module(self) -> bool:
return self._module is not None
@property
def module(self) -> Module:
"""The module that this Comm object is attached to."""
if self._module is None:
raise ValueError("No module.")
return self._module
@module.setter
def module(self, module: Module) -> None:
"""The module that this Comm object is attached to."""
# if we have a _set_module method, call it
self._set_module(module)
# store module
self._module = module
def _set_module(self, module: Module) -> None: ...
[docs]
async def open(self) -> None:
"""Open module."""
# add handler to global logger
root_logger = logging.getLogger()
if not any(isinstance(h, CommLoggingHandler) for h in root_logger.handlers):
from pyobs.utils.logging.context import ModuleNameFilter
handler = CommLoggingHandler(self)
handler.setLevel(logging.INFO)
handler.addFilter(ModuleNameFilter())
root_logger.addHandler(handler)
# start logging thread
self._logging_task = asyncio.create_task(self._logging())
# some events
await self.register_event(ModuleClosedEvent, self._client_disconnected)
[docs]
async def close(self) -> None:
"""Close module."""
self._closing.set()
# close thread
if self._logging_task:
self._logging_task.cancel()
self._logging_task = None
def _get_full_client_name(self, name: str) -> str:
"""Returns full name for given client.
Some Comm modules may use short names for their clients. This methods returns the full name
for a given short name.
Args:
name: Short name to get full name for.
Returns:
Full name for given client.
"""
# this base class doesn't have short names
return name
async def _get_client(self, client: str) -> Module | Proxy | None:
"""Get a proxy to the given client.
Args:
client: Name of client.
Returns:
Proxy class for given client.
"""
# return module, if "main" is requested
if client == "main":
return self.module
if client is None:
return None
# if client doesn't exist or we disabled caching, fetch a new proxy
if client not in self._proxies:
# get interfaces
try:
interfaces = await self.get_interfaces(client)
except IndexError:
return None
# create new proxy
proxy = Proxy(self, client, interfaces)
# fetch capabilities in the background, so a peer that isn't fully up yet doesn't block
# proxy construction -- Proxy.wait_for_capabilities() can be used to wait for the result
for interface in interfaces:
if interface.capabilities is not None:
asyncio.create_task(self._fetch_and_update_capabilities(client, interface, proxy))
# subscribe to state
for interface in interfaces:
if interface.has_own_state():
await self.subscribe_state(client, interface, functools.partial(proxy.update_state, interface))
# a proxy without interfaces is useless and would stay stale (e.g. peer's disco#info not
# answered yet), so do not cache it
if len(interfaces) == 0:
return proxy
self._proxies[client] = proxy
return self._proxies[client]
async def _fetch_and_update_capabilities(self, client: str, interface: type[Interface], proxy: Proxy) -> None:
"""Fetch capabilities for a single interface and push them into the given proxy once available."""
cap = await self._get_capabilities(client, interface)
if cap is not None:
proxy.update_capabilities(interface, cap)
async def _resolve_proxy(
self, name_or_object: str | object, obj_type: type[ProxyType] | None = None
) -> Any | ProxyType:
"""Returns object directly if it is of given type. Otherwise get proxy of client with given name and check type.
If name_or_object is an object:
- If it is of type (or derived), return object.
- Otherwise raise exception.
If name_name_or_object is string:
- Create proxy from name and raise exception, if it doesn't exist.
- Check type and raise exception if wrong.
- Return object.
Args:
name_or_object: Name of object or object itself.
obj_type: Expected class of object.
Returns:
Object or proxy to object.
Raises:
ValueError: If proxy does not exist or wrong type.
"""
if obj_type is not None and isinstance(name_or_object, obj_type):
# return directly
return name_or_object
elif isinstance(name_or_object, str):
# get proxy
try:
proxy = await self._get_client(name_or_object)
except KeyError:
raise ValueError(f"Could not get proxy for {name_or_object}.")
# check it
if proxy is None:
raise ValueError(f'Could not create proxy for given name "{name_or_object}".')
elif obj_type is None or isinstance(proxy, obj_type):
return proxy
else:
message = f'Proxy obtained from given name "{name_or_object}" is not of requested type "{obj_type}".'
hint = self._diagnose_missing_interface(name_or_object, obj_type)
if hint is not None:
message += f" {hint}"
raise ValueError(message)
else:
# completely wrong...
raise ValueError(f'Given parameter is neither a name nor an object of requested type "{obj_type}".')
def _diagnose_missing_interface(self, client: str, obj_type: type[Any]) -> str | None:
"""Backend hook, called when a proxy exists but doesn't implement obj_type.
Backends that can tell "genuinely missing" apart from "present at an
incompatible version" using data already in hand return a hint string
to append to the ValueError; other backends just return None.
"""
return None
async def _safe_resolve_proxy(
self, name_or_object: str | object, obj_type: type[ProxyType] | None = None
) -> Any | ProxyType | None:
"""Calls proxy() in a safe way and returns None instead of raising an exception."""
try:
return await self._resolve_proxy(name_or_object, obj_type)
except ValueError:
return None
@overload
def proxy(self, name_or_object: str | object, obj_type: type[ProxyType]) -> _ProxyContext[ProxyType]: ...
@overload
def proxy(self, name_or_object: str | object, obj_type: None = None) -> _ProxyContext[Any]: ...
[docs]
def proxy(self, name_or_object: str | object, obj_type: type[ProxyType] | None = None) -> _ProxyContext[Any]:
"""Returns a context manager; use as `async with self.proxy(...) as x:`."""
return _ProxyContext(self._resolve_proxy(name_or_object, obj_type))
@overload
def safe_proxy(
self, name_or_object: str | object, obj_type: type[ProxyType]
) -> _ProxyContext[ProxyType | None]: ...
@overload
def safe_proxy(self, name_or_object: str | object, obj_type: None = None) -> _ProxyContext[Any]: ...
[docs]
def safe_proxy(self, name_or_object: str | object, obj_type: type[ProxyType] | None = None) -> _ProxyContext[Any]:
"""Same as proxy(), but yields None inside the block instead of raising."""
return _ProxyContext(self._safe_resolve_proxy(name_or_object, obj_type))
[docs]
async def has_proxy(self, name_or_object: str | object, obj_type: type[Any] | None = None) -> bool:
"""True if a proxy of the given type can currently be resolved. Doesn't keep a reference
to it, so doesn't need async with the way proxy()/safe_proxy() do."""
return await self._safe_resolve_proxy(name_or_object, obj_type) is not None
async def _client_disconnected(self, event: Event, sender: str) -> bool:
"""Called when a client disconnects.
Args:
event: Disconnect event.
sender: Name of client that disconnected.
"""
# if a client disconnects, clear its proxy state then evict it
if sender in self._proxies:
self._proxies[sender].clear_state()
del self._proxies[sender]
# tear down any state subscriptions held for that client
for interface, callback in self._state_subscriptions.pop(sender, []):
await self.unsubscribe_state(sender, interface, callback)
# tear down any presence subscriptions held for that client -- otherwise a stale
# callback (e.g. bound to a since-destroyed GUI widget) lingers and can blow up
# the next presence update for this client, silently swallowing its reconnect
for callback in self._presence_subscriptions.pop(sender, []):
await self.unsubscribe_presence(sender, callback)
return True
@property
def name(self) -> str | None:
"""Name of this client."""
raise NotImplementedError
@property
def clients(self) -> list[str]:
"""Returns list of currently connected clients.
Returns:
(list) list of currently connected clients.
"""
raise NotImplementedError
[docs]
async def clients_with_interface(self, interface: type[Interface]) -> list[str]:
"""Returns list of currently connected clients that implement the given interface.
Args:
interface: Interface to search for.
Returns:
(list) list of currently connected clients that implement the given interface.
"""
return [c for c in self.clients if await self._supports_interface(c, interface)]
[docs]
async def get_interfaces(self, client: str) -> list[type[Interface]]:
"""Returns list of interfaces for given client.
Args:
client: Name of client.
Returns:
list of supported interfaces.
Raises:
IndexError: If client cannot be found.
"""
raise NotImplementedError
async def _supports_interface(self, client: str, interface: type[Interface]) -> bool:
"""Checks, whether the given client supports the given interface.
Args:
client: Client to check.
interface: Interface to check.
Returns:
Whether or not interface is supported.
"""
raise NotImplementedError
@staticmethod
def _interface_names_to_classes(interfaces: list[str]) -> list[type[Interface]]:
"""Converts a list of interface names to interface classes.
Args:
interfaces: list of interface names.
Returns:
list of interface classes.
"""
# loop interfaces
interface_classes = []
for interface_name in interfaces:
cls = get_registered_interface(interface_name)
if cls is not None:
interface_classes.append(cls)
else:
log.error('Could not find interface "%s" for client.', interface_name)
return interface_classes
[docs]
async def execute(self, client: str, method: str, annotation: dict[str, Any], *args: Any) -> Any:
"""Execute a given method on a remote client.
Args:
client (str): ID of client.
method (str): Method to call.
annotation: Method annotation.
*args: list of parameters for given method.
Returns:
Passes through return from method call.
"""
raise NotImplementedError
async def _logging(self) -> None:
"""Background thread for handling the logging."""
# run until closing
while True:
# get item (maybe wait for it) and send it
try:
entry = self._log_queue.get_nowait()
await self.send_event(entry)
except asyncio.QueueEmpty:
# if queue is empty, sleep a little
await asyncio.sleep(1)
except asyncio.CancelledError:
return
except Exception:
# don't forward this back through CommLoggingHandler -- it would re-queue
# itself onto _log_queue, and if the send failure is due to a sustained
# outage, every retry generates another one of these, forever
log.exception("Something went wrong", extra={"pyobs_no_forward": True})
pass
[docs]
def log_message(self, entry: LogEvent) -> None:
"""Send a log message to other clients.
Args:
entry (LogEvent): Log event to send.
"""
self._log_queue.put_nowait(entry)
[docs]
async def send_event(self, event: Event) -> None:
"""Send an event to other clients.
Args:
event (Event): Event to send
"""
pass
def _get_derived_events(self, event: type[Event]) -> list[type[Event]]:
"""Return list of given event itself and all events derived from it.
Args:
event: Event class to check.
Returns:
list of event classes.
"""
return [cls for cls in registered_events().values() if issubclass(cls, event)]
[docs]
async def register_event(self, event_class: type[Event], handler: EventHandler | None = None) -> None:
"""Register an event type. If a handler is given, we also receive those events, otherwise we just
send them. Event classes derived from the given one are registered, too, but only those that have been
imported (and therefore registered) by now.
Args:
event_class: Class of event to register.
handler: Event handler method.
"""
# we also want to register all events derived from the given one
event_classes = self._get_derived_events(event_class)
# do we have a handler? then we subscribe, otherwise we just declare that we send it
if handler:
self._events_subscribed.update(event_classes)
# loop classes
for ev in event_classes:
# initialize list
if ev not in self._event_handlers:
self._event_handlers[ev] = []
# avoid duplicates
if handler not in self._event_handlers[ev]:
# add handler
self._event_handlers[ev].append(handler)
else:
self._events_sent.update(event_classes)
# if event is not a local one, we also need to do some XMPP stuff
if not event_class.local:
await self._register_events(event_classes, handler)
[docs]
async def unregister_event(self, event_class: type[Event], handler: EventHandler) -> None:
"""Remove a handler previously added via register_event().
Leaves the event type itself registered as *sent* if the module also declared it via a
handler-less register_event() call, only stops calling this handler and, once no handler
is left for the event, drops it from what's advertised as *subscribed*.
Also cancels any dispatch of this handler that _send_event_to_module() already scheduled
via asyncio.create_task() but that hasn't run yet, so a handler bound to something torn
down right before this call can never fire against it afterwards (issue #871). If that
dispatch had already started and is currently suspended at an await, cancellation raises
CancelledError into it there. This guarantee only holds if the caller awaits
unregister_event() directly in the same synchronous stretch as the teardown it's guarding
-- scheduling it as a separate task (e.g. `asyncio.create_task(unregister_event(...))`)
can still lose the race against an already-scheduled dispatch.
Args:
event_class: Class of event that was registered.
handler: The exact handler that was passed to register_event().
"""
# same derived-events expansion as register_event(), so this mirrors exactly what was added
unsubscribed: list[type[Event]] = []
for ev in self._get_derived_events(event_class):
handlers = self._event_handlers.get(ev)
if handlers is not None and handler in handlers:
handlers.remove(handler)
if not handlers:
self._events_subscribed.discard(ev)
unsubscribed.append(ev)
# cancel any dispatch for this handler that _send_event_to_module() already
# scheduled via asyncio.create_task() but that hasn't run yet -- otherwise it
# fires later against whatever the handler is bound to, even though the handler
# is gone from _event_handlers as of this call (see issue #871)
pending = self._event_handler_tasks.pop((ev, handler), None)
if pending is not None:
for task in pending:
task.cancel()
# only event classes that just lost their last handler need the comm layer to actually
# tear anything down (e.g. XmppComm unsubscribing from peers' event nodes)
if unsubscribed and not event_class.local:
await self._unregister_events(unsubscribed)
async def _register_events(self, events: list[type[Event]], handler: EventHandler | None = None) -> None:
pass
async def _unregister_events(self, events: list[type[Event]]) -> None:
pass
[docs]
async def set_state(self, interface: type[Interface], state: Any) -> None:
"""Publish state for this module.
Args:
interface: Interface type for the state.
state: State object to publish.
"""
self._published_state.add(interface)
await self._set_state(interface, state)
async def _set_state(self, interface: type[Interface], state: Any) -> None:
pass
[docs]
def missing_published_state(self, interfaces: list[type[Interface]]) -> list[type[Interface]]:
"""Return interfaces (from the given list) that declare state via has_own_state() but
haven't had set_state() called for them yet."""
return [i for i in interfaces if i.has_own_state() and i not in self._published_state]
[docs]
async def set_capabilities(self, interface: type[Interface], capabilities: Any) -> None:
"""Publish capabilities for this module.
Called by Module.open() for each interface that defines a capabilities
dataclass. Not intended to be called directly by module authors after
that point — capabilities are fixed for the module lifetime.
Args:
interface: Interface type the capabilities belong to.
capabilities: Capabilities dataclass instance.
"""
await self._set_capabilities(interface, capabilities)
async def _set_capabilities(self, interface: type[Interface], capabilities: Any) -> None:
pass
[docs]
async def get_capabilities(self, module: str, interface: type[Interface]) -> Any | None:
"""Fetch and deserialize capabilities for a remote module's interface.
Args:
module: Module name (e.g. "camera").
interface: Interface class whose Capabilities dataclass to fetch.
Returns:
Deserialized Capabilities dataclass instance, or None if not published.
"""
return await self._get_capabilities(module, interface)
async def _get_capabilities(self, module: str, interface: type[Interface]) -> Any | None:
return None
[docs]
async def set_presence(self, state: ModuleState, error_string: str = "") -> None:
"""Publish presence for this module (module lifecycle state).
Called automatically by Module.set_state — not intended to be called
directly by module authors.
Args:
state: Current module lifecycle state.
error_string: Error message, used when state is ERROR.
"""
await self._set_presence(state, error_string)
async def _set_presence(self, state: ModuleState, error_string: str = "") -> None:
pass
[docs]
async def mark_ready(self) -> None:
"""Signal that the module has finished its own startup (reached ModuleState.READY) and is
safe for peers to discover.
Called by Module once it transitions to READY. Transports that announce a module to peers
independently of RPC dispatch (e.g. XmppComm's XMPP presence, or LocalComm's direct
ModuleOpenedEvent dispatch to already-connected peers) hold that announcement back until
this fires, so peers never read capabilities that are still mid-publish. Transports without
such an announcement don't need to override this.
"""
await self._mark_ready()
async def _mark_ready(self) -> None:
pass
[docs]
def get_client_state(self, module: str) -> tuple[ModuleState, str] | None:
"""Return the last known presence state of a connected module.
Returns (ModuleState, error_string) or None if the module is not connected.
This replaces the old get_state()/get_error_string() RPC pattern.
"""
return self._get_client_state(module)
def _get_client_state(self, module: str) -> tuple[ModuleState, str] | None:
return None
[docs]
def get_own_state(self, interface: type[Interface]) -> Any:
"""Return the last state published by this module for the given interface, or None."""
return self._get_own_state(interface)
def _get_own_state(self, interface: type[Interface]) -> Any:
return None
[docs]
def get_own_capabilities(self, interface: type[Interface]) -> Any:
"""Return the capabilities published by this module for the given interface, or None."""
return self._get_own_capabilities(interface)
def _get_own_capabilities(self, interface: type[Interface]) -> Any:
return None
[docs]
async def subscribe_state(
self,
module: str,
interface: type[Interface],
callback: StateCallback,
) -> None:
"""Subscribe to state updates for a given module and interface.
Delivers the current value immediately on subscribe.
Args:
module: Name of remote module.
interface: Interface type to subscribe to.
callback: Called with state object on each update.
"""
self._state_subscriptions.setdefault(module, []).append((interface, callback))
await self._subscribe_state(module, interface, callback)
async def _subscribe_state(
self,
module: str,
interface: type[Interface],
callback: StateCallback,
) -> None:
pass
[docs]
async def unsubscribe_state(
self,
module: str,
interface: type[Interface],
callback: StateCallback,
) -> None:
"""Unsubscribe from state updates.
Args:
module: Name of remote module.
interface: Interface type to unsubscribe from.
callback: Callback that was registered.
"""
await self._unsubscribe_state(module, interface, callback)
async def _unsubscribe_state(
self,
module: str,
interface: type[Interface],
callback: StateCallback,
) -> None:
pass
[docs]
async def subscribe_presence(self, module: str, callback: PresenceCallback) -> None:
"""Subscribe to presence updates for a given module.
Delivers the current value immediately, then on every change.
Args:
module: Name of remote module.
callback: Called with (ModuleState, error_string) on each update.
"""
self._presence_subscriptions.setdefault(module, []).append(callback)
await self._subscribe_presence(module, callback)
async def _subscribe_presence(self, module: str, callback: PresenceCallback) -> None:
pass
[docs]
async def unsubscribe_presence(self, module: str, callback: PresenceCallback) -> None:
"""Unsubscribe from presence updates.
Args:
module: Name of remote module.
callback: Callback that was registered.
"""
subs = self._presence_subscriptions.get(module)
if subs is not None and callback in subs:
subs.remove(callback)
await self._unsubscribe_presence(module, callback)
async def _unsubscribe_presence(self, module: str, callback: PresenceCallback) -> None:
pass
def _send_event_to_module(self, event: Event, from_client: str) -> None:
"""Send an event to all connected modules.
Args:
event: Event to send.
from_client: Client that sent the event.
"""
# send it
if event.__class__ in self._event_handlers:
for handler in self._event_handlers[event.__class__]:
# handle it
ret = handler(event, from_client)
if asyncio.iscoroutine(ret):
task = asyncio.create_task(ret)
self._event_handler_tasks.setdefault((event.__class__, handler), set()).add(task)
task.add_done_callback(functools.partial(self._log_handler_exception, handler=handler, event=event))
def _log_handler_exception(self, task: asyncio.Task[Any], handler: EventHandler, event: Event) -> None:
"""Log an event handler's exception with context, instead of leaving it to asyncio's own
"Task exception was never retrieved" warning at garbage-collection time, which gives no
indication of which handler or event actually failed -- every event handler is dispatched
as a fire-and-forget task (see _send_event_to_module), so this is the only place any of
them get their exceptions surfaced at all.
Also drops the task from _event_handler_tasks, the bookkeeping unregister_event() uses to
cancel still-pending dispatches for a handler being removed -- without this, a
long-running module would accumulate an entry per (event_class, handler) forever.
"""
tasks = self._event_handler_tasks.get((event.__class__, handler))
if tasks is not None:
tasks.discard(task)
if not tasks:
del self._event_handler_tasks[(event.__class__, handler)]
if task.cancelled():
return
exc = task.exception()
if exc is not None:
log.error(
"Unhandled exception in event handler %s for %s.",
getattr(handler, "__qualname__", handler),
event.__class__.__name__,
exc_info=exc,
)
__all__ = ["Comm"]