convert quick time music movies into midi files!

mov2mid.py

#!/usr/bin/env python3

# convert quicktime music movie into midi file

from __future__ import annotations

from collections import namedtuple
from abc import ABC
import itertools

import sys
import io
import struct

def variable_length_quantity(value, level=0) -> bytes:
    msb, lsb = divmod(value, 128)
    if level > 0:
        lsb |= 0x80
    return (variable_length_quantity(msb, level + 1) if msb > 0 else b'') + bytes([lsb])

class MIDIChunk(ABC):
    def __init__(self, type, data):
        self.type = type
        self.data = data

    def __bytes__(self):
        return self.type[:4].encode('ascii') + struct.pack('> I', len(self.data)) + self.data

class MIDIChunkHeader(MIDIChunk):
    def __init__(self, format, n_tracks, division):
        data = struct.pack('> H H H', format, n_tracks, division)
        super().__init__('MThd', data)

class MIDIChunkTrack(MIDIChunk):
    def __init__(self, events):
        data = b''.join(map(bytes, events))
        super().__init__('MTrk', data)

class MIDIFile:
    def __init__(self, chunks):
        self.chunks = chunks

    @classmethod
    def from_data(cls, format, division, tracks):
        return cls([MIDIChunkHeader(format, len(tracks), division)] + \
                   [MIDIChunkTrack(events) for events in tracks])

    def __bytes__(self):
        return b''.join(map(bytes, self.chunks))

class MIDIEvent:
    def __init__(self, delta_time, data):
        self.delta_time = delta_time
        self.data = data

    def __bytes__(self):
        d = variable_length_quantity(self.delta_time) + self.data
        return variable_length_quantity(self.delta_time) + self.data

class MIDIEventChannel(MIDIEvent):
    def __init__(self, delta_time, status, channel, *data):
        data = bytes([(status & 0xf0) | (channel & 0x0f)]) + bytes([x & 0x7f for x in data])
        super().__init__(delta_time, data)

class MIDIEventNoteOff(MIDIEventChannel):
    def __init__(self, delta_time, channel, note, velocity):
        super().__init__(delta_time, 0b10000000, channel, note, velocity)

class MIDIEventNoteOn(MIDIEventChannel):
    def __init__(self, delta_time, channel, note, velocity):
        super().__init__(delta_time, 0b10010000, channel, note, velocity)

class MIDIEventAfterTouch(MIDIEventChannel):
    def __init__(self, delta_time, channel, note, velocity):
        super().__init__(delta_time, 0b10100000, channel, note, velocity)

class MIDIEventControlChange(MIDIEventChannel):
    def __init__(self, delta_time, channel, controller, velocity):
        super().__init__(delta_time, 0b10110000, channel, controller, velocity)

class MIDIEventProgramChange(MIDIEventChannel):
    def __init__(self, delta_time, channel, program):
        super().__init__(delta_time, 0b11000000, channel, program);

class MIDIEventPitchWheelChange(MIDIEventChannel):
    def __init__(self, delta_time, channel, value):
        # value is between -1 and 1
        value = int((value + 1) * 0x2000)
        msb, lsb = divmod(value, 0x80)
        super().__init__(delta_time, 0b11100000, channel, lsb, msb)

class MIDIEventMeta(MIDIEvent):
    def __init__(self, delta_time, type, meta_data=b''):
        data = bytes([0xff, type, len(meta_data)]) + meta_data
        super().__init__(delta_time, data)

class MIDIEventMetaEndOfTrack(MIDIEventMeta):
    def __init__(self, delta_time):
        super().__init__(delta_time, 0x2f)

def unpack(stream, fmt):
    size = struct.calcsize(fmt)
    buf = stream.read(size)
    return struct.unpack(fmt, buf)

class Atom:
    def __init__(self, type, data):
        self.type = type
        self.data = data

    @classmethod
    def from_stream(cls, stream) -> Atom:
        try:
            size, type = unpack(stream, '>I 4s')
            data = stream.read(size - 8) if size > 0 else stream.read()
            return cls(type.decode('ascii'), data)
        except (UnicodeDecodeError):
            pass # malformed
        except struct.error:
            pass # end of atom

    @classmethod
    def all_from_stream(cls, stream):
        while atom := cls.from_stream(stream):
            yield atom
        
    @classmethod
    def from_bytes(cls, string) -> Atom:
        return cls.from_stream(io.BytesIO(string))

    @classmethod
    def all_from_bytes(cls, string):
        return cls.all_from_stream(io.BytesIO(string))

    @property
    def children(self):
        return Atom.all_from_bytes(self.data)

    def find_child(self, child_type):
        for child in self.children:
            if child.type == child_type:
                return child
            else:
                if grandchild := child.find_child(child_type):
                    return grandchild

