Add type stubs for mido MIDI library
- Created type stubs for various modules in the mido library including messages, midifiles, parser, ports, sockets, syx, tokenizer, and version. - Implemented type hints for functions and classes to improve type checking and code clarity. - Added support for MIDI over TCP/IP in sockets module. - Included methods for reading and writing SYX files in syx module. - Enhanced the parser functionality with a dedicated Parser class for MIDI byte streams. - Established a structure for MIDI file handling with MidiFile and MidiTrack classes.
This commit is contained in:
@@ -0,0 +1,237 @@
|
|||||||
|
"""
|
||||||
|
MIDI Objects for Python
|
||||||
|
|
||||||
|
Mido is a library for working with MIDI messages and ports. It's
|
||||||
|
designed to be as straight forward and Pythonic as possible.
|
||||||
|
|
||||||
|
Creating messages:
|
||||||
|
|
||||||
|
Message(type, **parameters) -- create a new message
|
||||||
|
MetaMessage(type, **parameters) -- create a new meta message
|
||||||
|
UnknownMetaMessage(type_byte, data=None, time=0)
|
||||||
|
|
||||||
|
Ports:
|
||||||
|
|
||||||
|
open_input(name=None, virtual=False, callback=None) -- open an input port
|
||||||
|
open_output(name=None, virtual=False, -- open an output port
|
||||||
|
autoreset=False)
|
||||||
|
open_ioport(name=None, virtual=False, -- open an I/O port (capable
|
||||||
|
callback=None, autoreset=False) of both input and output)
|
||||||
|
|
||||||
|
get_input_names() -- return a list of names of available input ports
|
||||||
|
get_output_names() -- return a list of names of available output ports
|
||||||
|
get_ioport_names() -- return a list of names of available I/O ports
|
||||||
|
|
||||||
|
MIDI files:
|
||||||
|
|
||||||
|
MidiFile(filename, **kwargs) -- open a MIDI file
|
||||||
|
MidiTrack() -- a MIDI track
|
||||||
|
bpm2tempo() -- convert beats per minute to MIDI file tempo
|
||||||
|
tempo2bpm() -- convert MIDI file tempo to beats per minute
|
||||||
|
merge_tracks(tracks) -- merge tracks into one track
|
||||||
|
|
||||||
|
SYX files:
|
||||||
|
|
||||||
|
read_syx_file(filename) -- read a SYX file
|
||||||
|
write_syx_file(filename, messages,
|
||||||
|
plaintext=False) -- write a SYX file
|
||||||
|
Parsing MIDI streams:
|
||||||
|
|
||||||
|
parse(bytes) -- parse a single message bytes
|
||||||
|
(any iterable that generates integers in 0..127)
|
||||||
|
parse_all(bytes) -- parse all messages bytes
|
||||||
|
Parser -- MIDI parser class
|
||||||
|
|
||||||
|
Parsing objects serialized with str(message):
|
||||||
|
|
||||||
|
parse_string(string) -- parse a string containing a message
|
||||||
|
parse_string_stream(iterable) -- parse strings from an iterable and
|
||||||
|
generate messages
|
||||||
|
|
||||||
|
Sub modules:
|
||||||
|
|
||||||
|
ports -- useful tools for working with ports
|
||||||
|
|
||||||
|
For more on MIDI, see:
|
||||||
|
|
||||||
|
http://www.midi.org/
|
||||||
|
|
||||||
|
|
||||||
|
Getting started:
|
||||||
|
|
||||||
|
>>> import mido
|
||||||
|
>>> m = mido.Message('note_on', note=60, velocity=64)
|
||||||
|
>>> m
|
||||||
|
<message note_on channel=0, note=60, velocity=64, time=0>
|
||||||
|
>>> m.type
|
||||||
|
'note_on'
|
||||||
|
>>> m.channel = 6
|
||||||
|
>>> m.note = 19
|
||||||
|
>>> m.copy(velocity=120)
|
||||||
|
<message note_on channel=0, note=60, velocity=64, time=0>
|
||||||
|
>>> s = mido.Message('sysex', data=[byte for byte in range(5)])
|
||||||
|
>>> s.data
|
||||||
|
(0, 1, 2, 3, 4)
|
||||||
|
>>> s.hex()
|
||||||
|
'F0 00 01 02 03 04 F7'
|
||||||
|
>>> len(s)
|
||||||
|
7
|
||||||
|
|
||||||
|
>>> default_input = mido.open_input()
|
||||||
|
>>> default_input.name
|
||||||
|
'MPK mini MIDI 1'
|
||||||
|
>>> output = mido.open_output('SD-20 Part A')
|
||||||
|
>>>
|
||||||
|
>>> for message in default_input:
|
||||||
|
... output.send(message)
|
||||||
|
|
||||||
|
>>> get_input_names()
|
||||||
|
['MPK mini MIDI 1', 'SH-201']
|
||||||
|
"""
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
|
||||||
|
from . import ports, sockets
|
||||||
|
from .backends.backend import Backend
|
||||||
|
from .messages import (
|
||||||
|
MAX_PITCHWHEEL,
|
||||||
|
MAX_SONGPOS,
|
||||||
|
MIN_PITCHWHEEL,
|
||||||
|
MIN_SONGPOS,
|
||||||
|
BaseMessage,
|
||||||
|
Message,
|
||||||
|
format_as_string,
|
||||||
|
parse_string,
|
||||||
|
parse_string_stream,
|
||||||
|
)
|
||||||
|
from .midifiles import (
|
||||||
|
KeySignatureError,
|
||||||
|
MetaMessage,
|
||||||
|
MidiFile,
|
||||||
|
MidiTrack,
|
||||||
|
UnknownMetaMessage,
|
||||||
|
bpm2tempo,
|
||||||
|
merge_tracks,
|
||||||
|
second2tick,
|
||||||
|
tempo2bpm,
|
||||||
|
tick2second,
|
||||||
|
)
|
||||||
|
from .parser import Parser, parse, parse_all
|
||||||
|
from .syx import read_syx_file, write_syx_file
|
||||||
|
from .version import version_info
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"MAX_PITCHWHEEL",
|
||||||
|
"MAX_SONGPOS",
|
||||||
|
"MIN_PITCHWHEEL",
|
||||||
|
"MIN_SONGPOS",
|
||||||
|
"KeySignatureError",
|
||||||
|
"Message",
|
||||||
|
"MetaMessage",
|
||||||
|
"MidiFile",
|
||||||
|
"MidiTrack",
|
||||||
|
"Parser",
|
||||||
|
"UnknownMetaMessage",
|
||||||
|
"bpm2tempo",
|
||||||
|
"format_as_string",
|
||||||
|
"merge_tracks",
|
||||||
|
"parse",
|
||||||
|
"parse_all",
|
||||||
|
"parse_string",
|
||||||
|
"parse_string_stream",
|
||||||
|
"ports",
|
||||||
|
"read_syx_file",
|
||||||
|
"second2tick",
|
||||||
|
"sockets",
|
||||||
|
"tempo2bpm",
|
||||||
|
"tick2second",
|
||||||
|
"version_info",
|
||||||
|
"write_syx_file",
|
||||||
|
]
|
||||||
|
|
||||||
|
def set_backend(name: str | Backend | None = ..., load: bool = False) -> None:
|
||||||
|
"""Set current backend.
|
||||||
|
|
||||||
|
name can be a module name like `'mido.backends.rtmidi'` or
|
||||||
|
a `Backend` object.
|
||||||
|
|
||||||
|
If no name is passed, the default backend will be used.
|
||||||
|
|
||||||
|
This will replace all the `open_*()` and `get_*_name()` functions
|
||||||
|
in top level mido module. The module will be loaded the first
|
||||||
|
time one of those functions is called."""
|
||||||
|
|
||||||
|
backend: Backend
|
||||||
|
|
||||||
|
def open_input(
|
||||||
|
name: str | None = ...,
|
||||||
|
virtual: bool = False,
|
||||||
|
callback: Callable[[BaseMessage], object] | None = ...,
|
||||||
|
**kwargs: object,
|
||||||
|
) -> ports.BaseInput:
|
||||||
|
"""Open an input port.
|
||||||
|
|
||||||
|
If the environment variable MIDO_DEFAULT_INPUT is set,
|
||||||
|
it will override the default port.
|
||||||
|
|
||||||
|
virtual=False
|
||||||
|
Passing True opens a new port that other applications can
|
||||||
|
connect to. Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
callback=None
|
||||||
|
A callback function to be called when a new message arrives.
|
||||||
|
The function should take one argument (the message).
|
||||||
|
Raises IOError if not supported by the backend.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def open_output(
|
||||||
|
name: str | None = ..., virtual: bool = False, autoreset: bool = False, **kwargs: object
|
||||||
|
) -> ports.BaseOutput:
|
||||||
|
"""Open an output port.
|
||||||
|
|
||||||
|
If the environment variable MIDO_DEFAULT_OUTPUT is set,
|
||||||
|
it will override the default port.
|
||||||
|
|
||||||
|
virtual=False
|
||||||
|
Passing True opens a new port that other applications can
|
||||||
|
connect to. Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
autoreset=False
|
||||||
|
Automatically send all_notes_off and reset_all_controllers
|
||||||
|
on all channels. This is the same as calling `port.reset()`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def open_ioport[TIn: ports.BaseInput = ports.BaseInput, TOut: ports.BaseOutput = ports.BaseOutput](
|
||||||
|
name: str | None = ...,
|
||||||
|
virtual: bool = False,
|
||||||
|
callback: Callable[[BaseMessage], object] | None = ...,
|
||||||
|
autoreset: bool = False,
|
||||||
|
**kwargs: object,
|
||||||
|
) -> ports.BaseIOPort | ports.IOPort[TIn, TOut]:
|
||||||
|
"""Open a port for input and output.
|
||||||
|
|
||||||
|
If the environment variable MIDO_DEFAULT_IOPORT is set,
|
||||||
|
it will override the default port.
|
||||||
|
|
||||||
|
virtual=False
|
||||||
|
Passing True opens a new port that other applications can
|
||||||
|
connect to. Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
callback=None
|
||||||
|
A callback function to be called when a new message arrives.
|
||||||
|
The function should take one argument (the message).
|
||||||
|
Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
autoreset=False
|
||||||
|
Automatically send all_notes_off and reset_all_controllers
|
||||||
|
on all channels. This is the same as calling `port.reset()`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def get_input_names(**kwargs: object) -> list[str]:
|
||||||
|
"""Return a list of all input port names."""
|
||||||
|
|
||||||
|
def get_output_names(**kwargs: object) -> list[str]:
|
||||||
|
"""Return a list of all output port names."""
|
||||||
|
|
||||||
|
def get_ioport_names(**kwargs: object) -> list[str]:
|
||||||
|
"""Return a list of all I/O port names."""
|
||||||
@@ -0,0 +1,119 @@
|
|||||||
|
from collections.abc import Callable
|
||||||
|
from types import ModuleType
|
||||||
|
|
||||||
|
from ..messages import BaseMessage
|
||||||
|
from ..ports import BaseInput, BaseIOPort, BaseOutput, IOPort
|
||||||
|
|
||||||
|
DEFAULT_BACKEND: str = "mido.backends.rtmidi"
|
||||||
|
|
||||||
|
class Backend:
|
||||||
|
"""
|
||||||
|
Wrapper for backend module.
|
||||||
|
|
||||||
|
A backend module implements classes for input and output ports for
|
||||||
|
a specific MIDI library. The Backend object wraps around the
|
||||||
|
object and provides convenient 'open_*()' and 'get_*_names()'
|
||||||
|
functions.
|
||||||
|
"""
|
||||||
|
|
||||||
|
name: str
|
||||||
|
api: str | None
|
||||||
|
use_environ: bool
|
||||||
|
def __init__(
|
||||||
|
self, name: str | None = ..., api: str | None = ..., load: bool = False, use_environ: bool = True
|
||||||
|
) -> None: ...
|
||||||
|
@property
|
||||||
|
def module(self) -> ModuleType:
|
||||||
|
"""A reference module implementing the backend.
|
||||||
|
|
||||||
|
This will always be a valid reference to a module. Accessing
|
||||||
|
this property will load the module. Use .loaded to check if
|
||||||
|
the module is loaded.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@property
|
||||||
|
def loaded(self) -> bool:
|
||||||
|
"""Return True if the module is loaded."""
|
||||||
|
|
||||||
|
def load(self) -> None:
|
||||||
|
"""Load the module.
|
||||||
|
|
||||||
|
Does nothing if the module is already loaded.
|
||||||
|
|
||||||
|
This function will be called if you access the 'module'
|
||||||
|
property."""
|
||||||
|
|
||||||
|
def open_input(
|
||||||
|
self,
|
||||||
|
name: str | None = ...,
|
||||||
|
virtual: bool = False,
|
||||||
|
callback: Callable[[BaseMessage], object] | None = ...,
|
||||||
|
**kwargs: object,
|
||||||
|
) -> BaseInput:
|
||||||
|
"""Open an input port.
|
||||||
|
|
||||||
|
If the environment variable MIDO_DEFAULT_INPUT is set,
|
||||||
|
it will override the default port.
|
||||||
|
|
||||||
|
virtual=False
|
||||||
|
Passing True opens a new port that other applications can
|
||||||
|
connect to. Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
callback=None
|
||||||
|
A callback function to be called when a new message arrives.
|
||||||
|
The function should take one argument (the message).
|
||||||
|
Raises IOError if not supported by the backend.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def open_output(
|
||||||
|
self, name: str | None = ..., virtual: bool = False, autoreset: bool = False, **kwargs: object
|
||||||
|
) -> BaseOutput:
|
||||||
|
"""Open an output port.
|
||||||
|
|
||||||
|
If the environment variable MIDO_DEFAULT_OUTPUT is set,
|
||||||
|
it will override the default port.
|
||||||
|
|
||||||
|
virtual=False
|
||||||
|
Passing True opens a new port that other applications can
|
||||||
|
connect to. Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
autoreset=False
|
||||||
|
Automatically send all_notes_off and reset_all_controllers
|
||||||
|
on all channels. This is the same as calling `port.reset()`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def open_ioport[TIn: BaseInput = BaseInput, TOut: BaseOutput = BaseOutput](
|
||||||
|
self,
|
||||||
|
name: str | None = ...,
|
||||||
|
virtual: bool = False,
|
||||||
|
callback: Callable[[BaseMessage], object] | None = ...,
|
||||||
|
autoreset: bool = False,
|
||||||
|
**kwargs: object,
|
||||||
|
) -> BaseIOPort | IOPort[TIn, TOut]:
|
||||||
|
"""Open a port for input and output.
|
||||||
|
|
||||||
|
If the environment variable MIDO_DEFAULT_IOPORT is set,
|
||||||
|
it will override the default port.
|
||||||
|
|
||||||
|
virtual=False
|
||||||
|
Passing True opens a new port that other applications can
|
||||||
|
connect to. Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
callback=None
|
||||||
|
A callback function to be called when a new message arrives.
|
||||||
|
The function should take one argument (the message).
|
||||||
|
Raises IOError if not supported by the backend.
|
||||||
|
|
||||||
|
autoreset=False
|
||||||
|
Automatically send all_notes_off and reset_all_controllers
|
||||||
|
on all channels. This is the same as calling `port.reset()`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def get_input_names(self, **kwargs: object) -> list[str]:
|
||||||
|
"""Return a list of all input port names."""
|
||||||
|
|
||||||
|
def get_output_names(self, **kwargs: object) -> list[str]:
|
||||||
|
"""Return a list of all output port names."""
|
||||||
|
|
||||||
|
def get_ioport_names(self, **kwargs: object) -> list[str]:
|
||||||
|
"""Return a list of all I/O port names."""
|
||||||
@@ -0,0 +1,67 @@
|
|||||||
|
from typing import Never, TypeIs, overload, override
|
||||||
|
|
||||||
|
from .messages import Message
|
||||||
|
from .midifiles import MetaMessage, UnknownMetaMessage
|
||||||
|
from .midifiles.meta import _MetaMessage # pyright: ignore[reportPrivateUsage]
|
||||||
|
|
||||||
|
class Frozen:
|
||||||
|
@override
|
||||||
|
def __setattr__(self, *_) -> Never: ...
|
||||||
|
@override
|
||||||
|
def __hash__(self) -> int: ...
|
||||||
|
|
||||||
|
class FrozenMessage(Frozen, Message): ...
|
||||||
|
class FrozenMetaMessage(Frozen, _MetaMessage): ...
|
||||||
|
class FrozenUnknownMetaMessage(Frozen, UnknownMetaMessage): ...
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def is_frozen(msg: UnknownMetaMessage) -> TypeIs[FrozenUnknownMetaMessage]: ... # pyright: ignore[reportOverlappingOverload]
|
||||||
|
@overload
|
||||||
|
def is_frozen(msg: _MetaMessage) -> TypeIs[FrozenMetaMessage]: ...
|
||||||
|
@overload
|
||||||
|
def is_frozen(msg: Message) -> TypeIs[FrozenMessage]: ...
|
||||||
|
@overload
|
||||||
|
def is_frozen(msg: object) -> TypeIs[Frozen]:
|
||||||
|
"""Return True if message is frozen, otherwise False."""
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def freeze_message[T: Frozen](msg: T) -> T: ...
|
||||||
|
@overload
|
||||||
|
def freeze_message(msg: Message) -> FrozenMessage: ...
|
||||||
|
@overload
|
||||||
|
def freeze_message(msg: UnknownMetaMessage) -> FrozenUnknownMetaMessage: ...
|
||||||
|
@overload
|
||||||
|
def freeze_message(msg: MetaMessage) -> FrozenMetaMessage: ...
|
||||||
|
@overload
|
||||||
|
def freeze_message(msg: None) -> None: ...
|
||||||
|
@overload
|
||||||
|
def freeze_message[T: Frozen](
|
||||||
|
msg: T | Message | UnknownMetaMessage | MetaMessage | None,
|
||||||
|
) -> T | FrozenMessage | FrozenUnknownMetaMessage | FrozenMetaMessage | None:
|
||||||
|
"""Freeze message.
|
||||||
|
|
||||||
|
Returns a frozen version of the message. Frozen messages are
|
||||||
|
immutable, hashable and can be used as dictionary keys.
|
||||||
|
|
||||||
|
Will return None if called with None. This allows you to do things
|
||||||
|
like::
|
||||||
|
|
||||||
|
msg = freeze_message(port.poll())
|
||||||
|
"""
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def thaw_message(msg: FrozenUnknownMetaMessage) -> UnknownMetaMessage: ...
|
||||||
|
@overload
|
||||||
|
def thaw_message(msg: FrozenMetaMessage) -> MetaMessage: ...
|
||||||
|
@overload
|
||||||
|
def thaw_message(msg: FrozenMessage) -> Message: ...
|
||||||
|
@overload
|
||||||
|
def thaw_message(
|
||||||
|
msg: FrozenMessage | FrozenUnknownMetaMessage | FrozenMetaMessage,
|
||||||
|
) -> Message | UnknownMetaMessage | MetaMessage:
|
||||||
|
"""Thaw message.
|
||||||
|
|
||||||
|
Returns a mutable version of a frozen message.
|
||||||
|
|
||||||
|
Will return None if called with None.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
from .checks import check_time
|
||||||
|
from .messages import BaseMessage, Message, format_as_string, parse_string, parse_string_stream
|
||||||
|
from .specs import MAX_PITCHWHEEL, MAX_SONGPOS, MIN_PITCHWHEEL, MIN_SONGPOS, SPEC_BY_STATUS, SPEC_BY_TYPE, SPEC_LOOKUP
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"MAX_PITCHWHEEL",
|
||||||
|
"MAX_SONGPOS",
|
||||||
|
"MIN_PITCHWHEEL",
|
||||||
|
"MIN_SONGPOS",
|
||||||
|
"SPEC_BY_STATUS",
|
||||||
|
"SPEC_BY_TYPE",
|
||||||
|
"SPEC_LOOKUP",
|
||||||
|
"BaseMessage",
|
||||||
|
"Message",
|
||||||
|
"check_time",
|
||||||
|
"format_as_string",
|
||||||
|
"parse_string",
|
||||||
|
"parse_string_stream",
|
||||||
|
]
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
from collections.abc import Callable, Iterable
|
||||||
|
from numbers import Integral, Real
|
||||||
|
|
||||||
|
from .specs import MsgDict
|
||||||
|
|
||||||
|
def check_type(type_: str) -> None: ...
|
||||||
|
def check_channel(channel: int | Integral) -> None: ...
|
||||||
|
def check_pos(pos: int | Integral) -> None: ...
|
||||||
|
def check_pitch(pitch: int | Integral) -> None: ...
|
||||||
|
def check_data(data_bytes: Iterable[int | Integral]) -> None: ...
|
||||||
|
def check_frame_type(value: int | Integral) -> None: ...
|
||||||
|
def check_frame_value(value: int | Integral) -> None: ...
|
||||||
|
def check_data_byte(value: int | Integral) -> None: ...
|
||||||
|
def check_time(time: float | Real) -> None: ...
|
||||||
|
|
||||||
|
_CHECKS: dict[str, Callable[[object], None]]
|
||||||
|
|
||||||
|
def check_value(name: str, value: object) -> None: ...
|
||||||
|
def check_msgdict(msgdict: MsgDict) -> None: ...
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
from collections.abc import Callable, Iterable
|
||||||
|
from numbers import Integral
|
||||||
|
|
||||||
|
from .specs import MsgDict
|
||||||
|
|
||||||
|
_SPECIAL_CASES: dict[int, Callable[[Iterable[int | Integral]], dict[str, object]]]
|
||||||
|
|
||||||
|
def decode_message(msg_bytes: Iterable[int | Integral], time: int = 0, check: bool = True) -> MsgDict:
|
||||||
|
"""Decode message bytes and return messages as a dictionary.
|
||||||
|
|
||||||
|
Raises ValueError if the bytes are out of range or the message is
|
||||||
|
invalid.
|
||||||
|
|
||||||
|
This is not a part of the public API.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,14 @@
|
|||||||
|
from collections.abc import Callable
|
||||||
|
|
||||||
|
from .specs import MsgDict
|
||||||
|
|
||||||
|
_SPECIAL_CASES: dict[str, Callable[[MsgDict], list[object]]]
|
||||||
|
|
||||||
|
def encode_message(msg: MsgDict) -> list[int]:
|
||||||
|
"""Encode msg dict as a list of bytes.
|
||||||
|
|
||||||
|
TODO: Add type and value checking.
|
||||||
|
(Can be turned off with keyword argument.)
|
||||||
|
|
||||||
|
This is not a part of the public API.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,146 @@
|
|||||||
|
from abc import ABCMeta, abstractmethod
|
||||||
|
from collections.abc import Generator, Iterable
|
||||||
|
from typing import Literal, Never, Self, Unpack, override
|
||||||
|
|
||||||
|
from mido.messages.specs import MsgDict, MsgDictOverride
|
||||||
|
|
||||||
|
class BaseMessage(metaclass=ABCMeta):
|
||||||
|
"""Abstract base class for messages."""
|
||||||
|
|
||||||
|
@property
|
||||||
|
@abstractmethod
|
||||||
|
def is_meta(self) -> bool: ... # a normal attribute in runtime
|
||||||
|
@abstractmethod
|
||||||
|
def copy(self) -> Self: ...
|
||||||
|
@abstractmethod
|
||||||
|
def bytes(self) -> list[int]: ...
|
||||||
|
def bin(self) -> bytearray:
|
||||||
|
"""Encode message and return as a bytearray.
|
||||||
|
|
||||||
|
This can be used to write the message to a file.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def hex(self, sep: str = " ") -> str:
|
||||||
|
"""Encode message and return as a string of hex numbers,
|
||||||
|
|
||||||
|
Each number is separated by the string sep.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def dict(self) -> MsgDict:
|
||||||
|
"""Returns a dictionary containing the attributes of the message.
|
||||||
|
|
||||||
|
Example: `{'type': 'sysex', 'data': [1, 2], 'time': 0}`
|
||||||
|
|
||||||
|
Sysex data will be returned as a list.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_dict(cls, data: MsgDict) -> Self:
|
||||||
|
"""Create a message from a dictionary.
|
||||||
|
|
||||||
|
Only "type" is required. The other will be set to default
|
||||||
|
values.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_realtime(self) -> bool:
|
||||||
|
"""True if the message is a system realtime message."""
|
||||||
|
|
||||||
|
def is_cc(self, control: int | None = ...) -> bool:
|
||||||
|
"""Return True if the message is of type 'control_change'.
|
||||||
|
|
||||||
|
The optional control argument can be used to test for a specific
|
||||||
|
control number, for example:
|
||||||
|
|
||||||
|
>>> if msg.is_cc(7):
|
||||||
|
... # Message is control change 7 (channel volume).
|
||||||
|
"""
|
||||||
|
|
||||||
|
@override
|
||||||
|
def __delattr__(self, name: str) -> Never: ...
|
||||||
|
@override
|
||||||
|
def __setattr__(self, name: str, value: object) -> None: ...
|
||||||
|
@override
|
||||||
|
def __eq__(self, other: object) -> bool: ...
|
||||||
|
|
||||||
|
class SysexData(tuple[int, ...]):
|
||||||
|
"""Special kind of tuple accepts and converts any sequence in `+=`."""
|
||||||
|
def __iadd__(self, other: SysexData) -> Self: ...
|
||||||
|
|
||||||
|
class Message(BaseMessage):
|
||||||
|
type: str
|
||||||
|
time: int
|
||||||
|
|
||||||
|
@property
|
||||||
|
@override
|
||||||
|
def is_meta(self) -> Literal[False]: ... # a normal attribute in runtime
|
||||||
|
def __init__(self, type: str, skip_checks: bool = False, **args: Unpack[MsgDictOverride]) -> None: ...
|
||||||
|
@override
|
||||||
|
def copy(self, skip_checks: bool = False, **overrides: Unpack[MsgDictOverride]) -> Self:
|
||||||
|
"""Return a copy of the message.
|
||||||
|
|
||||||
|
Attributes will be overridden by the passed keyword arguments.
|
||||||
|
Only message specific attributes can be overridden. The message
|
||||||
|
type can not be changed.
|
||||||
|
|
||||||
|
The `skip_checks` arg can be used to bypass validation of message
|
||||||
|
attributes and should be used cautiously.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_bytes(cls, data: Iterable[int], time: int = 0) -> Self:
|
||||||
|
"""Parse a byte encoded message.
|
||||||
|
|
||||||
|
Accepts a byte string or any iterable of integers.
|
||||||
|
|
||||||
|
This is the reverse of `msg.bytes()` or `msg.bin()`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_hex(cls, text: str, time: int = 0, sep: str | None = ...) -> Self:
|
||||||
|
"""Parse a hex encoded message.
|
||||||
|
|
||||||
|
This is the reverse of `msg.hex()`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_str(cls, text: str) -> Self:
|
||||||
|
"""Parse a string encoded message.
|
||||||
|
|
||||||
|
This is the reverse of `str(msg)`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __len__(self) -> int: ...
|
||||||
|
@override
|
||||||
|
def __setattr__(self, name: str, value: object) -> None: ...
|
||||||
|
@override
|
||||||
|
def bytes(self) -> list[int]:
|
||||||
|
"""Encode message and return as a list of integers."""
|
||||||
|
|
||||||
|
def parse_string(text: str) -> Message:
|
||||||
|
"""Parse a string of text and return a message.
|
||||||
|
|
||||||
|
The string can span multiple lines, but must contain
|
||||||
|
one full message.
|
||||||
|
|
||||||
|
Raises ValueError if the string could not be parsed.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def parse_string_stream(stream: Iterable[str]) -> Generator[tuple[Message, None] | tuple[None, str], Never]:
|
||||||
|
"""Parse a stream of messages and yield `(message, error_message)`
|
||||||
|
|
||||||
|
stream can be any iterable that generates text strings, where each
|
||||||
|
string is a string encoded message.
|
||||||
|
|
||||||
|
If a string can be parsed, `(message, None)` is returned. If it
|
||||||
|
can't be parsed, `(None, error_message)` is returned. The error
|
||||||
|
message contains the line number where the error occurred.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def format_as_string(msg: Message, include_time: bool = True) -> str:
|
||||||
|
"""Format a message and return as a string.
|
||||||
|
|
||||||
|
This is equivalent to `str(message)`.
|
||||||
|
|
||||||
|
To leave out the time attribute, pass `include_time=False`.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,179 @@
|
|||||||
|
"""Definitions and lookup tables for MIDI messages.
|
||||||
|
|
||||||
|
TODO:
|
||||||
|
|
||||||
|
* add lookup functions for messages definitions by type and status
|
||||||
|
byte.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from collections.abc import Sequence
|
||||||
|
from typing import Literal, NotRequired, TypedDict, final, overload, type_check_only
|
||||||
|
|
||||||
|
CHANNEL_MESSAGES: set[int]
|
||||||
|
COMMON_MESSAGES: set[int]
|
||||||
|
REALTIME_MESSAGES: set[int]
|
||||||
|
|
||||||
|
SYSEX_START: int = 0xF0
|
||||||
|
SYSEX_END: int = 0xF7
|
||||||
|
|
||||||
|
# Pitchwheel is a 14 bit signed integer
|
||||||
|
MIN_PITCHWHEEL: int = -8192
|
||||||
|
MAX_PITCHWHEEL: int = 8191
|
||||||
|
|
||||||
|
# Song pos is a 14 bit unsigned integer
|
||||||
|
MIN_SONGPOS: int = 0
|
||||||
|
MAX_SONGPOS: int = 16383
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class SpecsDict(TypedDict):
|
||||||
|
status_byte: int
|
||||||
|
type: str
|
||||||
|
value_names: tuple[str, ...]
|
||||||
|
attribute_names: set[str]
|
||||||
|
length: int | float
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class MsgDictOverride(TypedDict):
|
||||||
|
channel: NotRequired[int]
|
||||||
|
control: NotRequired[int]
|
||||||
|
data: NotRequired[Sequence[int]]
|
||||||
|
frame_type: NotRequired[int]
|
||||||
|
frame_value: NotRequired[int]
|
||||||
|
note: NotRequired[int]
|
||||||
|
pitch: NotRequired[int]
|
||||||
|
pos: NotRequired[int]
|
||||||
|
program: NotRequired[int]
|
||||||
|
song: NotRequired[int]
|
||||||
|
value: NotRequired[int]
|
||||||
|
velocity: NotRequired[int]
|
||||||
|
time: NotRequired[int | float]
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class MsgDictBase[T: str](TypedDict):
|
||||||
|
type: T
|
||||||
|
time: int | float
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class ChannelMsgDictBase[
|
||||||
|
T: Literal["note_off", "note_on", "polytouch", "control_change", "program_change", "aftertouch", "pitchwheel"]
|
||||||
|
](MsgDictBase[T]):
|
||||||
|
channel: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class NoteMsgDict(ChannelMsgDictBase[Literal["note_off", "note_on"]]):
|
||||||
|
note: int
|
||||||
|
velocity: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class PolytouchMsgDict(ChannelMsgDictBase[Literal["polytouch"]]):
|
||||||
|
note: int
|
||||||
|
value: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class ControlChangeMsgDict(ChannelMsgDictBase[Literal["control_change"]]):
|
||||||
|
control: int
|
||||||
|
value: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class ProgramChangeMsgDict(ChannelMsgDictBase[Literal["program_change"]]):
|
||||||
|
program: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class AftertouchMsgDict(ChannelMsgDictBase[Literal["aftertouch"]]):
|
||||||
|
value: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class PitchwheelMsgDict(ChannelMsgDictBase[Literal["pitchwheel"]]):
|
||||||
|
pitch: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class SysExMsgDict(MsgDictBase[Literal["sysex"]]):
|
||||||
|
data: Sequence[int]
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class QuarterFrameMsgDict(MsgDictBase[Literal["quarter_frame"]]):
|
||||||
|
frame_type: int
|
||||||
|
frame_value: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class SongposMsgDict(MsgDictBase[Literal["songpos"]]):
|
||||||
|
pos: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class SongSelectMsgDict(MsgDictBase[Literal["song_select"]]):
|
||||||
|
song: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
@final
|
||||||
|
class RealTimeMsgDict(
|
||||||
|
MsgDictBase[Literal["tune_request", "clock", "start", "continue", "stop", "active_sensing", "reset"]]
|
||||||
|
): ...
|
||||||
|
|
||||||
|
type MsgDict = (
|
||||||
|
NoteMsgDict
|
||||||
|
| PolytouchMsgDict
|
||||||
|
| ControlChangeMsgDict
|
||||||
|
| ProgramChangeMsgDict
|
||||||
|
| AftertouchMsgDict
|
||||||
|
| PitchwheelMsgDict
|
||||||
|
| SysExMsgDict
|
||||||
|
| QuarterFrameMsgDict
|
||||||
|
| SongposMsgDict
|
||||||
|
| SongSelectMsgDict
|
||||||
|
| RealTimeMsgDict
|
||||||
|
)
|
||||||
|
|
||||||
|
SPECS: list[SpecsDict]
|
||||||
|
SPEC_LOOKUP: dict[str | int, SpecsDict]
|
||||||
|
SPEC_BY_STATUS: dict[int, SpecsDict]
|
||||||
|
SPEC_BY_TYPE: dict[str, SpecsDict]
|
||||||
|
REALTIME_TYPES: set[str]
|
||||||
|
DEFAULT_VALUES: MsgDict
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["note_off", "note_on"], overrides: MsgDictOverride) -> NoteMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["polytouch"], overrides: MsgDictOverride) -> PolytouchMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["control_change"], overrides: MsgDictOverride) -> ControlChangeMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["program_change"], overrides: MsgDictOverride) -> ProgramChangeMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["aftertouch"], overrides: MsgDictOverride) -> AftertouchMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["pitchwheel"], overrides: MsgDictOverride) -> PitchwheelMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["sysex"], overrides: MsgDictOverride) -> SysExMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["quarter_frame"], overrides: MsgDictOverride) -> QuarterFrameMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["songpos"], overrides: MsgDictOverride) -> SongposMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(type_: Literal["song_select"], overrides: MsgDictOverride) -> SongSelectMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict(
|
||||||
|
type_: Literal["tune_request", "clock", "start", "continue", "stop", "active_sensing", "reset"],
|
||||||
|
overrides: MsgDictOverride,
|
||||||
|
) -> RealTimeMsgDict: ...
|
||||||
|
@overload
|
||||||
|
def make_msgdict[T: str](type_: T, overrides: MsgDictOverride) -> MsgDictBase[T]: ...
|
||||||
|
def make_msgdict[T: str](type_: T, overrides: MsgDictOverride) -> MsgDictBase[T]:
|
||||||
|
"""Return a new message.
|
||||||
|
|
||||||
|
Returns a dictionary representing a message.
|
||||||
|
|
||||||
|
Message values can be overriden.
|
||||||
|
|
||||||
|
No type or value checking is done. The caller is responsible for
|
||||||
|
calling `check_msgdict()`.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
"""
|
||||||
|
This type stub file was generated by pyright.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from .specs import MsgDict
|
||||||
|
|
||||||
|
def msg2str(msg: MsgDict, include_time: bool = True) -> str:
|
||||||
|
...
|
||||||
|
|
||||||
|
def str2msg(text: str) -> MsgDict:
|
||||||
|
"""Parse str format and return message dict.
|
||||||
|
|
||||||
|
No type or value checking is done. The caller is responsible for
|
||||||
|
calling check_msgdict().
|
||||||
|
"""
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
"""
|
||||||
|
This type stub file was generated by pyright.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from .meta import KeySignatureError, MetaMessage, UnknownMetaMessage
|
||||||
|
from .midifiles import MidiFile
|
||||||
|
from .tracks import MidiTrack, merge_tracks
|
||||||
|
from .units import bpm2tempo, second2tick, tempo2bpm, tick2second
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"KeySignatureError",
|
||||||
|
"MetaMessage",
|
||||||
|
"MidiFile",
|
||||||
|
"MidiTrack",
|
||||||
|
"UnknownMetaMessage",
|
||||||
|
"bpm2tempo",
|
||||||
|
"merge_tracks",
|
||||||
|
"second2tick",
|
||||||
|
"tempo2bpm",
|
||||||
|
"tick2second",
|
||||||
|
]
|
||||||
@@ -0,0 +1,366 @@
|
|||||||
|
"""
|
||||||
|
Meta messages for MIDI files.
|
||||||
|
|
||||||
|
TODO:
|
||||||
|
- what if an unknown meta message is implemented and someone depends on
|
||||||
|
the 'data' attribute?
|
||||||
|
- is 'type_byte' a good name?
|
||||||
|
- 'values' is not a good name for a dictionary.
|
||||||
|
- type and value safety?
|
||||||
|
- copy().
|
||||||
|
- expose _key_signature_encode/decode?
|
||||||
|
"""
|
||||||
|
|
||||||
|
from abc import ABCMeta, abstractmethod
|
||||||
|
from collections.abc import Generator, Iterable, Sequence
|
||||||
|
from contextlib import contextmanager
|
||||||
|
from numbers import Integral
|
||||||
|
from typing import ClassVar, Literal, Never, Self, SupportsIndex, override, type_check_only
|
||||||
|
|
||||||
|
from _typeshed import ReadableBuffer
|
||||||
|
|
||||||
|
from ..messages import BaseMessage
|
||||||
|
|
||||||
|
_charset: str = "latin1"
|
||||||
|
|
||||||
|
class KeySignatureError(Exception):
|
||||||
|
"""Raised when key cannot be converted from key/mode to key letter"""
|
||||||
|
|
||||||
|
_key_signature_decode: dict[tuple[int, int], str]
|
||||||
|
_key_signature_encode: dict[str, tuple[int, int]]
|
||||||
|
_smpte_framerate_decode: dict[int, int | float]
|
||||||
|
_smpte_framerate_encode: dict[int | float, int]
|
||||||
|
|
||||||
|
type _EncodeTypeNameSigned = Literal["byte", "short", "long"]
|
||||||
|
type _EncodeTypeNameUnsigned = Literal["ubyte", "ushort", "ulong"]
|
||||||
|
|
||||||
|
def signed(to_type: _EncodeTypeNameSigned | _EncodeTypeNameUnsigned, n: int) -> int: ...
|
||||||
|
def unsigned(to_type: _EncodeTypeNameSigned, n: int) -> int: ...
|
||||||
|
def encode_variable_int(value: int) -> list[int]:
|
||||||
|
"""Encode variable length integer.
|
||||||
|
|
||||||
|
Returns the integer as a list of bytes,
|
||||||
|
where the last byte is < 128.
|
||||||
|
|
||||||
|
This is used for delta times and meta message payload
|
||||||
|
length.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def decode_variable_int(value: Iterable[int]) -> int:
|
||||||
|
"""Decode a list to a variable length integer.
|
||||||
|
|
||||||
|
Does the opposite of `encode_variable_int(value)`
|
||||||
|
"""
|
||||||
|
|
||||||
|
def encode_string(string: str) -> list[int]: ...
|
||||||
|
def decode_string(data: Iterable[SupportsIndex] | ReadableBuffer) -> str: ...
|
||||||
|
@contextmanager
|
||||||
|
def meta_charset(tmp_charset: str) -> Generator[None, Never]: ...
|
||||||
|
def check_int(value: int | Integral, low: int | Integral, high: int | Integral) -> None: ...
|
||||||
|
def check_str(value: str) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec(metaclass=ABCMeta): # not an ABC in runtime
|
||||||
|
type: str
|
||||||
|
@type_check_only
|
||||||
|
@abstractmethod
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ... # not exist in runtime
|
||||||
|
@type_check_only
|
||||||
|
@abstractmethod
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ... # not exist in runtime
|
||||||
|
def check(self, name: Never, value: Never) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_sequence_number(MetaSpec):
|
||||||
|
type: str = "sequence_number"
|
||||||
|
type_byte: int = 0x00
|
||||||
|
attributes: ClassVar[list[str]] = ["number"]
|
||||||
|
defaults: ClassVar[list[int]] = [0]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: Never, value: int | Integral) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_text(MetaSpec):
|
||||||
|
type: str = "text"
|
||||||
|
type_byte: int = 0x01
|
||||||
|
attributes: ClassVar[list[str]] = ["text"]
|
||||||
|
defaults: ClassVar[list[str]] = [""]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Iterable[SupportsIndex] | ReadableBuffer) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: Never, value: str) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_copyright(MetaSpec_text):
|
||||||
|
type: str = "copyright"
|
||||||
|
type_byte: int = 0x02
|
||||||
|
|
||||||
|
class MetaSpec_track_name(MetaSpec_text):
|
||||||
|
type: str = "track_name"
|
||||||
|
type_byte: int = 0x03
|
||||||
|
attributes: ClassVar[list[str]] = ["name"]
|
||||||
|
defaults: ClassVar[list[str]] = [""]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Iterable[SupportsIndex] | ReadableBuffer) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
|
||||||
|
class MetaSpec_instrument_name(MetaSpec_track_name):
|
||||||
|
type: str = "instrument_name"
|
||||||
|
type_byte: int = 0x04
|
||||||
|
|
||||||
|
class MetaSpec_lyrics(MetaSpec_text):
|
||||||
|
type: str = "lyrics"
|
||||||
|
type_byte: int = 0x05
|
||||||
|
|
||||||
|
class MetaSpec_marker(MetaSpec_text):
|
||||||
|
type: str = "marker"
|
||||||
|
type_byte: int = 0x06
|
||||||
|
|
||||||
|
class MetaSpec_cue_marker(MetaSpec_text):
|
||||||
|
type: str = "cue_marker"
|
||||||
|
type_byte: int = 0x07
|
||||||
|
|
||||||
|
class MetaSpec_device_name(MetaSpec_track_name):
|
||||||
|
type: str = "device_name"
|
||||||
|
type_byte: int = 0x09
|
||||||
|
|
||||||
|
class MetaSpec_channel_prefix(MetaSpec):
|
||||||
|
type: str = "channel_prefix"
|
||||||
|
type_byte: int = 0x20
|
||||||
|
attributes: ClassVar[list[str]] = ["channel"]
|
||||||
|
defaults: ClassVar[list[int]] = [0]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: Never, value: int) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_midi_port(MetaSpec):
|
||||||
|
type: str = "midi_port"
|
||||||
|
type_byte: int = 0x21
|
||||||
|
attributes: ClassVar[list[str]] = ["port"]
|
||||||
|
defaults: ClassVar[list[int]] = [0]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: Never, value: int) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_end_of_track(MetaSpec):
|
||||||
|
type: str = "end_of_track"
|
||||||
|
type_byte: int = 0x2F
|
||||||
|
attributes: ClassVar[list[Never]] = []
|
||||||
|
defaults: ClassVar[list[Never]] = []
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
|
||||||
|
class MetaSpec_set_tempo(MetaSpec):
|
||||||
|
type: str = "set_tempo"
|
||||||
|
type_byte: int = 0x51
|
||||||
|
attributes: ClassVar[list[str]] = ["tempo"]
|
||||||
|
defaults: ClassVar[list[int]] = [500000]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: Never, value: int) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_smpte_offset(MetaSpec):
|
||||||
|
type: str = "smpte_offset"
|
||||||
|
type_byte: int = 0x54
|
||||||
|
attributes: ClassVar[list[str]] = ["frame_rate", "hours", "minutes", "seconds", "frames", "sub_frames"]
|
||||||
|
# TODO: What are some good defaults?
|
||||||
|
defaults: ClassVar[list[int]] = [24, 0, 0, 0, 0, 0]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: str, value: int) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_time_signature(MetaSpec):
|
||||||
|
type: str = "time_signature"
|
||||||
|
type_byte: int = 0x58
|
||||||
|
# TODO: these need more sensible names.
|
||||||
|
attributes: ClassVar[list[str]] = ["numerator", "denominator", "clocks_per_click", "notated_32nd_notes_per_beat"]
|
||||||
|
defaults: ClassVar[list[int]] = [4, 4, 24, 8]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: str, value: int) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_key_signature(MetaSpec):
|
||||||
|
type: str = "key_signature"
|
||||||
|
type_byte: int = 0x59
|
||||||
|
attributes: ClassVar[list[str]] = ["key"]
|
||||||
|
defaults: ClassVar[list[str]] = ["C"]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Sequence[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
@override
|
||||||
|
def check(self, name: Never, value: str) -> None: ...
|
||||||
|
|
||||||
|
class MetaSpec_sequencer_specific(MetaSpec):
|
||||||
|
type: str = "sequencer_specific"
|
||||||
|
type_byte: int = 0x7F
|
||||||
|
attributes: ClassVar[list[str]] = ["data"]
|
||||||
|
defaults: ClassVar[list[Sequence[int]]]
|
||||||
|
@override
|
||||||
|
def decode(self, message: MetaMessage, data: Iterable[int]) -> None: ...
|
||||||
|
@override
|
||||||
|
def encode(self, message: MetaMessage) -> list[int]: ...
|
||||||
|
|
||||||
|
def add_meta_spec(klass: type[MetaSpec]) -> None: ...
|
||||||
|
|
||||||
|
_META_SPECS: dict[int | str, type[MetaSpec]]
|
||||||
|
_META_SPEC_BY_TYPE: dict[str, type[MetaSpec]]
|
||||||
|
|
||||||
|
def build_meta_message(
|
||||||
|
meta_type: int | str, data: Sequence[int], delta: int = 0
|
||||||
|
) -> MetaMessage | UnknownMetaMessage: ...
|
||||||
|
|
||||||
|
# NOTE: Generics not supported in runtime
|
||||||
|
class _MetaMessage[T: str = str](BaseMessage):
|
||||||
|
type: T
|
||||||
|
time: int
|
||||||
|
@property
|
||||||
|
@override
|
||||||
|
def is_meta(self) -> Literal[True]: ... # a normal attribute in runtime
|
||||||
|
def __init__(self, type: str, skip_checks: bool = False, **kwargs: object) -> None: ...
|
||||||
|
@override
|
||||||
|
def copy(self, **overrides: object) -> Self:
|
||||||
|
"""Return a copy of the message
|
||||||
|
|
||||||
|
Attributes will be overridden by the passed keyword arguments.
|
||||||
|
Only message specific attributes can be overridden. The message
|
||||||
|
type can not be changed.
|
||||||
|
"""
|
||||||
|
@override
|
||||||
|
def __setattr__(self, name: str, value: object) -> None: ...
|
||||||
|
@override
|
||||||
|
def bytes(self) -> list[int]: ...
|
||||||
|
@classmethod
|
||||||
|
def from_bytes(cls, msg_bytes: Sequence[int]) -> MetaMessage | UnknownMetaMessage: ...
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class SequenceNumberMetaMessage(_MetaMessage[Literal["sequence_number"]]):
|
||||||
|
number: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class TextMetaMessage(_MetaMessage[Literal["text"]]):
|
||||||
|
text: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class CopyrightMetaMessage(_MetaMessage[Literal["copyright"]]):
|
||||||
|
text: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class TrackNameMetaMessage(_MetaMessage[Literal["track_name"]]):
|
||||||
|
name: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class InstrumentNameMetaMessage(_MetaMessage[Literal["instrument_name"]]):
|
||||||
|
name: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class LyricsMetaMessage(_MetaMessage[Literal["lyrics"]]):
|
||||||
|
text: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class MarkerMetaMessage(_MetaMessage[Literal["marker"]]):
|
||||||
|
text: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class CueMarkerMetaMessage(_MetaMessage[Literal["cue_marker"]]):
|
||||||
|
text: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class DeviceNameMetaMessage(_MetaMessage[Literal["device_name"]]):
|
||||||
|
name: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class ChannelPrefixMetaMessage(_MetaMessage[Literal["channel_prefix"]]):
|
||||||
|
channel: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class MidiPortMetaMessage(_MetaMessage[Literal["midi_port"]]):
|
||||||
|
port: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class EndOfTrackMetaMessage(_MetaMessage[Literal["end_of_track"]]): ...
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class SetTempoMetaMessage(_MetaMessage[Literal["set_tempo"]]):
|
||||||
|
tempo: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class SmpteOffsetMetaMessage(_MetaMessage[Literal["smpte_offset"]]):
|
||||||
|
frame_rate: int
|
||||||
|
hours: int
|
||||||
|
minutes: int
|
||||||
|
seconds: int
|
||||||
|
frames: int
|
||||||
|
sub_frames: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class TimeSignatureMetaMessage(_MetaMessage[Literal["time_signature"]]):
|
||||||
|
numerator: int
|
||||||
|
denominator: int
|
||||||
|
clocks_per_click: int
|
||||||
|
notated_32nd_notes_per_beat: int
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class KeySignatureMetaMessage(_MetaMessage[Literal["key_signature"]]):
|
||||||
|
key: str
|
||||||
|
|
||||||
|
@type_check_only
|
||||||
|
class SequencerSpecificMetaMessage(_MetaMessage[Literal["sequencer_specific"]]):
|
||||||
|
data: Sequence[int]
|
||||||
|
|
||||||
|
class UnknownMetaMessage(_MetaMessage):
|
||||||
|
type_byte: Sequence[int]
|
||||||
|
data: tuple[int, ...]
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
type_byte: Sequence[int],
|
||||||
|
data: Sequence[int] | None = ...,
|
||||||
|
time: int = 0,
|
||||||
|
type: str = "unknown_meta",
|
||||||
|
**kwargs: Never,
|
||||||
|
) -> None: ...
|
||||||
|
@override
|
||||||
|
def __setattr__(self, name: str, value: object) -> None: ...
|
||||||
|
@override
|
||||||
|
def bytes(self) -> list[int]: ...
|
||||||
|
|
||||||
|
# NOTE: `MetaMessage` is a concrete type (actually the `_MetaMessage` in this stub) in runtime
|
||||||
|
type MetaMessage = (
|
||||||
|
SequenceNumberMetaMessage
|
||||||
|
| TextMetaMessage
|
||||||
|
| CopyrightMetaMessage
|
||||||
|
| TrackNameMetaMessage
|
||||||
|
| InstrumentNameMetaMessage
|
||||||
|
| LyricsMetaMessage
|
||||||
|
| MarkerMetaMessage
|
||||||
|
| CueMarkerMetaMessage
|
||||||
|
| DeviceNameMetaMessage
|
||||||
|
| ChannelPrefixMetaMessage
|
||||||
|
| MidiPortMetaMessage
|
||||||
|
| EndOfTrackMetaMessage
|
||||||
|
| SetTempoMetaMessage
|
||||||
|
| SmpteOffsetMetaMessage
|
||||||
|
| TimeSignatureMetaMessage
|
||||||
|
| KeySignatureMetaMessage
|
||||||
|
| SequencerSpecificMetaMessage
|
||||||
|
)
|
||||||
@@ -0,0 +1,154 @@
|
|||||||
|
"""
|
||||||
|
MIDI file reading and playback.
|
||||||
|
|
||||||
|
References:
|
||||||
|
|
||||||
|
http://home.roadrunner.com/~jgglatt/
|
||||||
|
http://home.roadrunner.com/~jgglatt/tech/miditech.htm
|
||||||
|
http://home.roadrunner.com/~jgglatt/tech/midifile.htm
|
||||||
|
|
||||||
|
http://www.sonicspot.com/guide/midifiles.html
|
||||||
|
http://www.ccarh.org/courses/253/assignment/midifile/
|
||||||
|
https://code.google.com/p/binasc/wiki/mainpage
|
||||||
|
http://stackoverflow.com/questions/2984608/midi-delta-time
|
||||||
|
http://www.recordingblogs.com/sa/tabid/82/EntryId/44/MIDI-Part-XIII-Delta-time-a
|
||||||
|
http://www.sonicspot.com/guide/midifiles.html
|
||||||
|
"""
|
||||||
|
|
||||||
|
from collections.abc import Callable, Generator, Iterable, Sequence
|
||||||
|
from numbers import Real
|
||||||
|
from typing import IO, Literal, Never, Self, overload
|
||||||
|
|
||||||
|
from ..messages import BaseMessage, Message
|
||||||
|
from .meta import MetaMessage, UnknownMetaMessage
|
||||||
|
from .tracks import MidiTrack
|
||||||
|
|
||||||
|
type _RealNumber = int | float | Real
|
||||||
|
|
||||||
|
# The default tempo is 120 BPM.
|
||||||
|
# (500000 microseconds per beat (quarter note).)
|
||||||
|
DEFAULT_TEMPO: int = 500000
|
||||||
|
DEFAULT_TICKS_PER_BEAT: int = 480
|
||||||
|
|
||||||
|
# Maximum message length to attempt to read.
|
||||||
|
MAX_MESSAGE_LENGTH: int = 1000000
|
||||||
|
|
||||||
|
def print_byte(byte: int, pos: int = 0) -> None: ...
|
||||||
|
|
||||||
|
class DebugFileWrapper:
|
||||||
|
def __init__(self, file: IO[bytes]) -> None: ...
|
||||||
|
def read(self, size: int) -> bytes: ...
|
||||||
|
def tell(self) -> int: ...
|
||||||
|
|
||||||
|
def read_byte(self: IO[bytes]) -> int: ...
|
||||||
|
def read_bytes(infile: IO[bytes], size: int) -> list[int]: ...
|
||||||
|
def read_chunk_header(infile: IO[bytes]) -> tuple[str, int]: ...
|
||||||
|
def read_file_header(infile: IO[bytes]) -> tuple[int, int, int]: ...
|
||||||
|
def read_message(
|
||||||
|
infile: IO[bytes], status_byte: int, peek_data: Sequence[int], delta: int, clip: bool = False
|
||||||
|
) -> Message: ...
|
||||||
|
def read_sysex(infile: IO[bytes], delta: int, clip: bool = False) -> Message: ...
|
||||||
|
def read_variable_int(infile: IO[bytes]) -> int: ...
|
||||||
|
def read_meta_message(infile: IO[bytes], delta: int) -> MetaMessage | UnknownMetaMessage: ...
|
||||||
|
def read_track(infile: IO[bytes], debug: bool = False, clip: bool = False) -> MidiTrack: ...
|
||||||
|
def write_chunk(outfile: IO[bytes], name: bytes, data: bytes) -> None:
|
||||||
|
"""Write an IFF chunk to the file.
|
||||||
|
|
||||||
|
`name` must be a bytestring."""
|
||||||
|
|
||||||
|
def write_track(outfile: IO[bytes], track: Iterable[BaseMessage]) -> None: ...
|
||||||
|
def get_seconds_per_tick(tempo: _RealNumber, ticks_per_beat: _RealNumber) -> float: ...
|
||||||
|
|
||||||
|
class MidiFile:
|
||||||
|
filename: str | None
|
||||||
|
type: Literal[0, 1, 2]
|
||||||
|
ticks_per_beat: int
|
||||||
|
charset: str
|
||||||
|
debug: bool
|
||||||
|
clip: bool
|
||||||
|
tracks: list[MidiTrack]
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
filename: str | None = ...,
|
||||||
|
file: IO[bytes] | None = ...,
|
||||||
|
type: Literal[0, 1, 2] = 1,
|
||||||
|
ticks_per_beat: int = DEFAULT_TICKS_PER_BEAT, # noqa: PYI011
|
||||||
|
charset: str = "latin1",
|
||||||
|
debug: bool = False,
|
||||||
|
clip: bool = False,
|
||||||
|
tracks: list[MidiTrack] | None = ...,
|
||||||
|
) -> None: ...
|
||||||
|
@property
|
||||||
|
def merged_track(self) -> MidiTrack: ...
|
||||||
|
@merged_track.deleter
|
||||||
|
def merged_track(self) -> None: ...
|
||||||
|
def add_track(self, name: str | None = ...) -> MidiTrack:
|
||||||
|
"""Add a new track to the file.
|
||||||
|
|
||||||
|
This will create a new `MidiTrack` object and append it to the
|
||||||
|
track list.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@property
|
||||||
|
def length(self) -> int | float:
|
||||||
|
"""Playback time in seconds.
|
||||||
|
|
||||||
|
This will be computed by going through every message in every
|
||||||
|
track and adding up delta times.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __iter__(self) -> Generator[Message | MetaMessage | UnknownMetaMessage, Never]: ...
|
||||||
|
@overload
|
||||||
|
def play(
|
||||||
|
self, meta_messages: Literal[False] = False, now: Callable[[], float] = ...
|
||||||
|
) -> Generator[Message, Never]: ...
|
||||||
|
@overload
|
||||||
|
def play(
|
||||||
|
self, meta_messages: bool, now: Callable[[], float] = ...
|
||||||
|
) -> Generator[Message | MetaMessage | UnknownMetaMessage, Never]: ...
|
||||||
|
def play(
|
||||||
|
self, meta_messages: bool = False, now: Callable[[], float] = ...
|
||||||
|
) -> Generator[Message | MetaMessage | UnknownMetaMessage, Never]:
|
||||||
|
"""Play back all tracks.
|
||||||
|
|
||||||
|
The generator will sleep between each message by
|
||||||
|
default. Messages are yielded with correct timing. The time
|
||||||
|
attribute is set to the number of seconds slept since the
|
||||||
|
previous message.
|
||||||
|
|
||||||
|
By default you will only get normal MIDI messages. Pass
|
||||||
|
`meta_messages=True` if you also want meta messages.
|
||||||
|
|
||||||
|
You will receive copies of the original messages, so you can
|
||||||
|
safely modify them without ruining the tracks.
|
||||||
|
|
||||||
|
By default the system clock is used for the timing of yielded
|
||||||
|
MIDI events. To use a different clock (e.g. to synchronize to
|
||||||
|
an audio stream), pass `now=time_fn` where `time_fn` is a zero
|
||||||
|
argument function that yields the current time in seconds.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def save(self, filename: str | None = ..., file: IO[bytes] | None = ...) -> None:
|
||||||
|
"""Save to a file.
|
||||||
|
|
||||||
|
If file is passed the data will be saved to that file. This is
|
||||||
|
typically an in-memory file or and already open file like `sys.stdout`.
|
||||||
|
|
||||||
|
If filename is passed the data will be saved to that file.
|
||||||
|
|
||||||
|
Raises `ValueError` if both file and filename are None,
|
||||||
|
or if a type 0 file has != one track.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def print_tracks(self, meta_only: bool = False) -> None:
|
||||||
|
"""Prints out all messages in a .midi file.
|
||||||
|
|
||||||
|
May take argument `meta_only` to show only meta messages.
|
||||||
|
|
||||||
|
Use:
|
||||||
|
`print_tracks()` -> will print all messages
|
||||||
|
`print_tracks(meta_only=True)` -> will print only MetaMessages
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __enter__(self) -> Self: ...
|
||||||
|
def __exit__(self, type: object, value: object, traceback: object) -> Literal[False]: ...
|
||||||
@@ -0,0 +1,46 @@
|
|||||||
|
from collections.abc import Generator, Iterable, Sequence
|
||||||
|
from typing import Never, Self, SupportsIndex, overload, override
|
||||||
|
|
||||||
|
from ..messages import BaseMessage
|
||||||
|
|
||||||
|
class MidiTrack(list[BaseMessage]):
|
||||||
|
@property
|
||||||
|
def name(self) -> str:
|
||||||
|
"""Name of the track.
|
||||||
|
|
||||||
|
This will return the name from the first `track_name` meta
|
||||||
|
message in the track, or `''` if there is no such message.
|
||||||
|
|
||||||
|
Setting this property will update the name field of the first
|
||||||
|
`track_name` message in the track. If no such message is found,
|
||||||
|
one will be added to the beginning of the track with a delta
|
||||||
|
time of 0."""
|
||||||
|
|
||||||
|
@name.setter
|
||||||
|
def name(self, name: str) -> None: ...
|
||||||
|
@override
|
||||||
|
def copy(self) -> Self: ...
|
||||||
|
@overload
|
||||||
|
def __getitem__(self, index_or_slice: SupportsIndex) -> BaseMessage: ...
|
||||||
|
@overload
|
||||||
|
def __getitem__(self, index_or_slice: slice[int, int, int]) -> Self: ...
|
||||||
|
@override
|
||||||
|
def __add__(self, other: Sequence[BaseMessage]) -> Self: ... # pyright: ignore[reportIncompatibleMethodOverride]
|
||||||
|
@override
|
||||||
|
def __mul__(self, other: SupportsIndex) -> Self: ...
|
||||||
|
|
||||||
|
def fix_end_of_track(messages: Iterable[BaseMessage], skip_checks: bool = False) -> Generator[BaseMessage, Never]:
|
||||||
|
"""Remove all end_of_track messages and add one at the end.
|
||||||
|
|
||||||
|
This is used by `merge_tracks()` and `MidiFile.save()`."""
|
||||||
|
|
||||||
|
def merge_tracks(tracks: Iterable[MidiTrack], skip_checks: bool = False) -> MidiTrack:
|
||||||
|
"""Returns a MidiTrack object with all messages from all tracks.
|
||||||
|
|
||||||
|
The messages are returned in playback order with delta times
|
||||||
|
as if they were all in one track.
|
||||||
|
|
||||||
|
Pass `skip_checks=True` to skip validation of messages before merging.
|
||||||
|
This should ONLY be used when the messages in tracks have already
|
||||||
|
been validated by mido.checks.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,38 @@
|
|||||||
|
from numbers import Real
|
||||||
|
|
||||||
|
type _RealNumber = int | float | Real
|
||||||
|
type _TimeSignature = tuple[int, int]
|
||||||
|
|
||||||
|
def tick2second(tick: _RealNumber, ticks_per_beat: _RealNumber, tempo: _RealNumber) -> float:
|
||||||
|
"""Convert absolute time in ticks to seconds.
|
||||||
|
|
||||||
|
Returns absolute time in seconds for a chosen MIDI file time resolution
|
||||||
|
(ticks/pulses per quarter note, also called PPQN) and tempo (microseconds
|
||||||
|
per quarter note).
|
||||||
|
"""
|
||||||
|
|
||||||
|
def second2tick(second: _RealNumber, ticks_per_beat: _RealNumber, tempo: _RealNumber) -> int:
|
||||||
|
"""Convert absolute time in seconds to ticks.
|
||||||
|
|
||||||
|
Returns absolute time in ticks for a chosen MIDI file time resolution
|
||||||
|
(ticks/pulses per quarter note, also called PPQN) and tempo (microseconds
|
||||||
|
per quarter note). Normal rounding applies.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def bpm2tempo(bpm: _RealNumber, time_signature: _TimeSignature = (4, 4)) -> int:
|
||||||
|
"""Convert BPM (beats per minute) to MIDI file tempo (microseconds per
|
||||||
|
quarter note).
|
||||||
|
|
||||||
|
Depending on the chosen time signature a bar contains a different number of
|
||||||
|
beats. These beats are multiples/fractions of a quarter note, thus the
|
||||||
|
returned BPM depend on the time signature. Normal rounding applies.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def tempo2bpm(tempo: _RealNumber, time_signature: _TimeSignature = (4, 4)) -> float:
|
||||||
|
"""Convert MIDI file tempo (microseconds per quarter note) to BPM (beats
|
||||||
|
per minute).
|
||||||
|
|
||||||
|
Depending on the chosen time signature a bar contains a different number of
|
||||||
|
beats. The beats are multiples/fractions of a quarter note, thus the
|
||||||
|
returned tempo depends on the time signature denominator.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,81 @@
|
|||||||
|
"""
|
||||||
|
MIDI Parser
|
||||||
|
|
||||||
|
There is no need to use this module directly. All you need is
|
||||||
|
available in the top level module.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from collections import deque
|
||||||
|
from collections.abc import Generator, Iterable
|
||||||
|
from numbers import Integral
|
||||||
|
from typing import Never
|
||||||
|
|
||||||
|
from .messages import Message
|
||||||
|
|
||||||
|
class Parser:
|
||||||
|
"""
|
||||||
|
MIDI byte stream parser
|
||||||
|
|
||||||
|
Parses a stream of MIDI bytes and produces messages.
|
||||||
|
|
||||||
|
Data can be put into the parser in the form of
|
||||||
|
integers, byte arrays or byte strings.
|
||||||
|
"""
|
||||||
|
|
||||||
|
messages: deque[Message]
|
||||||
|
|
||||||
|
def __init__(self, data: Iterable[int] | None = ...) -> None: ...
|
||||||
|
def feed(self, data: Iterable[int]) -> None:
|
||||||
|
"""Feed MIDI data to the parser.
|
||||||
|
|
||||||
|
Accepts any object that produces a sequence of integers in
|
||||||
|
range 0..255, such as:
|
||||||
|
|
||||||
|
```
|
||||||
|
[0, 1, 2]
|
||||||
|
(0, 1, 2)
|
||||||
|
[for i in range(256)]
|
||||||
|
(for i in range(256)]
|
||||||
|
bytearray()
|
||||||
|
```
|
||||||
|
"""
|
||||||
|
|
||||||
|
def feed_byte(self, byte: int | Integral) -> None:
|
||||||
|
"""Feed one MIDI byte into the parser.
|
||||||
|
|
||||||
|
The byte must be an integer in range 0..255.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def get_message(self) -> Message | None:
|
||||||
|
"""Get the first parsed message.
|
||||||
|
|
||||||
|
Returns None if there is no message yet. If you don't want to
|
||||||
|
deal with None, you can use `pending()` to see how many messages
|
||||||
|
you can get before you get None, or just iterate over the
|
||||||
|
parser.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def pending(self) -> int:
|
||||||
|
"""Return the number of pending messages."""
|
||||||
|
|
||||||
|
def __len__(self) -> int:
|
||||||
|
"""Return the number of pending messages."""
|
||||||
|
|
||||||
|
def __iter__(self) -> Generator[Message, Never]:
|
||||||
|
"""Yield messages that have been parsed so far."""
|
||||||
|
|
||||||
|
def parse_all(data: Iterable[int] | None) -> list[Message]:
|
||||||
|
"""Parse MIDI data and return a list of all messages found.
|
||||||
|
|
||||||
|
This is typically used to parse a little bit of data with a few
|
||||||
|
messages in it. It's best to use a Parser object for larger
|
||||||
|
amounts of data. Also, tt's often easier to use `parse()` if you
|
||||||
|
know there is only one message in the data.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def parse(data: Iterable[int] | None) -> Message | None:
|
||||||
|
"""Parse MIDI data and return the first message found.
|
||||||
|
|
||||||
|
Data after the first message is ignored. Use `parse_all()`
|
||||||
|
to parse more than one message.
|
||||||
|
"""
|
||||||
@@ -0,0 +1,234 @@
|
|||||||
|
"""
|
||||||
|
Useful tools for working with ports
|
||||||
|
"""
|
||||||
|
|
||||||
|
import threading
|
||||||
|
from collections import deque
|
||||||
|
from collections.abc import Collection, Generator, Iterable
|
||||||
|
from typing import Any, Literal, Never, Self, overload
|
||||||
|
|
||||||
|
from .messages import Message
|
||||||
|
from .parser import Parser
|
||||||
|
|
||||||
|
# How many seconds to sleep before polling again.
|
||||||
|
_DEFAULT_SLEEP_TIME: float
|
||||||
|
_sleep_time: int | float
|
||||||
|
|
||||||
|
def sleep() -> None:
|
||||||
|
"""Sleep for N seconds.
|
||||||
|
|
||||||
|
This is used in ports when polling and waiting for messages. N can
|
||||||
|
be set with `set_sleep_time()`."""
|
||||||
|
|
||||||
|
def set_sleep_time(seconds: int | float = ...) -> None:
|
||||||
|
"""Set the number of seconds `sleep()` will sleep."""
|
||||||
|
|
||||||
|
def get_sleep_time() -> int | float:
|
||||||
|
"""Get number of seconds `sleep()` will sleep."""
|
||||||
|
|
||||||
|
def reset_messages() -> Generator[Message, Never]:
|
||||||
|
"""Yield "All Notes Off" and "Reset All Controllers" for all channels"""
|
||||||
|
|
||||||
|
def panic_messages() -> Generator[Message, Never]:
|
||||||
|
"""Yield "All Sounds Off" for all channels.
|
||||||
|
|
||||||
|
This will mute all sounding notes regardless of
|
||||||
|
envelopes. Useful when notes are hanging and nothing else
|
||||||
|
helps.
|
||||||
|
"""
|
||||||
|
|
||||||
|
class DummyLock:
|
||||||
|
def __enter__(self) -> Self: ...
|
||||||
|
def __exit__(self, *_: object) -> Literal[False]: ...
|
||||||
|
|
||||||
|
class BasePort:
|
||||||
|
"""
|
||||||
|
Abstract base class for Input and Output ports.
|
||||||
|
"""
|
||||||
|
|
||||||
|
is_input: bool = False
|
||||||
|
is_output: bool = False
|
||||||
|
_locking: bool = True
|
||||||
|
|
||||||
|
name: bool
|
||||||
|
closed: bool
|
||||||
|
_lock: DummyLock | threading.RLock
|
||||||
|
|
||||||
|
def __init__(self, name: str | None = None, **kwargs: object) -> None: ...
|
||||||
|
def _open(self, **kwargs: object) -> None: ...
|
||||||
|
def _close(self) -> None: ...
|
||||||
|
def close(self) -> None:
|
||||||
|
"""Close the port.
|
||||||
|
|
||||||
|
If the port is already closed, nothing will happen. The port
|
||||||
|
is automatically closed when the object goes out of scope or
|
||||||
|
is garbage collected.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __del__(self) -> None: ...
|
||||||
|
def __enter__(self) -> Self: ...
|
||||||
|
def __exit__(self, *_) -> Literal[False]: ...
|
||||||
|
|
||||||
|
class BaseInput(BasePort):
|
||||||
|
"""Base class for input port.
|
||||||
|
|
||||||
|
Subclass and override `_receive()` to create a new input port type.
|
||||||
|
(See `portmidi.py` for an example of how to do this.)
|
||||||
|
"""
|
||||||
|
|
||||||
|
is_input: bool = True
|
||||||
|
|
||||||
|
_parser: Parser
|
||||||
|
_messages: deque[Message]
|
||||||
|
|
||||||
|
def __init__(self, name: str = "", **kwargs: Any) -> None:
|
||||||
|
"""Create an input port.
|
||||||
|
|
||||||
|
name is the port name, as returned by `input_names()`. If
|
||||||
|
name is not passed, the default input is used instead.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def _receive(self, block: bool = True) -> Message: ...
|
||||||
|
def iter_pending(self) -> Generator[Message, Never]:
|
||||||
|
"""Iterate through pending messages."""
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def receive(self, block: Literal[True] = True) -> Message:
|
||||||
|
"""Return the next message.
|
||||||
|
|
||||||
|
This will block until a message arrives.
|
||||||
|
|
||||||
|
If you pass `block=False` it will not block and instead return
|
||||||
|
None if there is no available message.
|
||||||
|
|
||||||
|
If the port is closed and there are no pending messages IOError
|
||||||
|
will be raised. If the port closes while waiting inside `receive()`,
|
||||||
|
IOError will be raised. TODO: this seems a bit inconsistent. Should
|
||||||
|
different errors be raised? What's most useful here?
|
||||||
|
"""
|
||||||
|
@overload
|
||||||
|
def receive(self, block: bool) -> Message | None: ...
|
||||||
|
def poll(self) -> Message | None:
|
||||||
|
"""Receive the next pending message or None
|
||||||
|
|
||||||
|
This is the same as calling `receive(block=False)`."""
|
||||||
|
|
||||||
|
def __iter__(self) -> Generator[Message, Never]:
|
||||||
|
"""Iterate through messages until the port closes."""
|
||||||
|
|
||||||
|
class BaseOutput(BasePort):
|
||||||
|
"""
|
||||||
|
Base class for output port.
|
||||||
|
|
||||||
|
Subclass and override `_send()` to create a new port type. (See
|
||||||
|
`portmidi.py` for how to do this.)
|
||||||
|
"""
|
||||||
|
|
||||||
|
is_output: bool = True
|
||||||
|
|
||||||
|
autoreset: bool
|
||||||
|
|
||||||
|
def __init__(self, name: str = "", autoreset: bool = False, **kwargs: object) -> None:
|
||||||
|
"""Create an output port
|
||||||
|
|
||||||
|
name is the port name, as returned by `output_names()`. If
|
||||||
|
name is not passed, the default output is used instead.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def _send(self, msg: Message) -> None: ...
|
||||||
|
def send(self, msg: Message) -> None:
|
||||||
|
"""Send a message on the port.
|
||||||
|
|
||||||
|
A copy of the message will be sent, so you can safely modify
|
||||||
|
the original message without any unexpected consequences.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def reset(self) -> None:
|
||||||
|
"""Send "All Notes Off" and "Reset All Controllers" on all channels"""
|
||||||
|
|
||||||
|
def panic(self) -> None:
|
||||||
|
"""Send "All Sounds Off" on all channels.
|
||||||
|
|
||||||
|
This will mute all sounding notes regardless of
|
||||||
|
envelopes. Useful when notes are hanging and nothing else
|
||||||
|
helps.
|
||||||
|
"""
|
||||||
|
|
||||||
|
class BaseIOPort(BaseInput, BaseOutput): # pyright: ignore[reportUnsafeMultipleInheritance]
|
||||||
|
def __init__(self, name: str = "", **kwargs: object) -> None:
|
||||||
|
"""Create an IO port.
|
||||||
|
|
||||||
|
name is the port name, as returned by ioport_names().
|
||||||
|
"""
|
||||||
|
|
||||||
|
# NOTE: Generics not supported in runtime
|
||||||
|
class IOPort[InT: BaseInput = BaseInput, OutT: BaseOutput = BaseOutput](BaseIOPort):
|
||||||
|
"""Input / output port.
|
||||||
|
|
||||||
|
This is a convenient wrapper around an input port and an output
|
||||||
|
port which provides the functionality of both. Every method call
|
||||||
|
is forwarded to the appropriate port.
|
||||||
|
"""
|
||||||
|
|
||||||
|
_locking: bool = False
|
||||||
|
|
||||||
|
input: InT
|
||||||
|
output: OutT
|
||||||
|
|
||||||
|
def __init__(self, input: InT, output: OutT) -> None: ...
|
||||||
|
|
||||||
|
class EchoPort(BaseIOPort): ...
|
||||||
|
|
||||||
|
class MultiPort(BaseIOPort):
|
||||||
|
def __init__(self, ports: Iterable[BaseIOPort], yield_ports: bool = False) -> None: ...
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def multi_receive[T: BaseInput](
|
||||||
|
ports: Collection[T], yield_ports: Literal[True], block: bool = True
|
||||||
|
) -> Generator[tuple[T, Message], Never]: ...
|
||||||
|
@overload
|
||||||
|
def multi_receive(
|
||||||
|
ports: Collection[BaseInput], yield_ports: Literal[False] = False, block: bool = True
|
||||||
|
) -> Generator[Message, Never]: ...
|
||||||
|
@overload
|
||||||
|
def multi_receive[T: BaseInput](
|
||||||
|
ports: Collection[T], yield_ports: bool, block: bool = True
|
||||||
|
) -> Generator[tuple[T, Message] | Message, Never]: ...
|
||||||
|
def multi_receive[T: BaseInput](
|
||||||
|
ports: Collection[T], yield_ports: bool, block: bool = True
|
||||||
|
) -> Generator[tuple[T, Message] | Message, Never]:
|
||||||
|
"""Receive messages from multiple ports.
|
||||||
|
|
||||||
|
Generates messages from ever input port. The ports are polled in
|
||||||
|
random order for fairness, and all messages from each port are
|
||||||
|
yielded before moving on to the next port.
|
||||||
|
|
||||||
|
If `yield_ports=True`, `(port, message)` is yielded instead of just
|
||||||
|
the message.
|
||||||
|
|
||||||
|
If `block=False` only pending messages will be yielded.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def multi_iter_pending[T: BaseInput](
|
||||||
|
ports: Collection[T], yield_ports: Literal[True]
|
||||||
|
) -> Generator[tuple[T, Message], Never]: ...
|
||||||
|
@overload
|
||||||
|
def multi_iter_pending(
|
||||||
|
ports: Collection[BaseInput], yield_ports: Literal[False] = False
|
||||||
|
) -> Generator[Message, Never]: ...
|
||||||
|
@overload
|
||||||
|
def multi_iter_pending[T: BaseInput](
|
||||||
|
ports: Collection[T], yield_ports: bool
|
||||||
|
) -> Generator[tuple[T, Message] | Message, Never]: ...
|
||||||
|
def multi_iter_pending[T: BaseInput](
|
||||||
|
ports: Collection[T], yield_ports: bool
|
||||||
|
) -> Generator[tuple[T, Message] | Message, Never]:
|
||||||
|
"""Iterate through all pending messages in ports.
|
||||||
|
|
||||||
|
This is the same as calling `multi_receive(ports, block=False)`.
|
||||||
|
The function is kept around for backwards compatability.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def multi_send(ports: Iterable[BaseOutput], msg: Message) -> None:
|
||||||
|
"""Send message on all ports."""
|
||||||
@@ -0,0 +1,4 @@
|
|||||||
|
"""
|
||||||
|
This type stub file was generated by pyright.
|
||||||
|
"""
|
||||||
|
|
||||||
@@ -0,0 +1,43 @@
|
|||||||
|
"""
|
||||||
|
MIDI over TCP/IP.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import socket
|
||||||
|
from typing import Literal, overload
|
||||||
|
|
||||||
|
from .ports import BaseIOPort, MultiPort
|
||||||
|
|
||||||
|
class PortServer(MultiPort):
|
||||||
|
def __init__(self, host: str, portno: int, backlog: int = 1) -> None: ...
|
||||||
|
@overload
|
||||||
|
def accept(self, block: Literal[True] = True) -> SocketPort:
|
||||||
|
"""
|
||||||
|
Accept a connection from a client.
|
||||||
|
|
||||||
|
Will block until there is a new connection, and then return a
|
||||||
|
`SocketPort` object.
|
||||||
|
|
||||||
|
If `block=False`, `None` will be returned if there is no
|
||||||
|
new connection waiting.
|
||||||
|
"""
|
||||||
|
@overload
|
||||||
|
def accept(self, block: bool) -> SocketPort | None: ...
|
||||||
|
|
||||||
|
class SocketPort(BaseIOPort):
|
||||||
|
def __init__(self, host: str, portno: int, conn: socket.socket | None = ...) -> None: ...
|
||||||
|
|
||||||
|
def connect(host: str, portno: int) -> SocketPort:
|
||||||
|
"""Connect to a socket port server.
|
||||||
|
|
||||||
|
The return value is a `SocketPort` object connected to another
|
||||||
|
`SocketPort` object at the server end. Messages can be sent either way.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def parse_address(address: str) -> tuple[str, int]:
|
||||||
|
"""Parse and address on the format host:port.
|
||||||
|
|
||||||
|
Returns a tuple (host, port). Raises ValueError if format is
|
||||||
|
invalid or port is not an integer or out of range.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def format_address(host: str, portno: int) -> str: ...
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
"""
|
||||||
|
Read and write SYX file format
|
||||||
|
"""
|
||||||
|
|
||||||
|
from _typeshed import FileDescriptorOrPath
|
||||||
|
|
||||||
|
from .messages import Message
|
||||||
|
|
||||||
|
def read_syx_file(filename: FileDescriptorOrPath) -> list[Message]:
|
||||||
|
"""Read sysex messages from SYX file.
|
||||||
|
|
||||||
|
Returns a list of sysex messages.
|
||||||
|
|
||||||
|
This handles both the text (hexadecimal) and binary
|
||||||
|
formats. Messages other than sysex will be ignored. Raises
|
||||||
|
ValueError if file is plain text and byte is not a 2-digit hex
|
||||||
|
number.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def write_syx_file(filename: FileDescriptorOrPath, messages: list[Message], plaintext: bool = False) -> None:
|
||||||
|
"""Write sysex messages to a SYX file.
|
||||||
|
|
||||||
|
Messages other than sysex will be skipped.
|
||||||
|
|
||||||
|
By default this will write the binary format. Pass
|
||||||
|
`plaintext=True` to write the plain text format (hex encoded
|
||||||
|
ASCII text).
|
||||||
|
"""
|
||||||
@@ -0,0 +1,26 @@
|
|||||||
|
from collections.abc import Generator, Iterable
|
||||||
|
from numbers import Integral
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
class Tokenizer:
|
||||||
|
"""
|
||||||
|
Splits a MIDI byte stream into messages.
|
||||||
|
"""
|
||||||
|
def __init__(self, data: Iterable[int] | None = ...) -> None:
|
||||||
|
"""Create a new decoder."""
|
||||||
|
|
||||||
|
def feed_byte(self, byte: int | Integral) -> None:
|
||||||
|
"""Feed MIDI byte to the decoder.
|
||||||
|
|
||||||
|
Takes an int in range [0..255].
|
||||||
|
"""
|
||||||
|
|
||||||
|
def feed(self, data: Iterable[int]) -> None:
|
||||||
|
"""Feed MIDI bytes to the decoder.
|
||||||
|
|
||||||
|
Takes an iterable of ints in in range [0..255].
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __len__(self) -> int: ...
|
||||||
|
def __iter__(self) -> Generator[Any, Any]:
|
||||||
|
"""Yield messages that have been parsed so far."""
|
||||||
@@ -0,0 +1,4 @@
|
|||||||
|
from packaging.version import Version
|
||||||
|
|
||||||
|
__version__: str
|
||||||
|
version_info: Version
|
||||||
Reference in New Issue
Block a user