Implement (Async)?Server.dump_tree() (#395)

* Implement (Async)?Server.dump_tree()

* Extend sync timeout

* Debug /g_dumpTree errors

* Flaky tests

* Remove spurious prints

supriya/contexts/core.py

@@ -411,11 +411,11 @@ def _setup_allocators(self) -> None:
         self._sync_id = self._sync_id_minimum
 
     @abc.abstractmethod
-    def _validate_can_request(self):
+    def _validate_can_request(self) -> None:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def _validate_moment_timestamp(self, seconds: Optional[float]):
+    def _validate_moment_timestamp(self, seconds: Optional[float]) -> None:
         raise NotImplementedError
 
     ### PUBLIC METHODS ###
@@ -728,7 +728,7 @@ def copy_buffer(
         source_starting_frame: int,
         target_starting_frame: int,
         frame_count: int,
-    ):
+    ) -> None:
         """
         Copy a buffer.
 
@@ -752,7 +752,7 @@ def copy_buffer(
         )
         self._add_requests(request)
 
-    def do_nothing(self):
+    def do_nothing(self) -> None:
         """
         Emit a no-op "nothing" command.
         """

supriya/contexts/realtime.py

@@ -64,6 +64,7 @@
     Synth,
 )
 from .requests import (
+    DumpTree,
     GetBuffer,
     GetBufferRange,
     GetControlBus,
@@ -678,6 +679,32 @@ def disconnect(self) -> "Server":
         self._disconnect()
         return self
 
+    def dump_tree(
+        self,
+        group: Optional[Group] = None,
+        include_controls: bool = True,
+        sync: bool = True,
+    ) -> Optional[QueryTreeGroup]:
+        """
+        Dump the server's node tree.
+
+        Emit ``/g_dumpTree`` requests.
+
+        :param group: The group whose tree to query. Defaults to the root node.
+        :param include_controls: Flag for including synth control values.
+        :param sync: If true, communicate the request immediately. Otherwise bundle it
+            with the current request context.
+        """
+        self._validate_can_request()
+        request = DumpTree(items=[(group or self.root_node, bool(include_controls))])
+        if sync:
+            with self.process_protocol.capture() as transcript:
+                request.communicate(server=self)
+                self.sync(timeout=10.0)
+                return QueryTreeGroup.from_string("\n".join(transcript.lines))
+        self._add_requests(request)
+        return None
+
     def get_buffer(
         self, buffer: Buffer, *indices: int, sync: bool = True
     ) -> Optional[Dict[int, float]]:
@@ -952,7 +979,7 @@ def reset(self) -> "Server":
         self.sync()
         return self
 
-    def sync(self, sync_id: Optional[int] = None) -> "Server":
+    def sync(self, sync_id: Optional[int] = None, timeout: float = 1.0) -> "Server":
         """
         Sync the server.
 
@@ -964,7 +991,7 @@ def sync(self, sync_id: Optional[int] = None) -> "Server":
             raise ServerOffline
         Sync(
             sync_id=sync_id if sync_id is not None else self._get_next_sync_id()
-        ).communicate(server=self)
+        ).communicate(server=self, timeout=timeout)
         return self
 
 
@@ -1188,6 +1215,32 @@ async def disconnect(self) -> "AsyncServer":
         await self._disconnect()
         return self
 
+    async def dump_tree(
+        self,
+        group: Optional[Group] = None,
+        include_controls: bool = True,
+        sync: bool = True,
+    ) -> Optional[QueryTreeGroup]:
+        """
+        Dump the server's node tree.
+
+        Emit ``/g_dumpTree`` requests.
+
+        :param group: The group whose tree to query. Defaults to the root node.
+        :param include_controls: Flag for including synth control values.
+        :param sync: If true, communicate the request immediately. Otherwise bundle it
+            with the current request context.
+        """
+        self._validate_can_request()
+        request = DumpTree(items=[(group or self.root_node, bool(include_controls))])
+        if sync:
+            with self.process_protocol.capture() as transcript:
+                await request.communicate_async(server=self)
+                await self.sync(timeout=10.0)
+                return QueryTreeGroup.from_string("\n".join(transcript.lines))
+        self._add_requests(request)
+        return None
+
     async def get_buffer(
         self, buffer: Buffer, *indices: int, sync: bool = True
     ) -> Optional[Dict[int, float]]:
