partyzone

#!/usr/bin/env python

from __future__ import print_function

import sys
import logging
import gi
gi.require_version('Gst', '1.0')
gi.require_version('GstNet', '1.0')
import time
import signal
import select
import argparse
import os

os.environ["PYRO_LOGFILE"] = "pyro.log"
os.environ["PYRO_LOGLEVEL"] = "WARNING"

logging.basicConfig(
    format = "%(levelname) -10s %(asctime)s %(module)s:%(lineno)s %(funcName)s %(message)s",
    level = logging.INFO
)

logging.getLogger("Pyro4").setLevel(logging.WARNING)

import os.path
import socket
import Pyro4

from gi.repository import GObject, Gst, GstNet
from socket import gethostname


@Pyro4.expose
class Player(object):

    def __init__(self, host, network_clock, name=None, card=None):
        if name:
            self._name = name
        else:
            self._name = gethostname()
        self._host = host
        self._network_clock = network_clock
        self._track_uri = None
        self._card = card
        self._callback = None
        self.playbin = Gst.ElementFactory.make('playbin', 'playbin')

        print("player started on host: %s" % (self.host,))

        self.basetime = Gst.CLOCK_TIME_NONE
        self.stop()

        if self._card:
            print("setting card to " + self._card)
            alsa_sink = Gst.ElementFactory.make("alsasink", "alsasink")
            alsa_sink.set_property('device', self._card)
            self.playbin.set_property("audio-sink", alsa_sink)

        # wait until things stop
        self.bus =  self.playbin.get_bus()
        self.bus.add_signal_watch()
        self.bus.connect("message", self.on_message) 

    @property
    def name(self):
        return self._name

    @property
    def host(self):
        return self._host

    @property
    def track(self):
        if not self._track_uri:
            raise Exception("No track set")
        return self._track_uri

    @track.setter
    def track(self, value):
        print("setting track to %s" % (value,))
        self._track_uri = value

    def on_message(self, bus, message):
        t = message.type
        if t == Gst.MessageType.EOS:
            self.playbin.set_state(Gst.State.NULL)
            if self._callback:
                self._callback.play_done(self.name)    # invoke the callback object

        elif t == Gst.MessageType.BUFFERING:
            percent = message.parse_buffering()
            print('\r ' + str(percent) + '% buffered', end='')
            if int(percent) == 100:
                print('\n')
            sys.stdout.flush()
        elif t == Gst.MessageType.STATE_CHANGED:
            if isinstance(message.src, Gst.Pipeline):
                old_state, new_state, pending_state = message.parse_state_changed()
                print("Pipeline state changed from %s to %s." % (old_state.value_nick, new_state.value_nick))

                # Pipeline state changed from null to ready.
                # Pipeline state changed from ready to paused.
                # Pipeline state changed from paused to playing.

                if new_state.value_nick == "playing":
                    if self._callback:
                        self._callback.play_started(self.name)    # invoke the callback object

        elif t == Gst.MessageType.ERROR:
            self.playbin.set_state(Gst.State.NULL)
            err, debug = message.parse_error()
            print("Error: %s" % err, debug)

    def set_volume(self, vol):
        self.playbin.set_property('volume', vol)

    def set_callback_uri(self, callback_uri):
        self._callback_uri = callback_uri
        print("setting callback %s" % (self._callback_uri,))
        self._callback = Pyro4.Proxy(self._callback_uri)

    def get_basetime(self):
        return self.basetime

    def state(self):
        state = int(self.playbin.get_state(Gst.CLOCK_TIME_NONE)[1])
        print(state)
        return state

    @Pyro4.oneway
    def stop(self):
        self.playbin.set_state(Gst.State.NULL)
        if self._callback:
            self._callback.play_done(self.name)    # invoke the callback object

    @Pyro4.oneway
    def test(self):
        print("oneway test call")

    def is_playing(self):
        return self.state() == Gst.State.PLAYING

    def play(self, basetime=None):

        if not basetime:
            basetime = self._network_clock.system_time

        print("connecting to net clock %s:%s" % (self._network_clock.host, self._network_clock.port))
        client_clock = GstNet.NetClientClock.new('clock', self._network_clock.host,
                                                          self._network_clock.port, basetime)
        
        time.sleep(1.0) # Wait for the clock to stabilise

        self.playbin.use_clock(client_clock)
        self.playbin.set_start_time(Gst.CLOCK_TIME_NONE)
        self.playbin.set_base_time(basetime)    
        print("setting basetime to %s" % (basetime,))
        self.basetime = basetime

        print("setting uri to %s" % (self.track,))
        self.playbin.set_property('uri', self.track)
      
        print("playbin.set_state to playing") 
        self.playbin.set_state(Gst.State.PLAYING)
        print("returning basetime")
       
        return basetime

    @Pyro4.oneway
    def play_no_wait(self, basetime=None):
        self.play(basetime)

    def is_paused(self):
        return self.state() == Gst.State.PAUSED

    def pause(self, state):
        if state:
            self.playbin.set_state(Gst.State.PAUSED)
        else:
            self.playbin.set_state(Gst.State.PLAYING)

    def install_pyro_event_callback(self, daemon):
        """
        Add a callback to the tkinter event loop that is invoked every so often.
        The callback checks the Pyro sockets for activity and dispatches to the
        daemon's event process method if needed.
        """

        def pyro_event():
            while True:
                # for as long as the pyro socket triggers, dispatch events
                s, _, _ = select.select(daemon.sockets, [], [], 0.01)
                if s:
                    daemon.events(s)
                else:
                    # no more events, stop the loop, we'll get called again soon anyway
                    break
            GObject.timeout_add(20, pyro_event)

        GObject.timeout_add(20, pyro_event)


