pyupgrade to Python 3.9+

Author: Jajcus

alsa_midi/address.py

@@ -1,6 +1,5 @@
-
 from collections import namedtuple
-from typing import TYPE_CHECKING, Any, Tuple, Union, overload
+from typing import TYPE_CHECKING, Any, Union, overload
 
 from ._ffi import alsa, ffi
 from .util import _check_alsa_error
@@ -10,7 +9,7 @@
     from .port import Port, PortInfo
 
 
-AddressType = Union['Address', 'Port', 'PortInfo', Tuple[int, int]]
+AddressType = Union['Address', 'Port', 'PortInfo', tuple[int, int]]
 
 
 class Address(namedtuple("Address", "client_id port_id")):
@@ -79,7 +78,7 @@ def __new__(cls,
             return tuple.__new__(cls, tple)
 
     @staticmethod
-    def _parse(arg: str) -> Tuple[int, int]:
+    def _parse(arg: str) -> tuple[int, int]:
         addr_p = ffi.new("snd_seq_addr_t *")
         result = alsa.snd_seq_parse_address(ffi.NULL, addr_p, arg.encode())
         _check_alsa_error(result)

alsa_midi/client.py

@@ -1,13 +1,12 @@
-
 import asyncio
 import errno
 import select
 import time
+from collections.abc import MutableMapping
 from dataclasses import dataclass, field
 from enum import IntEnum, IntFlag
 from functools import partial
-from typing import (Any, Callable, List, MutableMapping, NewType, Optional, Set, Tuple, Union,
-                    overload)
+from typing import Any, Callable, NewType, Optional, Union, overload
 from weakref import WeakValueDictionary
 
 from ._ffi import alsa, ffi
@@ -21,7 +20,7 @@
 from .util import _check_alsa_error
 
 _snd_seq_t = NewType("_snd_seq_t", object)
-_snd_seq_t_p = NewType("_snd_seq_t_p", Tuple[_snd_seq_t])
+_snd_seq_t_p = NewType("_snd_seq_t_p", tuple[_snd_seq_t])
 _snd_midi_event_t = NewType("_snd_midi_event_t", object)
 
 
@@ -90,7 +89,7 @@ class ClientInfo:
     pid: Optional[int] = None
     num_ports: int = 0
     event_lost: int = 0
-    event_filter: Optional[Set[EventType]] = None
+    event_filter: Optional[set[EventType]] = None
 
     @classmethod
     def _from_alsa(cls, info: _snd_seq_client_info_t):
@@ -661,7 +660,7 @@ def drop_output_buffer(self):
         err = alsa.snd_seq_drop_output_buffer(self.handle)
         _check_alsa_error(err)
 
-    def _event_input(self, prefer_bytes: bool = False) -> Tuple[int, Optional[Event]]:
+    def _event_input(self, prefer_bytes: bool = False) -> tuple[int, Optional[Event]]:
         buf = ffi.new("snd_seq_event_t**", ffi.NULL)
         result = alsa.snd_seq_event_input(self.handle, buf)
         if result < 0:
@@ -730,7 +729,7 @@ def _prepare_event(self,
                        queue: Union['Queue', int] = None,
                        port: Union['Port', int] = None,
                        dest: AddressType = None,
-                       remainder: Optional[Any] = None) -> Tuple[_snd_seq_event_t, Any]:
+                       remainder: Optional[Any] = None) -> tuple[_snd_seq_event_t, Any]:
         """Prepare ALSA :alsa:`snd_seq_event_t` for given `event` object for output.
 
         For :class:`alsa_midi.MidiBytesEvent` may need to be called more than
@@ -780,7 +779,7 @@ def _event_output(self,
                       queue: Union['Queue', int] = None,
                       port: Union['Port', int] = None,
                       dest: AddressType = None,
-                      remainder: Optional[Any] = None) -> Tuple[int, Any]:
+                      remainder: Optional[Any] = None) -> tuple[int, Any]:
         alsa_event, remainder = self._prepare_event(event,
                                                     queue=queue, port=port, dest=dest,
                                                     remainder=remainder)
@@ -864,7 +863,7 @@ def _event_output_direct(self,
                              queue: Union['Queue', int] = None,
                              port: Union['Port', int] = None,
                              dest: AddressType = None,
-                             remainder: Optional[Any] = None) -> Tuple[int, Any]:
+                             remainder: Optional[Any] = None) -> tuple[int, Any]:
         alsa_event, remainder = self._prepare_event(event,
                                                     queue=queue, port=port, dest=dest,
                                                     remainder=remainder)
@@ -1114,7 +1113,7 @@ def list_ports(self, *,
                    include_no_export: bool = True,
                    only_connectable: bool = True,
                    sort: Union[bool, Callable[[PortInfo], Any]] = True,
-                   ) -> List[PortInfo]:
+                   ) -> list[PortInfo]:
         """More friendly interface to list available ports.
 
         Queries ALSA for all clients and ports and returns those matching the selected criteria.
