RM Clock.default()

supriya/clocks/__init__.py

@@ -1,6 +1,6 @@
 from .asynchronous import AsyncClock
-from .bases import BaseClock
-from .ephemera import (
+from .core import (
+    BaseClock,
     CallbackEvent,
     ChangeEvent,
     ClockContext,

supriya/clocks/asynchronous.py

@@ -4,9 +4,9 @@
 import traceback
 from typing import Awaitable, Optional, Tuple
 
-from .bases import BaseClock
-from .ephemera import (
+from .core import (
     Action,
+    BaseClock,
     CallbackEvent,
     ChangeEvent,
     ClockContext,
@@ -19,6 +19,12 @@
 
 
 class AsyncClock(BaseClock):
+    """
+    An async clock.
+    """
+
+    ### INITIALIZER ###
+
     def __init__(self) -> None:
         BaseClock.__init__(self)
         self._task: Optional[Awaitable[None]] = None

supriya/clocks/bases.py → supriya/clocks/core.py

@@ -1,34 +1,207 @@
 import collections
 import dataclasses
+import enum
 import fractions
 import itertools
 import logging
 import queue
 import time
 import traceback
-from typing import Any, Callable, Deque, Dict, FrozenSet, Optional, Set, Tuple, Union
+from functools import total_ordering
+from typing import (
+    Any,
+    Callable,
+    Deque,
+    Dict,
+    FrozenSet,
+    List,
+    Literal,
+    NamedTuple,
+    Optional,
+    Set,
+    Tuple,
+    Union,
+)
 
 from .. import conversions
-from .ephemera import (
-    Action,
-    CallbackCommand,
-    CallbackEvent,
-    ChangeCommand,
-    ChangeEvent,
-    ClockContext,
-    ClockState,
-    Command,
-    Event,
-    EventType,
-    Moment,
-    Quantization,
-    TimeUnit,
-)
-from .eventqueue import EventQueue
 
 logger = logging.getLogger(__name__)
 
 
+Quantization = Literal[
+    "8M",
+    "4M",
+    "2M",
+    "1M",
+    "1/2",
+    "1/2T",
+    "1/4",
+    "1/4T",
+    "1/8",
+    "1/8T",
+    "1/16",
+    "1/16T",
+    "1/32",
+    "1/32T",
+    "1/64",
+    "1/64T",
+    "1/128",
+]
+
+
+class EventType(enum.IntEnum):
+    CHANGE = 0
+    SCHEDULE = 1
+
+
+class TimeUnit(enum.IntEnum):
+    BEATS = 0
+    SECONDS = 1
+    MEASURES = 2
+
+
+class ClockState(NamedTuple):
+    beats_per_minute: float
+    initial_seconds: float
+    previous_measure: int
+    previous_offset: float
+    previous_seconds: float
+    previous_time_signature_change_offset: float
+    time_signature: Tuple[int, int]
+
+
+@dataclasses.dataclass(frozen=True)
+class Moment:
+    __slots__ = (
+        "beats_per_minute",
+        "measure",
+        "measure_offset",
+        "offset",
+        "seconds",
+        "time_signature",
+    )
+    beats_per_minute: float
+    measure: int
+    measure_offset: float
+    offset: float  # the beat since zero
+    seconds: float  # the seconds since zero
+    time_signature: Tuple[int, int]
+
+
+@dataclasses.dataclass(frozen=True)
+class Action:
+    event_id: int
+    event_type: int
+
+
+@dataclasses.dataclass(frozen=True)
+class Command(Action):
+    quantization: Optional[Quantization]
+    schedule_at: float
+    time_unit: Optional[TimeUnit]
+
+
+@dataclasses.dataclass(frozen=True)
+class CallbackCommand(Command):
+    args: Optional[Tuple]
+    kwargs: Optional[Dict]
+    procedure: Callable[["ClockContext"], Union[None, float, Tuple[float, TimeUnit]]]
+
+
+@dataclasses.dataclass(frozen=True)
+class ChangeCommand(Command):
+    beats_per_minute: Optional[float]
+    time_signature: Optional[Tuple[int, int]]
+
+
+@total_ordering
+@dataclasses.dataclass(frozen=True, eq=False)
+class Event(Action):
+    seconds: float
+    measure: Optional[int]
+    offset: Optional[float]
+
+    def __eq__(self, other: object) -> bool:
+        # Need to act like a tuple here
+        if not isinstance(other, Event):
+            return NotImplemented
+        return (self.seconds, self.event_type, self.event_id) == (
+            other.seconds,
+            other.event_type,
+            other.event_id,
+        )
+
+    def __lt__(self, other: object) -> bool:
+        # Need to act like a tuple here
+        if not isinstance(other, Event):
+            return NotImplemented
+        return (self.seconds, self.event_type, self.event_id) < (
+            other.seconds,
+            other.event_type,
+            other.event_id,
+        )
+
+
+@dataclasses.dataclass(frozen=True, eq=False)
+class CallbackEvent(Event):
+    procedure: Callable[["ClockContext"], Union[None, float, Tuple[float, TimeUnit]]]
+    args: Optional[Tuple]
+    kwargs: Optional[Dict]
+    invocations: int
+
+    def __hash__(self) -> int:
+        return hash((type(self), self.event_id))
+
+
+@dataclasses.dataclass(frozen=True, eq=False)
+class ChangeEvent(Event):
+    beats_per_minute: Optional[float]
+    time_signature: Optional[Tuple[int, int]]
+
+    def __hash__(self) -> int:
+        return hash((type(self), self.event_id))
+
+
+class ClockContext(NamedTuple):
+    current_moment: Moment
+    desired_moment: Moment
+    event: Union[CallbackEvent, ChangeEvent]
+
+
+class _EventQueue(queue.PriorityQueue[Event]):
+    ### PRIVATE METHODS ###
+
+    def _init(self, maxsize: Optional[int]) -> None:
+        self.queue: List[Event] = []
+        self.flags: Dict[Event, bool] = {}
+
+    def _put(self, event: Event) -> None:
+        self.flags[event] = True
+        super()._put(event)
+
+    def _get(self) -> Event:
+        while self.queue:
+            if not self.flags.pop((event := super()._get()), None):
+                continue
+            return event
+        raise queue.Empty
+
+    ### PUBLIC METHODS ###
+
+    def clear(self) -> None:
+        with self.mutex:
+            self._init(None)
+
+    def peek(self) -> Event:
+        with self.mutex:
+            self._put(event := self._get())
+        return event
+
+    def remove(self, event: Event) -> None:
+        with self.mutex:
+            self.flags.pop(event, None)
+
+
 class BaseClock:
     ### CLASS VARIABLES ###
 
@@ -60,7 +233,7 @@ def __init__(self) -> None:
         self._name = None
         self._counter = itertools.count()
         self._command_deque: Deque[Command] = collections.deque()
-        self._event_queue = EventQueue()
+        self._event_queue = _EventQueue()
         self._is_running = False
         self._slop = 0.001
         self._actions_by_id: Dict[int, Action] = {}