modules/communication/zmq/adapter.py

+import threading
+import zmq
+import time
+import sys
+import zmq.auth
+import redis
+import random
+import string
+import simplejson
+import traceback
+
+from datetime import datetime
+
+from zmq.auth.thread import ThreadAuthenticator
+from threading import Event
+from libraries.security.whitelist import AuraWhitelist
+from libraries.security.user import AuraUser
+from modules.scheduling.scheduler import AuraScheduler
+
+
+class ServerZMQAdapter(threading.Thread):
+    debug = False
+    config = None
+    auracontroller = None
+
+    def __init__(self, controller, config, debug):
+        self.auracontroller = controller
+        self.ip = config.get('zmqhostip')
+        self.port = config.get('zmqport')
+        self.debug = debug
+
+        # init
+        threading.Thread.__init__ (self)
+        self.shutdown_event = Event()
+
+        self.config = config
+
+        self.context = zmq.Context().instance()
+        self.authserver = ThreadAuthenticator(self.context)
+        self.socket = self.context.socket(zmq.REP)
+        self.can_send = False
+
+        # start thread
+        self.start()
+
+    # ------------------------------------------------------------------------------------------ #
+    def run(self):
+        """
+        run runs on function start
+        """
+        self.auracontroller.set_sender(self)
+        self.start_auth_server()
+        self.socket.plain_server = True
+#        print("binding to: tcp://"+self.ip+":"+self.port)
+        self.socket.bind("tcp://"+self.ip+":"+self.port)
+        self.shutdown_event.clear()
+        self.auracontroller.messenger.set_mail_addresses(self.config.get('frommail'), self.config.get('adminmail'))
+
+        # Process tasks forever
+        while not self.shutdown_event.is_set():
+            print("waiting for zmq message on socket tcp://"+self.ip+":"+self.port+"\n")
+            data = self.socket.recv().decode("utf-8")
+
+            msg = ''
+            retval = None
+
+            self.can_send = True
+
+            datasplit = data.split(' ')
+
+            command = str(datasplit.pop(0))
+            params = "()" if len(datasplit) < 1 else  "'" + "','".join(datasplit) + "'"
+
+            try:
+                statement = "self.auracontroller." + command + params
+
+                if self.debug:
+                    print("zmq received message '" + data + "' and trying to call " + statement)
+
+                if params == "()":
+                    # if there are no params => easily use getattr and get a retval
+                    f = getattr(self.auracontroller, command)
+                    retval = f()  # json string
+                else:
+                    # if there are params => use exec and get no retval
+                    exec("retval = " + statement)
+
+            except SyntaxError as e:
+                traceback.print_exc()
+
+                msg = 'Warning: Syntax Error'
+                self.auracontroller.message(msg)
+            except AttributeError as e:
+                traceback.print_exc()
+
+                msg = 'Warning: Method ' + command + ' does not exist'
+                self.auracontroller.message(msg)
+            except TypeError as e:
+                traceback.print_exc()
+
+                msg = 'Warning: Wrong number of params'
+                self.auracontroller.message(msg)
+            except Exception as e:
+                traceback.print_exc()
+                msg = 'Warning: Unknown Error'
+                self.auracontroller.message(msg)
+
+            if msg == '':
+                msg = 'OK'
+
+            if retval is None:
+                self.socket.send(msg.encode())
+            else:
+                self.socket.send(simplejson.dumps(retval).encode())
+
+        return
+
+    # ------------------------------------------------------------------------------------------ #
+    def join_comm(self):
+        try:
+            while self.is_alive():
+                print(str(datetime.now())+" joining")
+                self.join()
+                print("join out")
+
+                # if cnt % 30 == 0:
+                #    print(datetime.datetime.now().isoformat())
+                #    server.printLastMessages()
+                #    cnt = 0
+                # cnt = cnt + 1
+
+        except (KeyboardInterrupt, SystemExit):
+            # Dem Server den Shutdown event setzen
+            # server.shutdown_event.set()
+            # Der Server wartet auf Eingabe
+            # Daher einen Client initiieren, der eine Nachricht schickt
+            self.halt()
+            sys.exit('Terminated')
+
+    # ------------------------------------------------------------------------------------------ #
+    def halt(self):
+        """
+        Stop the server
+        """
+        if self.shutdown_event.is_set():
+            return
+        try:
+            del self.auracontroller
+        except:
+            pass
+        self.shutdown_event.set()
+        result = 'failed'
+        try:
+            result = self.socket.unbind("tcp://"+self.ip+":"+self.port)
+        except:
+            pass
+        #self.socket.close()
+
+    # ------------------------------------------------------------------------------------------ #
+    def reload(self):
+        """
+        stop, reload config and startagaing
+        """
+        if self.shutdown_event.is_set():
+            return
+        self.loadConfig()
+        self.halt()
+        time.sleep(3)
+        self.run()
+
+    # ------------------------------------------------------------------------------------------ #
+    def send(self, message):
+        """
+        Send a message to the client
+        :param message: string
+        """
+        if not self.can_send:
+            print("sending a "+str(len(message))+" long message via ZMQ.")
+            self.socket.send(message.encode("utf-8"))
+            self.can_send = False
+        else:
+            print("cannot send message via ZMQ: "+str(message))
+
+    # ------------------------------------------------------------------------------------------ #
+    def start_auth_server(self):
+        """
+        Start zmq authentification server
+        """
+        # stop auth server if running
+        if self.authserver.is_alive():
+            self.authserver.stop()
+
+        if self.config.get("securitylevel") > 0:
+            # Authentifizierungsserver starten.
+
+            self.authserver.start()
+
+            # Bei security level 2 auch passwort und usernamen verlangen
+            if self.securitylevel > 1:
+                try:
+
+                    addresses = AuraWhitelist().getList()
+                    for address in addresses:
+                        self.authserver.allow(address)
+
+                except:
+                    print("Exception passed!!!")
+                    pass
+
+            # Instruct authenticator to handle PLAIN requests
+            self.authserver.configure_plain(domain='*', passwords=self.get_accounts())
+
+    # ------------------------------------------------------------------------------------------ #
+    @staticmethod
+    def get_accounts(self):
+        """
+        Get accounts from redis db
+        :return: llist - a list of accounts
+        """
+        accounts = AuraUser().getLogins()
+        db = redis.Redis()
+
+        internaccount = db.get('internAccess')
+        if not internaccount:
+            user = ''.join(random.sample(string.ascii_lowercase,10))
+            password = ''.join(random.sample(string.ascii_lowercase+string.ascii_uppercase+string.digits,22))
+            db.set('internAccess', user + ':' + password)
+            intern = [user, password]
+        else:
+            intern =  internaccount.split(':')
+
+        accounts[intern[0]] = intern[1]
+
+        return accounts
+
+
+class ClientZMQAdapter:
+    debug = False
+
+    def __init__(self, host, port, debug=True):
+        self.host = host
+        self.port = port
+        self.debug = debug
+
+        self.username = ""
+        self.password = ""
+
+        # Socket to receive messages on
+        context = zmq.Context()
+        self.socket = context.socket(zmq.REQ)
+
+        self.socket.RCVTIMEO = 10000
+        self.message = ''
+
+    # ------------------------------------------------------------------------------------------ #
+    def send(self, message, blocking=True):
+        """
+        Send a message to the server
+        :param message: string
+        :return:boolean
+        """
+        db = redis.Redis()
+        account = db.get('internAccess')
+        (username, password) = account.decode().split(':')
+
+        self.socket.plain_username = username.encode()
+        self.socket.plain_password = password.encode()
+        self.socket.connect("tcp://"+self.host+":"+self.port)
+        self.socket.send_string(str(message))
+
+        try:
+            if self.debug:
+                if blocking:
+                    print("message '" + str(message) + "' sent to auraserver and waiting for reply")
+                else:
+                    print("message '" + str(message) + "' sent to auraserver. i am not waiting for a reply")
+
+            if blocking:
+                self.message = self.socket.recv().decode("utf-8")
+            else:
+                self.socket.recv(zmq.NOBLOCK)
+
+            if self.debug:
+                print("got reply from '"+str(message)+"': "+self.message)
+
+        except zmq.Again as e:
+            self.message = 'Not waiting for a reply'
+        except zmq.ZMQError as e:
+            self.message = ''
+            print(e)
+            return False
+        except Exception as e:
+            self.message = ''
+            print(e)
+            return False
+
+        return self.message
+
+    # ------------------------------------------------------------------------------------------ #
+    def set_username(self, username):
+        """
+        authentification to the server with username
+        :param username: string - the username
+        """
+        self.username = username
+
+    # ------------------------------------------------------------------------------------------ #
+    def set_password(self, password):
+        """
+        authentification to the server with password
+        :param password: string - the password
+        """
+        self.password = password
+
+    # ------------------------------------------------------------------------------------------ #
+    def receive(self):
+        """
+        Return the currently received message
+        :return:string - message from server
+        """
+        return self.message
+

modules/controller/controller.py

