377 lines
13 KiB
Python
377 lines
13 KiB
Python
from argparse import ArgumentParser
|
|
from typing import Generator, List, Tuple
|
|
|
|
import pygame as pg
|
|
import sys
|
|
|
|
from effects.effect import Effect
|
|
from effects.presets import Presets
|
|
from util.color import Colors
|
|
from util.audio import AudioProcess, find_all_inputs
|
|
|
|
|
|
def print_displays() -> None:
|
|
pg.display.init()
|
|
display = pg.display.get_desktop_sizes()
|
|
for i, d in enumerate(display):
|
|
print(f"#{i} - {d[0]}x{d[1]}")
|
|
sys.exit(0)
|
|
|
|
|
|
def print_inputs() -> None:
|
|
devs = find_all_inputs()
|
|
print("\n".join((x["name"] for x in devs)))
|
|
sys.exit(0)
|
|
|
|
|
|
class Beamshow:
|
|
def __init__(
|
|
self,
|
|
audio_device_name: str,
|
|
render3d: bool,
|
|
window: bool,
|
|
trails: bool,
|
|
randomize_interval: float,
|
|
fps: bool,
|
|
display: int,
|
|
beat_reactive: bool,
|
|
) -> 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()
|
|
self.beat_reactive = beat_reactive
|
|
self.beat_factor = 1
|
|
self.beat_skip_counter = 0
|
|
self.last_preset = ""
|
|
|
|
pg.display.init()
|
|
self.window, self.background = self.initialize()
|
|
self.audio = AudioProcess(audio_device_name)
|
|
|
|
def initialize(self) -> Tuple[pg.Surface, pg.Surface]:
|
|
recreate_effect = (
|
|
not self.effects or pg.display.is_fullscreen() == self.windowed
|
|
)
|
|
displays = pg.display.get_desktop_sizes()
|
|
if not 0 <= self.display_id < len(displays):
|
|
raise ValueError(
|
|
f"Display ID {self.display_id} invalid. Must be between 0 and {len(displays)}!"
|
|
)
|
|
|
|
win = pg.display.set_mode(
|
|
size=displays[self.display_id]
|
|
if not self.windowed
|
|
else (displays[self.display_id][0] // 2, displays[self.display_id][1] // 2),
|
|
flags=pg.FULLSCREEN if not self.windowed else 0,
|
|
display=self.display_id,
|
|
)
|
|
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)
|
|
|
|
if recreate_effect:
|
|
self.presets = Presets(
|
|
bounds=win.get_rect(), beat_reactive=self.beat_reactive
|
|
)
|
|
if self.last_preset and self.last_preset in self.presets:
|
|
self.effects = self.presets[self.last_preset]
|
|
else:
|
|
if self.randomize:
|
|
self.last_preset = self.presets.randomize()
|
|
self.effects = self.presets[self.last_preset]
|
|
else:
|
|
self.effects = self.presets.default()
|
|
|
|
return win, background
|
|
|
|
def render_loop_normal(
|
|
self,
|
|
) -> Generator[List[pg.event.Event], Tuple[bool, float], None]:
|
|
is_beat = False
|
|
frames_per_beat = 0.0
|
|
while True:
|
|
unhandled_events: List[pg.event.Event] = []
|
|
for event in pg.event.get():
|
|
unhandled_events.append(event)
|
|
|
|
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(
|
|
self,
|
|
) -> Generator[List[pg.event.Event], Tuple[bool, float], None]:
|
|
stage = pg.Surface(size=self.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)
|
|
]
|
|
|
|
scaled_positions = [
|
|
((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():
|
|
unhandled_events.append(event)
|
|
|
|
stage.blit(self.background, (0, 0))
|
|
for e in self.effects:
|
|
e.update(is_beat=is_beat, frames_per_beat=frames_per_beat)
|
|
e.draw(stage)
|
|
|
|
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))
|
|
self.window.blit(
|
|
pg.transform.scale(stage, scaled_sizes[int(100 * (i / 100) ** 2)]),
|
|
scaled_positions[int(100 * (i / 100) ** 2)],
|
|
)
|
|
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
|
|
strobe = False
|
|
fps_slidewindow = []
|
|
frames_per_beat = 0
|
|
pause_beat = False
|
|
while True:
|
|
with self.audio.lock:
|
|
is_beat = self.audio.is_beat
|
|
self.audio.is_beat = False
|
|
if is_beat:
|
|
if self.beat_skip_counter > 0:
|
|
is_beat = False
|
|
# print(" skip")
|
|
self.beat_skip_counter -= 1
|
|
else:
|
|
# print("beat")
|
|
self.beat_skip_counter = self.beat_factor - 1
|
|
|
|
fps_mean = (
|
|
sum(fps_slidewindow) / len(fps_slidewindow) if fps_slidewindow else 0
|
|
)
|
|
|
|
common_events = loop.send((is_beat and not pause_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
|
|
):
|
|
self.audio.stop()
|
|
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_F6:
|
|
print("Reload")
|
|
self.effects.clear()
|
|
reinitialize = True
|
|
elif event.key == pg.K_F7:
|
|
pause_beat = not pause_beat
|
|
print(f"Pause beat: {pause_beat}")
|
|
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")
|
|
self.beat_skip_counter = 0
|
|
elif event.key == pg.K_PAGEUP:
|
|
self.beat_factor = (
|
|
self.beat_factor // 2 if self.beat_factor > 1 else 1
|
|
)
|
|
print(f"Trigger on every {self.beat_factor} beat")
|
|
elif event.key == pg.K_PAGEDOWN:
|
|
self.beat_factor = self.beat_factor * 2
|
|
print(f"Trigger on every {self.beat_factor} beat")
|
|
elif event.key == pg.K_RETURN:
|
|
strobe = True
|
|
elif event.type == pg.KEYUP:
|
|
if event.key == pg.K_RETURN:
|
|
strobe = False
|
|
|
|
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 strobe:
|
|
self.window.fill(
|
|
Colors.White if (framecounter % 4 <= 1) else Colors.Black
|
|
)
|
|
|
|
if blackout:
|
|
self.window.fill(Colors.Black)
|
|
|
|
if is_beat:
|
|
pg.draw.rect(self.window, Colors.White, (0, 0, 20, 20))
|
|
|
|
current_fps = self.clock.get_fps()
|
|
fps_slidewindow.append(current_fps)
|
|
while len(fps_slidewindow) > 120:
|
|
fps_slidewindow.pop(0)
|
|
|
|
if self.print_fps and (framecounter & 0x3F == 0):
|
|
print(f"FPS: {current_fps:5.2f}, avg {fps_mean:5.2f}")
|
|
if fps_mean < 50:
|
|
pg.draw.rect(
|
|
self.window,
|
|
Colors.Yellow,
|
|
(self.window.get_width() - 20, 0, 20, 20),
|
|
)
|
|
elif fps_mean < 40:
|
|
pg.draw.rect(
|
|
self.window,
|
|
Colors.Red,
|
|
(self.window.get_width() - 20, 0, 20, 20),
|
|
)
|
|
|
|
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()
|
|
print(new_preset)
|
|
|
|
self.last_preset = new_preset
|
|
self.effects.clear()
|
|
self.effects.extend(self.presets[self.last_preset])
|
|
|
|
pg.display.flip()
|
|
self.clock.tick(60)
|
|
framecounter += 1
|
|
|
|
|
|
def app_main() -> None:
|
|
argparser = ArgumentParser(
|
|
description="beamshow - Render a light show for a video projector"
|
|
)
|
|
argparser.add_argument(
|
|
"-a",
|
|
"--audio",
|
|
metavar="NAME",
|
|
type=str,
|
|
required=True,
|
|
help="The audio device to use. Can be any substring",
|
|
)
|
|
argparser.add_argument(
|
|
"--3d",
|
|
dest="render3d",
|
|
action="store_true",
|
|
help="Render a 3D preview instead of the normal output",
|
|
)
|
|
argparser.add_argument(
|
|
"--list-displays", action="store_true", help="Show available displays"
|
|
)
|
|
argparser.add_argument(
|
|
"--list-inputs", action="store_true", help="Show available audio inputs"
|
|
)
|
|
argparser.add_argument(
|
|
"-w", "--window", action="store_true", help="Display in a window"
|
|
)
|
|
argparser.add_argument(
|
|
"-b", "--beat", action="store_true", help="Effects should react to beats"
|
|
)
|
|
argparser.add_argument(
|
|
"--trails", action="store_true", help="Fade patterns out (trail mode)"
|
|
)
|
|
argparser.add_argument(
|
|
"--randomize",
|
|
metavar="N",
|
|
type=int,
|
|
nargs="?",
|
|
default=0,
|
|
const=5,
|
|
help="Select random effect presets after <N> seconds",
|
|
)
|
|
argparser.add_argument("--fps", action="store_true", help="Show FPS in console")
|
|
argparser.add_argument(
|
|
"-d", "--display", type=int, default=0, help="ID of the display to use"
|
|
)
|
|
|
|
# print some nice output after the pygame banner to separate our stuff from theirs
|
|
print("")
|
|
print("-" * (len(str(argparser.description)) + 4))
|
|
print(f" {argparser.description} ")
|
|
print("-" * (len(str(argparser.description)) + 4))
|
|
print("")
|
|
|
|
args = argparser.parse_args()
|
|
|
|
if args.list_displays:
|
|
print_displays()
|
|
|
|
if args.list_displays:
|
|
print_inputs()
|
|
|
|
show = Beamshow(
|
|
audio_device_name=args.audio,
|
|
render3d=args.render3d,
|
|
window=args.window,
|
|
trails=args.trails,
|
|
randomize_interval=args.randomize,
|
|
fps=args.fps,
|
|
display=args.display,
|
|
beat_reactive=args.beat,
|
|
)
|
|
show.main()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app_main()
|