path: root/helpers/music_file.py

import threading
import pydub
import pygame
import math
import time
from transitions.extensions import HierarchicalMachine as Machine

class MusicFile(Machine):
    def __init__(self, filename, lock, channel_id, name = None, gain = 1):
        states = [
            'initial',
            'loading',
            'failed',
            { 'name': 'loaded', 'children': ['stopped', 'playing', 'paused'] }
        ]
        transitions = [
            { 'trigger': 'load', 'source':  'initial', 'dest': 'loading'},
            { 'trigger': 'fail', 'source':  'loading', 'dest': 'failed'},
            { 'trigger': 'success', 'source':  'loading', 'dest': 'loaded_stopped'},
            #{ 'trigger': 'play', 'source':  'loaded_stopped', 'dest': 'loaded_playing'},
            #{ 'trigger': 'pause', 'source':  'loaded_playing', 'dest': 'loaded_paused'},
            #{ 'trigger': 'stop', 'source':  ['loaded_playing','loaded_paused'], 'dest': 'loaded_stopped'}
        ]

        Machine.__init__(self, states=states, transitions=transitions, initial='initial')

        self.filename = filename
        self.channel_id = channel_id
        self.name = name or filename
        self.raw_data = None
        self.gain = gain

        self.flag_paused = False
        threading.Thread(name = "MSMusicLoad", target = self.load, kwargs = {'lock': lock}).start()

    def on_enter_loading(self, lock=None):
        lock.acquire()
        try:
            print("Loading « {} »".format(self.name))
            volume_factor = 20 * math.log10(self.gain)
            audio_segment = pydub.AudioSegment.from_file(self.filename).set_frame_rate(44100).apply_gain(volume_factor)
            self.sound_duration = audio_segment.duration_seconds
            self.raw_data = audio_segment.raw_data
        except Exception as e:
            print("failed to load « {} »: {}".format(self.name, e))
            self.loading_error = e
            self.fail()
        else:
            self.success()
            print("Loaded « {} »".format(self.name))
        finally:
            lock.release()

    def check_is_loaded(self):
        return self.state.startswith('loaded_')

    def is_playing(self):
        return self.channel().get_busy()

    def is_paused(self):
        return self.flag_paused

    @property
    def sound_position(self):
        if self.is_playing() and not self.is_paused():
            return min(time.time() - self.started_at, self.sound_duration)
        elif self.is_playing() and self.is_paused():
            return min(self.paused_at - self.started_at, self.sound_duration)
        else:
            return 0

    def play(self, fade_in = 0, volume = 100, start_at = 0):
        self.set_volume(volume)

        if start_at > 0:
            raw_data_length = len(self.raw_data)
            start_offset = int((raw_data_length / self.sound_duration) * start_at)
            start_offset = start_offset - (start_offset % 2)
            sound = pygame.mixer.Sound(self.raw_data[start_offset:])
        else:
            sound = pygame.mixer.Sound(self.raw_data)

        self.started_at = time.time()
        self.channel().play(sound, fade_ms = int(fade_in * 1000))
        self.flag_paused = False

    def pause(self):
        self.paused_at = time.time()
        self.channel().pause()
        self.flag_paused = True

    def unpause(self):
        self.started_at += (time.time() - self.paused_at)
        self.channel().unpause()
        self.flag_paused = False

    def stop(self, fade_out = 0):
        if fade_out > 0:
            self.channel().fadeout(int(fade_out * 1000))
        else:
            self.channel().stop()

    def set_volume(self, value):
        if value < 0:
            value = 0
        if value > 100:
            value = 100
        self.channel().set_volume(value / 100)

    def wait_end(self):
        pass

    def channel(self):
        return pygame.mixer.Channel(self.channel_id)