emmental/audio/bass.py

# Copyright 2021 (c) Anna Schumaker.
import gi
gi.require_version("Gst", "1.0")

import lib
import sys
from gi.repository import GObject
from gi.repository import GLib
from gi.repository import Gst
Gst.init(sys.argv)

from . import replaygain

TIMEOUT = 100


class BassPlayer(GObject.GObject):
    def __init__(self):
        GObject.GObject.__init__(self)
        lib.settings.initialize("audio.replaygain", "disabled")
        lib.settings.initialize("audio.volume", 1.0)

        self.audio = replaygain.ReplayGainSink()
        self.video  = Gst.ElementFactory.make("fakesink")

        self.playbin = Gst.ElementFactory.make("playbin")
        self.playbin.set_property("audio-sink", self.audio)
        self.playbin.set_property("video-sink", self.video)
        self.playbin.set_property("volume", lib.settings.get_float("audio.volume"))
        self.playbin.set_state(Gst.State.READY)
        self.set_property("replaygain", lib.settings.get("audio.replaygain"))

        self.bus.add_signal_watch()
        self.bus.connect("message::eos", self.__eos__)
        self.bus.connect("message::state-changed", self.state_changed)
        self.bus.connect("message::stream-start", self.stream_start)
        self.bus.connect("message::state-changed", self.state_changed)
        self.bus.connect("message::tag", self.tag)

        self.timeout = None

    def __eos__(self, bus, message):
        self.emit("eos")

    @GObject.Property
    def bus(self):
        return self.playbin.get_bus()

    @GObject.Property
    def duration(self):
        (res, dur) = self.playbin.query_duration(Gst.Format.TIME)
        return dur if res == True else 0

    @GObject.Property
    def playing(self):
        (ret, state, pending) = self.playbin.get_state(Gst.CLOCK_TIME_NONE)
        return state == Gst.State.PLAYING

    @playing.setter
    def playing(self, playing):
        state = Gst.State.PLAYING if playing else Gst.State.PAUSED
        self.playbin.set_state(state)

    @GObject.Property
    def play_percent(self):
        if self.playbin.clock == None or self.duration == 0:
            return 0
        runtime = self.playbin.clock.get_time() - self.playbin.base_time
        return runtime / self.duration

    @GObject.Property
    def position(self):
        (res, pos) = self.playbin.query_position(Gst.Format.TIME)
        return pos if res == True else 0

    @position.setter
    def position(self, pos):
        self.playbin.seek_simple(Gst.Format.TIME,
                                 Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT,
                                 pos)

    @GObject.Property
    def replaygain(self):
        return self.audio.get_property("mode")

    @replaygain.setter
    def replaygain(self, mode):
        lib.settings.set("audio.replaygain", mode)
        self.audio.set_property("mode", mode)

    @GObject.Property
    def uri(self):
        return self.playbin.get_property("uri")

    @uri.setter
    def uri(self, uri):
        if uri:
            self.playbin.set_property("uri", uri)
        else:
            self.playbin.set_state(Gst.State.READY)

    @GObject.Property
    def volume(self):
        return self.playbin.get_property("volume")

    @volume.setter
    def volume(self, vol):
        self.playbin.set_property("volume", vol)
        lib.settings.set("audio.volume", vol)

    def state_changed(self, bus, message):
        if message.src == self.playbin:
            (old, new, pending) = message.parse_state_changed()
            if new == Gst.State.PLAYING:
                self.emit("playback-start")
            else:
                self.emit("playback-paused")

    def stream_start(self, bus, message):
        self.emit("duration-changed")

    def tag(self, bus, message):
        (res, sample) = message.parse_tag().get_sample("image")
        if res:
            self.emit("artwork", sample)

    def timeout_function(self):
        self.emit("position-changed")
        return GLib.SOURCE_CONTINUE

    @GObject.Signal
    def about_to_finish(self):
        pass

    @GObject.Signal(arg_types=(Gst.Sample,))
    def artwork(self, sample):
        pass

    @GObject.Signal
    def duration_changed(self):
        pass

    @GObject.Signal
    def eos(self):
        pass

    @GObject.Signal
    def playback_start(self):
        if not self.timeout:
            self.timeout = GLib.timeout_add(TIMEOUT, self.timeout_function)

    @GObject.Signal
    def playback_paused(self):
        if self.timeout:
            GLib.source_remove(self.timeout)
            self.timeout = None

    @GObject.Signal
    def position_changed(self):
        remaining = self.duration - self.position
        if remaining < 2 * Gst.SECOND:
            if remaining + (TIMEOUT * Gst.MSECOND) >= 2 * Gst.SECOND:
                self.emit("about-to-finish")