🔒 Parsing incoming data as it comes in

[little moon]

I'm writing a simple MIDI parser as a learning exercise and I'd like it to interpret data piece-by-piece. Data is sent in as an iterable of integers (probably just bytes, when it's done, but I'm using a deque[int] for now).

I've currently got a simple generator-based event parser implementation, but I'm not sure what the best way is for me to actually use it, since the function yields until it's got all the data it needs, and then returns the newly-created event. So my questions are:

1. Are generators a good way to do this sort of thing?
2. Are there any other ways to achieve the same results?
3. What's the best way to use the current implementation?

Apologies if the code isn't layed out too well.. I've not done much using generators before.

The event parser function (irrelevant parts removed):

def parse_event(last_status_byte: int | None = None) -> "Generator[None, Any, MidiEvent]":
    first = yield # Get the first byte
    # Do some initial processing...

    if is_channel_event:
        # ...
        for _ in range(num_data_bytes): # num_data_bytes is calculated beforehand
            data_bytes.append((yield))

        # Return the newly-created event.
        return channel_event_classes[event_type](...)

    raise TypeError(f"Unsupported event: {status_byte:02X}.")

How I'm testing it out:

data = deque((0x90, 0x3C, 0x64))
event_generator = parse_event()

next(event_generator)
while True:
    try:
        event_generator.send(data.popleft())
    except StopIteration as exc:
        evt: NoteOnEvent = exc.value
        break

print(evt.channel, evt.note, evt.velocity, evt._bytes)

Thanks. :)

[paper marlin]

@little moon

Python help channel opened

Remember to:

• Ask your Python question, not if you can ask or if there's an expert who can help.
• Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
• Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.

[gleaming frigate]

I don't object to a generator here particularly, though I'd probably just have a while loop that reads from a stream and sends bytes into a function until the stream is exhausted.

[little moon]

hmm, I see.

[little moon]

What about if the stream is exhausted while parsing an event? i.e. if only the first byte is available at parser call time?

Cause the messages differ in length

That's why I thought I'd go for a generator-based parser

[gleaming frigate]

That while loop could also just be adapted to a function that yields the events as they come:

def process_midi(source):
  while source.has_data(): 
    header = source.read(4)
    chunk_len = source.read(4)
    
    # ... parse the chunk somewhere
    yield Chunk(...)

for chunk in process_midi(source):
  # do things with the chunk

And if you don't want to directly provide a source, then you could also just send the bytes as you're doing.

[fallen grotto]

am i following right, your generator outputs info you want to process as its created?

if so, perhaps a queue to send the generator output to

[paper marlin]

Python help channel closed for inactivity

This help channel has been closed. Feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.