@@ -1327,7 +1326,7 @@ def query_port_subscribers(self,
     def list_port_subscribers(self,
                               port: AddressType,
                               type: Optional[SubscriptionQueryType] = None,
-                              ) -> List[SubscriptionQuery]:
+                              ) -> list[SubscriptionQuery]:
         """Lists subscribers accessing a port.
 
         Wraps :alsa:`snd_seq_query_port_subscribers`.

alsa_midi/event.py

@@ -1,7 +1,7 @@
-
+from collections.abc import Iterable
 from enum import IntEnum, IntFlag
 from functools import total_ordering
-from typing import TYPE_CHECKING, Any, Iterable, NewType, Optional, Union
+from typing import TYPE_CHECKING, Any, NewType, Optional, Union
 
 from ._ffi import alsa, ffi
 from .address import Address, AddressType

alsa_midi/mido_backend.py

@@ -13,7 +13,8 @@
 import queue
 import sys
 import threading
-from typing import Any, Callable, List, MutableMapping, Optional
+from collections.abc import MutableMapping
+from typing import Any, Callable, Optional
 from weakref import WeakValueDictionary
 
 from mido.messages import Message
@@ -92,7 +93,7 @@ def get_devices(*args, **kwargs):
     return client.get_devices(*args, **kwargs)
 
 
-def _find_port(ports: List[alsa_midi.PortInfo], name: str) -> alsa_midi.PortInfo:
+def _find_port(ports: list[alsa_midi.PortInfo], name: str) -> alsa_midi.PortInfo:
     try:
         addr = alsa_midi.Address(name)
     except (ValueError, alsa_midi.exceptions.ALSAError):
@@ -104,7 +105,7 @@ def _find_port(ports: List[alsa_midi.PortInfo], name: str) -> alsa_midi.PortInfo
             if addr == alsa_midi.Address(port):
                 return port
         else:
-            raise IOError(f"unknown port {name!r}")
+            raise OSError(f"unknown port {name!r}")
 
     # check for exact match with name from get_devices()
     for port in ports:
@@ -128,10 +129,10 @@ def _find_port(ports: List[alsa_midi.PortInfo], name: str) -> alsa_midi.PortInfo
         if name == port.name:
             return port
 
-    raise IOError(f"unknown port {name!r}")
+    raise OSError(f"unknown port {name!r}")
 
 
-class PortCommon(object):
+class PortCommon:
     _last_num = 0
     _name_prefix = "inout"
     _caps = alsa_midi.PortCaps.READ | alsa_midi.PortCaps.WRITE
@@ -182,7 +183,7 @@ def _open(self, virtual=False, **kwargs):
             ports = client.client.list_ports(input=self._for_input, output=self._for_output)
 
             if not ports:
-                raise IOError('no ports available')
+                raise OSError('no ports available')
 
             if self.name is None:
                 dest_port = ports[0]
@@ -200,9 +201,9 @@ def _open(self, virtual=False, **kwargs):
         self._dest_port = dest_port
 
         api = 'seq'
-        self._device_type = 'AlsaMidi/{}'.format(api)
+        self._device_type = f'AlsaMidi/{api}'
         if virtual:
-            self._device_type = 'virtual {}'.format(self._device_type)
+            self._device_type = f'virtual {self._device_type}'
 
         client.ports[self._port.port_id] = self
 

alsa_midi/port.py

@@ -1,7 +1,6 @@
-
 from dataclasses import InitVar, dataclass, field
 from enum import IntFlag
-from typing import TYPE_CHECKING, Any, Callable, List, NewType, Optional
+from typing import TYPE_CHECKING, Any, Callable, NewType, Optional
 
 from ._ffi import alsa, ffi
 from .address import Address, AddressType
@@ -176,7 +175,7 @@ def set_info(self, info: 'PortInfo'):
             raise StateError("Already closed")
         return self.client.set_port_info(self, info)
 
-    def list_subscribers(self, type: 'SubscriptionQueryType' = None) -> List['SubscriptionQuery']:
+    def list_subscribers(self, type: 'SubscriptionQueryType' = None) -> list['SubscriptionQuery']:
         """Lists subscribers accessing a port.
 
         Wraps :alsa:`snd_seq_query_port_subscribers`.