+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import base64
+import logging
+import random
+import string
+import sys
+import traceback
+import simplejson
+
+from libraries.security.user import AuraUser
+from libraries.base.calendar import AuraCalendarService
+from libraries.base.schedulerconfig import AuraSchedulerConfig
+from libraries.exceptions.auraexceptions import NoProgrammeLoadedException
+
+from modules.scheduling.scheduler import AuraScheduler
+from modules.communication.redis.messenger import RedisMessenger
+from modules.communication.liquidsoap.communicator import LiquidSoapCommunicator
+
+
+"""
+    AuraController Class
+    Communicates with the liquidsoap server and the scheduler
+"""
+
+
+class AuraController:
+    messenger = RedisMessenger()
+    liquidsoapcommunicator = None
+    scheduler = None
+    userdb = AuraUser()
+
+    is_intern = False
+    lq_error = ''
+    config = None
+    sender = None
+    debug = False
+    redisclient = None
+
+    # Constructor
+    def __init__(self, config):
+        """
+        Constructor
+        @type    config: object
+        @param   config: Die geladene aura.ini
+        """
+        self.config = config
+        self.debug = self.config.get("debug")
+
+        # Felder die Liquidsoap fuer einen Track (rid) zurueckliefert
+        self.knownfields = ["status", "album", "time", "title", "artist", "comment", "filename", "on_air", "source", "rid", "genre"]
+        self.job_result = ['', '', '', '', '', '']
+
+#        self.messenger.set_channel('controller')
+#        self.messenger.set_section('execjob')
+
+        errors_file = self.config.get("install_dir")+"/errormessages/controller_error.js"
+        self.errorData = simplejson.load(open(errors_file))
+
+        # create scheduler and ls_communicator
+        self.liquidsoapcommunicator = LiquidSoapCommunicator(self.config)
+        self.scheduler = AuraScheduler(self.config)
+
+        # give both a reference of another
+        self.liquidsoapcommunicator.scheduler = self.scheduler
+        self.scheduler.liquidsoapcommunicator = self.liquidsoapcommunicator
+
+        # fetch act programme and write into database
+        # tell scheduler to reload from database
+        self.fetch_new_programme()
+
+    # ------------------------------------------------------------------------------------------ #
+    def fetch_new_programme(self):
+        acs = AuraCalendarService(self.config, "", "", False)
+        queue = acs.get_queue()
+
+        # start fetching thread
+        acs.start()
+
+        # wait for the end
+        response = queue.get()
+
+        if response == "fetching_finished":
+            # acs.get_calendar_data()
+            self.scheduler.load_programme_from_db()
+            return self.scheduler.get_act_programme_as_string()
+        else:
+            print("Got an unknown response from AuraCalendarService: "+response)
+
+    # ------------------------------------------------------------------------------------------ #
+    def get_act_programme(self):
+        try:
+            programme = self.scheduler.get_act_programme_as_string()
+
+        except NoProgrammeLoadedException as e:
+            if self.debug:
+                print("WARNING: no programme in memory. i have to reload it!")
+
+            # refetch the programme from pv and importer
+            return self.fetch_new_programme()
+
+        except Exception as e:
+            traceback.print_exc()
+        else:
+            return programme
+
+    # ------------------------------------------------------------------------------------------ #
+    def init_player(self):
+        return self.liquidsoapcommunicator.init_player()
+
+    # ------------------------------------------------------------------------------------------ #
+    def scheduler_data(self):
+        """
+        Scheduler Config ausliefern
+        """
+        jobs = []
+
+        try:
+            # Das scheduler.xml laden
+            schedulerconfig = AuraSchedulerConfig(self.sender.schedule_config)
+            jobs = schedulerconfig.getJobs()
+        except:
+            # Das scheint kein gültiges XML zu sein
+            self.warning('01', False)
+
+        self.success('00', simplejson.dumps(jobs))
+        self.notify_client()
+
+    # ------------------------------------------------------------------------------------------ #
+    def scheduler_store(self, adminuser, adminpassword, json):
+        """
+        Scheduler Config zurückschreiben
+        """
+        if not self.userdb.hasAdminRights(adminuser, adminpassword):
+            self.warning('01', False)
+            self.notify_client()
+            return
+        try:
+            schedulerconfig = AuraSchedulerConfig(self.sender.schedule_config)
+        except:
+            self.warning('02', False)
+            self.notify_client()
+        try:
+            schedulerconfig.storeJsonToXml( base64.b64decode(json))
+        except:
+            self.warning('02', False)
+            self.notify_client()
+        else:
+            if self.scheduler_reload():
+                self.success('00', True)
+            else:
+                self.warning('02', False)
+        self.notify_client()
+
+    # ------------------------------------------------------------------------------------------ #
+    def set_password(self, adminuser, adminpassword, username, password):
+        """
+        Ein Userpasswort setzen
+        TODO: Passwörter verschlüsselt übertragen
+        """
+        if self.userdb.hasAdminRights(adminuser, adminpassword):
+            self.userdb.setPassword(username, password)
+            self.success('00', password)
+        else:
+            self.warning('01', False)
+
+        self.notify_client()
+        self.sender.reload_config()
+
+    # ------------------------------------------------------------------------------------------ #
+    def add_user(self, adminuser, adminpassword, username):
+        """
+        Einen User hinzufügen
+        TODO: Passwort verschlüsselt übertragen
+        """
+        if self.userdb.hasAdminRights(adminuser, adminpassword):
+            password = ''.join(random.sample(string.lowercase+string.uppercase+string.digits,14))
+            self.userdb.insertUser(username, password, 'user')
+            self.success('00', password)
+        # TODO admin rechte checken user und passwort setzen, passwort zurückgeben
+        else:
+            self.warning('01', False)
+
+        self.notify_client()
+        self.sender.reload_config()
+
+    # ------------------------------------------------------------------------------------------ #
+    def del_user(self, adminuser, adminpassword, username):
+        """
+        Einen User löschen
+        TODO: Passwort verschlüsselt übertragen
+        """
+        # TODO admin rechte checken user löschen
+        if self.userdb.hasAdminRights(adminuser, adminpassword):
+            self.userdb.delete(username)
+            self.success('00', True)
+        else:
+            self.warning('01', False)
+
+        self.notify_client()
+        self.sender.reload_config()
+
+    # ------------------------------------------------------------------------------------------ #
+    def get_user_list(self, adminuser, adminpassword):
+        """
+        Einen User löschen
+        TODO: Passwort verschlüsselt übertragen
+        """
+        # TODO admin rechte checken user löschen
+        if self.userdb.hasAdminRights(adminuser, adminpassword):
+            userlist = self.userdb.getUserlist()
+            self.success('00', simplejson.dumps(userlist))
+        else:
+            self.warning('01', False)
+
+        self.notify_client()
+
+
+    # ------------------------------------------------------------------------------------------ #
+    def __metadata_format__(self, metadata):
+        """
+        Private: Vereinheitlicht die Metadaten von Playlist und anderen Kanälen und entfernt Anführungszeichen in den Feldern
+        @rtype:   boolean/dict
+        @return:  False/Metadaten
+        """
+        mdata = {}
+        try:
+            for key,val in metadata.items('root'):
+                if key in self.knownfields:
+                    mdata[key] = val.strip().replace('"', '')
+            return mdata
+        except:
+            return False
+
+    # ------------------------------------------------------------------------------------------ #
+    def __get_error__(self, errornumber):
+        """
+        Privat: Ermittelt Fehlermeldung, Job-Name (Klassenmethode) und Fehlercode für den Job aus error/controller_error.js
+        @type errornumber:  string
+        @param errornumber: Die interne Fehlernummer der aufrufenden Methode
+        """
+        f = sys._getframe(2)
+
+        job = f.f_code.co_name
+        data = {'message':'', 'job':job, 'code':'unknown'}
+        if job in self.errorData:
+            errMsg = self.errorData[job][errornumber]
+            errID = self.errorData[job]['id'] + str(errornumber)
+            args = {x:f.f_locals[x] if not x == 'self' else '' for x in f.f_code.co_varnames[:f.f_code.co_argcount]}
+
+            for key in args.keys():
+                errMsg = errMsg.replace('::' + key + '::', str(args[key]))
+
+            data['message'] = errMsg
+            data['job'] = job
+            data['code'] = errID
+
+        return data
+
+    # ------------------------------------------------------------------------------------------ #
+    def success(self, errnum='00', value='', section='main'):
+        """
+        Erfolgsmeldung loggen
+        @type errnum:    string
+        @param errnum:   Errornummer der aufrufenden Funktion
+        @type value:     string
+        @param value:    Optionaler Wert
+        @type section:   string
+        @param section:  Gültigkeitsbereich
+        """
+        error = self.__get_error__(errnum)
+        self.job_result = {'message':error['message'], 'code':error['code'], 'success':'success', 'job':error['job'], 'value':value, 'section':section}
+        self.messenger.send(error['message'], error['code'], 'success', error['job'], value, section)
+
+    # ------------------------------------------------------------------------------------------ #
+    def info(self, errnum='01', value='', section='main'):
+        """
+        Info loggen
+        @type errnum:    string
+        @param errnum:   Errornummer der aufrufenden Funktion
+        @type value:     string
+        @param value:    Optionaler Wert
+        @type section:   string
+        @param section:  Gültigkeitsbereich
+        """
+        error = self.__get_error__(errnum)
+        self.job_result = {'message':error['message'], 'code':error['code'], 'success':'info', 'job':error['job'], 'value':value, 'section':section}
+        self.messenger.send(error['message'], error['code'], 'info', error['job'], value, section)
+
+    # ------------------------------------------------------------------------------------------ #
+    def warning(self, errnum='01', value='', section='main'):
+        """
+        Warnung loggen
+        @type errnum:    string
+        @param errnum:   Errornummer der aufrufenden Funktion
+        @type value:     string
+        @param value:    Optionaler Wert
+        @type section:   string
+        @param section:  Gültigkeitsbereich
+        """
+        error = self.__get_error__(errnum)
+        self.job_result = {'message':error['message'], 'code':error['code'], 'success':'warning', 'job':error['job'], 'value':value, 'section':section}
+        self.messenger.send(error['message'], error['code'], 'warning', error['job'], value, section)
+
+    # ------------------------------------------------------------------------------------------ #
+    def error(self, errnum='01', value='', section='main'):
+        """
+        Error loggen
+        @type errnum:    string
+        @param errnum:   Errornummer der aufrufenden Funktion
+        @type value:     string
+        @param value:    Optionaler Wert
+        @type section:   string
+        @param section:  Gültigkeitsbereich
+        """
+        error = self.__get_error__(errnum)
+        self.job_result = {'message':error['message'], 'code':error['code'], 'success':'error', 'job':error['job'], 'value':value, 'section':section}
+        self.messenger.send(error['message'], error['code'], 'error', error['job'], value, section)
+
+    # ------------------------------------------------------------------------------------------ #
+    def fatal(self, errnum='01', value='', section='main'):
+        """
+        Fatal error loggen
+        @type errnum:    string
+        @param errnum:   Errornummer der aufrufenden Funktion
+        @type value:     string
+        @param value:    Optionaler Wert
+        @type section:   string
+        @param section:  Gültigkeitsbereich
+        """
+        error = self.__get_error__(errnum)
+        self.job_result = {'message':error['message'], 'code':error['code'], 'success':'fatal', 'job':error['job'], 'value':value, 'section':section}
+        self.messenger.send(error['message'], error['code'], 'fatal', error['job'], value, section)
+
+    # ------------------------------------------------------------------------------------------ #
+    def notify_client(self):
+        """
+        Eine Nachricht als JSON-String an den Client senden
+        """
+        if not self.is_intern:
+            self.message(simplejson.dumps(self.job_result))
+
+    # ------------------------------------------------------------------------------------------ #
+    def __check_result__(self, result):
+        """
+        Fehlerbehandlung
+        @type     result: string
+        @param    result: Ein Json-String
+        """
+        self.lq_error = simplejson.loads(result)
+
+        try:
+            if self.lq_error['success'] == 'true':
+                return True
+            else:
+                return False
+        except:
+            return False
+
+    # ------------------------------------------------------------------------------------------ #
+    def __update_event_queue__(self, playlist):
+        """
+        Playlist Eventqueue updaten
+        @type playlist: dict
+        @param playlist: Playlist
+        """
+        # eventuell noch bestehende Events im Queue löschen
+        self.messenger.queue_remove_events('playtrack', 'player')
+        # Für jeden Tack einen Event ankündigen
+        for track in playlist['playlist']['trackList']['track']:
+
+            if 'time' in track and 'start' in track:
+                starts = str(track['start'] + 'T' + track['time'])
+                event = {'job':'play', 'location': track['location'],'length': track['length'], 'date': track['start'], 'time': track['time']}
+                self.messenger.queue_add_event('playtrack', starts, event, 'player')
\ No newline at end of file

