camp-atmo/campatmo.py

#!/usr/bin/env python3
import time
import sys
import random
from glob import glob
from pygame import mixer


VERBOSITY = 0


class CampAtmo:
    tracks_per_type = 2
    single_track_types = ["lines"]
    sound_volume = 0.5

    def __init__(self):
        mixer.init(frequency=22050 * 2)
        self.load_sounds()

    def load_sounds(self):
        type_names = glob("samples/*/")
        type_files = {c.split("/")[1]: glob(c + "/*.ogg") for c in type_names}
        self.types = {}
        i = 0
        mixer.set_num_channels(
            sum(
                [
                    1 if name in self.single_track_types else self.tracks_per_type
                    for name in type_files
                ]
            )
        )
        v(f"{mixer.get_num_channels()} channels set")
        for name in type_files:
            self.types[name] = []
            for k in range(
                1 if name in self.single_track_types else self.tracks_per_type
            ):
                self.types[name].append(mixer.Channel(i))
                i += 1
        self.sounds = {}
        for name, files in type_files.items():
            self.sounds[name] = []
            for f in files:
                sound = mixer.Sound(f)
                sound.set_volume(self.sound_volume)
                self.sounds[name].append(sound)
            random.shuffle(self.sounds[name])

    def manage_type_queue(self, name, type_channels):
        try:
            next_sound_index = max(
                [
                    self.sounds[name].index(tt.get_queue() or tt.get_sound())
                    for tt in type_channels
                ]
            )
        except ValueError:
            next_sound_index = 0
        if next_sound_index >= len(self.sounds[name]):
            random.shuffle(self.sounds[name])
            next_sound_index = 0
        for c in type_channels:
            if c.get_queue() is not None and len(self.sounds[name]) > 1:
                continue
            c.queue(self.sounds[name][next_sound_index])
            if c.get_queue() is None and len(self.sounds[name]) > 1:
                c.queue(self.sounds[name][next_sound_index + 1])

    def run_forever(self):
        while True:
            for name, type_channels in self.types.items():
                if len(self.sounds[name]) == 0 or name in self.single_track_types:
                    continue
                self.manage_type_queue(name, type_channels)
            time.sleep(1)
            self.load_sounds()


def main():
    atmo = CampAtmo()
    try:
        atmo.run_forever()
    except KeyboardInterrupt:
        print("exit")
        sys.exit(0)


def v(*msg):
    if VERBOSITY >= 1:
        print(*msg)


if __name__ == "__main__":
    main()