diff --git a/sardine/UserConfig.py b/sardine/UserConfig.py index ed694ab9..2dc37214 100644 --- a/sardine/UserConfig.py +++ b/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 diff --git a/sardine_core/base/clock.py b/sardine_core/base/clock.py index 872a064f..11a199f3 100644 --- a/sardine_core/base/clock.py +++ b/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,7 +191,7 @@ 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, @@ -199,7 +199,7 @@ def get_beat_time( """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,7 +237,7 @@ 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, @@ -245,7 +245,7 @@ def get_bar_time( """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. diff --git a/sardine_core/clock/internal_clock.py b/sardine_core/clock/internal_clock.py index 7ed28800..b825f236 100644 --- a/sardine_core/clock/internal_clock.py +++ b/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): diff --git a/sardine_core/clock/link_clock.py b/sardine_core/clock/link_clock.py index 24102410..4387cbf0 100644 --- a/sardine_core/clock/link_clock.py +++ b/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",) diff --git a/sardine_core/fish_bowl.py b/sardine_core/fish_bowl.py index 55281ec1..ba0ef8c0 100644 --- a/sardine_core/fish_bowl.py +++ b/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) diff --git a/sardine_core/handlers/midi_in.py b/sardine_core/handlers/midi_in.py index 1b3891b0..73e5cfc5 100644 --- a/sardine_core/handlers/midi_in.py +++ b/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 """ diff --git a/sardine_core/handlers/missile.py b/sardine_core/handlers/missile.py index 803cc950..6a509095 100644 --- a/sardine_core/handlers/missile.py +++ b/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 diff --git a/sardine_core/handlers/osc_in.py b/sardine_core/handlers/osc_in.py index d43143be..ae0f509d 100644 --- a/sardine_core/handlers/osc_in.py +++ b/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] diff --git a/sardine_core/handlers/sender.py b/sardine_core/handlers/sender.py index 3db6c9c9..88d08d3b 100644 --- a/sardine_core/handlers/sender.py +++ b/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] diff --git a/sardine_core/handlers/sleep_handler/__init__.py b/sardine_core/handlers/sleep_handler/__init__.py index 2efd04a3..0c1d69df 100644 --- a/sardine_core/handlers/sleep_handler/__init__.py +++ b/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): diff --git a/sardine_core/handlers/superdirt.py b/sardine_core/handlers/superdirt.py index dfc53c5a..079bac85 100644 --- a/sardine_core/handlers/superdirt.py +++ b/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: diff --git a/sardine_core/io/UserConfig.py b/sardine_core/io/UserConfig.py index 537ad312..6cd8ca72 100644 --- a/sardine_core/io/UserConfig.py +++ b/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 diff --git a/sardine_core/run.py b/sardine_core/run.py index bd990afd..77441b0b 100644 --- a/sardine_core/run.py +++ b/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 diff --git a/sardine_core/scheduler/async_runner.py b/sardine_core/scheduler/async_runner.py index a3f1010f..e0eb0c45 100644 --- a/sardine_core/scheduler/async_runner.py +++ b/sardine_core/scheduler/async_runner.py @@ -6,7 +6,7 @@ import traceback from collections import deque from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, MutableSequence, NamedTuple, Optional, Union +from typing import TYPE_CHECKING, Any, MutableSequence, NamedTuple, Optional from rich.panel import Panel @@ -50,7 +50,7 @@ def _discard_kwargs(sig: inspect.Signature, kwargs: dict[str, Any]) -> dict[str, def _extract_new_period( sig: inspect.Signature, kwargs: dict[str, Any], default_period: int | float -) -> Union[float, int]: +) -> float | int: period = kwargs.get("p") # Assign a default period if necessary @@ -104,7 +104,7 @@ class FunctionState: class DeferredState(NamedTuple): - deadline: Union[float, int] + deadline: float | int index: int state: FunctionState @@ -187,7 +187,7 @@ class AsyncRunner: running at different phases. To synchronize these functions together, their interval shifts should be set to the same value (usually 0). """ - snap: Optional[Union[float, int]] + snap: Optional[float | int] """ The absolute time that the next interval should start at. @@ -361,7 +361,7 @@ def push(self, func: "MaybeCoroFunc", *args, **kwargs): self.states.append(new_state) def push_deferred( - self, deadline: Union[float, int], func: "MaybeCoroFunc", *args, **kwargs + self, deadline: float | int, func: "MaybeCoroFunc", *args, **kwargs ): """Adds a function to a queue to eventually be run. @@ -372,7 +372,7 @@ def push_deferred( priority and replace the `snap` attribute to ensure they run on time. Args: - time (Union[float, int]): + time (float | int): The absolute clock time to wait before the function state is pushed. func (MaybeCoroFunc): The function to add. @@ -481,7 +481,7 @@ def allow_interval_correction(self) -> None: """Allows the interval to be corrected in the next iteration.""" self._can_correct_interval = True - def delay_interval(self, deadline: Union[float, int], period: Union[float, int]): + def delay_interval(self, deadline: float | int, period: float | int): """Delays the next iteration until the given deadline has passed. This is equivalent to setting the runner's `snap` attribute @@ -496,8 +496,8 @@ def delay_interval(self, deadline: Union[float, int], period: Union[float, int]) to skip the current iteration. Args: - time (Union[float, int]): The absolute time to wait. - period (Union[float, int]): The period to synchronize to. + time (float | int): The absolute time to wait. + period (float | int): The period to synchronize to. Raises: RuntimeError: A function must be pushed before this can be used. @@ -510,7 +510,7 @@ def _check_snap(self, time: float) -> None: if self.snap is not None and time + self._last_interval >= self.snap: self.snap = None - def _correct_interval(self, period: Union[float, int]) -> None: + def _correct_interval(self, period: float | int) -> None: """Checks if the interval should be corrected. Interval correction occurs when `allow_interval_correction()` @@ -519,7 +519,7 @@ def _correct_interval(self, period: Union[float, int]) -> None: did not change, interval correction must be requested again. Args: - period (Union[float, int]): + period (float | int): The period being used in the current iteration. """ # NOTE: this should account for tempo change, weird... @@ -531,7 +531,7 @@ def _correct_interval(self, period: Union[float, int]) -> None: self._last_interval = interval self._can_correct_interval = False - def _correct_interval_background_job(self, period: Union[float, int]) -> None: + def _correct_interval_background_job(self, period: float | int) -> None: """ Alternative version for fixed-rate background jobs. The interval or period is not indexed on the clock like with the _correct_interval @@ -545,7 +545,7 @@ def _correct_interval_background_job(self, period: Union[float, int]) -> None: self._last_interval = interval self._can_correct_interval = False - def _get_next_deadline(self, period: Union[float, int]) -> float: + def _get_next_deadline(self, period: float | int) -> float: """Returns the amount of time until the next interval. The base interval is determined by the `period` argument, @@ -575,7 +575,7 @@ def _get_next_deadline(self, period: Union[float, int]) -> float: the above solution could potentially trigger non-missed iterations too early. Args: - period (Union[float, int]): + period (float | int): The number of beats in the interval. Returns: @@ -756,7 +756,7 @@ async def _call_func(self, func, args, kwargs): return await maybe_coro(func, *args, **valid_kwargs) - def _get_period(self, state: Optional[FunctionState]) -> Union[float, int]: + def _get_period(self, state: Optional[FunctionState]) -> float | int: """ TODO: ??? @@ -765,7 +765,7 @@ def _get_period(self, state: Optional[FunctionState]) -> Union[float, int]: most recent one). Returns: - Union[float, int]: The period to use for the next iteration. + float | int: The period to use for the next iteration. """ # If we don't have a state, we can't extract a period given by the user if state is None: @@ -817,11 +817,11 @@ def _maybe_print_new_state(self, state: FunctionState) -> None: print(f"[yellow][Saving [red]{self.name}[/red] from crash]") self._has_reverted = False - async def _sleep_until(self, deadline: Union[float, int]) -> bool: + async def _sleep_until(self, deadline: float | int) -> bool: """Sleeps until the given deadline or until the runner is reloaded. Args: - duration (Union[float, int]): The amount of time to sleep. + duration (float | int): The amount of time to sleep. Returns: bool: True if the runner was reloaded, False otherwise. @@ -857,7 +857,7 @@ async def _sleep_until(self, deadline: Union[float, int]) -> bool: ) ) - async def _sleep_unless_jump_started(self, deadline: Union[float, int]) -> bool: + async def _sleep_unless_jump_started(self, deadline: float | int) -> bool: if self._jump_start: self._jump_start = False return False diff --git a/sardine_core/scheduler/constants.py b/sardine_core/scheduler/constants.py index 8403c10a..c693a75e 100644 --- a/sardine_core/scheduler/constants.py +++ b/sardine_core/scheduler/constants.py @@ -1,6 +1,6 @@ -from typing import Awaitable, Callable, TypeVar, Union +from typing import Awaitable, Callable, TypeVar __all__ = ("MaybeCoroFunc", "T") T = TypeVar("T") -MaybeCoroFunc = Callable[..., Union[T, Awaitable[T]]] +MaybeCoroFunc = Callable[..., T | Awaitable[T]] diff --git a/sardine_core/sequences/iterators.py b/sardine_core/sequences/iterators.py index def6f924..dbc3e223 100644 --- a/sardine_core/sequences/iterators.py +++ b/sardine_core/sequences/iterators.py @@ -1,6 +1,5 @@ from itertools import count from string import ascii_letters -from typing import Union __all__ = ("Iterator",) @@ -18,7 +17,7 @@ def __init__(self): for c in ascii_letters: self._iterators[c] = count(0) - def reset(self, iterator: Union[str, None] = None): + def reset(self, iterator: str | None = None): if not iterator: self._iterators = {} for c in ascii_letters: diff --git a/sardine_core/sequences/sardine_parser/funclib.py b/sardine_core/sequences/sardine_parser/funclib.py index 3312a344..8ab261f5 100644 --- a/sardine_core/sequences/sardine_parser/funclib.py +++ b/sardine_core/sequences/sardine_parser/funclib.py @@ -5,7 +5,7 @@ from math import asin, atan, cos, pi, sin, tan from random import shuffle from time import time -from typing import Optional, Union +from typing import Optional from sardine_core.sequences.sequence import euclid @@ -457,7 +457,7 @@ def negative_euclidian_rhythm( def find_voice_leading( self, collection, - divider: Optional[Union[list, int]] = 4, + divider: Optional[list | int] = 4, ) -> list: """Simple voice leading algorithm""" # Splitting the collection with divider @@ -619,7 +619,7 @@ def expand(self, collection: list, factor: list, **kwargs) -> list: """ factor = factor[0] - def expand_number(number: Union[int, float]) -> int | float: + def expand_number(number: int | float) -> int | float: expansions = [0, -12, 12] return [number + (random.choice(expansions) * factor)] diff --git a/sardine_core/sequences/sardine_parser/tree_calc.py b/sardine_core/sequences/sardine_parser/tree_calc.py index e8f60157..9deea399 100644 --- a/sardine_core/sequences/sardine_parser/tree_calc.py +++ b/sardine_core/sequences/sardine_parser/tree_calc.py @@ -310,8 +310,8 @@ def extend(self, left, right): etc.. Args: - left (Union[list, float, int]): A token that will be extended - right (Union[list, float, int]): A token used as an extension rule + left (list | float | int): A token that will be extended + right (list | float | int): A token used as an extension rule Returns: list: A list of integers after applying the expansion rule. diff --git a/sardine_core/sequences/variables.py b/sardine_core/sequences/variables.py index 09bfaccf..465f60e0 100644 --- a/sardine_core/sequences/variables.py +++ b/sardine_core/sequences/variables.py @@ -1,6 +1,5 @@ from itertools import count from string import ascii_letters -from typing import Union __all__ = ("Variables",) @@ -16,7 +15,7 @@ def __init__(self): for c in ascii_letters: self._iterators[c] = 0 - def reset(self, iterator: Union[str, None] = None): + def reset(self, iterator: str | None = None): if not iterator: self._iterators = {} for c in ascii_letters: diff --git a/sardine_core/superdirt/process.py b/sardine_core/superdirt/process.py index 5520b338..2532c528 100644 --- a/sardine_core/superdirt/process.py +++ b/sardine_core/superdirt/process.py @@ -4,11 +4,10 @@ import platform import shutil import subprocess -import sys import tempfile from os import path, walk from pathlib import Path -from typing import Optional, Union +from typing import Optional import psutil from appdirs import * @@ -49,7 +48,7 @@ def _find_vanilla_startup_file(self): cur_path = Path(__file__).parent.resolve() return "".join([str(cur_path), "/default_superdirt.scd"]) - def _find_startup_file(self, user_file: Union[str, None] = None) -> Path: + def _find_startup_file(self, user_file: str | None = None) -> Path: """Find the SuperDirt startup file""" if not user_file: file_path = self._user_dir / "default_superdirt.scd" diff --git a/sardine_core/utils/Messages.py b/sardine_core/utils/Messages.py index bfdcb5f9..463161e9 100644 --- a/sardine_core/utils/Messages.py +++ b/sardine_core/utils/Messages.py @@ -2,6 +2,8 @@ from rich.panel import Panel +from sardine_core.io.UserConfig import Config + sardine_intro = """[red] ░██████╗░█████╗░██████╗░██████╗░██╗███╗░░██╗███████╗ ██╔════╝██╔══██╗██╔══██╗██╔══██╗██║████╗░██║██╔════╝ @@ -22,7 +24,7 @@ def _ticked(condition: bool): return "[X]" if condition else "[ ]" -def greeter_printer(intro_logo: str, config: dict): +def greeter_printer(intro_logo: str, config: Config): os.system("cls" if os.name == "nt" else "clear") midi_port = "Automatic" if config.midi == "Sardine" else config.midi config_message = ( diff --git a/sardine_core/utils/__init__.py b/sardine_core/utils/__init__.py index 95be0d42..6e450855 100644 --- a/sardine_core/utils/__init__.py +++ b/sardine_core/utils/__init__.py @@ -1,6 +1,6 @@ import functools import inspect -from typing import TYPE_CHECKING, Callable, Literal, Optional, ParamSpec, TypeVar, Union +from typing import TYPE_CHECKING, Callable, Literal, Optional, ParamSpec, TypeVar from .Messages import * @@ -10,9 +10,9 @@ P = ParamSpec("P") T = TypeVar("T") -Number = Union[float, int] -Quant = Optional[Union[Number, Literal["now", "beat", "bar"]]] -Span = Optional[Union[Number, Literal["inf"]]] +Number = float | int +Quant = Optional[Number | Literal["now", "beat", "bar"]] +Span = Optional[Number | Literal["inf"]] MISSING = object() diff --git a/tests/fish_bowl/__init__.py b/tests/fish_bowl/__init__.py index 199676ef..d384ef19 100644 --- a/tests/fish_bowl/__init__.py +++ b/tests/fish_bowl/__init__.py @@ -11,7 +11,6 @@ Optional, Sequence, TypeVar, - Union, ) import pytest_asyncio @@ -44,7 +43,7 @@ def __init__(self, *, whitelist: Optional[Collection[str]] = None): # Analysis methods - def filter(self, events: Union[str, Collection[str]]) -> Iterator[EventLogEntry]: + def filter(self, events: str | Collection[str]) -> Iterator[EventLogEntry]: if isinstance(events, str): events = (events,) diff --git a/tests/q b/tests/q index 99fab6e9..db4c7604 100644 --- a/tests/q +++ b/tests/q @@ -1,11 +1,11 @@ import functools -from typing import Sequence, Union +from typing import Sequence import pytest from sardine_core import utils -Number = Union[float, int] +Number = float | int @pytest.mark.parametrize( diff --git a/tests/test_utils.py b/tests/test_utils.py index 99fab6e9..cd5e92bb 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,11 +1,11 @@ import functools -from typing import Sequence, Union +from typing import Sequence import pytest from sardine_core import utils -Number = Union[float, int] +Number = float | int @pytest.mark.parametrize(