Refactor into class; add beat control

beamshow.py

@@ -1,11 +1,11 @@
 from argparse import ArgumentParser
-from typing import Generator, Iterable, Tuple
+import time
+from typing import Generator, List, Tuple
 
 import pygame as pg
 import sys
 
 from effects.effect import Effect, color_wheel, Colors
-from effects.moonflower import Moonflower
 from effects.presets import Presets
 
 
@@ -17,98 +17,263 @@ def print_displays() -> None:
     sys.exit(0)
 
 
-def initialize(
+class Beamshow:
-    display_id: int, windowed: bool, trails: bool
+    def __init__(
-) -> Tuple[pg.Surface, pg.Surface]:
+        self,
+        render3d: bool,
+        window: bool,
+        trails: bool,
+        randomize_interval: float,
+        fps: bool,
+        display: int,
+    ) -> None:
+        self.render3d = render3d
+        self.windowed = window
+        self.trails = trails
+        self.randomize = bool(randomize_interval)
+        self.randomize_interval = randomize_interval or 5
+        self.print_fps = fps
+        self.display_id = display
+        self.effects: List[Effect] = []
+        self.clock = pg.time.Clock()
+
         pg.display.init()
+        self.window, self.background = self.initialize()
+
+    def initialize(self) -> Tuple[pg.Surface, pg.Surface]:
         displays = pg.display.get_desktop_sizes()
