1
0
Fork 0
Você não pode selecionar mais de 25 tópicos Os tópicos devem começar com uma letra ou um número, podem incluir traços ('-') e podem ter até 35 caracteres.

166 linhas
4.9 KiB

#!/usr/bin/env python3
from glob import glob
import os
import random
import sys
from threading import Thread
import time
from typing import Union, List, Mapping
os.environ["PYGAME_HIDE_SUPPORT_PROMPT"] = "hide"
from pygame import mixer
from collection import SampleCollection
from atmorepl import AtmoReplRunner
VERBOSITY = 0
channel_index = 0
class CampAtmo:
tracks_per_type = 2
single_track_types = ["lines", "reactions"]
background_track_types = ["background", "buzzwords"]
type_mix = {
"lines": 0.8,
"reactions": 0.7,
"buzzwords": 0.6,
"background": 0.3,
"other": 0.5,
}
lines_delta_sec = (5, 15)
line_reaction_wait_sec = 0.8
def __init__(self):
self.stopped = False
mixer.init(frequency=22050 * 2)
type_dirs = glob("samples/*/")
self.types = {}
for type_dir in type_dirs:
type_name = type_dir.split("/")[1]
self.types[type_name] = SampleCollection(
type_dir, volume=self.type_mix[type_name]
)
i = 0
mixer.set_num_channels(
sum(
[
1 if name in self.single_track_types else self.tracks_per_type
for name in self.types
]
)
)
v(f"{mixer.get_num_channels()} channels set")
self.type_channels = {}
for type_name in self.types:
self.type_channels[type_name] = []
for k in range(
1 if type_name in self.single_track_types else self.tracks_per_type
):
self.type_channels[type_name].append(mixer.Channel(i))
i += 1
def start(self):
self.background_thread = Thread(
None, self.run_forever, "atmo-background", daemon=True
)
self.background_thread.start()
self.lines_thread = Thread(
None, self.run_lines_forever, "atmo-lines", daemon=True
)
self.lines_thread.start()
def stop(self):
self.stopped = True
self.background_thread.join(1.2)
self.lines_thread.join(2)
def load_sounds(self):
for _, collection in self.types.items():
collection.load()
def run_forever(self):
while not self.stopped:
for type_name, channels in self.type_channels.items():
if (
type_name not in self.background_track_types
or len(self.types[type_name]) == 0
):
continue
self.play_type_sounds(type_name)
time.sleep(1)
def run_lines_forever(self):
while not self.stopped:
time.sleep(random.randrange(*self.lines_delta_sec))
length_sec = self.play_type_sounds("lines")
time.sleep(length_sec + self.line_reaction_wait_sec)
length_sec = self.play_type_sounds("reactions")
time.sleep(length_sec + self.line_reaction_wait_sec)
self.play_type_sounds("reactions")
def play_type_sounds(self, type_name) -> int:
if type_name in self.types and len(self.types[type_name]) == 0:
return
max_sound_length = 0
for channel in self.type_channels[type_name]:
if channel.get_queue() is not None and len(self.types[type_name]) > 1:
continue
sample = self.types[type_name].next()
max_sound_length = max(max_sound_length, sample.sound.get_length())
channel.queue(sample.sound)
if (
type_name not in self.single_track_types
and channel.get_queue() is None
and len(self.types[type_name]) > 1
):
channel.queue(self.types[type_name].next_sound())
return max_sound_length
def get_status(self):
ret_strs = []
for type_name, channels in self.type_channels.items():
samples = []
for chan in channels:
sample = self.types[type_name].get_sample_from_channel(chan)
if sample is not None:
samples.append(sample)
if samples:
sample_infos = [
f"{s.path} @{s.sound.get_volume():.0%}vol" for s in samples
]
ret_strs.append(f"{type_name}: {sample_infos}")
return "\n".join(ret_strs)
def pause(self):
if mixer.get_busy():
mixer.pause()
return False
else:
mixer.unpause()
return True
def main():
atmo = CampAtmo()
atmoRepl = AtmoReplRunner(atmo, port=7723)
atmoRepl.start()
atmo.start()
try:
while True:
time.sleep(10)
except KeyboardInterrupt:
print("exiting...")
atmo.stop()
sys.exit(0)
def v(*msg):
if VERBOSITY >= 1:
print(*msg)
if __name__ == "__main__":
main()