Build SMF (Standard MIDI File) from winrt_midi_in_timing

build_smf_from_winrt_midi_in_timing.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Build SMF (Standard MIDI File) from winrt_midi_in_timing.

https://gist.github.com/trueroad/74f3a5e6d73af6c3a6b294350b9253f6

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.
"""

# See
# https://gist.github.com/trueroad/2ffa736d0b206973a80c99e177273795

import sys
from typing import Final, List, TextIO, Tuple

import mido  # type: ignore[import]


# 10 MHz, 100 ns
UWP_FREQ: Final[int] = 10000000


def parse_line(line: str) -> Tuple[int, int, List[int]]:
    """
    INPUT.tsv の 2 行目以降をパース.

    Args:
      line (str): INPUT.tsv の行文字列

    Returns:
      Tuple:
        int: MIDI メッセージ受信時の QueryPerformanceCounter (QPC) 値
        int: UWP MIDI タイムスタンプ
        List[int]: MIDI メッセージ
    """
    items: List[str] = line.split('\t')
    message_str: List[str] = items[2].split()
    message: List[int] = []
    m: str
    for m in message_str:
        message.append(int(m, 16))
    return (int(items[0]), int(items[1]), message)


# Overflow low が発生したか否か
b_overflow_low: bool = False


def fix_uwp_delta(qpc: int, uwp: int, qpc_delta: int, uwp_delta: int) -> int:
    """
    UWP タイムスタンプのデルタを修正する.

    Args:
      qpc (int): MIDI メッセージ受信時の QueryPerformanceCounter (QPC) 値
      uwp (int): UWP MIDI タイムスタンプ
      qpc_delta (int): QPC 値によるデルタタイム（単位：ms）
      uwp_delta (int): UWP MIDI タイムスタンプによるデルタタイム（単位：ms）

    Returns:
      int: 修正した UWP MIDI タイムスタンプのデルタタイム（単位：ms）
    """
    global b_overflow_low

    # Overflow low 未発生かつ誤差一定以下の場合は修正対象外
    if not b_overflow_low and abs(qpc_delta - uwp_delta) < 4096:
        return uwp_delta

    print(f'QPC {qpc}, UWP {uwp}: '
          f'QPC delta {qpc_delta}, UWP delta {uwp_delta}')

    uwp_delta_fixed: int = uwp_delta

    # Fix after overflow low
    # https://github.com/trueroad/BLE_MIDI_packet_data_set#page-7-overflow-both
    if b_overflow_low and uwp_delta >= 128:
        # Overflow low 発生後かつ UWP デルタが 128 ms 以上の場合

        # ・本件修正を要するメッセージは
        # 　発生した次のパケットに入っている最初のメッセージ
        # ・そのパケットでは timestampHigh がパーサの想定より +1 されている
        # 　（発生パケットで桁上がりして +1 されたハズがなっていない）
        # 　ので UWP デルタが余計に +128 されてしまう
        # ・発生後でも同一パケット内の後続メッセージの場合は
        # 　デルタタイムが 128 ms 未満となるので、そういうものはスキップ

        # 余計な +128 を減算して修正
        uwp_delta_fixed -= 128
        # 発生後は 1 回だけ修正すれば OK
        b_overflow_low = False

        print(f'  -> Fix after overflow low: UWP delta {uwp_delta_fixed}')

    # Fix overflow low
    # https://github.com/trueroad/BLE_MIDI_packet_data_set#page-7-overflow-low
    if (8192 - 128) < uwp_delta and uwp_delta < 8192 and qpc_delta < 4096:
        # Overflow low が発生した場合

        # ・同一パケット内で timestampLow が小さくなったら桁上がりとみなして
        # 　timestampHigh を +1 する必要があるが、パーサがそう解釈せず
        # 　そのままの値であるとしている模様
        # ・するとタイムスタンプが 1 回転 (8192 ms) 近くしたものと解釈され
        # 　QPC デルタとは大きな乖離が発生する
        # ・本件発生した場合は UWP デルタに 1 回転 8192 ms が余計に加算され、
        # 　桁上がり分 128 ms が加算されない状態となるため、
        # 　UWP デルタは (8192 - 128) ms を超えたものとなる
        # ・本件は同一パケット内でのみ発生するが UWP デルタが 8192 ms 以上は
        # 　別パケットになるため修正対象外
        # ・よって UWP デルタが修正対象範囲内かつ QPC デルタが一定以下
        # 　（修正後の UWP デルタの方が近くなる）場合に修正を発動する

        # 余計な 1 回転を減算、未加算の桁上がり分 128 を加算して修正
        uwp_delta_fixed -= (8192 - 128)
        # 次のパケットで桁上がり分の再度修正が必要
        b_overflow_low = True

        print(f'  -> Fix overflow low: UWP delta {uwp_delta_fixed}')

    return uwp_delta_fixed


def main() -> None:
    """Test main."""
    print('Build SMF from winrt_midi_in_timing\n\n'
          'https://gist.github.com/trueroad/'
          '74f3a5e6d73af6c3a6b294350b9253f6\n\n'
          'Copyright (C) 2022 Masamichi Hosoda.\n'
          'All rights reserved.\n')

    if len(sys.argv) != 5:
        # INPUT.tsv ファイルには
        # WinRT MIDI Transfer with python winrt
        # https://gist.github.com/trueroad/6beaf87280afb2b5e33b4838d73be6ed
        # で得られたファイルが使用できる
        print('Usage: ./build_smf.py [INPUT.tsv OUTPUT_QPC.mid '
              'OUTPUT_UWP.mid OUTPUT_UWP_FIXED.mid]')
        return

    # QueryPerformanceCounter (QPC) の値を基準にした SMF
    # TPQN 480, テンポ四分音符＝ 125 設定により 1 tick = 1 ms とする
    mid_qpc: mido.MidiFile = mido.MidiFile(type=0, ticks_per_beat=480)
    track_qpc: mido.MidiTrack = mido.MidiTrack()
    mid_qpc.tracks.append(track_qpc)
    track_qpc.append(mido.MetaMessage('set_tempo', tempo=480000))

    # UWP MIDI タイムスタンプを基準にした SMF
    mid_uwp: mido.MidiFile = mido.MidiFile(type=0, ticks_per_beat=480)
    track_uwp: mido.MidiTrack = mido.MidiTrack()
    mid_uwp.tracks.append(track_uwp)
    track_uwp.append(mido.MetaMessage('set_tempo', tempo=480000))

    # UWP MIDI タイムスタンプを修正したものを基準にした SMF
    mid_uwpfix: mido.MidiFile = mido.MidiFile(type=0, ticks_per_beat=480)
    track_uwpfix: mido.MidiTrack = mido.MidiTrack()
    mid_uwpfix.tracks.append(track_uwpfix)
    track_uwpfix.append(mido.MetaMessage('set_tempo', tempo=480000))

    f: TextIO
    with open(sys.argv[1]) as f:
        # QPC の周波数を取得
        qpc_freq: int = int(f.readline())
        print(f'QPC freq.: {qpc_freq}')

        # 最初の MIDI メッセージを取得
        # デルタ計算用に QPC 値、UWP MIDI タイムスタンプを保存
        before_qpc: int
        before_uwp: int
        message: List[int]
        before_qpc, before_uwp, message = parse_line(f.readline())
        print(f'first: QPC {before_qpc}, UWP {before_uwp}, {message}')

        # 最初の MIDI メッセージを SMF に追加
        msg_qpc: mido.Message = mido.Message.from_bytes(message, time=0)
        track_qpc.append(msg_qpc)
        msg_uwp: mido.Message = mido.Message.from_bytes(message, time=0)
        track_uwp.append(msg_uwp)
        msg_uwpfix: mido.Message = mido.Message.from_bytes(message, time=0)
        track_uwpfix.append(msg_uwpfix)

        for line in f:
            # 2 番目以降の MIDI メッセージを取得
            qpc: int
            uwp: int
            qpc, uwp, message = parse_line(line)

            # QPC デルタ、 UWP デルタを計算（単位：ms）
            qpc_delta: int = (qpc - before_qpc) * 1000 // qpc_freq
            uwp_delta: int = (uwp - before_uwp) * 1000 // UWP_FREQ

            # UWP デルタを修正
            uwp_delta_fixed: int = fix_uwp_delta(qpc, uwp,
                                                 qpc_delta, uwp_delta)

            # 各デルタとともに MIDI メッセージを SMF へ追加
            msg_qpc = mido.Message.from_bytes(message, time=qpc_delta)
            track_qpc.append(msg_qpc)
            msg_uwp = mido.Message.from_bytes(message, time=uwp_delta)
            track_uwp.append(msg_uwp)
            msg_uwpfix = mido.Message.from_bytes(message, time=uwp_delta_fixed)
            track_uwpfix.append(msg_uwpfix)

            # 前回の QPC 値、UWP MIDI タイムスタンプを保存
            before_qpc = qpc
            before_uwp = uwp

        # 各 SMF を保存
        mid_qpc.save(sys.argv[2])
        mid_uwp.save(sys.argv[3])
        mid_uwpfix.save(sys.argv[4])


if __name__ == '__main__':
    main()