circuitpython/shared-module/synthio/MidiTrack.c

// This file is part of the CircuitPython project: https://circuitpython.org
//
// SPDX-FileCopyrightText: Copyright (c) 2021 Artyom Skrobov
//
// SPDX-License-Identifier: MIT

#include "py/runtime.h"
#include "shared-bindings/synthio/MidiTrack.h"
#include "shared-bindings/audiocore/__init__.h"


static void record_midi_stream_error(synthio_miditrack_obj_t *self) {
    self->error_location = self->pos;
    self->pos = self->track.len;
}

static mp_obj_t parse_note(synthio_miditrack_obj_t *self) {
    uint8_t *buffer = self->track.buf;
    size_t len = self->track.len;
    if (self->pos + 1 >= len) {
        record_midi_stream_error(self);
    }
    uint8_t note = buffer[(self->pos)++];
    if (note > 127 || buffer[(self->pos)++] > 127) {
        record_midi_stream_error(self);
    }
    return MP_OBJ_NEW_SMALL_INT(note);
}

static int decode_duration(synthio_miditrack_obj_t *self) {
    uint8_t *buffer = self->track.buf;
    size_t len = self->track.len;
    uint8_t c;
    uint32_t delta = 0;
    do {
        c = buffer[self->pos++];
        delta <<= 7;
        delta |= c & 0x7f;
    } while ((c & 0x80) && (self->pos < len));

    // errors cannot be raised from the background task, so simply end the track.
    if (c & 0x80) {
        self->pos = self->track.len;
        record_midi_stream_error(self);
    }
    return delta * self->synth.base.sample_rate / self->tempo;
}

// invariant: pointing at a MIDI message
static void decode_until_pause(synthio_miditrack_obj_t *self) {
    uint8_t *buffer = self->track.buf;
    size_t len = self->track.len;
    do {
        switch (buffer[self->pos++] >> 4) {
            case 8: { // Note Off
                mp_obj_t note = parse_note(self);
                synthio_span_change_note(&self->synth, note, SYNTHIO_SILENCE);
                break;
            }
            case 9: { // Note On
                mp_obj_t note = parse_note(self);
                synthio_span_change_note(&self->synth, SYNTHIO_SILENCE, note);
                break;
            }
            case 10:
            case 11:
            case 14: // two data bytes to ignore
                parse_note(self);
                break;
            case 12:
            case 13: // one data byte to ignore
                if (self->pos >= len || buffer[self->pos++] > 127) {
                    record_midi_stream_error(self);
                }
                break;
            case 15: // the full syntax is too complicated, just assume it's "End of Track" event
                self->pos = len;
                break;
            default: // invalid event
                record_midi_stream_error(self);
        }
        if (self->pos < len) {
            self->synth.span.dur = decode_duration(self);
        }
    } while (self->pos < len && self->synth.span.dur == 0);
}

static void start_parse(synthio_miditrack_obj_t *self) {
    self->pos = 0;
    self->error_location = -1;
    self->synth.span.dur = decode_duration(self);
    if (self->synth.span.dur == 0) {
        // the usual case: the file starts with some MIDI event, not a delay
        decode_until_pause(self);
    }
}

void common_hal_synthio_miditrack_construct(synthio_miditrack_obj_t *self,
    const uint8_t *buffer, uint32_t len, uint32_t tempo, uint32_t sample_rate,
    mp_obj_t waveform_obj, mp_obj_t filter_obj, mp_obj_t envelope_obj) {

    self->tempo = tempo;
    self->track.buf = (void *)buffer;
    self->track.len = len;

    synthio_synth_init(&self->synth, sample_rate, 1, waveform_obj, envelope_obj);

    start_parse(self);
}

void common_hal_synthio_miditrack_deinit(synthio_miditrack_obj_t *self) {
    synthio_synth_deinit(&self->synth);
}

mp_int_t common_hal_synthio_miditrack_get_error_location(synthio_miditrack_obj_t *self) {
    return self->error_location;
}

void synthio_miditrack_reset_buffer(synthio_miditrack_obj_t *self,
    bool single_channel_output, uint8_t channel) {
    synthio_synth_reset_buffer(&self->synth, single_channel_output, channel);
    start_parse(self);
}

audioio_get_buffer_result_t synthio_miditrack_get_buffer(synthio_miditrack_obj_t *self,
    bool single_channel_output, uint8_t channel, uint8_t **buffer, uint32_t *buffer_length) {
    if (audiosample_deinited(&self->synth.base)) {
        *buffer_length = 0;
        return GET_BUFFER_ERROR;
    }

    synthio_synth_synthesize(&self->synth, buffer, buffer_length, single_channel_output ? 0 : channel);
    if (self->synth.span.dur == 0) {
        if (self->pos == self->track.len) {
            return GET_BUFFER_DONE;
        } else {
            decode_until_pause(self);
        }
    }
    return GET_BUFFER_MORE_DATA;
}