jack_of_hearts

#!/usr/bin/python3

# Example JACK midi event generator
#
# Play a drum pattern over JACK

import argparse
import asyncio
import json
import statistics
import sys
from collections import deque
from typing import NamedTuple, Type

import pyeep.app
from pyeep.generative import (DRUM_CHANNEL, DRUM_CLOSED_HIHAT, DRUM_CRASH1,
                              DRUM_LOW_TOM, DRUM_SIDE_STICK, GenerativeScore)
from pyeep.jackmidi import MidiPlayer

# See:
# https://soundprogramming.net/file-formats/general-midi-instrument-list/
# https://www.pgmusic.com/tutorial_gm.htm


class HeartSample(NamedTuple):
    # UNIX timestamp in nanoseconds
    time: int
    rate: float
    rr: tuple[float] = ()


class DrumLoop(GenerativeScore):
    def __init__(self, *, player: MidiPlayer, heart: "App", bpm: int = 60, **kw):
        super().__init__(player=player, bpm=bpm)
        self.heart = heart
        self.channel = DRUM_CHANNEL

    def beat(self):
        super().beat()
        self.bpm = self.heart.last_sample.rate

        if self.heart.improvised_delta > 2:
            self.drum(DRUM_LOW_TOM, 0, 1/4)
            self.drum(DRUM_CRASH1, 1/2, 1/4)
        elif self.heart.improvised_delta > 0.5:
            self.drum(DRUM_LOW_TOM, 0, 1/4)
            self.drum(DRUM_CLOSED_HIHAT, 1/2, 1/4)
        elif self.heart.improvised_delta < -0.5:
            self.drum(DRUM_SIDE_STICK, 0, 1/4, velocity=64)
        else:
            self.drum(DRUM_LOW_TOM, 0, 1/4)


class App(pyeep.app.JackApp):
    def __init__(self, args: argparse.Namespace):
        super().__init__(args)
        self.player = self.add_jack_component(MidiPlayer)
        self.score: GenerativeScore | None = None

        self.last_sample: HeartSample | None = None
        self.last_window: deque[float] = deque(maxlen=10)
        self.improvised_delta: float = 0

    async def read_socket(self):
        # Read heart beats from https://www.enricozini.org/blog/2023/debian/monitoring-a-heart-rate-monitor/
        reader, writer = await asyncio.open_unix_connection(self.args.socket)

        # Skip the initial line with recent heartbeat history
        initial = json.loads(await reader.readline())
        for sample in (HeartSample(*s) for s in initial["last"]):
            self.last_window.append(sample.rate)

        while not self.shutting_down and (line := await reader.readline()):
            self.last_sample = HeartSample(*json.loads(line))
            self.on_sample()

    def on_sample(self):
        if len(self.last_window) > 5:
            mean = statistics.mean(self.last_window)
            variance = statistics.variance(self.last_window)
            if self.last_sample.rate > mean + variance:
                self.improvised_delta = self.last_sample.rate - mean - variance
            elif self.last_sample.rate < mean - variance:
                self.improvised_delta = -(mean - variance - self.last_sample.rate)
            else:
                self.improvised_delta = 0.0
            print("Improvised delta:", self.improvised_delta)
        self.last_window.append(self.last_sample.rate)

    def set_score(self, score_cls: Type[GenerativeScore]):
        if self.score is not None:
            self.score.stop()
        self.score = score_cls(player=self.player, heart=self)

    async def composer(self):
        while not self.shutting_down:
            if self.score is None or self.last_sample is None:
                await asyncio.sleep(0.2)
                continue

            self.score.beat()

            await asyncio.sleep(60 / self.score.bpm)

        if self.score.stop():
            await asyncio.sleep(60 / self.score.bpm)

    async def aio_main(self):
        await asyncio.gather(self.read_socket(), self.composer())

    def ui_main(self):
        try:
            while not self.shutting_down:
                cmd = input("> ")
                if cmd.startswith("q"):
                    break
        except EOFError:
            pass

    def main(self, score_cls: Type[GenerativeScore]):
        self.set_score(score_cls)
        super().main()


def main():
    parser = App.argparser("JACK of Hearts", "Generate a drum pattern based on heart beat")
    parser.add_argument("socket", action="store", help="path to the socket to use to read heart beats")
    args = parser.parse_args()

    with App(args) as app:
        app.main(DrumLoop)


if __name__ == "__main__":
    sys.exit(main())