Clean: Remove Union type from codebase

I have removed the verbose 'Union' from typing and replaced it by the "|"
symbol that denotes the same concept.

sardine/UserConfig.py

@@ -1,7 +1,6 @@
 import json
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Union
 
 from appdirs import *
 from rich import print
@@ -48,7 +47,7 @@ def _recursive_update(dest: dict, src: dict):
 
 @dataclass
 class Config:
-    midi: Union[str, None]
+    midi: str | None
     beats: int
     parameters: list
     bpm: int

sardine_core/base/clock.py

@@ -1,6 +1,6 @@
 import math
 from abc import ABC, abstractmethod
-from typing import Optional, Union
+from typing import Optional
 
 from .runner import BaseRunnerHandler
 
@@ -191,15 +191,15 @@ def can_sleep(self) -> bool:
 
     def get_beat_time(
         self,
-        n_beats: Union[int, float],
+        n_beats: int | float,
         *,
         time: Optional[float] = None,
         sync: bool = True,
     ) -> float:
         """Determines the amount of time to wait for N beats to pass.
 
         Args:
-            n_beats (Union[int, float]): The number of beats to wait for.
+            n_beats (int | float): The number of beats to wait for.
             time (Optional[float]):
                 The exact time to use for calculations.
                 If not provided, this defaults to `shifted_time`.
@@ -237,15 +237,15 @@ def get_beat_time(
 
     def get_bar_time(
         self,
-        n_bars: Union[int, float],
+        n_bars: int | float,
         *,
         time: Optional[float] = None,
         sync: bool = True,
     ) -> float:
         """Determines the amount of time to wait for N bars to pass.
 
         Args:
-            n_bars (Union[int, float]): The number of bars to wait for.
+            n_bars (int | float): The number of bars to wait for.
             time (Optional[float]):
                 The exact time to use for calculations.
                 If not provided, this defaults to `shifted_time`.