modules/liquidsoap/__init__.py

+__author__ = 'michel'

modules/liquidsoap/alsa.liq

+
+input_fs = single(id="fs", "/var/audio/fallback/output.flac")
+input_http = input.http(id="http", "http://stream.fro.at/fro128.mp3")
+linein_alsa_1 = input.alsa(id="linein", device = input_device_0, bufferize = false)
+mixer = mix(id="mixer", [input_fs, input_http, linein_alsa_1])
+output.alsa(id="lineout", device = output_device_0, bufferize = false, mixer)

modules/liquidsoap/alsa_settings_tester.liq

+set("log.file.path", "./<script>.log")
+
+set("server.telnet", true)
+set("server.telnet.bind_addr", "0.0.0.0")
+set("server.telnet.port", 1234)
+
+# ALSA / pulse settings
+# durch ausprobieren herausgefunden für asus xonar dgx 5.1
+# chip: CMI8788
+# driver: snd_oxygen
+
+set("frame.duration", 0.30)
+set("alsa.alsa_buffer", 8192) # 7168) # 6144) # 8192) # 10240) #15876
+set("alsa.buffer_length", 25)
+set("alsa.periods", 0) # assertion error when setting periods other than 0 => alsa default
+
+input_linein = input.alsa(id="linein", bufferize = false)
+input_fs = single(id="fs", "/var/audio/fallback/output.flac")
+input_http = input.http(id="http", "http://stream.fro.at/fro-128.ogg")
+
+mixer = mix(id="mixer", [input_fs, input_http, input_linein])
+
+output.alsa(id="lineout", bufferize = false, mixer)
+

modules/liquidsoap/dynamicsources.liq

