voicemeeter-api-python/voicemeeterlib/updater.py

import threading
import time

from .util import comp


class Updater(threading.Thread):
    def __init__(self, remote):
        super().__init__(name="updater", target=self.update, daemon=True)
        self._remote = remote
        self._remote._strip_comp = [False] * (
            2 * self._remote.kind.phys_in + 8 * self._remote.kind.virt_in
        )
        self._remote._bus_comp = [False] * (self._remote.kind.num_bus * 8)

    def _update_comps(self, strip_level, bus_level):
        self._remote._strip_comp, self._remote._bus_comp = (
            tuple(not x for x in comp(self._remote.cache["strip_level"], strip_level)),
            tuple(not x for x in comp(self._remote.cache["bus_level"], bus_level)),
        )

    def update(self):
        """
        Continously update observers of dirty states.

        Generate _strip_comp, _bus_comp and update level cache if ldirty.

        Runs updates at a rate of self.ratelimit.
        """
        while self._remote.running:
            start = time.time()
            if self._remote.event.pdirty and self._remote.pdirty:
                self._remote.subject.notify("pdirty")
            if self._remote.event.mdirty and self._remote.mdirty:
                self._remote.subject.notify("mdirty")
            if self._remote.event.midi and self._remote.get_midi_message():
                self._remote.subject.notify("midi")
            if self._remote.event.ldirty and self._remote.ldirty:
                self._update_comps(self._remote._strip_buf, self._remote._bus_buf)
                self._remote.cache["strip_level"] = self._remote._strip_buf
                self._remote.cache["bus_level"] = self._remote._bus_buf
                self._remote.subject.notify("ldirty")

            elapsed = time.time() - start
            if self._remote.event.any() and self._remote.ratelimit - elapsed > 0:
                time.sleep(self._remote.ratelimit - elapsed)
            else:
                time.sleep(0.1)