class AtomSTCO:
    def __init__(self, offsets):
        self.offsets = offsets

    @classmethod
    def from_stream(cls, stream):
        flags, length = unpack(stream, '> 4s I')
        offsets = [unpack(stream, '> I')[0] for i in range(length)]
        return AtomSTCO(offsets)

    @classmethod
    def from_bytes(cls, string):
        return cls.from_stream(io.BytesIO(string))
        
class AtomSTSD:
    def __init__(self, entries):
        self.entries = entries

    @classmethod
    def from_stream(cls, stream):
        def unpack_entry(stream):
            size, data_format, data_reference_index = unpack(stream, '>I 4s x x x x x x H')
            description = stream.read(size - 16)
            return {'data_format': data_format,
                    'data_reference_index': data_reference_index,
                    'description': description}

        flags, length = unpack(stream, '> 4s I')
        return AtomSTSD([unpack_entry(stream) for i in range(length)])

    @classmethod
    def from_bytes(cls, string):
        return cls.from_stream(io.BytesIO(string))
        
class Chunk:
    def __init__(self, data):
        self.data = data

    def __iter__(self):
        stream = io.BytesIO(self.data)
        while long_word := stream.read(4):
            if long_word == b'\x60\x00\x00\x00':
                return
            yield long_word

class ChunkStream:
    def __init__(self, chunks):
        self.chunks = chunks

    def __iter__(self):
        return itertools.chain(*self.chunks)

    @classmethod
    def from_paths(cls, resource_path, data_path):
        with open(resource_path, 'rb') as resource_stream:
            with open(data_path, 'rb') as data_stream:

                atom_moov = Atom.from_stream(resource_stream)
                mdat = data_stream.read()
                atom_stco = AtomSTCO.from_bytes(atom_moov.find_child('stco').data)
                atom_stsd = AtomSTSD.from_bytes(atom_moov.find_child('stsd').data)

                chunks = [Chunk(entry['description'][4:]) for entry in atom_stsd.entries] \
                       + [Chunk(mdat[offset:]) for offset in atom_stco.offsets]

                return ChunkStream(chunks)

class AbsoluteEvent:
    def __init__(self, time, function):
        self.time = time
        self.function = function

    def relative_event(self, time, time_scale):
        return self.function(int((self.time - time) * time_scale))

class ConversionContext:
    def __init__(self, chunk_stream):
        self.chunk_stream = iter(chunk_stream)

    def process_word(self, long_word):
        for handler in handlers:
            if handler.match(long_word):
                return handler.handler(self, long_word)
        print(f'unknown event: {list(map(hex, long_word))}')
        return 0

    def read_long_words(self, n):
        return b''.join([next(self.chunk_stream) for i in range(n)])

    @property
    def midi_events(self):
        time = 0
        for event in sorted(self.events, key=lambda x: x.time):
            yield event.relative_event(time, self.time_scale)
            time = event.time

    def to_midi(self) -> MIDIFile:
        self.time_scale = 960 * 2
        self.time = 0
        self.channels = {}

        self.events = []

        for long_word in self.chunk_stream:
            self.process_word(long_word)

        self.add_end_of_track()

        division = 960
        return MIDIFile.from_data(0, division, [self.midi_events])

    def add_event(self, time, function):
        self.events.append(AbsoluteEvent(time, function))

    def add_note(self, channel, note, velocity, duration):
        def function_on(delta_time):
            return MIDIEventNoteOn(delta_time, channel, note, velocity)
        def function_off(delta_time):
            return MIDIEventNoteOff(delta_time, channel, note, velocity)
        self.add_event(self.time, function_on)
        self.add_event(self.time + duration, function_off)

    def add_program_change(self, channel, program):
        def function(delta_time):
            return MIDIEventProgramChange(delta_time, channel, program)
        self.add_event(self.time, function)

    def add_pitch_wheel_change(self, channel, value):
        def function(delta_time):
            return MIDIEventPitchWheelChange(delta_time, channel, value)
        self.add_event(self.time, function)

    def add_end_of_track(self):
        def function(delta_time):
            return MIDIEventMetaEndOfTrack(delta_time)
        self.add_event(self.time, function)