+%include "readini.liq"
+
+inifile = '/etc/aura/aura.ini'
+ini = read_ini(inifile)
+
+set("log.file.path", "./<script>.log")
+
+set("server.telnet", true)
+set("server.telnet.bind_addr", "0.0.0.0")
+set("server.telnet.port", 1234)
+
+use_alsa           = list.assoc("use_alsa", ini) == "y"
+frame_duration     = float_of_string(list.assoc("frame_duration", ini))
+frame_size         = int_of_string(list.assoc("frame_size", ini))
+alsa_buffer        = int_of_string(list.assoc("alsa_buffer", ini))
+alsa_buffer_length = int_of_string(list.assoc("alsa_buffer_length", ini))
+alsa_periods       = int_of_string(list.assoc("alsa_periods", ini))
+
+if use_alsa then
+    if frame_duration > 0.0 then
+        print("setting frame.duration to #{frame_duration}")
+        set("frame.duration", frame_duration)
+    end
+    if frame_size > 0 then
+        print("setting frame.size to #{frame_size}")
+        set("frame.size", frame_size)
+    end
+    if alsa_buffer > 0 then
+        print("setting alsa.buffer to #{alsa_buffer}")
+        set("alsa.alsa_buffer", alsa_buffer)
+    end
+    if alsa_buffer > 0 then
+        print("setting alsa.buffer_length to #{alsa_buffer_length}")
+        set("alsa.buffer_length", alsa_buffer_length)
+    end
+    if alsa_periods > 0 then
+        print("setting alsa.periods to #{alsa_periods}")
+        set("alsa.periods", alsa_periods) # assertion error when setting periods other than 0 => alsa default
+    end
+end
+
+# First, we create a list referencing the dynamic sources:
+dyn_sources = ref []
+
+# This is our icecast output.
+# It is a partial application: the source needs to be given!
+out = output.icecast(%mp3,
+                     host="test",
+                     password="hackme",
+                     fallible=true)
+                     
+
+
+# Now we write a function to create 
+# a playlist source and output it.
+def create_playlist(uri) =
+  # The playlist source 
+  s = playlist(uri)
+
+  # The output
+  output = out(s)
+
+  # We register both source and output 
+  # in the list of sources
+  dyn_sources := 
+      list.append( [(uri,s),(uri,output)],
+                    !dyn_sources )
+  "Done!"
+end
+
+# And a function to destroy a dynamic source
+def destroy_playlist(uri) = 
+  # We need to find the source in the list,
+  # remove it and destroy it. Currently, the language
+  # lacks some nice operators for that so we do it
+  # the functional way
+
+  # This function is executed on every item in the list
+  # of dynamic sources
+  def parse_list(ret, current_element) = 
+    # ret is of the form: (matching_sources, remaining_sources)
+    # We extract those two:
+    matching_sources = fst(ret)
+    remaining_sources = snd(ret)
+
+    # current_element is of the form: ("uri", source) so 
+    # we check the first element
+    current_uri = fst(current_element)
+    if current_uri == uri then
+      # In this case, we add the source to the list of
+      # matched sources
+      (list.append( [snd(current_element)], 
+                     matching_sources),
+       remaining_sources)
+    else
+      # In this case, we put the element in the list of remaining
+      # sources
+      (matching_sources,
+       list.append([current_element], 
+                    remaining_sources))
+    end
+  end
+    
+  # Now we execute the function:
+  result = list.fold(parse_list, ([], []), !dyn_sources)
+  matching_sources = fst(result)
+  remaining_sources = snd(result)
+
+  # We store the remaining sources in dyn_sources
+  dyn_sources := remaining_sources
+
+  # If no source matched, we return an error
+  if list.length(matching_sources) == 0 then
+    "Error: no matching sources!"
+  else
+    # We stop all sources
+    list.iter(source.shutdown, matching_sources)
+    # And return
+    "Done!"
+  end
+end
+
+def backup_pool () =
+  result = get_process_lines("python ./helpers/next_song_from_pool.py")
+  log("next song: #{result}")
+  request.create(list.hd(result))
+end
+
+backup_pool = request.dynamic(backup_pool)
+
+
+# Now we register the telnet commands:
+server.register(namespace="dynamic_playlist",
+                description="Start a new dynamic playlist.",
+                usage="start <uri>",
+                "start",
+                create_playlist)
+server.register(namespace="dynamic_playlist",
+                description="Stop a dynamic playlist.",
+                usage="stop <uri>",
+                "stop",
+                destroy_playlist)
+server.register(namespace="dyn",
+                description="load next song from pool",
+                usage="next <folder>",
+                "next",
+                
+                
+                
+output.dummy(blank())

modules/liquidsoap/fallback.liq

+# Custom crossfade to deal with jingles.
+def smart_crossfade (~start_next=5.,~fade_in=3.,~fade_out=3.,
+                     ~default=(fun (a,b) -> sequence([a, b])),
+                     ~high=-15., ~medium=-32., ~margin=4.,
+                     ~width=2.,~conservative=false,s)
+  fade.out = fade.out(type="sin",duration=fade_out)
+  fade.in  = fade.in(type="sin",duration=fade_in)
+  add = fun (a,b) -> add(normalize=false,[b, a])
+  log = log(label="smart_crossfade")
+
+  def transition(a,b,ma,mb,sa,sb)
+
+    list.iter(fun(x)-> log(level=4,"Before: #{x}"),ma)
+    list.iter(fun(x)-> log(level=4,"After : #{x}"),mb)
+
+    if ma["type"] == "jingles" or mb["type"] == "jingles" then
+      log("Old or new file is a jingle: sequenced transition.")
+      sequence([sa, sb])
+    elsif
+      # If A and B are not too loud and close, fully cross-fade them.
+      a <= medium and b <= medium and abs(a - b) <= margin
+    then
+      log("Old <= medium, new <= medium and |old-new| <= margin.")
+      log("Old and new source are not too loud and close.")
+      log("Transition: crossed, fade-in, fade-out.")
+      add(fade.out(sa),fade.in(sb))
+
+    elsif
+      # If B is significantly louder than A, only fade-out A.
+      # We don't want to fade almost silent things, ask for >medium.
+      b >= a + margin and a >= medium and b <= high
+    then
+      log("new >= old + margin, old >= medium and new <= high.")
+      log("New source is significantly louder than old one.")
+      log("Transition: crossed, fade-out.")
+      add(fade.out(sa),sb)
+
+    elsif
+      # Opposite as the previous one.
+      a >= b + margin and b >= medium and a <= high
+    then
+      log("old >= new + margin, new >= medium and old <= high")
+      log("Old source is significantly louder than new one.")
+      log("Transition: crossed, fade-in.")
+      add(sa,fade.in(sb))
+
+    elsif
+      # Do not fade if it's already very low.
+      b >= a + margin and a <= medium and b <= high
+    then
+      log("new >= old + margin, old <= medium and new <= high.")
+      log("Do not fade if it's already very low.")
+      log("Transition: crossed, no fade.")
+      add(sa,sb)
+
+    # What to do with a loud end and a quiet beginning ?
+    # A good idea is to use a jingle to separate the two tracks,
+    # but that's another story.
+
+    else
+      # Otherwise, A and B are just too loud to overlap nicely,
+      # or the difference between them is too large and overlapping would
+      # completely mask one of them.
+      log("No transition: using default.")
+      default(sa, sb)
+    end
+  end
+
+  smart_cross(width=width, duration=start_next, conservative=conservative, transition, s)
+end
+
+# create a pool
+def fallback_create(~skip=true, name, requestor)
+  log("Creating channel #{name}")
+
+  # Create the request.dynamic source
+  # Set conservative to true to queue
+  # several songs in advance
+  #source = request.dynamic(conservative=true, length=50., id="pool_"^name, requestor, timeout=60.)
+  source = request.dynamic(length=50., id="pool_"^name, requestor, timeout=60.)
+
+  # Apply normalization using replaygain information
+  source = amplify(1., override="replay_gain", source)
+
+  # Skip blank when asked to
+  source =
+    if skip then
+      skip_blank(source, max_blank=10., threshold=-40.)
+    else
+      source
+    end
+
+  # Tell the system when a new track
+  # is played
+  source = on_metadata(fun (meta) ->
+    system('#{list.assoc("install_dir", ini)}/modules/liquidsoap/helpers/message.py -c aura -t liquid_startup'), source)
+
+  # Finally apply a smart crossfading
+  smart_crossfade(source)
+end
+
+def create_dynamic_playlist(next)
+  log("next song is: #{next}")
+  request.create(list.hd(next))
+end
+
+def create_station_fallback() =
+  log("requesting next song for STATION fallback")
+  result = get_process_lines('#{list.assoc("install_dir", ini)}/guru.py --get-next-fallback-file-for "station" --quiet')
+  create_dynamic_playlist(result)
+end
+
+def create_show_fallback() =
+  log("requesting next song for SHOW fallback")
+  result = get_process_lines('#{list.assoc("install_dir", ini)}/guru.py --get-next-fallback-file-for "show" --quiet')
+  create_dynamic_playlist(result)
+end
+
+def create_timeslot_fallback() =
+  log("requesting next song for TIMESLOT fallback")
+  result = get_process_lines('#{list.assoc("install_dir", ini)}/guru.py --get-next-fallback-file-for "timeslot" --quiet')
+  create_dynamic_playlist(result)
+end
\ No newline at end of file

modules/liquidsoap/helpers/__init__.py

+__author__ = 'michel'
+
+import os
+import sys

modules/liquidsoap/helpers/message.py

+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import sys
+import getopt
+
+scriptdir = os.path.dirname(os.path.abspath(__file__))
+
+# import our libraries
+package_dir = os.path.join(scriptdir, '../../../')
+path = list(sys.path)
+sys.path.insert(0, package_dir)
+
+#from libraries.reporting.messenger import AuraMessenger
+from libraries.base.config import ConfigReader
+from modules.communication.zmq.adapter import ClientZMQAdapter
+
+def main(argv):
+    #m = AuraMessenger()
+    task = ''
+    channel = ''
+    name= ''
+    value = ''
+    time = ''
+
+    config = ConfigReader()
+    config.loadConfig()
+    zmqclient = ClientZMQAdapter(config.get('zmqhostip'), config.get('zmqport'))
+
+    try:
+        opts, args = getopt.getopt(argv,"c:t:n:v:t",["channel=","task=", "name=","value=","time="])
+    except getopt.GetoptError:
+        print('message.py -c <channel> -t <task> [-n <name> -v <value> [-t <time>]]')
+        sys.exit(2)
+    for opt, arg in opts:
+        if opt == '-h':
+            print('message.py -c <channel> -t <task>')
+            sys.exit()
+        elif opt in ("-c", "--channel"):
+            channel = arg
+        elif opt in ("-t", "--task"):
+            task = arg
+        elif opt in ("-n", "--name"):
+            name = arg
+        elif opt in ("-v", "--value"):
+            value = arg
+        elif opt in ("-t", "--time"):
+            eventtime = arg
+
+
+    #m.setChannel(channel)
+    if task == 'setEvent':
+        #setEvent(self, name, eventtime, value)
+        #opts, args = getopt.getopt(argv,"c:t:",["channel=","task="])
+        pass
+    elif task == 'setState':
+        if name == '':
+            sys.exit(2)
+        zmqclient.send("startup")
+    #    m.setState(name, value)
+    elif task == 'liquid_startup':
+        zmqclient.send(task, False)
+    else:
+        print(task)
+        #zmqclient.send(task)
+    #    m.sayAlive()
+
+if __name__ == "__main__":
+    main(sys.argv[1:])
+    sys.path[:] = path
\ No newline at end of file

modules/liquidsoap/helpers/silence.sh

+#!/bin/bash
+
+pack_int(){ printf "%08X\n" $1 | sed 's/\([0-9A-F]\{2\}\)\([0-9A-F]\{2\}\)\([0-9A-F]\{2\}\)\([0-9A-F]\{2\}\)/\\\\\\x\4\\\\\\x\3\\\\\\x\2\\\\\\x\1/I' | xargs printf; }
+
+pack_short(){ printf "%04X\n" $1 | sed 's/\([0-9A-F]\{2\}\)\([0-9A-F]\{2\}\)/\\\\\\x\2\\\\\\x\1/I' | xargs printf; }
+
+duration=1800
+if [[ $# -eq 1 ]]; then
+    duration=$1
+fi
+
+channels=2
+bps=16
+sample=44100
+Subchunk1Size=18
+Subchunk2Size=$(echo "$duration*$sample*$channels*$bps/8" | bc)
+ChunkSize=$((20 + $Subchunk1Size + $Subchunk2Size))
+
+echo -n RIFF
+pack_int $ChunkSize
+echo -n "WAVEfmt "
+pack_int $Subchunk1Size
+pack_short 1
+pack_short $channels
+pack_int $sample
+pack_int $((bps/8 * channels * sample))
+pack_short $((bps/8 * channels))
+pack_short $bps
+pack_short 0
+echo -n data
+pack_int $Subchunk2Size
+
+dd if=/dev/zero bs=1 count=$Subchunk2Size 2>/dev/null

modules/liquidsoap/inToOut.liq

+set("frame.size", 1881)
+#set("decoding.buffer_length", 20.)
+#set("frame.channels", 2)
+#set("frame.samplerate", 44100)
+
+set("server.telnet", true)
+set("server.telnet.port", 1234)
+
+set("alsa.alsa_buffer", 2048)
+#set("alsa.periods", 8)
+#set("alsa.buffer_length", 4096)
+
+# works, but every now and then buffer overruns in alsa.in on node
+# more or less the same with or w/o clock_safe set. maybe a bit better with clock_safe=false
+input_alsa = input.alsa(bufferize=false, clock_safe=false)
+output.alsa(bufferize=false, clock_safe=false, input_alsa)
+
+#output.alsa(bufferize = false, input.alsa(bufferize = false))
+
+# works
+#url = "http://stream.fro.at/fro64.mp3"
+#i = strip_blank(input.http(id="stream", url))
+#input_http = mksafe(i)
+#output.alsa(bufferize=false, input_http)
+
+# works
+#input_fs = fallback(track_sensitive=false,
+#            [ single("/var/audio/fallback/music.flac"), playlist.safe("/var/audio/fallback/NightmaresOnWax/Smokers Delight") ] )
+#ignore(input_fs)
+#output.alsa(bufferize=false, input_fs)
+
+#output.alsa(device="hw:1,0", bufferize=false, input_alsa)
+#output.alsa(device="hw:1,0", bufferize=false, input_http)
+#output.pulseaudio(input_http)
+#output.alsa(device="hw:0,0", bufferize=false, input_alsa)
+#output.pulseaudio(input_fs)
+
+#mixer = mix(id = "mixer", [input_alsa])
+#final_stream = fallback(id="station_fallback", [mixer, playlist.safe("/var/audio/fallback/NightmaresOnWax/Smokers Delight")])
+#output.alsa(device="hw:1,0", bufferize=false, final_stream)
+
+#ignore(mixer)
+#ignore(input_fs)
+#ignore(input_http)
+#ignore(input_alsa)
+
+
+

modules/liquidsoap/inputToOutput.liq

+#set("frame.size", 3763)
+#set("decoding.buffer_length", 20.)
+#set("frame.channels", 2)
+#set("frame.samplerate", 44100)
+
+set("server.telnet", true)
+set("server.telnet.port", 1234)
+
+input_alsa = input.alsa(bufferize=false)
+#output.alsa(device="hw:0,0", bufferize=false, input_alsa)
+
+#output.alsa(bufferize = false, input.alsa(bufferize = false))
+
+url = "http://stream.fro.at/fro64.mp3"
+#url = "http://mp3stream1.apasf.apa.at:8000/listen.pls"
+input_http = strip_blank(id="http_strip", input.http(id="stream", url))
+#input_http = mksafe(i)
+ignore(input_http)
+
+
+input_fs = strip_blank(id="fs_strip", single("/var/audio/fallback/music.flac"))
+#fallback(track_sensitive=false,
+#            [ single("/var/audio/fallback/music.flac"), playlist.safe("/var/audio/fallback/NightmaresOnWax/Smokers Delight") ] )
+ignore(input_fs)
+
+#output.alsa(device="hw:0,0", bufferize=false, input_alsa)
+#output.alsa(device="hw:0,0", bufferize=false, input_http)
+#output.pulseaudio(input_http)
+#output.alsa(device="hw:0,0", bufferize=false, input_fs)
+#output.pulseaudio(input_fs)
+
+mixer = mix(id = "mixer", [input_fs, input_http, input_alsa])
+#final_stream = fallback(id="station_fallback", [mixer, playlist.safe("/var/audio/fallback/NightmaresOnWax/Smokers Delight")])
+output.alsa(bufferize=false, mixer)
+
+ignore(mixer)
+#ignore(input_fs)
+#ignore(input_http)
+#ignore(input_alsa)
+
+
+

modules/liquidsoap/library.liq

+%include "readini.liq"
+
+inifile = '/etc/aura/aura.ini'
+ini = read_ini(inifile)
+
+def get_task_number (taskname) =
+    tasks = [
+         ('seek_track', '10'),
+         ('load', '11'),
+         ('move', '12'),
+         ('insert', '13'),
+         ('remove', '14'),
+         ('push', '15'),
+         ('data', '16'),
+         ('skip', '17'),
+         ('pause', '19'),
+         ('play', '18'),
+         ('flush', '16'),
+         ]
+    if list.mem_assoc(taskname, tasks) then
+        tasks[taskname]
+    else
+        log("Warning: task #{taskname} does not exist")
+        "00"
+    end
+end
+
+def get_namespace_number(namespace) =
+    channelnames = ref string.split(separator=',', list.assoc("channels", ini))
+
+    namespaces = [
+         ('player', '10'),
+         ('common', '11'),
+         ('playlist', '11'),
+         ('request', '12'),
+         ('mixer', '13'),
+         ]
+    def addNamespace (ret, el)
+        number = string_of((list.length(ret) + 10))
+        list.append(ret,[("ch#{el}", "#{number}")])
+    end
+    namespaces = list.fold(addNamespace, namespaces, !channelnames)
+
+    if list.mem_assoc(namespace, namespaces) then
+        namespaces[namespace]
+    else
+        log("Warning: namespace #{namespace} does not exist")
+        log(json_of(namespaces))
+        "00"
+    end
+end
+
+def create_protocol () =
+    #pwd = list.hd(get_process_lines('pwd'))
+    #def getdirname() =
+
+    #    if dirname(argv(0)) == "." then
+    #        ""
+    #    else
+    #       '/'^dirname(argv(0))
+    #    end
+    #end
+    #cur = getdirname()
+    add_protocol(temporary=true, "silence", fun(arg,delay)-> begin system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/silence.sh 0.01 > /var/tmp/quiet.wav; qwavheaderdump -q -F /var/tmp/quiet.wav > /dev/null' ) ['/var/tmp/quiet.wav'] end)
+end
+
+#create_protocol()
+
+def message (~value="", namespace, success, message, code, error, ~level="error", ~type='internal') =
+
+    level = if success then 'success' else level end
+    namespace = if namespace == 'common' then
+                   'playlist'
+                else
+                    namespace
+                end
+    namespace_number = get_namespace_number(namespace)
+    code = "10#{namespace_number}#{code}"
+    result = [('section', '#{namespace}'),
+              ('success', '#{success}'),
+              ('message', '#{message}'),
+              ('value', '#{value}'),
+              ('code', '#{code}#{error}'),
+              ('level', '#{level}'),
+              ('type', '#{type}'),
+              ]
+    json_of(compact=true,result)
+
+end
+
+##Sorgt für unterbrechungsfreien Wechsel zwischen den Tracks
+def crossfade(a,b)
+    add(normalize=false,
+        [ fade.initial(duration=2., b),
+         sequence(merge=true,
+                  [ blank(duration=0.), fallback([]) ]),
+         fade.final(duration=2.,a) ])
+end
+
+def foldplaylist(l, el)=
+        # fold metadata to list
+        def tolist (ret, el) =
+            tmp = string.split(separator='=',el)
+            if list.length(tmp) > 1 then # found the separator in current el
+                # get the key -> value pair
+                key = list.nth(tmp,0)
+                value = list.nth(tmp,1)
+                # append to list
+                list.append(ret,[(key,value)])
+            else
+                # nothing to do
+                ret
+            end
+        end
+        # extract metadata part from annotation
+        extracted = string.extract(pattern="annotate:[,](.*):/",string.replace(pattern='"',(fun (s) -> ""),el))
+        # split extracted string by comma separator and create a list
+        list.append(l,[list.fold(tolist, [], string.split(separator=',',extracted["1"]))])
+end
+# The server seeking function
+def seek_track(source, t) =
+
+  t = float_of_string(default=0.,t)
+  ret = source.seek(source,t)
+  log("Seeked #{ret} seconds.")
+  success = ret > 0.
+  error = if success then '00' else '01' end
+  errors = [('00', 'Seeked #{ret} seconds on #{source.id(source)}'), ('01', 'Seek failed')]
+  message(value=string_of(ret), source.id(source), success, errors[error], get_task_number ('seek_track'), error, level="warning", type='unknown')
+  #"Seeked #{ret} seconds."
+end
+
+def reload_recorded(~skip, ~uri) =
+  print('param skip: #{skip} +++ param uri: #{uri}')
+  print("reload_recorded not implemented")
+  "nana"
+end
+
+

modules/liquidsoap/playd.liq

+#!/usr/bin/liquidsoap
+
+#set("log.stdout", true)
+
+inst = if argv(1) != "" then string_of(argv(1)) else 'playd' end
+instance = ref inst
+
+%include "readini.liq"
+
+inifile = '/etc/aura/aura.ini'
+ini = read_ini(inifile)
+
+# send alive 
+ignore(system('#{list.assoc("install_dir", ini)}/modules/liquidsoap/helpers/message.py -c #{!instance} -t sayAlive'))
+# send alive frequently
+exec_at(freq=20., pred={true}, {ignore(system('#{list.assoc("install_dir", ini)}/modules/liquidsoap/helpers/message.py -c #{!instance} -t sayAlive'))})
+
+# set current playlist
+ignore(system('#{list.assoc("install_dir", ini)}/modules/liquidsoap/helpers/message.py -c #{!instance} --task=setState -n playlistcurrent -v ""'))
+
+# load data from ini file
+#daemongroup = list.assoc("daemongroup", ini)
+#daemonuser = list.assoc("daemonuser", ini)
+
+
+socketdir = list.assoc("socketdir", ini)
+
+# ALSA settings
+use_alsa = list.assoc("use_alsa", ini)
+alsa_buffer = int_of_string(list.assoc("alsa_buffer", ini))
+alsa_periods = int_of_string(list.assoc("alsa_periods", ini))
+
+# set player i/o devices
+player_input = list.assoc("input_device", ini)
+#player_second_input = list.assoc("player_second")
+player_output = list.assoc("output_device", ini)
+
+# fallback settings
+fallback_audio_folder = list.assoc("fallback_audio_folder", ini)
+fallback_max_blank = float_of_string(list.assoc("fallback_max_blank", ini))
+fallback_min_noise = float_of_string(list.assoc("fallback_min_noise", ini))
+fallback_threshold = float_of_string(list.assoc("fallback_threshold", ini))
+
+# channel names from config
+channelnames = ref string.split(separator=',', list.assoc("channels", ini))
+
+# alsa settings
+# buffer - decrease latency: eg. alsa_buffer="2024"
+set("alsa.alsa_buffer", alsa_buffer)
+set("alsa.periods", alsa_periods)
+
+# enable socketserver
+set("server.socket", true)
+set("server.socket.path", socketdir^"/<script>.sock")
+set("server.telnet", true)
+set("server.telnet.port", 1234)
+
+# enable daemon
+#set("init.daemon", true)
+#set("init.daemon.change_user.group", daemongroup)
+#set("init.daemon.change_user.user", daemonuser)
+#set("init.daemon.pidfile.path", socketdir^"/<script>.pid")
+#set("init.daemon.pidfile.perms", 0o666)
+
+# set logging
+set("log.file",true)
+set("log.file.path", list.assoc("logdir", ini)^"/<script>.log")
+set("log.file.perms",0o640)
+set("log.level", 10)
+
+# allowed mime types
+set("playlists.mime_types.xml",["text/xml","application/xml","application/xspf+xml"])
+
+# load functions
+# dir = list.assoc("install_dir", ini)
+
+%include "library.liq"
+%include "playlist.liq"
+
+# Der input wie oben definiert
+def get_input()
+    def get_input()
+        if use_alsa == "y" then
+            if player_input == "soundcard" then
+                print("autodetect device")
+                input.alsa(id="sound_input", fallible=true, clock_safe=false)
+            else
+                print("manual set device: "^player_input)
+                input.alsa(id="sound_input", fallible=true, clock_safe=false, device=player_input)
+            end
+        else
+            input.pulseaudio(id="sound_input")
+        end
+    end
+    def get_fallback()
+        if fallback_audio_folder != "" then
+            print("fallbackfolder chosen: "^fallback_audio_folder)
+            playlist.safe("/var/audio/fallback/music.flac")
+            #playlist.safe(fallback_audio_folder)
+        else
+            blank(duration=20.0)
+        end
+    end
+
+    if player_input == "" then
+        blank(duration=0.1) # wait...
+    else
+        # start silence detector on input alsa and set a fallback
+        fallback(track_sensitive=false,
+                 [ strip_blank(max_blank=fallback_max_blank, min_noise=fallback_min_noise, threshold=fallback_threshold, get_input()),
+                   get_fallback() ]
+        )
+    end
+end
+
+playlistrunning = ref false
+
+playlist_recorded = playlist.xml(id='playlist', on_done=fun() -> begin ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n playlistcurrent -v ""')) ignore(server.execute('mixer.select 0 false')) end, 'none')
+# Die Source aus der Playlist
+recorded = snd(playlist_recorded)
+
+# Skippen erlauben
+add_skip_command(recorded)
+
+# Funktion zum laden einer neuen Playlist
+playlist_funcs = fst(playlist_recorded)
+
+# Flush functions
+flush_recorded = snd(fst(playlist_funcs))
+
+# Reload function
+reload_recorded = fst(fst(playlist_funcs))
+
+#up funtction
+move_recorded = fst(fst(snd(playlist_funcs)))
+
+#insert funtction
+insert_recorded = fst(snd(fst(snd(playlist_funcs))))
+
+# remove function
+remove_recorded = snd(snd(fst(snd(playlist_funcs))))
+
+#push function
+push_recorded = snd(snd(snd(playlist_funcs)))
+
+#show playlist function
+data_recorded = fst(snd(snd(playlist_funcs)))
+
+# create playlist source with smart crossfade
+playlist = fallback(id="common", track_sensitive=false, [                                    
+    switch(transitions = 
+        [crossfade,crossfade], 
+        [
+            ( {!playlistrunning == true}, recorded ), # play the playlist
+            ( {!playlistrunning == false}, get_input() )
+        ]
+    ),
+        if fallback_audio_folder != "" then 
+            playlist.safe(fallback_audio_folder)
+        else 
+            blank(duration=20.0)
+        end  
+])
+
+# method to dynamicaly create a channel
+def addChannel (ret, el, ~base="ch")
+    equeue = request.equeue(id = base^el) # create the equeue request
+    # add a seek function to the channel
+    server.register(namespace="ch"^el,
+                description="Seek to a relative position",
+                usage="seek <duration>",
+                "seek",fun (x) ->  begin seek_track(equeue, x) end)
+    # append and return the list
+    list.append(ret,[equeue])
+end
+
+channels = addChannel([], '1', base='auto')
+channels = addChannel(channels, '2', base='auto')
+channels = list.fold(addChannel, channels, !channelnames)
+
+mixer = mix(id = "mixer", list.append([playlist], channels))
+#ignore(s) # testing
+
+
+# User may load a XML-Playlist
+#server.register(namespace='playlist',
+#    description="Load Playlist",
+#    usage="load <uri>",
+#    "load",fun (p) ->  begin
+#        reload_recorded(skip=0, uri=p)
+#    end
+#)
+
+# Register the seek function
+server.register(namespace='playlist',
+    description="Seek to a relative position",
+    usage="seek <duration>",
+    "seek",fun (x) ->  begin seek_track(recorded, x) end
+)
+
+# The play function
+server.register(namespace='playlist',
+    description="Play recorded files",
+    usage="play",
+    "play",fun (x) -> if !playlistrunning == true then                                    
+        message('playlist', false, 'Playlist already playing', get_task_number ('play'), '01', level="info", type="user")
+    else 
+        ignore(server.execute('mixer.select 0 true')) 
+        playlistrunning := true 
+        message('playlist', true, 'Start playing' , get_task_number ('play'), '00')
+    end
+)
+
+# Flush current  playlist
+server.register(namespace='playlist',
+    description="Flush recorded playlist",
+    usage="flush",
+    "flush",fun (s) ->  begin flush_recorded() end)
+
+# The remove function
+server.register(namespace='playlist',
+    description="Remove item from playlist",
+    usage="remove <pos>",
+    "remove",fun (p) ->  begin remove_recorded(int_of_string(p)) end)
+
+# Let the user move up or down a track
+server.register(namespace='playlist',
+    description="Move an item up or down",
+    usage="move <from> <to>",
+    "move",fun (p) ->  begin  
+        params=string.split(separator=" ",p) 
+        src=if list.length(params)>0 then (int_of_string(list.nth(params,0))+0) else 0 end
+        target=if list.length(params)>1 then (int_of_string(list.nth(params,1))+0) else 0 end
+        move_recorded(src, target) 
+    end
+)
+
+# Insert an entry
+server.register(namespace='playlist',
+    description="Add an entry",
+    usage="insert <pos> <uri> [<title> <time>]",
+    "insert",fun (p) ->  begin  
+        params=string.split(separator=" ",p) 
+        pos=if list.length(params)>0 then list.nth(params,0) else '' end
+        uri=if list.length(params)>1 then list.nth(params,1) else '' end                                    
+        title=if list.length(params)>2 then list.nth(params,2) else '' end
+        time=if list.length(params)>3 then list.nth(params,3) else '' end
+        insert_recorded(pos=pos, title=title,time=time,uri) 
+    end
+)
+
+# Insert a track on top of playlist
+server.register(namespace='playlist',
+    description="Push an item to top and play immediately",
+    usage="push <pos>",
+    "push",fun (p) ->  begin
+        params=string.split(separator=" ",p) 
+        pos=if list.length(params)>0 then int_of_string(list.nth(params,0)) else 0 end 
+        push_recorded(pos) 
+    end
+)
+
+# Show metadata
+server.register(namespace='playlist',
+    description="Show remaining playlist data in json format",
+    usage="data",
+    "data",fun (s) ->  begin 
+        playlist = data_recorded() 
+        json_of(compact=true, playlist) 
+    end
+)
+
+# Pause/stop playlist
+server.register(namespace='playlist',
+    description="Pause playing recorded files",
+    usage="pause",
+    "pause", fun (x) -> if !playlistrunning == false then 
+        message('playlist', false, 'Playlist already stopped', get_task_number ('pause'), '01', level="info", type="user")
+    else 
+        playlistrunning := false 
+        ignore(server.execute('playlist.skip'))
+        ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n playlistcurrent -v ""'))
+        message('playlist', true, 'Playlist stopped', get_task_number ('pause'), '00') 
+    end
+)
+
+# get remaining time
+server.register(namespace='playlist',
+    description="Remaining time",
+    usage = "remaining",
+    "remaining", fun(x) -> string_of(source.remaining(recorded)) 
+)
+
+#streamm = single("/var/audio/fallback/music.flac")
+#ignore(streamm) # testing
+
+# Alsa output
+if use_alsa == "y" then
+    if player_output == 'soundcard'  then
+        output.alsa(id="player", fallible=true, mixer)
+    else
+        output.alsa(id="player", fallible=true, device=player_output, mixer)
+    end
+else
+    output.pulseaudio(id="player", mixer)
+end
+
+
+# %include "stream.liq"
+     

modules/liquidsoap/playlist.liq

+
+# Liest Playlist im XML-Format (XSPF)
+%include "readini.liq"
+%include "library.liq"
+
+inst = if argv(1) != "" then string_of(argv(1)) else 'playd' end
+instance = ref inst
+
+error_message = ref ""
+
+def playlist.xml(~id="",~skip=0,~on_done={()},uri)
+  # A reference to the playlist
+  
+  playlist = ref []  
+  
+  # A reference to the uri
+  playlist_uri = ref uri
+  # A reference to know if the source
+  # has been stopped
+  has_stopped = ref false
+  # The next function - see request.dynamic
+  def next () =
+    file =
+      if list.length(!playlist) > 0 then               
+        ret = list.hd(!playlist) # get first entry
+        playlist := list.tl(!playlist) # remove first entry from playlist
+        ret
+      else
+        # Playlist finished?
+        if not !has_stopped then
+          on_done () # call on_done method
+        end
+        has_stopped := true
+        ""
+      end
+    if file == '' then
+       request.create("silence:waiting") #
+    else              
+       track = file
+       extracted = string.extract(pattern="annotate:.*:(/.*)",string.replace(pattern='"',(fun (s) -> ""), track))
+       wavfile = extracted["1"]
+       
+       ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n playlistcurrent -v "#{wavfile}"'))
+       request.create(track)
+    end    
+  end
+  # Instanciate the source
+  source = request.dynamic(id=id,next)
+  
+  # flush function
+  def flush () =
+    log(label=id,"Flush "^string_of(list.length(!playlist))^" Items")     
+    playlist := []  # playlist is empty now
+    ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n playlistcurrent -v ""'))
+    message(value=json_of(!playlist), 'playlist', true, "Playlist flushed", get_task_number ('flush'), "00", type='internal')
+  end
+  
+  # Get its id.
+  id = source.id(source)
+  # The load function
+  def load_playlist (~skip=1) =
+
+    playlist_tmp = request.create.raw(!playlist_uri) # temporary try to create the playlist
+    
+    result =
+        if file.exists(!playlist_uri) and request.resolve(playlist_tmp) then # playlist should exist and be resolvable
+            playlist_tmp = request.filename(playlist_tmp) # create a temporary request
+            entries = playlist.parse(playlist_tmp) # parse the playlist request
+            
+            # create a annotation from playlist track
+            def annotate_track(el) =
+              meta = fst(el)
+              # remove protocol from filepath
+              filepath = string.replace(pattern='"',(fun (s) -> ""),string.replace(pattern='file://',(fun (s) -> ""),string.escape(snd(el))))
+              # fold metadata
+              s = list.fold(fun (cur, el) -> begin "#{cur},#{fst(el)}=#{string.escape(snd(el))}" end, "", meta)
+              # return annotation
+              "annotate:#{s}:#{filepath}"
+            end
+            # map entries
+            error_message := if list.length(entries) > 0 then "OK" else "empty" end
+            list.map(annotate_track,entries)
+        else
+            error_message := if file.exists(!playlist_uri) then "!resolve" else "!exist" end
+            log(label=id,"Couldn't read playlist: request resolution failed.")
+            # return an empty list
+            []
+        end
+    request.destroy(playlist_tmp)  # destroy the temporary request      
+    playlist := result
+  end
+
+  # Move function 
+  def move (fromPos, toPos) =
+    countx = ref -1 # counter
+    fromx = ref (fromPos - 1) # position of item       
+    tox = ref (toPos - 1) #  position to move
+    msg = ref ""
+    msg := message('playlist', true, "Item #{fromPos} moved to #{toPos} in Playlist", get_task_number ('move'), "00", type='internal')
+    # check if we can move
+    if (toPos) > list.length(!playlist) 
+        or (fromPos) > list.length(!playlist)
+        or toPos < 1 
+        or fromPos < 1 then         
+            msg := message('playlist', false, "Moving item #{fromPos} to position  #{toPos} is impossible.", get_task_number ('move'), "01", type='user', level="warning")
+    elsif toPos != fromPos then
+        # get the item to move
+        itemToMove = ref list.nth(!playlist, !fromx)
+    
+        # returns the resorted list
+        def resort (ret, el) =
+            countx := !countx + 1     
+            if !countx == !fromx then # das ist das Item, das verschoben wird -  es wird uebersprungen 
+                ret
+            elsif !countx == !tox then # Wir haben die Position erreicht, an der das zu verschiebende Item platziert wird             
+                if !fromx < !tox then # aber entry wurde bereits uebersprungen 
+                    list.append(ret,[el, !itemToMove]) # also muss hinter dem aktuellen Element eingefuegt werden                    
+                else  
+                    list.append(ret,[!itemToMove, el]) # ...andernfalls davor
+                end
+            
+            else
+                list.append(ret,[el]) #Liste um den aktuellen Eintrag erweitern
+            end            
+        end
+        playlist := list.fold(resort, [], !playlist)            
+    end    
+    !msg
+  end
+  
+  # method to delete an item from the list  
+  def remove (index) =
+    countx = ref 0
+    delx = ref (index - 1)        
+
+    def delete (ret, el) =
+        countx := !countx + 1  
+        if !countx == (!delx +1) then 
+            ret # the position to be removed
+        else
+            list.append(ret,[el]) # append evereything else
+        end              
+    end
+    
+    playlist := list.fold(delete, [], !playlist)     
+    message(value=json_of(!playlist), 'playlist', true, "Item #{index} removed from Playlist", get_task_number ('remove'), "00", type='internal')
+  end
+  
+  # method to insert an item 
+  def insert (~title='Unknown', ~time='00:00', ~pos='', uri) =
+    el = 'annotate:title="#{title}",time="#{time}",location="#{uri}":#{uri}'
+    playlist := list.append(!playlist,[el]) #Item erst mal an die Playlist anhaengen
+    if int_of_string(pos) > 0 then # Eine Position ist angegeben
+        result = move(list.length(!playlist), int_of_string(pos)) # Item dorthin verschieben
+        result_list = of_json(default=[('success','false')], result) # Die Ausgabe von "move" als assoziatives Array
+        if list.assoc('success', result_list) != "true" then # War move nicht erfolgreich?
+            result # ... dann geben wir die Ausgabe von move zurueck
+        else # ... andernfalls Erfolgsmeldung
+            message(value=json_of(!playlist), 'playlist', true, "#{title} item  inserted into #{pos} in Playlist", get_task_number ('insert'), "00", type='internal')               
+        end
+    elsif int_of_string(uri) > 0 then # uri ist ein int? Da hat der User was falsch gemacht...
+        message('playlist', true, "Syntax error: playlist.insert <pos> <uri> [<title> <time>]", get_task_number ('insert'), "01", type='user', level="info")
+    else # da auch, da ist pos 0 oder negativ    
+        message('playlist', true, "Cannot insert #{title} at position #{pos}. Put it on the end.", get_task_number ('insert'), "02", type='user', level="info")
+    end
+  end
+  
+  # returns the remaining list
+  def getplaylist () =
+    
+    list.fold(foldplaylist, [], !playlist)
+  end
+  
+  def push (index) =      
+      "Not implemented yet"
+  end  
+  
+  # The reload function
+  def reload(~blank=false, ~skip=0, ~uri="") =
+    if uri != "" then
+      playlist_uri := uri
+    end
+    log(label=id,"Reloading playlist with URI #{!playlist_uri}")
+    has_stopped := false
+    load_playlist(skip=skip)
+    #if !error_message == 'OK' then
+    #    message('playlist', true, 'Playlist #{!playlist_uri} loaded', get_task_number ('load'), '00', level="success")
+    #elsif !error_message == '!exist' then
+    #    message('playlist', false, "Playlist #{!playlist_uri} does not exist", get_task_number ('load'), '01', level="warning", type="external")
+    #elsif !error_message == 'empty' then
+    #    message('playlist', false, "Playlist #{!playlist_uri} is empty or wrong format", get_task_number ('load'), '02', level="warning", type="external")
+    #else
+    #    message('playlist', false, "Playlist #{!playlist_uri} doesn't resolve.", get_task_number ('load'), '03', level="warning", type="external")
+    #end
+  end
+  log(label=id,"Loading playlist with URI #{!playlist_uri}")
+  # Load the playlist
+  load_playlist(skip=skip)
+  # Return
+  (((reload,flush),((move,(insert,remove)),(getplaylist,push))),source)
+end
+

modules/liquidsoap/pulse.liq

+
+input_fs = single(id="fs", "/var/audio/fallback/output.flac")
+input_http = input.http(id="http", "http://stream.fro.at/fro128.mp3")
+linein_pulse_1 = input.pulseaudio(id="linein", device = input_device_0)
+mixer = mix(id="mixer", [input_fs, input_http, linein_pulse_1])
+out := output.pulseaudio(id="lineout", mixer)

modules/liquidsoap/readini.liq

+
+def read_ini(file)
+    ret = get_process_lines("cat "^file )
+    ret = list.map(string.split(separator="="), ret)
+
+    # l' => the filling list
+    def f(l',l)=
+      if list.length(l) >= 2 then
+        line = string.extract(pattern='"(.*)"', list.nth(l,1))
+        print(line)
+        print((list.hd(l),line['1']))
+        
+        list.append([(list.hd(l),line['1'])],l')
+      else
+        if list.length(l) >= 1 then
+          list.append([(list.hd(l),"")],l')
+        else
+          l'
+        end
+      end
+    end
+
+    list.fold(f,[],ret)
+end

modules/liquidsoap/record.liq

+inst = if argv(1) != "" then string_of(argv(1)) else 'record' end
+instance = ref inst
+
+%include "readini.liq"
+
+ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} -t sayAlive'))
+exec_at(freq=20., pred={true}, {ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} -t sayAlive'))})
+
+audiobase= if !instance == 'record' then list.assoc("audiobase", ini) else list.assoc("altaudiobase", ini)  end
+filenamepattern = ref audiobase^"/%Y-%m-%d/%Y-%m-%d-%H-%M.wav"
+daemongroup = list.assoc("daemongroup", ini)
+daemonuser = list.assoc("daemonuser", ini)
+socketdir = list.assoc("socketdir", ini)
+recinput = if !instance == 'record' then list.assoc("recinput", ini) else list.assoc("altrecinput", ini)  end
+recorder_device = if !instance == 'record' then list.assoc("recorder_device", ini) else list.assoc("altrecorder_device", ini) end
+set("init.daemon",true)
+set("init.daemon.change_user.group",daemongroup)
+set("init.daemon.change_user.user",daemonuser)
+set("server.socket",true)
+set("server.socket.path",socketdir^"/"^!instance^".sock")
+set("init.daemon.pidfile.path",socketdir^"/"^!instance^".pid")
+set("init.daemon",true)
+set("log.file",true)
+set("log.file.path",list.assoc("logdir", ini)^"/"^!instance^".log")
+set("log.file.perms",0o660)
+set("log.level",4)
+
+# Der aktuelle Dateiname für die Aufnahme
+recordingfile = ref ""
+#wir definieren eine Referenz für die Stop-Funktion, die aber bisher noch nichts tun kann
+stop_f = ref (fun () -> ())
+#bewahre uns davor, dass zweimal gleichzeitig die gleiche Date aufgenommen wird
+block_dump = ref false
+
+# Stop dump - wir definieren die Funktion, die stop_f ausführt
+def stop_dump() =
+  f = !stop_f
+  f ()
+end
+
+#Der input wie oben definiert
+def get_input() 
+    if recinput == 'soundcard' then
+        ## input ist Alsa  
+        if recorder_device != ''  then
+            input.alsa(device=recorder_device)
+        else
+            input.alsa()
+        end
+            
+    elsif recinput == 'nosound' then
+        mksafe(empty())
+    else
+        ## input ein via config definierter Stream
+        mksafe(input.http(recinput))        
+    end
+end
+
+#Wav header fixen und ggf. die Aufzeichnung beenden
+def on_close(filename) 
+    # es darf wieder aufgenommen werden
+    block_dump := false 
+    # Korrekten WAV-Header schreiben
+    system("qwavheaderdump -F #{filename}")
+    # event dumpend feuern
+    ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n dumpend -v #{filename}'))
+    log('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n dumpend -v #{filename}')
+    # Naechsten Dateinamen vormerken
+    recordingfile := list.hd(get_process_lines("date +#{!filenamepattern}"))
+    ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n dumpfile -v #{!recordingfile}'))
+end
+
+#Funktion gibt Auskunft welches File aktuell ist und wieviel Prozent bereits aufgenommen werden
+def currecording()
+    curfile = !recordingfile
+    if curfile != "" then           
+        procent_done = list.hd(get_process_lines("echo $(($(stat -c%s "^curfile^")/3174777))")) 
+        "#{curfile},#{procent_done}"
+    else 
+        ""
+    end        
+end
+
+#Funktion zum Start der Aufzeichnung
+def start_dump() =
+    log('start dump') 
+    # don't record twice
+    if !block_dump == false then        
+        block_dump := true
+        # Der Input - Alsa oder Stream
+        input=get_input()
+        record = output.file(
+               id="recorder",   
+               # Wav Stereo erzeugen         
+               %wav(stereo=true),
+               perm = 0o664,
+               # die aktuell aufnehmende Datei in 'recordingfile' festhalten
+               on_start={ begin                             
+                    recordingfile := list.hd(get_process_lines("date +#{!filenamepattern}"))
+                    log('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n dumpfile -v #{!recordingfile}')
+                    ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n dumpfile -v #{!recordingfile}'))
+                    end},
+               # Beim Stoppen den Status zurücksetzen
+               on_stop={ignore(system('#{list.assoc("install_dir", ini)}/modules/soundengine/helpers/message.py -c #{!instance} --task=setState -n dumpfile -v ""'))},
+               # Dateipfad
+               !filenamepattern,
+               # Funktion die beim Schließen der Datei aufgerufen wird
+               on_close=on_close,
+               # Alle 30 Minuten eine neue Datei
+               reopen_when={ if !instance == 'record' then int_of_float(gettimeofday()/60.) mod 30 == 0  else false end },
+               # Der Input
+               input
+        )
+           # Die Stopfunkton zeigt nun auf die Shutdown-Funktion der aktuellen Source
+        stop_f := fun () -> begin 
+                                log('stop recording') 
+                                # Aufnahme sauber beenden
+                                ignore(server.execute('recorder.stop'))
+                                # Source zerstören 
+                                source.shutdown(record) 
+                                # Variable zurücksetzen
+                                recordingfile := ""  
+                            end                                          
+     end    
+end    
+
+
+# Der Server wird durch 3 Funktionen bereichert
+# Der User darf die Aufzeichnung manuell starten
+server.register(namespace="record",
+                description="Start recording.",
+                usage="start",
+                "start",
+                fun (s) ->  begin start_dump() "OK" end)
+
+# Der User darf die Aufzeichnung manuell stoppen
+server.register(namespace="record",
+                description="Stop recording.",
+                usage="stop",
+                "stop",
+                fun (s) -> begin stop_dump() "OK" end)
+
+if !instance != 'record' then
+    # Der User darf einen Dateinamen für die Aufnahme definieren
+    server.register(namespace="record",
+                    description="Define filename for output.",
+                    usage="setfilename",
+                    "setfilename",
+                    fun (s) -> begin filenamepattern := audiobase^"/"^string_of(s)  "OK" end)
+end
+
+# Der USer kann sich über den Fortschritt der Aufnahme informieren
+server.register(namespace="record",
+                description="Show current file.",
+                usage="curfile",
+                "curfile",
+                fun (s) -> currecording() )
+
+output.dummy(blank(id="serve"))

modules/liquidsoap/serverfunctions.liq

+# shutdown server function
+#server.register(namespace='server',
+#    description="shutdown server",
+#    usage="stop",
+#    "stop",
+#    fun(x,y) -> stop_server )
\ No newline at end of file

modules/liquidsoap/settings.liq

+# LOG FILE SETTINGS
+set("log.file.path", "./<script>.log")
+
+# SERVER SETTINGS
+set("server.telnet", true)
+set("server.telnet.bind_addr", "0.0.0.0")
+set("server.telnet.port", 1234)
+
+set("server.socket", true)
+set("server.socket.path", "./<script>.sock")
+
+# SOUND CARD SETTINGS
+input_device_0 = list.assoc("input_device[0]", ini)
+#input_device_1 = list.assoc("input_device[1]", ini)
+#input_device_2 = list.assoc("input_device[2]", ini)
+output_device_0 = list.assoc("output_device[0]", ini)
+
+ignore(input_device_0)
+ignore(output_device_0)
+
+# ALSA / pulse settings
+use_alsa           = list.assoc("use_alsa", ini) == "y"
+frame_duration     = float_of_string(list.assoc("frame_duration", ini))
+frame_size         = int_of_string(list.assoc("frame_size", ini))
+alsa_buffer        = int_of_string(list.assoc("alsa_buffer", ini))
+alsa_buffer_length = int_of_string(list.assoc("alsa_buffer_length", ini))
+alsa_periods       = int_of_string(list.assoc("alsa_periods", ini))
+
+if use_alsa then
+    if frame_duration > 0.0 then
+        print("setting frame.duration to #{frame_duration}")
+        set("frame.duration", frame_duration)
+    end
+    if frame_size > 0 then
+        print("setting frame.size to #{frame_size}")
+        set("frame.size", frame_size)
+    end
+    if alsa_buffer > 0 then
+        print("setting alsa.buffer to #{alsa_buffer}")
+        set("alsa.alsa_buffer", alsa_buffer)
+    end
+    if alsa_buffer > 0 then
+        print("setting alsa.buffer_length to #{alsa_buffer_length}")
+        set("alsa.buffer_length", alsa_buffer_length)
+    end
+    if alsa_periods > 0 then
+        print("setting alsa.periods to #{alsa_periods}")
+        set("alsa.periods", alsa_periods) # assertion error when setting periods other than 0 => alsa default
+    end
+end
\ No newline at end of file