I'm working on a Python class that is a cyclic generator, like itertools.cycle, but it offers access to its current state and executes a callback after the completion of each cycle.

I'm looking for feedback on the way I've implemented this so that I might improve it or find a better way.

In this version, I've used itertools.cycle, but I'm wondering if there might be any advantage to subclassing collections.abc.Iterator and building it from the ground up. My thoughts are that I might be able to improve performance by using modulus division to constrain the output, without the overhead of importing other libraries. My primary use for this will be to drive other clocks, so timing is crucial.

I'm also unsure if this is the best way to implement the callback. It works and is simple, but I'm interested in the opinions and experience of others regarding this feature.

All guidance and opinions welcomed!



I've updated the code to provide some context. I've added here the class that processes incoming midi byte stream. This MidiStream object is where I plan to 'plug in' the EventClock, using the midi bytes to trigger the iterator. However I haven't yet connected the two objects.

import asyncio
from time import time
from itertools import cycle
from functools import partial
from typing import Callable, Optional, Tuple

from meta.nebula import Singleton
from src.midiconf import clock, clsattrs

class EventClock:
    """cyclic generator with access to current position
       executes a callback at the start/end of each cycle """
    def __init__(self, size: int, callback: Callable, args: Optional[Tuple]=None):
        self._size = size
        self._position = 0
        self._loop = cycle(range(1, size + 1))
        self._callback = partial(callback, *args)

    def __repr__(self):
        return f"<_class {type(self).__name__}(size={self._size}, position={self._position})_>"
    def nextstep(self):
        self._position = next(self._loop)
        if self._position == 1:
        return self._position

    def position(self):
        return self._position

    def reset(self):
        """reset generator's currrent state to -1"""
        for index in self._loop:
            if index == self.length:

class MidiStream(metaclass=Singleton):
    """process incoming midi byte stream 
       port: rtmidi_port, policy: asyncio loop_policy
    def __init__(self, port, policy=None):
        self.port = port
        self.policy = policy
    def enqueue(self, midipacket, data=None):
        """syncronization callback, fetch and enqueue"""
        timestamp = time()
        msg, delta = midipacket
        queue_packet = (msg, delta, timestamp)
            self.loop.call_soon_threadsafe(self.midistream.put_nowait, queue_packet)

        except BaseException as failure:
            print(f"callback exc: {type(failure)} {failure}")

    async def dequeue(self):
        while True:
            msg, delta, timestamp = await self.midistream.get()
            if msg[0] == self.clock:
                # Trigger the EventClock
                # EventClock cycles and triggers callback
                # Callback does something like, 
                # updates a display or triggers other midi events, etc. 

    async def clocksync(self):
        """main coroutine """
        self.loop = asyncio.get_event_loop()
        self.midistream = asyncio.Queue(maxsize=256)

            await self.dequeue()

        except asyncio.CancelledError:
            self.recyle(self.ppbr, resolution)

    def syncronize(self):

input = MidiStream(midiin)

  • \$\begingroup\$ Welcome on Code Review! Could you please give us a simple use case for the above code? When & how would you use the EventClock ? \$\endgroup\$ Dec 18, 2020 at 0:28
  • \$\begingroup\$ Thanks! The idea came out of the need to filter midi beat clock messages, which will trigger other midi clocks to subdivide the beat and ultimately trigger midi sequences and other types of events. The event clock can be set to cycle at various resolutions, such as 96 pulses per bar or 24 ppqn etc.. The callback is in place to leave it open as far as what the triggered event might be. I always strive to make utilities that might be useful in many applications. \$\endgroup\$
    – Кафка
    Dec 18, 2020 at 2:19
  • \$\begingroup\$ # Trigger the EventClock is not enough for me to see how the clock is used. Please just show all of your code. \$\endgroup\$
    – Reinderien
    Dec 18, 2020 at 22:27
  • \$\begingroup\$ That's as far as I've gotten. I pass the midi byte to a function that advances the EventClock. I may alter EventClock to accept the byte directly. The EventClock cycles and executes the callback. I haven't coded anything beyond what's here aside from some other midi event classes that aren't relevant as they just format messages. I appreciate your input, though. I'll certainly explore what you've contributed so far. I like the terse statements. I'm curious know if there might be any performance advantages by not using abstractions like cycle. \$\endgroup\$
    – Кафка
    Dec 18, 2020 at 22:48

1 Answer 1


Without seeing how this is called, I can't properly understand why this isn't just a "normal" Python iterator-type class:

  • Do not have a reset; simply have the caller restart their own iteration, which will re-invoke __iter__
  • Do not store iteration state on the class; keep it within the __iter__ method
  • Calling args Optional is incorrect because a None will make your expansion crash; so give it a default instead

Something like:

from functools import partial
from typing import Callable, Tuple

class EventClock:
    """cyclic generator with access to current position
       executes a callback at the start/end of each cycle """

    def __init__(self, size: int, callback: Callable, args: Tuple = ()):
        self._loop = range(size)
        self._callback = partial(callback, *args)

    def __iter__(self):
        while True:
            yield from self._loop

def my_cb():

i = 0
ev = EventClock(3, my_cb)
for p in ev:
    i += 1
    if i >= 5:


  • \$\begingroup\$ Thanks for the input! I realized the error with assigning None to args after I posted. My use for this is to filter midi clock messages at various resolutions in order to trigger events in sync with external devices. It's called by pulling a midi message from a queue and passing it to the EventClock, which would trigger the next step in cycle object. After 24 ticks for example, It would trigger another midi event of some sort (callback). There are 5 EventClocks processing the incoming midi stream at different resolutions. I chose a reset option to reset them all whenever the sync restarts. \$\endgroup\$
    – Кафка
    Dec 18, 2020 at 17:22
  • \$\begingroup\$ That's fine, and the code I've suggested can easily accommodate what you've described, but until you show more usage code I can't show you how. \$\endgroup\$
    – Reinderien
    Dec 18, 2020 at 17:30
  • \$\begingroup\$ I've updated the code to provide some context. It's as far as I've gotten with it - just two classes. I haven't put them together yet. \$\endgroup\$
    – Кафка
    Dec 18, 2020 at 20:13
  • 1
    \$\begingroup\$ Ordinarily we'd rollback the edit to the question, but in this case I'd say the answer was a bit premature. Would you please update the answer? \$\endgroup\$
    – Mast
    Dec 18, 2020 at 20:41
  • \$\begingroup\$ Why do you suggest not storing the state as a class property? How would it be accessible from the __ iter__ method if I needed to obtain it at some later time? I like the idea of having the iter restart after exhaustion, but I'll have to implement a function to call it I suppose. Do you think doing it this way might improve performance? \$\endgroup\$
    – Кафка
    Dec 19, 2020 at 7:30

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.