def find_network_clock(ns):
    clocks = ns.list(prefix="partyzone.clock").items()
    network_clock = None
    if clocks:
        name, uri = clocks[0]
        network_clock = Pyro4.Proxy(uri)
    return network_clock


if __name__ == '__main__':

    # default host name
    default_host = '127.0.0.1'
    default_friendly_name = None

    parser = argparse.ArgumentParser(description='Syncronise music accross machines.')

    parser.add_argument('--player', dest='player', action='store_true',
        help='player type. controller or player')

    parser.add_argument('--host', type=str, default=default_host, help="Host ip you wish to bind to")

    parser.add_argument('--pyro-port', dest='pyro_port', type=int, default=6564, help="Pyro port you wish to bind to")

    parser.add_argument('--card', type=str, default=None, help="Alsa card to use")

    parser.add_argument('--name', type=str, default=default_friendly_name, help="Friendly name for the player")

    parser.add_argument('--play', type=str, default=None, help="Play file for testing")

    parser.add_argument('--test-sync', dest='testsync', action='store_true', help="Play test file to check sync")
    parser.set_defaults(testsync=False)

    # parser.add_argument('-r', '--recursive-play', dest='recursive', action=readable_dir, default=None, help="Directory of items to recursively play")

    args = parser.parse_args()
   
    if not args.play: 
        while True:  # Debian at least fails dns at startup with systemd. Even specifying after network
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                s.connect(('google.com', 0))
                default_host = s.getsockname()[0]
                default_friendly_name = gethostname()
                break
            except:
                time.sleep(5)
            
    print("host: " + str(args.host))

    Gst.init(sys.argv)

    register_name = None

    try:

        if args.player:

            if not args.host:
                raise AttributeError("host parameter required")

            ns = None

            while True:
                try:
                    ns = Pyro4.naming.locateNS()
                    break
                except Pyro4.errors.NamingError:
                    print("Can't find Pyro nameserver!")
                except ex:
                    print(str(ex))
                    sys.exit(1)
                time.sleep(5)


            network_clock = find_network_clock(ns)
                
            while not network_clock:
                network_clock = find_network_clock(ns)
		time.sleep(2)

            register_name = "partyzone.players.(%s)" % args.name
            player = Player(args.host, network_clock=network_clock, name=args.name, card=args.card)

            try:
                existing = ns.lookup(register_name)
                print("Player still exists in Name Server with id: %s" % existing.object)
                print("Previous daemon socket port: %d" % existing.port)
                # start the daemon on the previous port
                daemon = Pyro4.core.Daemon(host=args.host, port=existing.port)
                # register the object in the daemon with the old objectId
                daemon.register(player, objectId=existing.object)
            except Pyro4.errors.NamingError:
                # just start a new daemon on a random port
                daemon = Pyro4.core.Daemon(host=args.host, port=args.pyro_port)
                # register the object in the daemon and let it get a new objectId
                # also need to register in name server because it's not there yet.
                uri = daemon.register(player)
                ns.register(register_name, uri)
                print("Player started.")

            # add a Pyro event callback to the gui's mainloop
            player.install_pyro_event_callback(daemon)
            GObject.MainLoop().run()

        elif args.testsync:

            # Example controller
            with Pyro4.locateNS() as ns:
                player_uris = ns.list(prefix="partyzone.players")
                players = []
                for name, uri in player_uris.items():
                    players.append(Pyro4.Proxy(uri))
                print("players: " + str(players))

            #path = os.path.dirname(os.path.realpath(__file__))
            #filepath = 'file://' + os.path.join(path, 'testing/test/mp3')
            filepath = 'file:///home/glenn/test.mp3'

            if len(players) > 0:
                try:
                    p = players[0]
                    p.track = filepath
                    print("setting basetime for %s to None" % (p.name,))
                    basetime = p.play(None)

                    for p in players[1:]:
                        p.track = filepath
                        print("setting basetime for %s to %s" % (p.name, str(basetime)))
                        p.play(basetime)

                except Pyro4.errors.CommunicationError as ex:
                    pass

        elif args.play:

            ns = None

            while True:
                try:
                    ns = Pyro4.naming.locateNS()
                    break
                except Pyro4.errors.NamingError:
                    print("Can't find Pyro nameserver!")
                except ex:
                    print(str(ex))
                    sys.exit(1)
                time.sleep(5)

            network_clock = find_network_clock(ns)
                
            if not network_clock:
                raise AttributeError("network clock required")

            player = Player(args.host, network_clock=network_clock, name=args.name, card=args.card)

            filepath = 'file://' + args.play

            print("playing " + filepath)
            player.track = filepath
            player.play()

            GObject.MainLoop().run()

        else:  # Controller

            # Example controller
            with Pyro4.locateNS() as ns:
                players = ns.list(prefix="partyzone.players")
                player_objects = []
                for name, uri in players.items():
                    player_objects.append(Pyro4.Proxy(uri))

                if not player_objects:
                    print("No players found")
                    sys.exit(0)

            while True:
                print("call...")
                for p in player_objects:
                    try:
                        print(p.name)
                        p.test()
                        print("Sleeping 1 second")
                        time.sleep(1)
                    except Pyro4.errors.ConnectionClosedError:     # or possibly even ProtocolError
                        print("Connection lost. REBINDING...")
                        print("(restart the server now)")
                        p._pyroReconnect()


    except KeyboardInterrupt:

        #with Pyro4.locateNS() as ns:
        #    players = ns.list(prefix="partyzone")
            
        #    for name, uri in players.items():
        #        player = Pyro4.Proxy(uri)
        #        player.cleanup()

        #if register_name:
        #    with Pyro4.locateNS() as ns:
        #        ns.remove(name=register_name)

        sys.exit()