handlers = []

class EventHandler:
    def __init__(self, signature, mask, handler):
        self.signature = signature
        self.mask = mask
        self.handler = handler

    def match(self, long_word):
        return long_word[0] & self.mask == self.signature & self.mask

def handler(signature, mask):
    def decorator(handler):
        handlers.append(EventHandler(signature, mask, handler))
        return handler
    return decorator

@handler(0b10010000, 0b11110000)
def event_long_note_request(context, long_word):
    long_word += context.read_long_words(1)
    part     = (long_word[0] & 0x0f) << 8 \
             | (long_word[1] & 0xff) << 0
    pitch    = (long_word[2] & 0xff) << 8 \
             | (long_word[3] & 0xff) << 0
    velocity = (long_word[4] & 0x1f) << 2 \
             | (long_word[5] & 0xc0) >> 6
    duration = (long_word[5] & 0x3f) << 16 \
             | (long_word[6] & 0xff) << 8 \
             | (long_word[7] & 0xff) << 0
    ## TODO microtonal..
    if pitch > 128:
        pitch //= 256
    context.add_note(context.channels[part], pitch, velocity, duration / 600)
    #print(f'note request: {long_word} {context.channels[part]} {part} {pitch} {velocity} {duration / 600}')

@handler(0b00100000, 0b11100000)
def event_short_note_request(context, long_word):
    part     = (long_word[0] & 0x1f) >> 0
    pitch    = (long_word[1] & 0xfc) >> 2
    velocity = (long_word[1] & 0x03) << 5 \
             | (long_word[2] & 0xf8) >> 3
    duration = (long_word[2] & 0x07) << 8 \
             | (long_word[3] & 0xff) >> 0
    context.add_note(context.channels[part], pitch + 32, velocity, duration / 600)
    #print(f'note request: {part} {pitch} {velocity} {duration / 600}')

@handler(0b01000000, 0b11100000)
def event_short_controller_request(context, long_word):
    part       = (long_word[0] & 0x1f) << 0
    controller = (long_word[1] & 0xff) << 0
    value = struct.unpack('>h', long_word[2:])[0]
    if controller == 32: # bend
        context.add_pitch_wheel_change(context.channels[part], value / 512)
    else:
        #context.midi.addControllerEvent(0, context.channels[part], context.time, controller, value // 256)
        print(f'WARNIN UNKNOWN CONTRCOLLER AAAAAAA {controller}')
    print(f'controller request: {part} {controller} {value}')

@handler(0b00000000, 0b11100000)
def event_rest(context, long_word):
    duration = long_word[1] << 16 \
             | long_word[2] << 8  \
             | long_word[3] << 0
    #print(f'rest: {duration / 600}')
    context.time += duration / 600

@handler(0b11110000, 0b11110000)
def event_general(context, long_word):
    part    = (long_word[0] & 0x0f) << 8 \
            | (long_word[1] & 0xff) << 0
    length  = (long_word[2] & 0xff) << 8 \
            | (long_word[3] & 0xff) << 0
    payload = context.read_long_words(length - 1)
    program = struct.unpack('>I', payload[-8:-4])[0]
    if program < 128:
        if len(context.channels) == 9:
            context.channels[part] = 10
        else:
            context.channels[part] = len(context.channels)
        if program > 1:
            program -= 1
        context.add_program_change(context.channels[part], program)
    else:
        context.channels[part] = 9
        context.add_program_change(context.channels[part], program - 16385)
    print(f'general event {part} = {context.channels[part]} = {program}')

def convert_with_paths(resource_path, data_path, midi_path):
    with open(midi_path, 'wb') as midi_stream:
        chunk_stream = ChunkStream.from_paths(resource_path, data_path)
        midi_stream.write(bytes(ConversionContext(chunk_stream).to_midi()))

if __name__ == '__main__':
    if len(sys.argv) < 4:
        print(f'Usage: {sys.argv[0]} [path to resource fork] [path to data fork] [path to output file]')
    else:
        convert_with_paths(*sys.argv[1:])