@@ -1468,7 +1521,9 @@ async def reset(self) -> "AsyncServer":
         await self.sync()
         return self
 
-    async def sync(self, sync_id: Optional[int] = None) -> "AsyncServer":
+    async def sync(
+        self, sync_id: Optional[int] = None, timeout: float = 1.0
+    ) -> "AsyncServer":
         """
         Sync the server.
 
@@ -1480,5 +1535,5 @@ async def sync(self, sync_id: Optional[int] = None) -> "AsyncServer":
             raise ServerOffline
         await Sync(
             sync_id=sync_id if sync_id is not None else self._get_next_sync_id()
-        ).communicate_async(server=self)
+        ).communicate_async(server=self, timeout=timeout)
         return self

supriya/contexts/requests.py

@@ -439,15 +439,15 @@ class DumpTree(Request):
         >>> from supriya.contexts.requests import DumpTree
         >>> request = DumpTree(items=[(0, True)])
         >>> request.to_osc()
-        OscMessage('/g_dumpTree', 0, True)
+        OscMessage('/g_dumpTree', 0, 1)
     """
 
     items: Sequence[Tuple[SupportsInt, bool]]
 
     def to_osc(self) -> OscMessage:
         contents = []
         for node_id, flag in self.items:
-            contents.extend([int(node_id), bool(flag)])
+            contents.extend([int(node_id), int(flag)])
         return OscMessage(RequestName.GROUP_DUMP_TREE, *contents)
 
 

supriya/contexts/responses.py

