-
Notifications
You must be signed in to change notification settings - Fork 818
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
54cec67
commit 2f2f314
Showing
6 changed files
with
310 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,34 @@ | ||
from __future__ import annotations | ||
|
||
from typing import NamedTuple | ||
|
||
from textual.geometry import Region, Size, Spacing | ||
|
||
|
||
class MapGeometry(NamedTuple): | ||
"""Defines the absolute location of a Widget.""" | ||
|
||
region: Region | ||
"""The (screen) [region][textual.geometry.Region] occupied by the widget.""" | ||
order: tuple[tuple[int, int, int], ...] | ||
"""Tuple of tuples defining the painting order of the widget. | ||
Each successive triple represents painting order information with regards to | ||
ancestors in the DOM hierarchy and the last triple provides painting order | ||
information for this specific widget. | ||
""" | ||
clip: Region | ||
"""A [region][textual.geometry.Region] to clip the widget by (if a Widget is within a container).""" | ||
virtual_size: Size | ||
"""The virtual [size][textual.geometry.Size] (scrollable area) of a widget if it is a container.""" | ||
container_size: Size | ||
"""The container [size][textual.geometry.Size] (area not occupied by scrollbars).""" | ||
virtual_region: Region | ||
"""The [region][textual.geometry.Region] relative to the container (but not necessarily visible).""" | ||
dock_gutter: Spacing | ||
"""Space from the container reserved by docked widgets.""" | ||
|
||
@property | ||
def visible_region(self) -> Region: | ||
"""The Widget region after clipping.""" | ||
return self.clip.intersection(self.region) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,75 @@ | ||
"""A command palette command provider for Textual system commands. | ||
This is a simple command provider that makes the most obvious application | ||
actions available via the [command palette][textual.command.CommandPalette]. | ||
""" | ||
|
||
from __future__ import annotations | ||
|
||
from .command import DiscoveryHit, Hit, Hits, Provider | ||
from .types import IgnoreReturnCallbackType | ||
|
||
|
||
class SystemCommands(Provider): | ||
"""A [source][textual.command.Provider] of command palette commands that run app-wide tasks. | ||
Used by default in [`App.COMMANDS`][textual.app.App.COMMANDS]. | ||
""" | ||
|
||
@property | ||
def _system_commands(self) -> tuple[tuple[str, IgnoreReturnCallbackType, str], ...]: | ||
"""The system commands to reveal to the command palette.""" | ||
return ( | ||
( | ||
"Toggle light/dark mode", | ||
self.app.action_toggle_dark, | ||
"Toggle the application between light and dark mode", | ||
), | ||
( | ||
"Quit the application", | ||
self.app.action_quit, | ||
"Quit the application as soon as possible", | ||
), | ||
( | ||
"Ring the bell", | ||
self.app.action_bell, | ||
"Ring the terminal's 'bell'", | ||
), | ||
) | ||
|
||
async def discover(self) -> Hits: | ||
"""Handle a request for the discovery commands for this provider. | ||
Yields: | ||
Commands that can be discovered. | ||
""" | ||
for name, runnable, help_text in self._system_commands: | ||
yield DiscoveryHit( | ||
name, | ||
runnable, | ||
help=help_text, | ||
) | ||
|
||
async def search(self, query: str) -> Hits: | ||
"""Handle a request to search for system commands that match the query. | ||
Args: | ||
query: The user input to be matched. | ||
Yields: | ||
Command hits for use in the command palette. | ||
""" | ||
# We're going to use Textual's builtin fuzzy matcher to find | ||
# matching commands. | ||
matcher = self.matcher(query) | ||
|
||
# Loop over all applicable commands, find those that match and offer | ||
# them up to the command palette. | ||
for name, runnable, help_text in self._system_commands: | ||
if (match := matcher.match(name)) > 0: | ||
yield Hit( | ||
match, | ||
matcher.highlight(name), | ||
runnable, | ||
help=help_text, | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,181 @@ | ||
""" | ||
A class to manage [workers](/guide/workers) for an app. | ||
You access this object via [App.workers][textual.app.App.workers] or [Widget.workers][textual.dom.DOMNode.workers]. | ||
""" | ||
|
||
from __future__ import annotations | ||
|
||
import asyncio | ||
from collections import Counter | ||
from operator import attrgetter | ||
from typing import TYPE_CHECKING, Any, Iterable, Iterator | ||
|
||
import rich.repr | ||
|
||
from .worker import Worker, WorkerState, WorkType | ||
|
||
if TYPE_CHECKING: | ||
from .app import App | ||
from .dom import DOMNode | ||
|
||
|
||
@rich.repr.auto(angular=True) | ||
class WorkerManager: | ||
"""An object to manager a number of workers. | ||
You will not have to construct this class manually, as widgets, screens, and apps | ||
have a worker manager accessibly via a `workers` attribute. | ||
""" | ||
|
||
def __init__(self, app: App) -> None: | ||
"""Initialize a worker manager. | ||
Args: | ||
app: An App instance. | ||
""" | ||
self._app = app | ||
"""A reference to the app.""" | ||
self._workers: set[Worker] = set() | ||
"""The workers being managed.""" | ||
|
||
def __rich_repr__(self) -> rich.repr.Result: | ||
counter: Counter[WorkerState] = Counter() | ||
counter.update(worker.state for worker in self._workers) | ||
for state, count in sorted(counter.items()): | ||
yield state.name, count | ||
|
||
def __iter__(self) -> Iterator[Worker[Any]]: | ||
return iter(sorted(self._workers, key=attrgetter("_created_time"))) | ||
|
||
def __reversed__(self) -> Iterator[Worker[Any]]: | ||
return iter( | ||
sorted(self._workers, key=attrgetter("_created_time"), reverse=True) | ||
) | ||
|
||
def __bool__(self) -> bool: | ||
return bool(self._workers) | ||
|
||
def __len__(self) -> int: | ||
return len(self._workers) | ||
|
||
def __contains__(self, worker: object) -> bool: | ||
return worker in self._workers | ||
|
||
def add_worker( | ||
self, worker: Worker, start: bool = True, exclusive: bool = True | ||
) -> None: | ||
"""Add a new worker. | ||
Args: | ||
worker: A Worker instance. | ||
start: Start the worker if True, otherwise the worker must be started manually. | ||
exclusive: Cancel all workers in the same group as `worker`. | ||
""" | ||
if exclusive and worker.group: | ||
self.cancel_group(worker.node, worker.group) | ||
self._workers.add(worker) | ||
if start: | ||
worker._start(self._app, self._remove_worker) | ||
|
||
def _new_worker( | ||
self, | ||
work: WorkType, | ||
node: DOMNode, | ||
*, | ||
name: str | None = "", | ||
group: str = "default", | ||
description: str = "", | ||
exit_on_error: bool = True, | ||
start: bool = True, | ||
exclusive: bool = False, | ||
thread: bool = False, | ||
) -> Worker: | ||
"""Create a worker from a function, coroutine, or awaitable. | ||
Args: | ||
work: A callable, a coroutine, or other awaitable. | ||
name: A name to identify the worker. | ||
group: The worker group. | ||
description: A description of the worker. | ||
exit_on_error: Exit the app if the worker raises an error. Set to `False` to suppress exceptions. | ||
start: Automatically start the worker. | ||
exclusive: Cancel all workers in the same group. | ||
thread: Mark the worker as a thread worker. | ||
Returns: | ||
A Worker instance. | ||
""" | ||
worker: Worker[Any] = Worker( | ||
node, | ||
work, | ||
name=name or getattr(work, "__name__", "") or "", | ||
group=group, | ||
description=description or repr(work), | ||
exit_on_error=exit_on_error, | ||
thread=thread, | ||
) | ||
self.add_worker(worker, start=start, exclusive=exclusive) | ||
return worker | ||
|
||
def _remove_worker(self, worker: Worker) -> None: | ||
"""Remove a worker from the manager. | ||
Args: | ||
worker: A Worker instance. | ||
""" | ||
self._workers.discard(worker) | ||
|
||
def start_all(self) -> None: | ||
"""Start all the workers.""" | ||
for worker in self._workers: | ||
worker._start(self._app, self._remove_worker) | ||
|
||
def cancel_all(self) -> None: | ||
"""Cancel all workers.""" | ||
for worker in self._workers: | ||
worker.cancel() | ||
|
||
def cancel_group(self, node: DOMNode, group: str) -> list[Worker]: | ||
"""Cancel a single group. | ||
Args: | ||
node: Worker DOM node. | ||
group: A group name. | ||
Returns: | ||
A list of workers that were cancelled. | ||
""" | ||
workers = [ | ||
worker | ||
for worker in self._workers | ||
if (worker.group == group and worker.node == node) | ||
] | ||
for worker in workers: | ||
worker.cancel() | ||
return workers | ||
|
||
def cancel_node(self, node: DOMNode) -> list[Worker]: | ||
"""Cancel all workers associated with a given node | ||
Args: | ||
node: A DOM node (widget, screen, or App). | ||
Returns: | ||
List of cancelled workers. | ||
""" | ||
workers = [worker for worker in self._workers if worker.node == node] | ||
for worker in workers: | ||
worker.cancel() | ||
return workers | ||
|
||
async def wait_for_complete(self, workers: Iterable[Worker] | None = None) -> None: | ||
"""Wait for workers to complete. | ||
Args: | ||
workers: An iterable of workers or None to wait for all workers in the manager. | ||
""" | ||
try: | ||
await asyncio.gather(*[worker.wait() for worker in (workers or self)]) | ||
except asyncio.CancelledError: | ||
pass |