WinRT MIDI Transfer with python winrt

winrt_midi_transfer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WinRT MIDI Transfer with python winrt.

https://gist.github.com/trueroad/6beaf87280afb2b5e33b4838d73be6ed

Copyright (C) 2022 Masamichi Hosoda.
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

* Redistributions of source code must retain the above copyright notice,
  this list of conditions and the following disclaimer.

* Redistributions in binary form must reproduce the above copyright notice,
  this list of conditions and the following disclaimer in the documentation
  and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED.
IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.
"""

# 参考
# 似たような機能を C++/WinRT で実装したもの
# https://gist.github.com/trueroad/9c5317af5f212b2de7c7012e76b9e66b

import asyncio
import re
from typing import Any, Dict, List, Optional, Tuple, Union

from winrt.windows.devices.enumeration import (  # type: ignore[import]
    DeviceInformation, DeviceInformationCollection
)
from winrt.windows.devices.midi import (  # type: ignore[import]
    MidiInPort, MidiOutPort,
    MidiMessageReceivedEventArgs,
    IMidiMessage, MidiMessageType,
    MidiNoteOffMessage, MidiNoteOnMessage, MidiPolyphonicKeyPressureMessage,
    MidiControlChangeMessage, MidiProgramChangeMessage,
    MidiChannelPressureMessage, MidiPitchBendChangeMessage,
    MidiSystemExclusiveMessage,
    MidiTimeCodeMessage, MidiSongPositionPointerMessage,
    MidiSongSelectMessage, MidiTuneRequestMessage,
    MidiTimingClockMessage, MidiStartMessage, MidiContinueMessage,
    MidiStopMessage, MidiActiveSensingMessage, MidiSystemResetMessage
)
from winrt.windows.foundation import TimeSpan  # type: ignore[import]
from winrt.windows.security.cryptography import (  # type: ignore[import]
        CryptographicBuffer
)
from winrt.windows.storage.streams import IBuffer  # type: ignore[import]


class winrt_midi_transfer():
    """WinRT MIDI Transfer class."""

    def __init__(self) -> None:
        """__init__."""
        # MIDI IN port
        self.in_port: Optional[MidiInPort] = None
        # MIDI OUT port
        self.out_port: Optional[MidiOutPort] = None

        # hex ID pattern
        self.hex_id_pattern: re.Pattern[str] = \
            re.compile('#MIDII_([0-9A-F]{8})\\..+#')

        # MIDI IN port list
        self.in_ports: List[Tuple[str, str, str, str]] = []
        # MIDI OUT port list
        self.out_ports: List[Tuple[str, str, str, str]] = []

    async def list_ports(self,
                         kind: Union[MidiInPort, MidiOutPort] = MidiInPort
                         ) -> List[Tuple[str, str, str, str]]:
        """
        ポートのリストを取得する.

        Args:
        kind (Union[MidiInPort, MidiOutPort]): 取得するポートの種類

        Returns:
          List:
            Tuple:
              str: ポートの名前
              str: ポートの ID
              str: ID の 16 進数 8 桁部分
              str: ポートの表示名（ID の 16 進数 8 桁部分を含む）
        """
        devs: DeviceInformationCollection = \
            await DeviceInformation.find_all_async(
                kind.get_device_selector(), [])

        retval: List[Tuple[str, str, str, str]] = []
        d: DeviceInformation
        for d in devs:
            regex_result: Optional[re.Match[str]] = \
                self.hex_id_pattern.search(d.id)
            hex_id: str
            display_name: str
            if regex_result is None:
                hex_id = ''
                display_name = d.name
            else:
                hex_id = regex_result.group(1)
                display_name = f'{d.name} [ {hex_id} ]'
                if len(self.in_ports) > 0 and kind is MidiOutPort:
                    display_name_in: str
                    hex_id_in: str
                    for _, _, hex_id_in, display_name_in in self.in_ports:
                        if hex_id == hex_id_in:
                            display_name = display_name_in
                            break
            retval.append((d.name, d.id, hex_id, display_name))

        return retval

    def __note_off(self, message: MidiNoteOffMessage) -> str:
        channel: int = message.channel
        note: int = message.note
        velocity: int = message.velocity
        return (f'note_off: channel {channel}, '
                f'note {note}, velocity {velocity}')

    def __note_on(self, message: MidiNoteOnMessage) -> str:
        channel: int = message.channel
        note: int = message.note
        velocity: int = message.velocity
        return (f'note_on: channel {channel}, '
                f'note {note}, velocity {velocity}')

    def __polytouch(self, message: MidiPolyphonicKeyPressureMessage) -> str:
        channel: int = message.channel
        note: int = message.note
        pressure: int = message.pressure
        return (f'polytouch: channel {channel}, note {note}, '
                f'pressure {pressure}')

    def __control_change(self, message: MidiControlChangeMessage) -> str:
        channel: int = message.channel
        controller: int = message.controller
        control_value: int = message.control_value
        return (f'control_change: channel {channel}, '
                f'controller {controller}, '
                f'control_value {control_value}')

    def __program_change(self, message: MidiProgramChangeMessage) -> str:
        channel: int = message.channel
        program: int = message.program
        return f'program_change: channel {channel}, program {program}'

    def __aftertouch(self, message: MidiChannelPressureMessage) -> str:
        channel: int = message.channel
        pressure: int = message.pressure
        return f'aftertouch: channel {channel}, pressure {pressure}'

    def __pitchwheel(self, message: MidiPitchBendChangeMessage) -> str:
        channel: int = message.channel
        bend: int = message.bend
        return f'pitchwheel: channel {channel}, bend {bend}'

    def __sysex(self, message: MidiSystemExclusiveMessage) -> str:
        raw_data: IBuffer = message.raw_data
        return 'sysex'

    def __quarter_frame(self, message: MidiTimeCodeMessage) -> str:
        frame_type: int = message.frame_type
        raw_data: IBuffer = message.raw_data
        return f'quarter_frame: frame_type {frame_type}'

    def __songpos(self, message: MidiSongPositionPointerMessage) -> str:
        beats: int = message.beats
        return f'songpos: beats {beats}'

    def __song_select(self, message: MidiSongSelectMessage) -> str:
        song: int = message.song
        return f'song_select: song {song}'

    def __tune_request(self, message: MidiTuneRequestMessage) -> str:
        return 'tune_request'

    def __sysex_eox(self, message: MidiSystemExclusiveMessage) -> str:
        return 'sysex_eox'

    def __clock(self, message: MidiTimingClockMessage) -> str:
        return 'clock'

    def __start(self, message: MidiStartMessage) -> str:
        return 'start'

    def __continue(self, message: MidiContinueMessage) -> str:
        return 'continue'

    def __stop(self, message: MidiStopMessage) -> str:
        return 'stop'

    def __active_sensing(self, message: MidiActiveSensingMessage) -> str:
        return 'active_sensing'

    def __system_reset(self, message: MidiSystemResetMessage) -> str:
        return 'system_reset'

    def __message_to_str(self, message: IMidiMessage) -> str:
        """
        MIDI メッセージを文字列へ変換する.

        人間が見て中身が読めるようにする。

        Args:
          message (IMidiMessage): WinRT の MIDI メッセージ

        Returns:
          str: MIDI メッセージを文字列化したもの
        """
        t: int = message.type
        buff: str
        if t == MidiMessageType.NOTE_OFF:
            buff = self.__note_off(
                MidiNoteOffMessage._from(message))
        elif t == MidiMessageType.NOTE_ON:
            buff = self.__note_on(
                MidiNoteOnMessage._from(message))
        elif t == MidiMessageType.POLYPHONIC_KEY_PRESSURE:
            buff = self.__polytouch(
                MidiPolyphonicKeyPressureMessage._from(message))
        elif t == MidiMessageType.CONTROL_CHANGE:
            buff = self.__control_change(
                MidiControlChangeMessage._from(message))
        elif t == MidiMessageType.PROGRAM_CHANGE:
            buff = self.__program_change(
                MidiProgramChangeMessage._from(message))
        elif t == MidiMessageType.CHANNEL_PRESSURE:
            buff = self.__aftertouch(
                MidiChannelPressureMessage._from(message))
        elif t == MidiMessageType.PITCH_BEND_CHANGE:
            buff = self.__pitchwheel(
                MidiPitchBendChangeMessage._from(message))
        elif t == MidiMessageType.SYSTEM_EXCLUSIVE:
            buff = self.__sysex(
                MidiSystemExclusiveMessage._from(message))
        elif t == MidiMessageType.MIDI_TIME_CODE:
            buff = self.__quarter_frame(
                MidiTimeCodeMessage._from(message))
        elif t == MidiMessageType.SONG_POSITION_POINTER:
            buff = self.__songpos(
                MidiSongPositionPointerMessage._from(message))
        elif t == MidiMessageType.SONG_SELECT:
            buff = self.__song_select(
                MidiSongSelectMessage._from(message))
        elif t == MidiMessageType.TUNE_REQUEST:
            buff = self.__tune_request(
                MidiTuneRequestMessage._from(message))
        elif t == MidiMessageType.END_SYSTEM_EXCLUSIVE:
            buff = self.__sysex_eox(
                MidiSystemExclusiveMessage._from(message))
        elif t == MidiMessageType.TIMING_CLOCK:
            buff = self.__clock(
                MidiTimingClockMessage._from(message))
        elif t == MidiMessageType.START:
            buff = self.__start(
                MidiStartMessage._from(message))
        elif t == MidiMessageType.CONTINUE:
            buff = self.__continue(
                MidiContinueMessage._from(message))
        elif t == MidiMessageType.STOP:
            buff = self.__stop(
                MidiStopMessage._from(message))
        elif t == MidiMessageType.ACTIVE_SENSING:
            buff = self.__active_sensing(
                MidiActiveSensingMessage._from(message))
        elif t == MidiMessageType.SYSTEM_RESET:
            buff = self.__system_reset(
                MidiSystemResetMessage._from(message))
        else:
            buff = f'unknown ({t}: {MidiMessageType(t)})'

        raw_data: IBuffer = message.raw_data
        byte_list: List[int] = \
            CryptographicBuffer.copy_to_byte_array(raw_data)
        buff += f', {[hex(x) for x in byte_list]}'

        return buff

    def __midi_in_callback(self,
                           sender: MidiInPort,
                           e: MidiMessageReceivedEventArgs
                           ) -> None:
        """
        MIDI IN イベントで呼ばれるコールバック.

        Args:
          sender (MidiInPort): イベントが発生した MIDI IN ポート
          e (MidiMessageReceivedEventArgs): メッセージを含む
        """
        message: IMidiMessage = e.message

        if self.out_port is not None:
            self.out_port.send_message(message)

        timestamp: TimeSpan = message.timestamp
        duration: int = timestamp.duration
        print(f'{duration}, {self.__message_to_str(message)}')

    async def select_midi_in_port(self) -> Optional[MidiInPort]:
        """
        MIDI IN ポートを選択する.

        Returns:
          Optional[MidiInPort]: 選択された MIDI IN ポート
            None は選択されなかったことを示す
        """
        self.in_ports = await self.list_ports()
        i: int
        print('\nMIDI IN ports\n')
        for i in range(len(self.in_ports)):
            print(f'{i}: {self.in_ports[i][3]}')
        choice: int = int(input('\nSelect number > '))
        if 0 <= choice and choice < len(self.in_ports):
            print(f'\nSelected: {i}, {self.in_ports[choice]}')
            return await MidiInPort.from_id_async(self.in_ports[choice][1])
        print(f'Error: {choice}')
        return None

    async def select_midi_out_port(self) -> Optional[MidiOutPort]:
        """
        MIDI OUT ポートを選択する.

        Returns:
          Optional[MidiOutPort]: 選択された MIDI OUT ポート
            None は選択されなかったことを示す
        """
        self.out_ports = await self.list_ports(MidiOutPort)
        i: int
        print('\nMIDI OUT ports\n')
        for i in range(len(self.out_ports)):
            print(f'{i}: {self.out_ports[i][3]}')
        choice: int = int(input('\nSelect number > '))
        if 0 <= choice and choice < len(self.out_ports):
            print(f'\nSelected: {i}, {self.out_ports[choice]}')
            return await MidiOutPort.from_id_async(self.out_ports[choice][1])
        print(f'Error: {choice}')
        return None

    async def transfer(self) -> None:
        """MIDI IN ポートと MIDI OUT ポートを選んで転送."""
        self.in_port = await self.select_midi_in_port()
        self.out_port = await self.select_midi_out_port()

        if self.in_port is None:
            print('Input port is None.')
            return

        print('\nCallback starting...')
        self.in_port.add_message_received(self.__midi_in_callback)
        print('Started\n')

        while True:
            await asyncio.sleep(1.0)


def main() -> None:
    """Test main."""
    print('WinRT MIDI Transfer with python winrt\n\n'
          'https://gist.github.com/trueroad/'
          '6beaf87280afb2b5e33b4838d73be6ed\n\n'
          'Copyright (C) 2022 Masamichi Hosoda.\n'
          'All rights reserved.\n')

    wmt: winrt_midi_transfer = winrt_midi_transfer()
    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(wmt.transfer())
    except KeyboardInterrupt:
        print('Interrupted')


if __name__ == '__main__':
    main()