-    if not 0 <= display_id < len(displays):
+        if not 0 <= self.display_id < len(displays):
             raise ValueError(
-            f"Display ID {display_id} invalid. Must be between 0 and {len(displays)}!"
+                f"Display ID {self.display_id} invalid. Must be between 0 and {len(displays)}!"
             )
 
         win = pg.display.set_mode(
-        size=displays[display_id]
+            size=displays[self.display_id]
-        if not windowed
+            if not self.windowed
-        else (displays[display_id][0] // 2, displays[display_id][1] // 2),
+            else (displays[self.display_id][0] // 2, displays[self.display_id][1] // 2),
-        flags=pg.FULLSCREEN if not windowed else 0,
+            flags=pg.FULLSCREEN if not self.windowed else 0,
-        display=display_id,
+            display=self.display_id,
         )
-    if trails:
+        if self.trails:
             background = pg.Surface(win.get_size(), flags=pg.SRCALPHA)
             background.fill(pg.Color(0, 0, 0, 5))
         else:
             background = pg.Surface(win.get_size())
             background.fill(Colors.Black)
+
+        self.presets = Presets(bounds=win.get_rect())
+        if self.randomize:
+            self.effects = self.presets.randomize()
+        else:
+            from effects.moonflower import Moonflower
+
+            self.effects = [
+                Moonflower(
+                    bounds=win.get_rect(),
+                    colors=(
+                        color_wheel(increase=75),
+                        color_wheel(hue=180, increase=75),
+                    ),
+                    beat_color=True,
+                    size=win.get_height() // 4,
+                    outer=5,
+                    velocity=(1, 1),
+                    rot_speed=1.5,
+                    x_factor=(1, 1),
+                    y_factor=(2.2, 2.2),
+                )
+            ]
+
         return win, background
 
-
     def render_loop_normal(
-    window: pg.Surface,
+        self,
-    background: pg.Surface,
+    ) -> Generator[List[pg.event.Event], Tuple[bool, float], None]:
-    effects: Iterable[Effect],
+        is_beat = False
-    clock: pg.time.Clock,
+        frames_per_beat = 0.0
-) -> Generator[None, None, None]:
-    blackout = False
-
         while True:
+            unhandled_events: List[pg.event.Event] = []
             for event in pg.event.get():
-            if event.type == pg.K_SPACE:
+                unhandled_events.append(event)
-                blackout = not blackout
-            if event.type == pg.QUIT:
-                pg.quit()
-                sys.exit()
-
-        window.blit(background, (0, 0))
-        if not blackout:
-            for e in effects:
-                e.update()
-                e.draw(window)
-        yield
 
+            self.window.blit(self.background, (0, 0))
+            for e in self.effects:
+                e.update(is_beat=is_beat, frames_per_beat=frames_per_beat)
+                e.draw(self.window)
+            is_beat, frames_per_beat = yield unhandled_events
 
     def render_loop_3d(
-    window: pg.Surface,
+        self,
-    background: pg.Surface,
+    ) -> Generator[List[pg.event.Event], Tuple[bool, float], None]:
-    effects: Iterable[Effect],
+        stage = pg.Surface(size=self.window.get_size())
-    clock: pg.time.Clock,
-) -> Generator[None, None, None]:
-    stage = pg.Surface(size=window.get_size())
         stage.fill(Colors.Black)
 
         full_size = stage.get_size()
         scaled_sizes = [
-        (((full_size[0] * i) // 100), ((full_size[1] * i) // 100)) for i in range(101)
+            (((full_size[0] * i) // 100), ((full_size[1] * i) // 100))
+            for i in range(101)
         ]
 
         scaled_positions = [
-        ((full_size[0] - s[0]) // 2, (full_size[1] - s[1]) // 2) for s in scaled_sizes
+            ((full_size[0] - s[0]) // 2, (full_size[1] - s[1]) // 2)
+            for s in scaled_sizes
         ]
 
+        is_beat = False
+        frames_per_beat = 0.0
         while True:
+            unhandled_events: List[pg.event.Event] = []
             for event in pg.event.get():
-            if event.type == pg.QUIT:
+                unhandled_events.append(event)
-                pg.quit()
-                sys.exit()
 
-        stage.blit(background, (0, 0))
+            stage.blit(self.background, (0, 0))
-        for e in effects:
+            for e in self.effects:
-            e.update()
+                e.update(is_beat=is_beat, frames_per_beat=frames_per_beat)
                 e.draw(stage)
 
-        window.fill(Colors.Black)
+            self.window.fill(Colors.Black)
 
             stage.set_colorkey(Colors.Black)
             for i in range(0, 101, 4):
                 stage.set_alpha(192 - 192 * ((i / 100) ** 0.5))
-            window.blit(
+                self.window.blit(
                     pg.transform.scale(stage, scaled_sizes[int(100 * (i / 100) ** 2)]),
                     scaled_positions[int(100 * (i / 100) ** 2)],
                 )
-        yield
+            is_beat, frames_per_beat = yield unhandled_events
+
+    def main(self):
+
+        if self.render3d:
+            loop = self.render_loop_3d()
+        else:
+            loop = self.render_loop_normal()
+        next(loop)
+
+        framecounter = 0
+        blackout = False
+        single_random = False
+        taps = []
+        tap_mean = 0
+        last_beat = 0
+        fps_slidewindow = []
+        frames_per_beat = 0
+        while True:
+            is_beat = False
+            fps_mean = (
+                sum(fps_slidewindow) / len(fps_slidewindow) if fps_slidewindow else 0
+            )
+            if tap_mean and last_beat:
+                this_beat = time.time()
+                is_beat = this_beat >= last_beat + tap_mean
+                frames_per_beat = fps_mean * tap_mean
+                if is_beat:
+                    last_beat = this_beat
+
+            common_events = loop.send((is_beat, frames_per_beat))
+            reinitialize = False
+            for event in common_events:
+                if event.type == pg.QUIT or (
+                    event.type == pg.KEYDOWN and event.key == pg.K_ESCAPE
+                ):
+                    pg.quit()
+                    sys.exit()
+                elif event.type == pg.KEYDOWN:
+                    if event.key == pg.K_F5:
+                        single_random = True
+                        print("Switching to new random preset")
+                    elif event.key == pg.K_F8:
+                        self.randomize = not self.randomize
+                        state_str = (
+                            f"on, {self.randomize_interval} s"
+                            if self.randomize
+                            else "off"
+                        )
+                        print(f"Random preset switching: [{state_str}]")
+                    elif event.key == pg.K_F9:
+                        self.render3d = not self.render3d
+                        print(f'Pseudo-3d view [{"on" if self.render3d else "off"}]')
+                        reinitialize = True
+                    elif event.key == pg.K_F10:
+                        self.trails = not self.trails
+                        print(f'Trails [{"on" if self.trails else "off"}]')
+                        reinitialize = True
+                    elif event.key == pg.K_F11:
+                        self.windowed = not self.windowed
+                        print(f'Windowed mode [{"on" if self.windowed else "off"}]')
+                        reinitialize = True
+                    elif event.key == pg.K_SPACE:
+                        blackout = not blackout
+                        print(f'BLACKOUT [{"ON" if blackout else "off"}]')
+                    elif event.key == pg.K_HOME:
+                        print("resetting beat timing")
+                        this_tap = time.time()
+                        last_beat = this_tap
+                    elif event.key == pg.K_END:
+                        print("starting new tap measurement")
+                        taps.clear()
+                    elif event.key == pg.K_PAGEUP:
+                        tap_mean /= 2
+                        print(
+                            f"Taps: Mean {tap_mean:5.3f} s, {60/tap_mean:5.3f} BPM, {fps_mean * tap_mean:5.3f} FPB [x2]"
+                        )
+                    elif event.key == pg.K_PAGEDOWN:
+                        tap_mean *= 2
+                        print(
+                            f"Taps: Mean {tap_mean:5.3f} s, {60/tap_mean:5.3f} BPM, {fps_mean * tap_mean:5.3f} FPB [/2]"
+                        )
+                    elif event.key == pg.K_KP_ENTER or event.key == pg.K_RETURN:
+                        this_tap = time.time()
+                        if taps and this_tap - taps[-1] > 10:
+                            print("starting new tap measurement")
+                            taps.clear()
+                        taps.append(this_tap)
+                        last_beat = this_tap
+                        if len(taps) >= 2:
+                            tap_mean = sum(
+                                map(lambda t1, t2: t2 - t1, taps, taps[1:])
+                            ) / (len(taps) - 1)
+                            print(
+                                f"Taps: Mean {tap_mean:5.3f} s, {60/tap_mean:5.3f} BPM, {fps_mean * tap_mean:5.3f} FPB"
+                            )
+
+            if reinitialize:
+                self.window, self.background = self.initialize()
+                if self.render3d:
+                    loop = self.render_loop_3d()
+                else:
+                    loop = self.render_loop_normal()
+                next(loop)
+
+            if blackout:
+                self.window.fill(Colors.Black)
+
+            if is_beat:
+                pg.draw.rect(self.window, Colors.White, (0, 0, 30, 30))
+
+            pg.display.flip()
+            self.clock.tick(60)
+            framecounter += 1
+
+            if (
+                self.randomize and (framecounter % (self.randomize_interval * 60)) == 0
+            ) or single_random:
+                single_random = False
+                new_preset = self.presets.randomize()
+                while new_preset == self.effects:
+                    new_preset = self.presets.randomize()
+                self.effects.clear()
+                self.effects.extend(self.presets.randomize())
+
+            current_fps = self.clock.get_fps()
+            fps_slidewindow.append(current_fps)
+            while len(fps_slidewindow) > 30:
+                fps_slidewindow.pop(0)
+            if self.print_fps:
+                print(f"FPS: {current_fps:3d}, avg {fps_mean:3d}")
 
 
-def main() -> None:
+def app_main() -> None:
     argparser = ArgumentParser(
         description="beamshow - Render a light show for a video projector"
     )
@@ -153,50 +318,16 @@ def main() -> None:
     if args.list_displays:
         print_displays()
 
-    window, background = initialize(
+    show = Beamshow(
-        display_id=args.display, windowed=args.window, trails=args.trails
+        render3d=args.render3d,
+        window=args.window,
+        trails=args.trails,
+        randomize_interval=args.randomize,
+        fps=args.fps,
+        display=args.display,
     )
-
+    show.main()
-    presets = Presets(bounds=window.get_rect())
-    if args.randomize:
-        effects = presets.randomize()
-    else:
-        effects = [
-            Moonflower(
-                bounds=window.get_rect(),
-                # colors=(Colors.Red,Colors.Blue),
-                colors=(color_wheel(), color_wheel(hue=180)),
-                size=window.get_height() / 4,
-                outer=5,
-                velocity=(1, 1),
-                rot_speed=1.5,
-                x_factor=(1, 1),
-                y_factor=(2.2, 2.2),
-            ),
-        ]
-
-    clock = pg.time.Clock()
-
-    if args.render3d:
-        loop = render_loop_3d(window, background, effects, clock)
-    else:
-        loop = render_loop_normal(window, background, effects, clock)
-
-    framecounter = 0
-    while True:
-        next(loop)
-        pg.display.flip()
-        clock.tick(60)
-        framecounter += 1
-
-        if args.randomize:
-            if (framecounter % (args.randomize * 60)) == 0:
-                effects.clear()
-                effects.extend(presets.randomize())
-
-        if args.fps:
-            print(clock.get_fps())
 
 
 if __name__ == "__main__":
-    main()
+    app_main()

effects/doublespot.py

@@ -61,6 +61,7 @@ class DoubleSpot(Effect):
         hold: int = 60 * 1,
         fade_out: bool = False,
         fade_in: bool = False,
+        beat_adapt: bool = False,
         *groups: pg.sprite.Group
     ) -> None:
         self.color = color
@@ -68,6 +69,7 @@ class DoubleSpot(Effect):
         self.fade_in_time: int = hold // 4 if fade_in else 0
         self.fade_out_time: int = hold // 4 if fade_out else 0
         self.hold_time: int = hold - (self.fade_in_time + self.fade_out_time)
+        self.beat_adapt: bool = beat_adapt
         self.spots: List[Spot] = []
 
         self.randrange = (
@@ -79,7 +81,7 @@ class DoubleSpot(Effect):
         image.set_colorkey(Colors.Black)
         super().__init__(image, bounds, *groups)
 
-        self.update()
+        self.update(is_beat=False, frames_per_beat=0.0)
 
     def add_spot(self, spot_color):
         position = (
@@ -111,14 +113,29 @@ class DoubleSpot(Effect):
     def update(self, *args: Any, **kwargs: Any) -> None:
         self.image.fill(Colors.Black)
 
+        if self.beat_adapt and kwargs["frames_per_beat"]:
+            hold = kwargs["frames_per_beat"] - 2
+            self.fade_in_time = int(hold // 4 if self.fade_in_time else 0)
+            self.fade_out_time = int(hold // 4 if self.fade_out_time else 0)
+            self.hold_time = int(hold - self.fade_in_time)
+
+        if not self.beat_adapt:
             if len(self.spots) == 0 or (
-            self.fade_in_time != 0 and self.spots[0].start_next and len(self.spots) == 2
+                self.fade_in_time != 0
+                and self.spots[0].start_next
+                and len(self.spots) == 2
             ):
                 spot_color = (
                     self.color if isinstance(self.color, pg.Color) else next(self.color)
                 )
                 self.add_spot(spot_color)
                 self.add_spot(spot_color)
+        elif kwargs["is_beat"]:
+            spot_color = (
+                self.color if isinstance(self.color, pg.Color) else next(self.color)
+            )
+            self.add_spot(spot_color)
+            self.add_spot(spot_color)
 
         for spot in self.spots:
             if not spot.finished:

effects/effect.py

@@ -1,10 +1,8 @@
 from dataclasses import dataclass
-from enum import Enum
 import math
 import random
 from typing import Generator, Tuple
 import pygame as pg
-from abc import abstractmethod
 
 
 def copy_color(source: pg.Color) -> pg.Color:
@@ -47,28 +45,28 @@ def color_randomize() -> Generator[pg.Color, None, None]:
 def transform_bounce(
     bounds: pg.Rect,
     velocity: Tuple[int, int],
-    x_factor: Tuple[int, int],
+    x_factor: Tuple[float, float],
-    y_factor: Tuple[int, int],
+    y_factor: Tuple[float, float],
 ) -> Generator[Tuple[int, int], Tuple[int, int], None]:
     min_velocity = velocity[0]
     max_velocity = velocity[1]
     current_velocity = random.randint(min_velocity, max_velocity)
-    ticks = random.randint(0, 360)
+    phase = random.uniform(0, 360)
     current_x_factor = random.uniform(x_factor[0], x_factor[1])
     current_y_factor = random.uniform(y_factor[0], y_factor[1])
 
     size_x, size_y = yield (bounds.centerx, bounds.centery)
     while True:
         pos_x = int(
-            math.cos(current_x_factor * ticks) * (bounds.width - size_x) // 2
+            math.cos(current_x_factor * phase) * (bounds.width - size_x) // 2
             + bounds.centerx
         )
         pos_y = int(
-            math.sin(current_y_factor * ticks) * (bounds.height - size_y) // 2
+            math.sin(current_y_factor * phase) * (bounds.height - size_y) // 2
             + bounds.centery
         )
 
-        ticks += int(current_velocity / 180 * math.pi)
+        phase += current_velocity / 180 * math.pi
         current_velocity = random.randint(min_velocity, max_velocity)
         size_x, size_y = yield (pos_x, pos_y)
 

effects/moonflower.py

@@ -20,22 +20,24 @@ class Moonflower(Effect):
             Union[pg.Color, Generator[pg.Color, None, None]],
             Union[pg.Color, Generator[pg.Color, None, None]],
         ],
-        size=100,
+        beat_color: bool = False,
-        velocity=(1, 10),
+        size: int = 100,
-        rot_speed=5,
+        velocity: Tuple[int, int] = (1, 10),
-        outer=5,
+        rot_speed: float = 5,
-        x_factor=(0.1, 1),
+        outer: int = 5,
-        y_factor=(0.1, 1),
+        x_factor: Tuple[float, float] = (0.1, 1),
+        y_factor: Tuple[float, float] = (0.1, 1),
         *groups: pg.sprite.Group
     ) -> None:
         self.min_velocity = velocity[0]
         self.max_velocity = velocity[1]
 
         self.rot_speed = rot_speed
-        self.rotation = 0
+        self.rotation = 0.0
         self.velocity = random.randint(self.min_velocity, self.max_velocity)
-        self.ticks = random.randint(0, 360)
+        # self.ticks = random.randint(0, 360)
         self.colors = colors
+        self.beat_color = beat_color
         self.o_count = outer
 
         self.o_f = 1 / math.sin(math.pi / self.o_count)
@@ -61,26 +63,28 @@ class Moonflower(Effect):
             bounds=bounds, velocity=velocity, x_factor=x_factor, y_factor=y_factor
         )
         next(self.bouncer)
-        self.update()
+        self.o_color = (
-
-    def update(self, *args: Any, **kwargs: Any) -> None:
-        self.rect.center = self.bouncer.send(self.rect.size)
-
-        self.image.fill(Colors.Blue)
-        pg.draw.rect(
-            self.image, Colors.Black, (1, 1, self.rect.width - 2, self.rect.height - 2)
-        )
-
-        o_color = (
             self.colors[0]
             if isinstance(self.colors[0], pg.Color)
             else next(self.colors[0])
         )
-        i_color = (
+        self.i_color = (
             self.colors[1]
             if isinstance(self.colors[1], pg.Color)
             else next(self.colors[1])
         )
+        self.update(is_beat=False)
+
+    def update(self, *args: Any, **kwargs: Any) -> None:
+        self.rect.center = self.bouncer.send(self.rect.size)
+
+        self.image.fill(Colors.Black)
+
+        if not self.beat_color or kwargs["is_beat"]:
+            if isinstance(self.colors[0], Generator):
+                self.o_color = next(self.colors[0])
+            if isinstance(self.colors[1], Generator):
+                self.i_color = next(self.colors[1])
 
         for phi in range(0, 360, 360 // self.o_count):
             rads = math.pi * (phi + self.rotation) / 180
@@ -88,17 +92,17 @@ class Moonflower(Effect):
             pos_y = self.rect.height // 2 + self.o_radius * math.sin(rads)
             pg.draw.circle(
                 self.image,
-                o_color,
+                self.o_color,
                 (pos_x, pos_y),
                 self.spot_radius,
             )
         pg.draw.circle(
             self.image,
-            i_color,
+            self.i_color,
             self.image.get_rect().center,
             self.i_radius,
         )
 
         self.rotation += self.rot_speed
-        self.ticks += int(self.velocity / 180 * math.pi)
+        # self.ticks += int(self.velocity / 180 * math.pi)
         self.velocity = random.randint(self.min_velocity, self.max_velocity)

effects/presets.py

@@ -52,6 +52,3 @@ class Presets:
                 y_factor=(2.2, 2.2),
             ),
         ]
-
-
-#