Signal (Textualize#4012)

CHANGELOG.md

@@ -18,6 +18,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - Added `DOMNode.has_pseudo_classes` https://github.com/Textualize/textual/pull/3970
 - Added `Widget.allow_focus` and `Widget.allow_focus_children` https://github.com/Textualize/textual/pull/3989
+- Added `Query.blur` and `Query.focus` https://github.com/Textualize/textual/pull/4012
+- Added `MessagePump.message_queue_size` https://github.com/Textualize/textual/pull/4012
+- Added `TabbedContent.active_pane` https://github.com/Textualize/textual/pull/4012
 
 ### Fixed
 
@@ -27,6 +30,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - `SelectionList` option IDs are usable as soon as the widget is instantiated https://github.com/Textualize/textual/issues/3903
 - Fix issue with `Strip.crop` when crop window start aligned with strip end https://github.com/Textualize/textual/pull/3998
 - Fixed Strip.crop_extend https://github.com/Textualize/textual/pull/4011
+- Fixed declaration after nested rule set causing a parse error https://github.com/Textualize/textual/pull/4012
 - ID and class validation was too lenient https://github.com/Textualize/textual/issues/3954
 - Fixed a crash if the `TextArea` language was set but tree-sitter lanuage binaries were not installed https://github.com/Textualize/textual/issues/4045
 

src/textual/css/query.py

@@ -430,3 +430,28 @@ def refresh(
         for node in self:
             node.refresh(repaint=repaint, layout=layout)
         return self
+
+    def focus(self) -> DOMQuery[QueryType]:
+        """Focus the first matching node that permits focus.
+
+        Returns:
+            Query for chaining.
+        """
+        for node in self:
+            if node.allow_focus():
+                node.focus()
+                break
+        return self
+
+    def blur(self) -> DOMQuery[QueryType]:
+        """Blur the first matching node that is focused.
+
+        Returns:
+            Query for chaining.
+        """
+        focused = self._node.screen.focused
+        if focused is not None:
+            nodes: list[Widget] = list(self)
+            if focused in nodes:
+                self._node.screen._reset_focus(focused, avoiding=nodes)
+        return self

src/textual/css/tokenize.py

@@ -210,7 +210,7 @@ def __call__(self, code: str, read_from: CSSLocation) -> Iterable[Token]:
                 nest_level += 1
             elif name == "declaration_set_end":
                 nest_level -= 1
-                expect = expect_root_nested if nest_level else expect_root_scope
+                expect = expect_declaration if nest_level else expect_root_scope
                 yield token
                 continue
             expect = get_state(name, expect)

src/textual/css/tokenizer.py

@@ -248,11 +248,14 @@ def get_token(self, expect: Expect) -> Token:
         line = self.lines[line_no]
         match = expect.match(line, col_no)
         if match is None:
+            error_line = line[col_no:].rstrip()
+            error_message = (
+                f"{expect.description} (found {error_line.split(';')[0]!r})."
+            )
+            if not error_line.endswith(";"):
+                error_message += "; Did you forget a semicolon at the end of a line?"
             raise TokenError(
-                self.read_from,
-                self.code,
-                (line_no + 1, col_no + 1),
-                f"{expect.description} (found {line[col_no:].rstrip()!r}).; Did you forget a semicolon at the end of a line?",
+                self.read_from, self.code, (line_no + 1, col_no + 1), error_message
             )
         iter_groups = iter(match.groups())
 

src/textual/message.py

@@ -14,8 +14,8 @@
 from .case import camel_to_snake
 
 if TYPE_CHECKING:
+    from .dom import DOMNode
     from .message_pump import MessagePump
-    from .widget import Widget
 
 
 @rich.repr.auto
@@ -77,7 +77,7 @@ def __init_subclass__(
         cls.handler_name = f"on_{namespace}_{name}" if namespace else f"on_{name}"
 
     @property
-    def control(self) -> Widget | None:
+    def control(self) -> DOMNode | None:
         """The widget associated with this message, or None by default."""
         return None
 

src/textual/message_pump.py

@@ -188,6 +188,11 @@ def has_parent(self) -> bool:
         """Does this object have a parent?"""
         return self._parent is not None
 
+    @property
+    def message_queue_size(self) -> int:
+        """The current size of the message queue."""
+        return self._message_queue.qsize()
+
     @property
     def app(self) -> "App[object]":
         """

src/textual/signal.py

@@ -0,0 +1,88 @@
+"""
+Signals are a simple pub-sub mechanism.
+
+DOMNodes can subscribe to a signal, which will invoke a callback when the signal is published.
+
+This is experimental for now, for internal use. It may be part of the public API in a future release.
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from weakref import WeakKeyDictionary
+
+import rich.repr
+
+from textual import log
+
+if TYPE_CHECKING:
+    from ._types import IgnoreReturnCallbackType
+    from .dom import DOMNode
+
+
+class SignalError(Exception):
+    """Base class for a signal."""
+
+
+@rich.repr.auto(angular=True)
+class Signal:
+    """A signal that a widget may subscribe to, in order to invoke callbacks when an associated event occurs."""
+
+    def __init__(self, owner: DOMNode, name: str) -> None:
+        """Initialize a signal.
+
+        Args:
+            owner: The owner of this signal.
+            name: An identifier for debugging purposes.
+        """
+        self._owner = owner
+        self._name = name
+        self._subscriptions: WeakKeyDictionary[
+            DOMNode, list[IgnoreReturnCallbackType]
+        ] = WeakKeyDictionary()
+
+    def __rich_repr__(self) -> rich.repr.Result:
+        yield "owner", self._owner
+        yield "name", self._name
+        yield "subscriptions", list(self._subscriptions.keys())
+
+    def subscribe(self, node: DOMNode, callback: IgnoreReturnCallbackType) -> None:
+        """Subscribe a node to this signal.
+
+        When the signal is published, the callback will be invoked.
+
+        Args:
+            node: Node to subscribe.
+            callback: A callback function which takes no arguments, and returns anything (return type ignored).
+        """
+        if not node.is_running:
+            raise SignalError(
+                f"Node must be running to subscribe to a signal (has {node} been mounted)?"
+            )
+        callbacks = self._subscriptions.setdefault(node, [])
+        if callback not in callbacks:
+            callbacks.append(callback)
+
+    def unsubscribe(self, node: DOMNode) -> None:
+        """Unsubscribe a node from this signal.
+
+        Args:
+            node: Node to unsubscribe,
+        """
+        self._subscriptions.pop(node, None)
+
+    def publish(self) -> None:
+        """Publish the signal (invoke subscribed callbacks)."""
+
+        for node, callbacks in list(self._subscriptions.items()):
+            if not node.is_running:
+                # Removed nodes that are no longer running
+                self._subscriptions.pop(node)
+            else:
+                # Call callbacks
+                for callback in callbacks:
+                    try:
+                        callback()
+                    except Exception as error:
+                        log.error(f"error publishing signal to {node} ignored; {error}")

src/textual/widget.py

@@ -2806,10 +2806,10 @@ def _get_scrollable_region(self, region: Region) -> Region:
         scrollbar_size_horizontal = styles.scrollbar_size_horizontal
         scrollbar_size_vertical = styles.scrollbar_size_vertical
 
-        show_vertical_scrollbar: bool = (
+        show_vertical_scrollbar: bool = bool(
             show_vertical_scrollbar and scrollbar_size_vertical
         )
-        show_horizontal_scrollbar: bool = (
+        show_horizontal_scrollbar: bool = bool(
             show_horizontal_scrollbar and scrollbar_size_horizontal
         )
 
@@ -2843,10 +2843,10 @@ def _arrange_scrollbars(self, region: Region) -> Iterable[tuple[Widget, Region]]
         scrollbar_size_horizontal = self.scrollbar_size_horizontal
         scrollbar_size_vertical = self.scrollbar_size_vertical
 
-        show_vertical_scrollbar: bool = (
+        show_vertical_scrollbar: bool = bool(
             show_vertical_scrollbar and scrollbar_size_vertical
         )
-        show_horizontal_scrollbar: bool = (
+        show_horizontal_scrollbar: bool = bool(
             show_horizontal_scrollbar and scrollbar_size_horizontal
         )
 

src/textual/widgets/_tabbed_content.py

@@ -321,6 +321,11 @@ def __init__(
         self._initial = initial
         super().__init__(name=name, id=id, classes=classes, disabled=disabled)
 
+    @property
+    def active_pane(self) -> TabPane | None:
+        """The currently active pane, or `None` if no pane is active."""
+        return self.get_pane(self.active)
+
     def validate_active(self, active: str) -> str:
         """It doesn't make sense for `active` to be an empty string.
 

tests/snapshot_tests/snapshot_apps/nested_specificity.py

@@ -31,14 +31,15 @@ class NestedCSS(BaseTester):
     DEFAULT_CSS = """
     NestedCSS {
         width: 1fr;
-        height: 1fr;
-        background: green 10%;
-        border: blank;
+        height: 1fr;        
 
         &:focus {
             background: green 20%;
             border: round green;
         }
+
+        background: green 10%;
+        border: blank;
     }
     """
 

tests/test_message_pump.py

@@ -3,6 +3,7 @@
 from textual.app import App, ComposeResult
 from textual.errors import DuplicateKeyHandlers
 from textual.events import Key
+from textual.message import Message
 from textual.widget import Widget
 from textual.widgets import Input
 
@@ -70,6 +71,25 @@ def on_input_changed(self, event: Input.Changed) -> None:
         self.input_changed_events.append(event)
 
 
+async def test_message_queue_size():
+    """Test message queue size property."""
+    app = App()
+    assert app.message_queue_size == 0
+
+    class TestMessage(Message):
+        pass
+
+    async with app.run_test() as pilot:
+        assert app.message_queue_size == 0
+        app.post_message(TestMessage())
+        assert app.message_queue_size == 1
+        app.post_message(TestMessage())
+        assert app.message_queue_size == 2
+        # A pause will process all the messages
+        await pilot.pause()
+        assert app.message_queue_size == 0
+
+
 async def test_prevent() -> None:
     app = PreventTestApp()
 

tests/test_query.py

@@ -11,7 +11,7 @@
     WrongType,
 )
 from textual.widget import Widget
-from textual.widgets import Label
+from textual.widgets import Input, Label
 
 
 def test_query():
@@ -313,3 +313,33 @@ def compose(self):
     async with app.run_test() as pilot:
         app.query(MyWidget).refresh(repaint=args[0], layout=args[1])
         assert refreshes[-1] == args
+
+
+async def test_query_focus_blur():
+    class FocusApp(App):
+        AUTO_FOCUS = None
+
+        def compose(self) -> ComposeResult:
+            yield Input(id="foo")
+            yield Input(id="bar")
+            yield Input(id="baz")
+
+    app = FocusApp()
+    async with app.run_test() as pilot:
+        # Nothing focused
+        assert app.focused is None
+        # Focus first input
+        app.query(Input).focus()
+        await pilot.pause()
+        assert app.focused.id == "foo"
+        # Blur inputs
+        app.query(Input).blur()
+        await pilot.pause()
+        assert app.focused is None
+        # Focus another
+        app.query("#bar").focus()
+        await pilot.pause()
+        assert app.focused.id == "bar"
+        # Focus non existing
+        app.query("#egg").focus()
+        assert app.focused.id == "bar"