@@ -286,7 +285,7 @@ def _to_alsa(self) -> _snd_seq_port_info_t:
         return info
 
 
-def get_port_info_sort_key(preferred_types: List[PortType] = []
+def get_port_info_sort_key(preferred_types: list[PortType] = []
                            ) -> Callable[[PortInfo], Any]:
     """Return a :class:`PortInfo` sorting key function for given type
     preference."""

alsa_midi/queue.py

@@ -1,4 +1,3 @@
-
 from dataclasses import dataclass, field
 from enum import IntEnum
 from typing import TYPE_CHECKING, NamedTuple, NewType, Optional, Union

alsa_midi/util.py

@@ -1,4 +1,3 @@
-
 from ._ffi import alsa, ffi
 from .exceptions import ALSAError
 

setup.py

@@ -1,4 +1,3 @@
-
 import os
 
 from setuptools import setup

tests/conftest.py

@@ -1,11 +1,10 @@
-
 import asyncio
 import os
 import re
 import subprocess
 import sys
 from statistics import mean
-from typing import Dict, List, Optional, Tuple
+from typing import Optional
 
 import pytest
 import pytest_asyncio
@@ -16,8 +15,8 @@ class AlsaPortState:
     port_id: int
     name: str
     flags: str
-    connected_from: List[Tuple[int, int]]
-    connected_to: List[Tuple[int, int]]
+    connected_from: list[tuple[int, int]]
+    connected_to: list[tuple[int, int]]
 
     def __init__(self, client_id, port_id, name, flags):
         self.client_id = client_id
@@ -44,8 +43,8 @@ class AlsaClientState:
     name: str
     type: str
     version: Optional[str]
-    ports: Dict[int, AlsaPortState]
-    queues: Dict[int, 'AlsaQueueState']
+    ports: dict[int, AlsaPortState]
+    queues: dict[int, 'AlsaQueueState']
 
     def __init__(self, client_id, name, client_type, midi_version):
         self.client_id = client_id
@@ -80,7 +79,7 @@ class AlsaQueueState:
     timer_ppq: int
     current_tempo: int
     current_bpm: int
-    current_time: Tuple[int, int]
+    current_time: tuple[int, int]
     current_tick: int
 
     def __init__(self, lines):
@@ -138,7 +137,7 @@ def __str__(self):
 PROC_CONN_FROM_LINE_RE = re.compile(r'\s+Connected From:\s*(\S.*)$')
 
 
-def parse_port_list(string: str) -> List[Tuple[int, int]]:
+def parse_port_list(string: str) -> list[tuple[int, int]]:
     port_list = [p.strip() for p in string.split(",")]
     port_list = [p.split("[", 1)[0] for p in port_list]
     port_list = [p.split(":", 1) for p in port_list]
@@ -147,9 +146,9 @@ def parse_port_list(string: str) -> List[Tuple[int, int]]:
 
 
 class AlsaSequencerState:
-    clients: Dict[int, AlsaClientState]
-    ports: Dict[Tuple[int, int], AlsaPortState]
-    queues: Dict[int, AlsaQueueState]
+    clients: dict[int, AlsaClientState]
+    ports: dict[tuple[int, int], AlsaPortState]
+    queues: dict[int, AlsaQueueState]
     max_clients: int
     cur_clients: int
 
@@ -174,7 +173,7 @@ def load(self):
         self.queues = {}
 
         print("clients:")
-        with open("/proc/asound/seq/clients", "r") as proc_f:
+        with open("/proc/asound/seq/clients") as proc_f:
             client = None
             port = None
             for line in proc_f:
@@ -215,7 +214,7 @@ def load(self):
                     continue
 
         print("queues:")
-        with open("/proc/asound/seq/queues", "r") as proc_f:
+        with open("/proc/asound/seq/queues") as proc_f:
             queues = []
             queue_lines = []
             for line in proc_f:
@@ -247,10 +246,10 @@ def alsa_seq_state():
 if not alsa_seq_present:
     try:
         # try triggering snd-seq module load
-        with open("/dev/snd/seq", "r"):
+        with open("/dev/snd/seq"):
             pass
         alsa_seq_present = os.path.exists("/proc/asound/seq/clients")
-    except IOError:
+    except OSError:
         pass
 
 
@@ -328,7 +327,7 @@ def aseqdump():
 
     class Aseqdump:
         process: Optional[subprocess.Popen]
-        output: List[Tuple[Address, str]]
+        output: list[tuple[Address, str]]
         port: Address
 
         def __init__(self, process: subprocess.Popen):

tests/test_address.py

@@ -1,4 +1,3 @@
-
 import errno
 
 import pytest