Parse VT inputs as bytes instead of strings (#83)

counterweight/_utils.py

@@ -52,14 +52,14 @@ def partition_int(total: int, weights: tuple[int]) -> list[int]:
     """Partition an integer into a list of integers, with each integer in the list corresponding to the weight at the same index in the weights list."""
     # https://stackoverflow.com/questions/62914824/c-sharp-split-integer-in-parts-given-part-weights-algorithm
 
+    if any(w < 0 for w in weights):
+        raise ValueError("Weights must be non-negative")
+
     if total == 0:  # optimization
         return [0] * len(weights)
 
     total_weight = sum(weights)
 
-    if not total_weight > 0:
-        raise ValueError("Total weight must be positive")
-
     partition = []
     accumulated_diff = 0.0
     for w in weights:

counterweight/app.py

@@ -38,9 +38,9 @@
 from counterweight.output import (
     CLEAR_SCREEN,
     paint_to_instructions,
-    start_mouse_reporting,
+    start_mouse_tracking,
     start_output_control,
-    stop_mouse_reporting,
+    stop_mouse_tracking,
     stop_output_control,
 )
 from counterweight.paint import Paint, paint_layout, svg
@@ -128,7 +128,7 @@ def put_event(event: AnyEvent) -> None:
         if not headless:
             start_handling_resize_signal(put_event=put_event)
             start_output_control(stream=output_stream)
-            start_mouse_reporting(stream=output_stream)
+            start_mouse_tracking(stream=output_stream)
 
             key_thread = Thread(target=read_keys, args=(input_stream, put_event), daemon=True)
             key_thread.start()
@@ -361,7 +361,7 @@ def handle_control(control: AnyControl | None) -> None:
         logger.info("Application stopping...")
 
         if not headless:
-            stop_mouse_reporting(stream=output_stream)
+            stop_mouse_tracking(stream=output_stream)
             stop_output_control(stream=output_stream)
             stop_input_control(stream=input_stream, original=original)
             stop_handling_resize_signal()

counterweight/cli.py

@@ -12,7 +12,7 @@
 from counterweight.events import AnyEvent
 from counterweight.input import read_keys, start_input_control, stop_input_control
 from counterweight.logging import tail_devlog
-from counterweight.output import start_mouse_reporting, stop_mouse_reporting
+from counterweight.output import start_mouse_tracking, stop_mouse_tracking
 
 cli = Typer(
     name=PACKAGE_NAME,
@@ -66,7 +66,7 @@ def put_event(event: AnyEvent) -> None:
 
     original = start_input_control(stream=input_stream)
     if mouse:
-        start_mouse_reporting(stream=output_stream)
+        start_mouse_tracking(stream=output_stream)
     try:
         while True:
             print("Waiting for input...")
@@ -77,4 +77,8 @@ def put_event(event: AnyEvent) -> None:
     finally:
         stop_input_control(stream=input_stream, original=original)
         if mouse:
-            stop_mouse_reporting(stream=output_stream)
+            stop_mouse_tracking(stream=output_stream)
+
+
+if __name__ == "__main__":
+    cli()

counterweight/events.py

@@ -1,41 +1,48 @@
-from time import monotonic_ns
+from dataclasses import dataclass
 from typing import Union
 
-from pydantic import Field
-
 from counterweight.geometry import Position
-from counterweight.types import FrozenForbidExtras
 
 
-class _Event(FrozenForbidExtras):
-    timestamp_ns: int = Field(default_factory=monotonic_ns)
+@dataclass(frozen=True)
+class _Event:
+    pass
 
 
+@dataclass(frozen=True)
 class TerminalResized(_Event):
     pass
 
 
+@dataclass(frozen=True)
 class KeyPressed(_Event):
     key: str
 
 
+@dataclass(frozen=True)
 class MouseMoved(_Event):
     position: Position
+    button: int | None
 
 
+@dataclass(frozen=True)
 class MouseDown(_Event):
     position: Position
     button: int
 
 
+@dataclass(frozen=True)
 class MouseUp(_Event):
     position: Position
+    button: int
 
 
+@dataclass(frozen=True)
 class StateSet(_Event):
     pass
 
 
+@dataclass(frozen=True)
 class Dummy(_Event):
     pass
 

counterweight/input.py

@@ -9,7 +9,6 @@
 from time import perf_counter_ns
 from typing import TextIO
 
-from parsy import ParseError
 from structlog import get_logger
 
 from counterweight.events import AnyEvent
@@ -31,29 +30,24 @@ def read_keys(stream: TextIO, put_event: Callable[[AnyEvent], None]) -> None:
         # so we can dispense with the ceremony of actually using the results of the select.
 
         start_parsing = perf_counter_ns()
-        b = os.read(stream.fileno(), 1_000)
-        bytes = list(b)
-        buffer = b.decode("utf-8")
+        bytes = os.read(stream.fileno(), 1_000)
         try:
-            inputs = vt_inputs.parse(buffer)
+            inputs = vt_inputs.parse(bytes)
 
             for i in inputs:
                 put_event(i)
 
             logger.debug(
                 "Parsed user input",
                 inputs=inputs,
-                buffer=repr(buffer),
                 bytes=bytes,
-                len_buffer=len(buffer),
                 elapsed_ns=f"{perf_counter_ns() - start_parsing:_}",
             )
-        except (ParseError, KeyError) as e:
+        except Exception as e:
             logger.error(
                 "Failed to parse input",
-                error=str(e),
-                buffer=repr(buffer),
-                len_buffer=len(buffer),
+                error=repr(e),
+                bytes=bytes,
                 elapsed_ns=f"{perf_counter_ns() - start_parsing:_}",
             )