@@ -3,6 +3,7 @@
 """
 
 import dataclasses
+import re
 from collections import deque
 from typing import Deque, Dict, List, Optional, Sequence, Tuple, Type, Union
 
@@ -372,7 +373,9 @@ def _get_str_format_pieces(self, unindexed=False):
 @dataclasses.dataclass
 class QueryTreeGroup:
     node_id: int
-    children: List[Union["QueryTreeGroup", QueryTreeSynth]]
+    children: List[Union["QueryTreeGroup", QueryTreeSynth]] = dataclasses.field(
+        default_factory=list
+    )
 
     ### SPECIAL METHODS ###
 
@@ -428,6 +431,45 @@ def recurse(
             deque(response.items),
         )
 
+    @classmethod
+    def from_string(cls, string) -> "QueryTreeGroup":
+        node_pattern = re.compile(r"^\s*(\d+) (\S+)$")
+        control_pattern = re.compile(r"\w+: \S+")
+        lines = string.splitlines()
+        if not lines[0].startswith("NODE TREE"):
+            raise ValueError
+        stack: List[QueryTreeGroup] = [
+            QueryTreeGroup(node_id=int(lines.pop(0).rpartition(" ")[-1]))
+        ]
+        for line in lines:
+            indent = line.count("   ")
+            if match := (node_pattern.match(line)):
+                while len(stack) > indent:
+                    stack.pop()
+                node_id = int(match.groups()[0])
+                if (name := match.groups()[1]) == "group":
+                    stack[-1].children.append(group := QueryTreeGroup(node_id=node_id))
+                    stack.append(group)
+                else:
+                    stack[-1].children.append(
+                        synth := QueryTreeSynth(node_id=node_id, synthdef_name=name)
+                    )
+            else:
+                for pair in control_pattern.findall(line):
+                    name_string, _, value_string = pair.partition(": ")
+                    try:
+                        name_or_index: Union[int, str] = int(name_string)
+                    except ValueError:
+                        name_or_index = name_string
+                    try:
+                        value: Union[float, str] = float(value_string)
+                    except ValueError:
+                        value = value_string
+                    synth.controls.append(
+                        QueryTreeControl(name_or_index=name_or_index, value=value)
+                    )
+        return stack[0]
+
 
 @dataclasses.dataclass
 class StatusInfo(Response):

supriya/patterns/testutils.py

@@ -75,8 +75,4 @@ def run_pattern_test(pattern, expected, is_infinite, stop_at):
         sanitized_actual = sanitize(actual[: len(expected)])
     else:
         sanitized_actual = sanitize(actual)
-    # expected_string = "\n".join(repr(x) for x in expected)
-    # actual_string = "\n".join(repr(x) for x in sanitized_actual)
-    # for line in unified_diff(expected_string.splitlines(), actual_string.splitlines()):
-    #     print(line)
     assert sanitized_actual == expected, sanitized_actual

supriya/scsynth.py

@@ -10,7 +10,7 @@
 import threading
 from dataclasses import dataclass
 from pathlib import Path
-from typing import IO, Callable, Dict, List, Optional, Tuple, cast
+from typing import IO, Callable, Dict, Iterator, List, Optional, Set, Tuple, cast
 
 import psutil
 import uqbar.io
@@ -260,6 +260,26 @@ class BootStatus(enum.IntEnum):
     QUITTING = 3
 
 
+class Capture:
+
+    def __init__(self, process_protocol: "ProcessProtocol") -> None:
+        self.process_protocol = process_protocol
+        self.lines: List[str] = []
+
+    def __enter__(self) -> "Capture":
+        self.process_protocol.captures.add(self)
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
+        self.process_protocol.captures.remove(self)
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self.lines)
+
+    def __len__(self) -> int:
+        return len(self.lines)
+
+
 class ProcessProtocol:
     def __init__(
         self,
@@ -270,6 +290,7 @@ def __init__(
         on_quit_callback: Optional[Callable] = None,
     ) -> None:
         self.buffer_ = ""
+        self.captures: Set[Capture] = set()
         self.error_text = ""
         self.name = name
         self.on_boot_callback = on_boot_callback
@@ -310,6 +331,8 @@ def _handle_data_received(
         if "\n" in text:
             text, _, self.buffer_ = text.rpartition("\n")
             for line in text.splitlines():
+                for capture in self.captures:
+                    capture.lines.append(line)
                 line_status = self._parse_line(line)
                 if line_status == LineStatus.READY:
                     boot_future.set_result(True)
@@ -356,6 +379,9 @@ def _quit(self) -> bool:
         self.status = BootStatus.QUITTING
         return True
 
+    def capture(self) -> Capture:
+        return Capture(self)
+
 
 class SyncProcessProtocol(ProcessProtocol):
     def __init__(
@@ -401,15 +427,10 @@ def _run_process_thread(self, options: Options) -> None:
             self.on_panic_callback()
 
     def _run_read_thread(self) -> None:
-        while self.status == BootStatus.BOOTING:
+        while self.status in (BootStatus.BOOTING, BootStatus.ONLINE):
             if not (text := cast(IO[bytes], self.process.stdout).readline().decode()):
                 continue
             _, _ = self._handle_data_received(boot_future=self.boot_future, text=text)
-        while self.status == BootStatus.ONLINE:
-            if not (text := cast(IO[bytes], self.process.stdout).readline().decode()):
-                continue
-            # we can capture /g_dumpTree output here
-            # do something
 
     def _shutdown(self) -> None:
         self.process.terminate()

supriya/ugens/core.py

@@ -5935,11 +5935,7 @@ def _remap_controls(
         return ugens
 
     def _sort_topologically(self, ugens: List[UGen]) -> List[UGen]:
-        try:
-            sort_bundles = self._initiate_topological_sort(ugens)
-        except Exception:
-            print(ugens)
-            raise
+        sort_bundles = self._initiate_topological_sort(ugens)
         available_ugens: List[UGen] = []
         output_stack: List[UGen] = []
         for ugen in reversed(ugens):