@@ -261,7 +261,7 @@ def get_bar_time(
         """
         return self.get_beat_time(n_bars * self.beats_per_bar, time=time, sync=sync)
 
-    async def sleep(self, duration: Union[float, int]) -> None:
+    async def sleep(self, duration: float | int) -> None:
         """Sleeps for the given duration.
 
         This method can be optionally overridden by subclasses.

sardine_core/clock/internal_clock.py

@@ -1,11 +1,10 @@
 import asyncio
 import math
 import time
-from typing import Union
 
 from sardine_core.base import BaseClock
 
-NUMBER = Union[int, float]
+NUMBER = int | float
 
 __all__ = ("InternalClock",)
 
@@ -132,7 +131,7 @@ def tempo(self, new_tempo: NUMBER):
 
     ## METHODS  ##############################################################
 
-    async def sleep(self, duration: Union[float, int]) -> None:
+    async def sleep(self, duration: NUMBER) -> None:
         return await asyncio.sleep(duration)
 
     async def run(self):

sardine_core/clock/link_clock.py

@@ -1,10 +1,10 @@
-from typing import Optional, Union
+from typing import Optional
 
 import link
 
 from sardine_core.base import BaseClock, BaseThreadedLoopMixin
 
-NUMBER = Union[int, float]
+NUMBER = int | float
 
 __all__ = ("LinkClock",)
 

sardine_core/fish_bowl.py

@@ -1,6 +1,6 @@
 import asyncio
 import collections
-from typing import Hashable, Iterable, Optional, Protocol, Union
+from typing import Hashable, Iterable, Optional, Protocol
 
 from exceptiongroup import BaseExceptionGroup
 
@@ -77,7 +77,9 @@ def __repr__(self) -> str:
         status = (
             "playing"
             if running and not paused
-            else "paused" if running and paused else "stopped"
+            else "paused"
+            if running and paused
+            else "stopped"
         )
 
         return "<{} {} clock={!r}>".format(
@@ -164,14 +166,14 @@ def is_running(self):
 
     ## SLEEPING MANAGEMENT ############################################################
 
-    async def sleep(self, duration: Union[int, float]):
+    async def sleep(self, duration: int | float):
         """Sleeps for the given duration.
 
         This method is simply a shorthand for `self.sleeper.sleep(duration)`.
         """
         return await self.sleeper.sleep(duration)
 
-    async def sleep_beats(self, beats: Union[int, float]):
+    async def sleep_beats(self, beats: int | float):
         """Sleeps for the given number of beats."""
         return await self.sleep(beats * self.clock.beat_duration)
 

sardine_core/handlers/midi_in.py

@@ -1,7 +1,7 @@
 import sys
 from collections import deque
 from dataclasses import dataclass
-from typing import Optional, Union
+from typing import Optional
 
 import mido
 from mido import Message, get_input_names, open_input, parse_string_stream
@@ -99,7 +99,7 @@ def _push_message_to_dict(index: str, message: mido.Message) -> None:
                 # Case where the message is just garbage to dispose of
                 return
 
-    def _extract_value(self, message: Union[mido.Message, None]) -> Union[Message, int]:
+    def _extract_value(self, message: mido.Message | None) -> Message | int:
         """
         Given a mido.Message, extract needed value based on message type
         """

sardine_core/handlers/missile.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Optional, Union
+from typing import Optional
 
 from sardine_core.base import BaseHandler
 
@@ -9,7 +9,7 @@
 class MissileMode(BaseHandler):
     """Maximize the current thread's wake time with a CPU-intensive task."""
 
-    def __init__(self, *, burn_rate: Union[float, int] = 1000):
+    def __init__(self, *, burn_rate: float | int = 1000):
         super().__init__()
         self.burn_interval = 1 / burn_rate
         self._running = False

sardine_core/handlers/osc_in.py

@@ -1,4 +1,4 @@
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional
 
 from osc4py3.as_eventloop import *
 from osc4py3.oscchannel import TransportChannel, get_channel
@@ -109,7 +109,7 @@ def event_dispatcher(address, *args) -> None:
 
         osc_method(address, event_dispatcher, argscheme=OSCARG_DATA)
 
-    def get(self, address: str) -> Union[Any, None]:
+    def get(self, address: str) -> Any | None:
         """Get a watched value. Return None if not found"""
         try:
             return self._watched_values[address]

sardine_core/handlers/sender.py

@@ -1,7 +1,7 @@
 import asyncio
 from math import floor
 from random import random
-from typing import Any, Callable, Generator, Optional, ParamSpec, TypeVar, Union
+from typing import Any, Callable, Generator, Optional, ParamSpec, TypeVar
 
 from sardine_core.base import BaseHandler
 from sardine_core.sequences import euclid
@@ -12,14 +12,14 @@
 P = ParamSpec("P")
 T = TypeVar("T")
 
-Number = Union[float, int]
+Number = float | int
 ReducedElement = TypeVar("ReducedElement")
-RecursiveElement = Union[ReducedElement, list]  # assume list is list[RecursiveElement]
-ParsableElement = Union[RecursiveElement, str]
+RecursiveElement = ReducedElement | list  # assume list is list[RecursiveElement]
+ParsableElement = RecursiveElement | str
 
 # Sub-types of ParsableElement
-NumericElement = Union[Number, list, str]
-StringElement = Union[str, list]  # assume list is list[StringElement]
+NumericElement = Number | list | str
+StringElement = str | list  # assume list is list[StringElement]
 
 Pattern = dict[str, list[ParsableElement]]
 ReducedPattern = dict[str, ReducedElement]

sardine_core/handlers/sleep_handler/__init__.py

@@ -1,7 +1,7 @@
 import asyncio
 import heapq
 from collections import deque
-from typing import Optional, Union
+from typing import Optional
 
 from exceptiongroup import BaseExceptionGroup
 
@@ -11,7 +11,7 @@
 
 __all__ = ("SleepHandler", "TimeHandle")
 
-NUMBER = Union[float, int]
+NUMBER = float | int
 
 
 class SleepHandler(BaseHandler):

sardine_core/handlers/superdirt.py

@@ -1,6 +1,6 @@
 import time
 from itertools import chain
-from typing import Any, Callable, List, Optional, Union
+from typing import Any, Callable, List, Optional
 
 from osc4py3 import oscbuildparse
 from osc4py3.as_eventloop import osc_send, osc_udp_client
@@ -159,14 +159,15 @@ def rename_keys(initial_dictionary: dict, aliases: dict) -> dict:
     @alias_param(name="rate", alias="r")
     def send(
         self,
-        sound: Union[
+        sound: Optional[StringElement | List[StringElement]]
+        | Callable[
+            [],
             Optional[StringElement | List[StringElement]],
-            Callable[[], Optional[StringElement | List[StringElement]]],
         ],
-        orbit: Union[NumericElement, Callable[[], NumericElement]] = 0,
-        iterator: Union[Number, Callable[[], Number]] = 0,
-        divisor: Union[NumericElement, Callable[[], NumericElement]] = 1,
-        rate: Union[NumericElement, Callable[[], NumericElement]] = 1,
+        orbit: NumericElement | Callable[[], NumericElement] = 0,
+        iterator: Number | Callable[[], Number] = 0,
+        divisor: NumericElement | Callable[[], NumericElement] = 1,
+        rate: NumericElement | Callable[[], NumericElement] = 1,
         **pattern: ParsableElement,
     ):
         if sound is None:

sardine_core/io/UserConfig.py

@@ -1,7 +1,6 @@
 import json
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Union
 
 from appdirs import *
 
@@ -50,7 +49,7 @@ def _recursive_update(dest: dict, src: dict):
 
 @dataclass
 class Config:
-    midi: Union[str, None]
+    midi: str | None
     beats: int
     bpm: int
     debug: bool

sardine_core/run.py

@@ -4,7 +4,7 @@
 from itertools import product
 from pathlib import Path
 from string import ascii_lowercase, ascii_uppercase
-from typing import Any, Callable, Optional, ParamSpec, TypeVar, Union, overload
+from typing import Any, Callable, Optional, ParamSpec, TypeVar, overload
 
 from ziffers import z
 
@@ -158,7 +158,7 @@ def wrapper(*args: ParamSpec.args, **kwargs: ParamSpec.kwargs) -> T:
 
 @overload
 def swim(
-    func: Union[Callable[ParamSpec, Any], AsyncRunner],
+    func: Callable[ParamSpec, Any] | AsyncRunner,
     /,
     # NOTE: AsyncRunner doesn't support generic args/kwargs
     *args: ParamSpec.args,
@@ -174,13 +174,13 @@ def swim(
     quant: Quant = "bar",
     until: Optional[int] = None,
     **kwargs,
-) -> Callable[[Union[Callable, AsyncRunner]], AsyncRunner]: ...
+) -> Callable[[Callable | AsyncRunner], AsyncRunner]: ...
 
 
 # FIXME: quant docstring is outdated
 # pylint: disable=keyword-arg-before-vararg  # signature is valid
 def swim(
-    func: Optional[Union[Callable, AsyncRunner]] = None,
+    func: Optional[Callable | AsyncRunner] = None,
     /,
     *args,
     quant: Quant = "bar",
@@ -193,7 +193,7 @@ def swim(
     declared and followed by the scheduler system to recurse in time if needed.
 
     Args:
-        func (Optional[Union[Callable[P, T], AsyncRunner]]):
+        func (Optional[Callable[P | T], AsyncRunner]]):
             The function to be scheduled. If this is an AsyncRunner,
             the current state is simply updated with new arguments.
         *args: Positional arguments to be passed to `func.`
@@ -213,8 +213,7 @@ def swim(
         **kwargs: Keyword arguments to be passed to `func.`
     """
 
-    def decorator(func: Union[Callable, AsyncRunner], /) -> AsyncRunner:
-
+    def decorator(func: Callable | AsyncRunner, /) -> AsyncRunner:
         # This is true when the function is already running on the scheduler
         if isinstance(func, AsyncRunner):
             func.update_state(*args, **kwargs)
@@ -277,7 +276,7 @@ def again(runner: AsyncRunner, *args, **kwargs):
     runner.reload()
 
 
-def die(func: Union[Callable, AsyncRunner]) -> AsyncRunner:
+def die(func: Callable | AsyncRunner) -> AsyncRunner:
     """
     Swimming decorator: remove a function from the scheduler. The function
     will not be called again and will likely stop recursing in time.
@@ -295,7 +294,7 @@ def die(func: Union[Callable, AsyncRunner]) -> AsyncRunner:
     return runner
 
 
-def sleep(n_beats: Union[int, float]):
+def sleep(n_beats: int | float):
     """Artificially sleep in the current function for `n_beats`.
 
     Example usage: ::
@@ -444,7 +443,7 @@ class Delay:
     extra indentation for marking visually where sleep takes effect.
     """
 
-    def __init__(self, duration: Union[int, float] = 1, delayFirst: bool = True):
+    def __init__(self, duration: int | float = 1, delayFirst: bool = True):
         self.duration = duration
         self.delayFirst = delayFirst