diff --git a/i3ipc/_private/pubsub.py b/i3ipc/_private/pubsub.py index 56fe574..2f52cfa 100644 --- a/i3ipc/_private/pubsub.py +++ b/i3ipc/_private/pubsub.py @@ -1,9 +1,23 @@ +from typing import Callable, Optional, TypeAlias, TypedDict + +from i3ipc.connection import Connection +from i3ipc.events import IpcBaseEvent + +Handler: TypeAlias = Callable[[Connection, IpcBaseEvent], None] + + +class Subscription(TypedDict): + event: str + detail: Optional[str] + handler: Handler + + class PubSub(object): - def __init__(self, conn): + def __init__(self, conn: Connection): self.conn = conn - self._subscriptions = [] + self._subscriptions: list[Subscription] = [] - def subscribe(self, detailed_event, handler): + def subscribe(self, detailed_event: str, handler: Handler): event = detailed_event.replace('-', '_') detail = '' @@ -12,10 +26,10 @@ def subscribe(self, detailed_event, handler): self._subscriptions.append({'event': event, 'detail': detail, 'handler': handler}) - def unsubscribe(self, handler): + def unsubscribe(self, handler: Handler): self._subscriptions = list(filter(lambda s: s['handler'] != handler, self._subscriptions)) - def emit(self, event, data): + def emit(self, event: str, data: Optional[IpcBaseEvent]): detail = '' if data and hasattr(data, 'change'): @@ -27,4 +41,4 @@ def emit(self, event, data): if data: s['handler'](self.conn, data) else: - s['handler'](self.conn) + s['handler'](self.conn) # type: ignore[call-arg] diff --git a/i3ipc/_private/types.py b/i3ipc/_private/types.py index b6ae544..c498d9c 100644 --- a/i3ipc/_private/types.py +++ b/i3ipc/_private/types.py @@ -43,11 +43,11 @@ class EventType(Enum): TICK = (1 << 7) INPUT = (1 << 21) - def to_string(self): + def to_string(self) -> str: return str.lower(self.name) @staticmethod - def from_string(val): + def from_string(val) -> 'EventType': match = [e for e in EventType if e.to_string() == val] if not match: @@ -55,7 +55,7 @@ def from_string(val): return match[0] - def to_list(self): + def to_list(self) -> list[str]: events_list = [] if self.value & EventType.WORKSPACE.value: events_list.append(EventType.WORKSPACE.to_string()) diff --git a/i3ipc/aio/connection.py b/i3ipc/aio/connection.py index 4755853..8b0732d 100644 --- a/i3ipc/aio/connection.py +++ b/i3ipc/aio/connection.py @@ -6,7 +6,7 @@ from .. import con import os import json -from typing import Optional, List, Tuple, Callable, Union +from typing import Callable, Coroutine, Optional, TypeAlias, TypedDict, Union import struct import socket import logging @@ -33,9 +33,22 @@ def ensure_future(obj): return future +Handler: TypeAlias = Union[ + Callable[['Connection', IpcBaseEvent], None], Callable[['Connection', IpcBaseEvent], Coroutine] +] + + +class _Subscription(TypedDict): + event: str + detail: Optional[str] + handler: Handler + + class _AIOPubSub(PubSub): - def queue_handler(self, handler, data=None): - conn = self.conn + _subscriptions: list[_Subscription] # type: ignore[assignment] + + def queue_handler(self, handler: Handler, data: Optional[IpcBaseEvent] = None): + conn: Connection = self.conn # type: ignore[assignment] async def handler_coroutine(): try: @@ -46,15 +59,15 @@ async def handler_coroutine(): handler(conn, data) else: if asyncio.iscoroutinefunction(handler): - await handler(conn) + await handler(conn) # type: ignore[call-arg] else: - handler(conn) + handler(conn) # type: ignore[call-arg] except Exception as e: conn.main_quit(_error=e) ensure_future(handler_coroutine()) - def emit(self, event, data): + def emit(self, event: str, data: Optional[IpcBaseEvent]): detail = '' if data and hasattr(data, 'change'): @@ -137,7 +150,9 @@ class Con(con.Con): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - async def command(self, command: str) -> List[CommandReply]: + _conn: 'Connection' # type: ignore[assignment] + + async def command(self, command: str) -> list[CommandReply]: # type: ignore[override] """Runs a command on this container. .. seealso:: https://i3wm.org/docs/userguide.html#list_of_commands @@ -148,7 +163,7 @@ async def command(self, command: str) -> List[CommandReply]: """ return await self._conn.command('[con_id="{}"] {}'.format(self.id, command)) - async def command_children(self, command: str) -> List[CommandReply]: + async def command_children(self, command: str) -> list[CommandReply]: # type: ignore[override] """Runs a command on the immediate children of the currently selected container. @@ -173,7 +188,7 @@ def _pack(msg_type: MessageType, payload: str) -> bytes: return b''.join((_MAGIC, s, pb)) -def _unpack_header(data: bytes) -> Tuple[bytes, int, int]: +def _unpack_header(data: bytes) -> tuple[bytes, int, int]: return struct.unpack(_struct_header, data[:_struct_header_size]) @@ -257,11 +272,11 @@ class Connection: def __init__(self, socket_path: Optional[str] = None, auto_reconnect: bool = False): self._socket_path = socket_path self._auto_reconnect = auto_reconnect - self._pubsub = _AIOPubSub(self) - self._subscriptions = set() - self._main_future = None - self._reconnect_future = None - self._synchronizer = None + self._pubsub = _AIOPubSub(self) # type: ignore[arg-type] + self._subscriptions: set = set() + self._main_future: Optional[Future] = None + self._reconnect_future: Optional[Future] = None + self._synchronizer: Optional[Synchronizer] = None def _sync(self): if self._synchronizer is None: @@ -270,7 +285,7 @@ def _sync(self): self._synchronizer.sync() @property - def socket_path(self) -> str: + def socket_path(self) -> Optional[str]: """The path of the socket this ``Connection`` is connected to. :rtype: str @@ -323,28 +338,29 @@ def _read_message(self): return - magic, message_length, event_type = _unpack_header(buf) + magic, message_length, event_type_int = _unpack_header(buf) assert magic == _MAGIC raw_message = self._sub_socket.recv(message_length) message = json.loads(raw_message) # events have the highest bit set - if not event_type & (1 << 31): + if not event_type_int & (1 << 31): # a reply return - event_type = EventType(1 << (event_type & 0x7f)) + event_type = EventType(1 << (event_type_int & 0x7f)) logger.info('got message on subscription socket: type=%s, message=%s', event_type, raw_message) + event: IpcBaseEvent if event_type == EventType.WORKSPACE: - event = WorkspaceEvent(message, self, _Con=Con) + event = WorkspaceEvent(message, self, _Con=Con) # type: ignore[arg-type] elif event_type == EventType.OUTPUT: event = OutputEvent(message) elif event_type == EventType.MODE: event = ModeEvent(message) elif event_type == EventType.WINDOW: - event = WindowEvent(message, self, _Con=Con) + event = WindowEvent(message, self, _Con=Con) # type: ignore[arg-type] elif event_type == EventType.BARCONFIG_UPDATE: event = BarconfigUpdateEvent(message) elif event_type == EventType.BINDING: @@ -409,6 +425,7 @@ async def do_reconnect(): error = e await asyncio.sleep(0.001) + assert self._reconnect_future is not None if error: self._reconnect_future.set_exception(error) else: @@ -464,7 +481,7 @@ async def _message(self, message_type: MessageType, payload: str = '') -> bytear logger.info('got message reply: %s', message) return message - async def subscribe(self, events: Union[List[Event], List[str]], force: bool = False): + async def subscribe(self, events: Union[list[Event], list[str]], force: bool = False): """Send a ``SUBSCRIBE`` command to the ipc subscription connection and await the result. To attach event handlers, use :func:`Connection.on() `. Calling this is only needed if you want @@ -487,7 +504,7 @@ async def subscribe(self, events: Union[List[Event], List[str]], force: bool = F for e in events: e = Event(e) - if e not in Event._subscribable_events: + if e not in (Event._subscribable_events): # type: ignore[attr-defined] correct_event = str.split(e.value, '::')[0].upper() raise ValueError( f'only nondetailed events are subscribable (use Event.{correct_event})') @@ -510,9 +527,7 @@ async def subscribe(self, events: Union[List[Event], List[str]], force: bool = F await self._loop.sock_sendall(self._sub_socket, _pack(MessageType.SUBSCRIBE, payload)) - def on(self, - event: Union[Event, str], - handler: Callable[['Connection', IpcBaseEvent], None] = None): + def on(self, event: Union[Event, str], handler: Optional[Handler] = None): def on_wrapped(handler): self._on(event, handler) return handler @@ -522,7 +537,7 @@ def on_wrapped(handler): else: return on_wrapped - def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBaseEvent], None]): + def _on(self, event: Union[Event, str], handler: Handler): """Subscribe to the event and call the handler when it is emitted by the i3 ipc. @@ -534,6 +549,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase if type(event) is Event: event = event.value + assert type(event) is str event = event.replace('-', '_') if event.count('::') > 0: @@ -543,7 +559,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase logger.info('adding event handler: event=%s, handler=%s', event, handler) - self._pubsub.subscribe(event, handler) + self._pubsub.subscribe(event, handler) # type: ignore[arg-type] ensure_future(self.subscribe([base_event])) def off(self, handler: Callable[['Connection', IpcBaseEvent], None]): @@ -554,9 +570,9 @@ def off(self, handler: Callable[['Connection', IpcBaseEvent], None]): :type handler: :class:`Callable` """ logger.info('removing event handler: handler=%s', handler) - self._pubsub.unsubscribe(handler) + self._pubsub.unsubscribe(handler) # type: ignore[arg-type] - async def command(self, cmd: str) -> List[CommandReply]: + async def command(self, cmd: str) -> list[CommandReply]: """Sends a command to i3. .. seealso:: https://i3wm.org/docs/userguide.html#list_of_commands @@ -567,10 +583,10 @@ async def command(self, cmd: str) -> List[CommandReply]: command given. :rtype: list(:class:`CommandReply `) """ - data = await self._message(MessageType.COMMAND, cmd) + data_raw = await self._message(MessageType.COMMAND, cmd) - if data: - data = json.loads(data) + if data_raw: + data = json.loads(data_raw) return CommandReply._parse_list(data) else: return [] @@ -581,11 +597,11 @@ async def get_version(self) -> VersionReply: :returns: The i3 version. :rtype: :class:`i3ipc.VersionReply` """ - data = await self._message(MessageType.GET_VERSION) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_VERSION) + data = json.loads(data_raw) return VersionReply(data) - async def get_bar_config_list(self) -> List[str]: + async def get_bar_config_list(self) -> list[str]: """Gets the names of all bar configurations. :returns: A list of all bar configurations. @@ -611,28 +627,28 @@ async def get_bar_config(self, bar_id=None) -> Optional[BarConfigReply]: return None bar_id = bar_config_list[0] - data = await self._message(MessageType.GET_BAR_CONFIG, bar_id) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_BAR_CONFIG, bar_id) + data = json.loads(data_raw) return BarConfigReply(data) - async def get_outputs(self) -> List[OutputReply]: + async def get_outputs(self) -> list[OutputReply]: """Gets the list of current outputs. :returns: A list of current outputs. :rtype: list(:class:`i3ipc.OutputReply`) """ - data = await self._message(MessageType.GET_OUTPUTS) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_OUTPUTS) + data = json.loads(data_raw) return OutputReply._parse_list(data) - async def get_workspaces(self) -> List[WorkspaceReply]: + async def get_workspaces(self) -> list[WorkspaceReply]: """Gets the list of current workspaces. :returns: A list of current workspaces :rtype: list(:class:`i3ipc.WorkspaceReply`) """ - data = await self._message(MessageType.GET_WORKSPACES) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_WORKSPACES) + data = json.loads(data_raw) return WorkspaceReply._parse_list(data) async def get_tree(self) -> Con: @@ -642,9 +658,9 @@ async def get_tree(self) -> Con: :rtype: :class:`i3ipc.Con` """ data = await self._message(MessageType.GET_TREE) - return Con(json.loads(data), None, self) + return Con(json.loads(data), None, self) # type: ignore[arg-type] - async def get_marks(self) -> List[str]: + async def get_marks(self) -> list[str]: """Gets the names of all currently set marks. :returns: A list of currently set marks. @@ -653,7 +669,7 @@ async def get_marks(self) -> List[str]: data = await self._message(MessageType.GET_MARKS) return json.loads(data) - async def get_binding_modes(self) -> List[str]: + async def get_binding_modes(self) -> list[str]: """Gets the names of all currently configured binding modes :returns: A list of binding modes @@ -668,8 +684,8 @@ async def get_config(self) -> ConfigReply: :returns: A class containing the config. :rtype: :class:`i3ipc.ConfigReply` """ - data = await self._message(MessageType.GET_CONFIG) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_CONFIG) + data = json.loads(data_raw) return ConfigReply(data) async def send_tick(self, payload: str = "") -> TickReply: @@ -678,28 +694,28 @@ async def send_tick(self, payload: str = "") -> TickReply: :returns: The reply to the tick command :rtype: :class:`i3ipc.TickReply` """ - data = await self._message(MessageType.SEND_TICK, payload) - data = json.loads(data) + data_raw = await self._message(MessageType.SEND_TICK, payload) + data = json.loads(data_raw) return TickReply(data) - async def get_inputs(self) -> List[InputReply]: + async def get_inputs(self) -> list[InputReply]: """(sway only) Gets the inputs connected to the compositor. :returns: The reply to the inputs command :rtype: list(:class:`i3ipc.InputReply`) """ - data = await self._message(MessageType.GET_INPUTS) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_INPUTS) + data = json.loads(data_raw) return InputReply._parse_list(data) - async def get_seats(self) -> List[SeatReply]: + async def get_seats(self) -> list[SeatReply]: """(sway only) Gets the seats configured on the compositor :returns: The reply to the seats command :rtype: list(:class:`i3ipc.SeatReply`) """ - data = await self._message(MessageType.GET_SEATS) - data = json.loads(data) + data_raw = await self._message(MessageType.GET_SEATS) + data = json.loads(data_raw) return SeatReply._parse_list(data) def main_quit(self, _error=None): diff --git a/i3ipc/con.py b/i3ipc/con.py index a3ba37f..4ea4207 100644 --- a/i3ipc/con.py +++ b/i3ipc/con.py @@ -1,12 +1,15 @@ import re import sys +from .connection import Connection from .model import Rect, Gaps from . import replies from collections import deque -from typing import List, Optional +from typing import Any, cast, Generic, Literal, Optional, TypeVar +_Con = TypeVar('_Con', bound='Con') -class Con: + +class Con(Generic[_Con]): """A container of a window and child containers gotten from :func:`i3ipc.Connection.get_tree()` or events. .. seealso:: https://i3wm.org/docs/ipc.html#_tree_reply @@ -85,7 +88,40 @@ class Con: :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data, parent, conn): + border: str + current_border_width: int + floating: Literal["auto_off", "auto_on", "user_off", "user_on"] + focus: list[int] + focused: bool + fullscreen_mode: int + id: int + layout: str + marks: list[str] + name: str + num: int + orientation: str + percent: float + scratchpad_state: str + shell: str + sticky: bool + type: str + urgent: bool + window: int + window_class: Optional[str] + window_instance: Optional[str] + window_role: Optional[str] + window_title: Optional[str] + rect: Rect + window_rect: Rect + deco_rect: Optional[Rect] + geometry: Optional[Rect] + app_id: str + pid: int + gaps: Optional[Gaps] + representation: str + visible: bool + + def __init__(self: _Con, data: dict[str, Any], parent: _Con, conn: Connection): self.ipc_data = data self._conn = conn self.parent = parent @@ -126,12 +162,12 @@ def __init__(self, data, parent, conn): self.nodes = [] if 'nodes' in data: for n in data['nodes']: - self.nodes.append(self.__class__(n, self, conn)) + self.nodes.append(cast(_Con, self.__class__(n, self, conn))) self.floating_nodes = [] if 'floating_nodes' in data: for n in data['floating_nodes']: - self.floating_nodes.append(self.__class__(n, self, conn)) + self.floating_nodes.append(cast(_Con, self.__class__(n, self, conn))) self.window_class = None self.window_instance = None @@ -185,7 +221,7 @@ def is_floating(self) -> bool: return True return False - def root(self) -> 'Con': + def root(self) -> _Con: """Gets the root container. :returns: The root container. @@ -193,7 +229,7 @@ def root(self) -> 'Con': """ if not self.parent: - return self + return cast(_Con, self) con = self.parent @@ -202,7 +238,7 @@ def root(self) -> 'Con': return con - def descendants(self) -> List['Con']: + def descendants(self) -> list[_Con]: """Gets a list of all child containers for the container in breadth-first order. @@ -211,7 +247,7 @@ def descendants(self) -> List['Con']: """ return [c for c in self] - def descendents(self) -> List['Con']: + def descendents(self) -> list[_Con]: """Gets a list of all child containers for the container in breadth-first order. @@ -224,7 +260,7 @@ def descendents(self) -> List['Con']: print('WARNING: descendents is deprecated. Use `descendants()` instead.', file=sys.stderr) return self.descendants() - def leaves(self) -> List['Con']: + def leaves(self) -> list[_Con]: """Gets a list of leaf child containers for this container in breadth-first order. Leaf containers normally contain application windows. @@ -240,7 +276,7 @@ def leaves(self) -> List['Con']: return leaves - def command(self, command: str) -> List[replies.CommandReply]: + def command(self, command: str) -> list[replies.CommandReply]: """Runs a command on this container. .. seealso:: https://i3wm.org/docs/userguide.html#list_of_commands @@ -251,7 +287,7 @@ def command(self, command: str) -> List[replies.CommandReply]: """ return self._conn.command('[con_id="{}"] {}'.format(self.id, command)) - def command_children(self, command: str) -> List[replies.CommandReply]: + def command_children(self, command: str) -> list[replies.CommandReply]: """Runs a command on the immediate children of the currently selected container. @@ -260,16 +296,14 @@ def command_children(self, command: str) -> List[replies.CommandReply]: :returns: A list of replies for each command that was executed. :rtype: list(:class:`CommandReply `) """ - if not len(self.nodes): - return commands = [] for c in self.nodes: commands.append('[con_id="{}"] {};'.format(c.id, command)) - self._conn.command(' '.join(commands)) + return self._conn.command(' '.join(commands)) - def workspaces(self) -> List['Con']: + def workspaces(self) -> list[_Con]: """Gets a list of workspace containers for this tree. :returns: A list of workspace containers. @@ -288,7 +322,7 @@ def collect_workspaces(con): collect_workspaces(self.root()) return workspaces - def find_focused(self) -> Optional['Con']: + def find_focused(self) -> Optional[_Con]: """Finds the focused container under this container if it exists. :returns: The focused container if it exists. @@ -300,7 +334,7 @@ def find_focused(self) -> Optional['Con']: except StopIteration: return None - def find_by_id(self, id: int) -> Optional['Con']: + def find_by_id(self, id: int) -> Optional[_Con]: """Finds a container with the given container id under this node. :returns: The container with this container id if it exists. @@ -312,7 +346,7 @@ def find_by_id(self, id: int) -> Optional['Con']: except StopIteration: return None - def find_by_pid(self, pid: int) -> List['Con']: + def find_by_pid(self, pid: int) -> list[_Con]: """Finds all the containers under this node with this pid. :returns: A list of containers with this pid. @@ -320,7 +354,7 @@ def find_by_pid(self, pid: int) -> List['Con']: """ return [c for c in self if c.pid == pid] - def find_by_window(self, window: int) -> Optional['Con']: + def find_by_window(self, window: int) -> Optional[_Con]: """Finds a container with the given window id under this node. :returns: The container with this window id if it exists. @@ -332,7 +366,7 @@ def find_by_window(self, window: int) -> Optional['Con']: except StopIteration: return None - def find_by_role(self, pattern: str) -> List['Con']: + def find_by_role(self, pattern: str) -> list[_Con]: """Finds all the containers under this node with a window role that matches the given regex pattern. @@ -342,7 +376,7 @@ def find_by_role(self, pattern: str) -> List['Con']: """ return [c for c in self if c.window_role and re.search(pattern, c.window_role)] - def find_named(self, pattern: str) -> List['Con']: + def find_named(self, pattern: str) -> list[_Con]: """Finds all the containers under this node with a name that matches the given regex pattern. @@ -352,7 +386,7 @@ def find_named(self, pattern: str) -> List['Con']: """ return [c for c in self if c.name and re.search(pattern, c.name)] - def find_titled(self, pattern: str) -> List['Con']: + def find_titled(self, pattern: str) -> list[_Con]: """Finds all the containers under this node with a window title that matches the given regex pattern. @@ -362,7 +396,7 @@ def find_titled(self, pattern: str) -> List['Con']: """ return [c for c in self if c.window_title and re.search(pattern, c.window_title)] - def find_classed(self, pattern: str) -> List['Con']: + def find_classed(self, pattern: str) -> list[_Con]: """Finds all the containers under this node with a window class, or app_id that matches the given regex pattern. @@ -375,7 +409,7 @@ def find_classed(self, pattern: str) -> List['Con']: return x11_windows + wayland_windows - def find_instanced(self, pattern: str) -> List['Con']: + def find_instanced(self, pattern: str) -> list[_Con]: """Finds all the containers under this node with a window instance that matches the given regex pattern. @@ -385,7 +419,7 @@ def find_instanced(self, pattern: str) -> List['Con']: """ return [c for c in self if c.window_instance and re.search(pattern, c.window_instance)] - def find_marked(self, pattern: str = ".*") -> List['Con']: + def find_marked(self, pattern: str = ".*") -> list[_Con]: """Finds all the containers under this node with a mark that matches the given regex pattern. @@ -393,10 +427,10 @@ def find_marked(self, pattern: str = ".*") -> List['Con']: pattern. :rtype: list(:class:`Con`) """ - pattern = re.compile(pattern) - return [c for c in self if any(pattern.search(mark) for mark in c.marks)] + cpattern = re.compile(pattern) + return [c for c in self if any(cpattern.search(mark) for mark in c.marks)] - def find_fullscreen(self) -> List['Con']: + def find_fullscreen(self) -> list[_Con]: """Finds all the containers under this node that are in fullscreen mode. @@ -405,7 +439,7 @@ def find_fullscreen(self) -> List['Con']: """ return [c for c in self if c.type == 'con' and c.fullscreen_mode] - def workspace(self) -> Optional['Con']: + def workspace(self) -> Optional[_Con]: """Finds the workspace container for this node if this container is at or below the workspace level. @@ -414,7 +448,7 @@ def workspace(self) -> Optional['Con']: workspace level. """ if self.type == 'workspace': - return self + return cast(_Con, self) ret = self.parent @@ -425,7 +459,7 @@ def workspace(self) -> Optional['Con']: return ret - def scratchpad(self) -> 'Con': + def scratchpad(self) -> Optional[_Con]: """Finds the scratchpad container. :returns: The scratchpad container. diff --git a/i3ipc/connection.py b/i3ipc/connection.py index 50d3d44..4bdef26 100644 --- a/i3ipc/connection.py +++ b/i3ipc/connection.py @@ -232,7 +232,7 @@ def get_version(self) -> VersionReply: data = json.loads(data) return VersionReply(data) - def get_bar_config(self, bar_id: str = None) -> Optional[BarConfigReply]: + def get_bar_config(self, bar_id: Optional[str] = None) -> Optional[BarConfigReply]: """Gets the bar configuration specified by the id. :param bar_id: The bar id to get the configuration for. If not given, @@ -391,7 +391,7 @@ def off(self, handler: Callable[['Connection', IpcBaseEvent], None]): def on(self, event: Union[Event, str], - handler: Callable[['Connection', IpcBaseEvent], None] = None): + handler: Optional[Callable[['Connection', IpcBaseEvent], None]] = None): def on_wrapped(handler): self._on(event, handler) return handler @@ -413,6 +413,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase if type(event) is Event: event = event.value + assert type(event) is str event = event.replace('-', '_') if event.count('::') > 0: @@ -426,7 +427,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase self._pubsub.subscribe(event, handler) return - event_type = 0 + event_type = EventType(0) if base_event == 'workspace': event_type = EventType.WORKSPACE elif base_event == 'output': @@ -453,18 +454,18 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase self._pubsub.subscribe(event, handler) - def _event_socket_setup(self): + def _event_socket_setup(self) -> None: self._sub_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self._sub_socket.connect(self._socket_path) self._subscribe(self.subscriptions) - def _event_socket_teardown(self): + def _event_socket_teardown(self) -> None: if self._sub_socket: self._sub_socket.shutdown(socket.SHUT_RDWR) self._sub_socket = None - def _event_socket_poll(self): + def _event_socket_poll(self: 'Connection'): if self._sub_socket is None: return True @@ -479,7 +480,7 @@ def _event_socket_poll(self): data = json.loads(data) msg_type = 1 << (msg_type & 0x7f) event_name = '' - event = None + event: IpcBaseEvent if msg_type == EventType.WORKSPACE.value: event_name = 'workspace' @@ -518,7 +519,7 @@ def _event_socket_poll(self): print(e) raise e - def main(self, timeout: float = 0.0): + def main(self, timeout: float = 0.0) -> None: """Starts the main loop for this connection to start handling events. :param timeout: If given, quit the main loop after ``timeout`` seconds. @@ -557,7 +558,7 @@ def main(self, timeout: float = 0.0): if loop_exception: raise loop_exception - def main_quit(self): + def main_quit(self) -> None: """Quits the running main loop for this connection.""" logger.info('shutting down the main loop') self._quitting = True diff --git a/i3ipc/events.py b/i3ipc/events.py index c80dddc..fa545d1 100644 --- a/i3ipc/events.py +++ b/i3ipc/events.py @@ -1,6 +1,7 @@ -from . import con +from . import con, connection from .replies import BarConfigReply, InputReply from enum import Enum +from typing import Any, Optional class IpcBaseEvent: @@ -45,7 +46,7 @@ class Event(Enum): INPUT_REMOVED = 'input::removed' -Event._subscribable_events = [e for e in Event if '::' not in e.value] +Event._subscribable_events = [e for e in Event if '::' not in e.value] # type: ignore[attr-defined] class WorkspaceEvent(IpcBaseEvent): @@ -65,11 +66,11 @@ class WorkspaceEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data, conn, _Con=con.Con): + def __init__(self, data: dict[str, Any], conn: connection.Connection, _Con=con.Con): self.ipc_data = data - self.change = data['change'] - self.current = None - self.old = None + self.change: str = data['change'] + self.current: Optional[con.Con] = None + self.old: Optional[con.Con] = None if 'current' in data and data['current']: self.current = _Con(data['current'], None, conn) @@ -89,9 +90,9 @@ class OutputEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data - self.change = data['change'] + self.change: str = data['change'] class ModeEvent(IpcBaseEvent): @@ -107,10 +108,10 @@ class ModeEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data - self.change = data['change'] - self.pango_markup = data.get('pango_markup', False) + self.change: str = data['change'] + self.pango_markup: bool = data.get('pango_markup', False) class WindowEvent(IpcBaseEvent): @@ -126,9 +127,9 @@ class WindowEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data, conn, _Con=con.Con): + def __init__(self, data: dict[str, Any], conn: connection.Connection, _Con=con.Con): self.ipc_data = data - self.change = data['change'] + self.change: str = data['change'] self.container = _Con(data['container'], None, conn) @@ -185,13 +186,13 @@ class BindingInfo: :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data - self.command = data['command'] - self.event_state_mask = data.get('event_state_mask', []) - self.input_code = data['input_code'] - self.symbol = data.get('symbol', None) - self.input_type = data['input_type'] + self.command: str = data['command'] + self.event_state_mask: list[str] = data.get('event_state_mask', []) + self.input_code: int = data['input_code'] + self.symbol: Optional[str] = data.get('symbol', None) + self.input_type: str = data['input_type'] # sway only self.symbols = data.get('symbols', []) # not included in sway @@ -211,9 +212,9 @@ class BindingEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data - self.change = data['change'] + self.change: str = data['change'] self.binding = BindingInfo(data['binding']) @@ -228,9 +229,9 @@ class ShutdownEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data - self.change = data['change'] + self.change: str = data['change'] class TickEvent(IpcBaseEvent): @@ -248,11 +249,11 @@ class TickEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data # i3 didn't include the 'first' field in 4.15. See i3/i3#3271. - self.first = data.get('first', None) - self.payload = data['payload'] + self.first: Optional[bool] = data.get('first', None) + self.payload: str = data['payload'] class InputEvent(IpcBaseEvent): @@ -265,7 +266,7 @@ class InputEvent(IpcBaseEvent): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data - self.change = data['change'] + self.change: str = data['change'] self.input = InputReply(data['input']) diff --git a/i3ipc/model.py b/i3ipc/model.py index 759f063..de79455 100644 --- a/i3ipc/model.py +++ b/i3ipc/model.py @@ -1,3 +1,6 @@ +from typing import Any, Optional + + class Rect: """Used by other classes to represent rectangular position and dimensions. @@ -10,11 +13,11 @@ class Rect: :ivar width: The width of the rectangle. :vartype width: int """ - def __init__(self, data): - self.x = data['x'] - self.y = data['y'] - self.height = data['height'] - self.width = data['width'] + def __init__(self, data: dict[str, Any]): + self.x: int = data['x'] + self.y: int = data['y'] + self.height: int = data['height'] + self.width: int = data['width'] class OutputMode: @@ -27,19 +30,19 @@ class OutputMode: :vartype refresh: The refresh rate of the output in this mode. :vartype refresh: int """ - def __init__(self, data): - self.width = data['width'] - self.height = data['height'] - self.refresh = data['refresh'] + def __init__(self, data: dict[str, Any]): + self.width: int = data['width'] + self.height: int = data['height'] + self.refresh: int = data['refresh'] - def __getitem__(self, item): + def __getitem__(self, item: str): # for backwards compatability because this used to be a dict if not hasattr(self, item): raise KeyError(item) return getattr(self, item) @classmethod - def _parse_list(cls, data): + def _parse_list(cls, data: list[dict[str, Any]]) -> list['OutputMode']: return [cls(d) for d in data] @@ -59,10 +62,10 @@ class Gaps: :ivar bottom: The bottom outer gaps. :vartype bottom: int or :class:`None` if not supported. """ - def __init__(self, data): - self.inner = data['inner'] - self.outer = data['outer'] - self.left = data.get('left', None) - self.right = data.get('right', None) - self.top = data.get('top', None) - self.bottom = data.get('bottom', None) + def __init__(self, data: dict[str, Any]): + self.inner: int = data['inner'] + self.outer: int = data['outer'] + self.left: Optional[int] = data.get('left', None) + self.right: Optional[int] = data.get('right', None) + self.top: Optional[int] = data.get('top', None) + self.bottom: Optional[int] = data.get('bottom', None) diff --git a/i3ipc/py.typed b/i3ipc/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/i3ipc/replies.py b/i3ipc/replies.py index 8a2fa6a..9bb9cf2 100644 --- a/i3ipc/replies.py +++ b/i3ipc/replies.py @@ -1,8 +1,11 @@ from .model import Rect, OutputMode +from typing import Any, Callable, cast, TypeVar + +_BaseReplyType = TypeVar('_BaseReplyType', bound='_BaseReply') class _BaseReply: - def __init__(self, data): + def __init__(self, data: dict[str, Any]): self.ipc_data = data for member in self.__class__._members: value = data.get(member[0], None) @@ -12,8 +15,11 @@ def __init__(self, data): setattr(self, member[0], None) @classmethod - def _parse_list(cls, data): - return [cls(d) for d in data] + def _parse_list(cls, data: list[dict[str, Any]]) -> list[_BaseReplyType]: + return [cast(_BaseReplyType, cls(d)) for d in data] + + ipc_data: dict[str, Any] + _members: list[tuple[str, Callable]] class CommandReply(_BaseReply): @@ -28,6 +34,8 @@ class CommandReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + success: bool + error: str _members = [ ('success', bool), ('error', str), @@ -61,6 +69,12 @@ class WorkspaceReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + num: int + name: str + visible: bool + focused: bool + rect: Rect + output: str _members = [ ('num', int), ('name', str), @@ -114,6 +128,21 @@ class OutputReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + name: str + active: bool + primary: bool + current_workspace: str + rect: Rect + make: str + model: str + serial: str + scale: float + transform: str + max_render_time: int + focused: bool + dpms: bool + modes: list[OutputMode] + current_mode: OutputMode _members = [ ('name', str), ('active', bool), @@ -147,11 +176,11 @@ class BarConfigGaps: :ivar bottom: The gap on the bottom. :vartype bottom: int """ - def __init__(self, data): - self.left = data['left'] - self.right = data['right'] - self.top = data['top'] - self.bottom = data['bottom'] + def __init__(self, data: dict[str, Any]): + self.left: int = data['left'] + self.right: int = data['right'] + self.top: int = data['top'] + self.bottom: int = data['bottom'] class BarConfigReply(_BaseReply): @@ -179,19 +208,30 @@ class BarConfigReply(_BaseReply): :ivar colors: Contains key/value pairs of colors. Each value is a color code in hex, formatted #rrggbb (like in HTML). :vartype colors: dict - :ivar tray_padding: The tray is shown on the right-hand side of the bar. By default, a padding of 2 pixels is used for the upper, lower and right-hand side of the tray area and between the individual icons. + :ivar tray_padding: The tray is shown on the right-hand side of the bar. By default, a padding + of 2 pixels is used for the upper, lower and right-hand side of the tray area and between + the individual icons. :vartype tray_padding: int - :ivar hidden_state: In order to control whether i3bar is hidden or shown in hide mode, there exists the hidden_state option, which has no effect in dock mode or invisible mode. It indicates the current hidden_state of the bar: (1) The bar acts like in normal hide mode, it is hidden and is only unhidden in case of urgency hints or by pressing the modifier key (hide state), or (2) it is drawn on top of the currently visible workspace (show state). + :ivar hidden_state: In order to control whether i3bar is hidden or shown in hide mode, there + exists the hidden_state option, which has no effect in dock mode or invisible mode. It + indicates the current hidden_state of the bar: (1) The bar acts like in normal hide mode, + it is hidden and is only unhidden in case of urgency hints or by pressing the modifier key + (hide state), or (2) it is drawn on top of the currently visible workspace (show state). :vartype hidden_state: str :ivar modifier: The modifier used to switch between hide/show mode. :vartype modifier: int - :ivar separator_symbol: Specifies a custom symbol to be used for the separator as opposed to the vertical, one pixel thick separator. + :ivar separator_symbol: Specifies a custom symbol to be used for the separator as opposed to the + vertical, one pixel thick separator. :vartype separator_symbol: str :ivar workspace_min_width: :vartype workspace_min_width: int - :ivar strip_workspace_numbers: When strip_workspace_numbers is set to yes, any workspace that has a name of the form "[n][:][NAME]" will display only the name. You could use this, for instance, to display Roman numerals rather than digits by naming your workspaces to "2:I", "2:II", "3:III", "4:IV", ... + :ivar strip_workspace_numbers: When strip_workspace_numbers is set to yes, any workspace that + has a name of the form "[n][:][NAME]" will display only the name. You could use this, for + instance, to display Roman numerals rather than digits by naming your workspaces to "2:I", + "2:II", "3:III", "4:IV", ... :vartype strip_workspace_numbers: bool - :ivar strip_workspace_name: When strip_workspace_name is set to yes, any workspace that has a name of the form "[n][:][NAME]" will display only the number. + :ivar strip_workspace_name: When strip_workspace_name is set to yes, any workspace that has a + name of the form "[n][:][NAME]" will display only the number. :vartype strip_workspace_name: bool :ivar gaps: (sway only) :vartype gaps: :class:`BarConfigGaps` @@ -204,6 +244,26 @@ class BarConfigReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + id: int + tray_padding: int + hidden_state: str + mode: str + modifier: int + position: str + status_command: str + font: str + workspace_buttons: bool + workspace_min_width: int + strip_workspace_name: bool + strip_workspace_numbers: bool + binding_mode_indicator: bool + separator_symbol: str + verbose: bool + colors: dict + gaps: BarConfigGaps + bar_height: int + status_padding: int + status_edge_padding: int _members = [ ('id', str), ('tray_padding', int), @@ -247,6 +307,11 @@ class VersionReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + major: int + minor: int + patch: int + human_readable: str + loaded_config_file_name: str _members = [ ('major', int), ('minor', int), @@ -267,6 +332,7 @@ class ConfigReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + config: str _members = [ ('config', str), ] @@ -282,6 +348,7 @@ class TickReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + success: bool _members = [ ('success', bool), ] @@ -314,6 +381,14 @@ class InputReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ + name: str + vendor: int + product: int + type: str + xkb_active_layout_name: str + xkb_layout_names: list + xkb_active_layout_index: int + libinput: dict _members = [ ('identifier', str), ('name', str), @@ -345,5 +420,13 @@ class SeatReply(_BaseReply): :ivar ipc_data: The raw data from the i3 ipc. :vartype ipc_data: dict """ - _members = [('name', str), ('capabilities', int), ('focus', int), - ('devices', InputReply._parse_list)] + name: str + capabilities: int + focus: int + devices: list[InputReply] + _members = [ + ('name', str), + ('capabilities', int), + ('focus', int), + ('devices', InputReply._parse_list) + ]