Cyclic generator with access to current local state and executes callback after each cycle

Question

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!

Thanks!

Update:

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})_>"
        
    @property
    def nextstep(self):
        self._position = next(self._loop)
        if self._position == 1:
            self._callback()
        return self._position

    @property
    def position(self):
        return self._position

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

@clsattrs(clock)
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)
        try:
            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)
        self.port.set_callback(self.enqueue)

        try:
            await self.dequeue()

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


    def syncronize(self):
        asyncio.set_event_loop_policy(self.policy)
        asyncio.run(self.clocksync())


input = MidiStream(midiin)
input.syncronize()

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:
            self._callback()
            yield from self._loop


def my_cb():
    print('Callback')

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

"""
Prints:

Callback
0
1
2
Callback
0
1
"""