requirements.txt

+Flask
+Flask-SQLAlchemy==2.3.2
+mysqlclient
+redis==2.10.6
+mutagen==1.42
+validators==0.12.4

systemd service/aura-engine.service

+[Unit]
+Description=Aura Engine Playout Server
+After=network.target
+
+[Service]
+Type=simple
+User=gg
+WorkingDirectory=/home/gg/PycharmProjects/engine
+ExecStart=/home/gg/PycharmProjects/engine/aura.py
+ExecStop=/home/gg/PycharmProjects/engine/guru.py --shutdown --quiet
+Restart=always
+
+[Install]
+WantedBy=multi-user.target

systemd service/aura-lqs.service

+[Unit]
+Description=Aura Engine Playout Server
+After=network.target aura-engine.service
+Wants=aura-engine.service
+
+[Service]
+Type=simple
+User=gg
+ExecStart=/usr/bin/liquidsoap /home/gg/PycharmProjects/engine/modules/liquidsoap/engine.liq
+Restart=always
+
+[Install]
+WantedBy=multi-user.target

test.py

+#!/usr/bin/python3
+
+#
+#  engine
+#
+#  Playout Daemon for autoradio project
+#
+#
+#  Copyright (C) 2017-2018 Gottfried Gaisbauer <gottfried.gaisbauer@servus.at>
+#
+#  This file is part of engine.
+#
+#  engine is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  any later version.
+#
+#  engine is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with engine. If not, see <http://www.gnu.org/licenses/>.
+#
+
+from libraries.database.broadcasts import *
+import json
+import logging
+import sqlalchemy
+import decimal
+
+from modules.communication.liquidsoap.communicator import LiquidSoapCommunicator
+from modules.monitoring.diskspace_watcher import DiskSpaceWatcher
+from libraries.base.config import AuraConfig
+from libraries.database.broadcasts import Schedule, ScheduleEntry
+from modules.scheduling.scheduler import AuraScheduler, AuraCalendarService
+
+def alchemyencoder(obj):
+    """JSON encoder function for SQLAlchemy special classes."""
+    if isinstance(obj, datetime.date):
+        return obj.isoformat()
+    elif isinstance(obj, decimal.Decimal):
+        return float(obj)
+    elif isinstance(obj, sqlalchemy.orm.state.InstanceState):
+        return ""
+    #elif isinstance(obj, Schedule):
+    #    return json.dumps([obj._asdict()], default=alchemyencoder)
+    else:
+        return str(obj)
+
+#    programme_as_string = json.dumps([se[0]._asdict()], default=alchemyencoder)
+#    print(programme_as_string)
+
+def start_diskspace_watcher():
+    config = AuraConfig()
+    config.read_config()
+    diskspace_watcher = DiskSpaceWatcher(config.config, logging.getLogger("AuraEngine"), LiquidSoapCommunicator(config.config))
+    diskspace_watcher.run()
+
+def select_act_programme():
+    # start_diskspace_watcher()
+    # select_act_programme()
+    config = AuraConfig()
+    config.read_config()
+
+    liquidsoapcommunicator = LiquidSoapCommunicator(config.config)
+    sched = AuraScheduler(config.config)
+
+    liquidsoapcommunicator.scheduler = sched
+    sched.liquidsoapcommunicator = liquidsoapcommunicator
+
+    programme = sched.load_programme_from_db()
+
+    for show in programme:
+        print(show)
+
+def fadeout(lsc):
+    entry = ScheduleEntry.select_act_programme()
+    lsc.fade_out(entry, 2)
+
+def fadein(lsc):
+    entry = ScheduleEntry.select_act_programme()
+    lsc.fade_in(entry, 1)
+
+def fetch_new_programme():
+    config = AuraConfig()
+    config.read_config()
+    acs = AuraCalendarService(config.config)
+
+
+    queue = acs.get_queue()
+
+    # start fetching thread
+    acs.start()
+
+    # wait for the end
+    response = queue.get()
+
+# # ## ## ## ## ## # #
+# # ENTRY FUNCTION # #
+# # ## ## ## ## ## # #
+def main():
+    fetch_new_programme()
+
+
+# # ## ## ## ## ## ## # #
+# # End ENTRY FUNCTION # #
+# # ## ## ## ## ## ## # #
+
+
+if __name__ == "__main__":
+    main()
+

testing/test.py

+import os
+import unittest
+import validators
+
+from datetime import datetime
+
+# libraries.base
+from libraries.base.logger import AuraLogger
+from libraries.base.config import AuraConfig
+# libraries.database
+from libraries.database.broadcasts import Schedule, ScheduleEntry, TrackService
+# libraries.security
+# from libraries.security.user import AuraUser
+
+# modules
+from modules.communication.liquidsoap.communicator import LiquidSoapCommunicator
+from modules.scheduling.scheduler import AuraScheduler
+
+class TestLogger(unittest.TestCase):
+    aura_logger = None
+
+    def setUp(self):
+        self.aura_logger = AuraLogger()
+
+    def test_logger(self):
+        self.assertTrue(self.aura_logger.logger.hasHandlers())
+
+
+class TestConfig(unittest.TestCase):
+    aura_config = None
+
+    def setUp(self):
+        self.aura_config = AuraConfig()
+
+    def test_config(self):
+        # is ini path correct set?
+        self.assertEqual(self.aura_config.config.ini_path, "/etc/aura/engine.ini")
+
+        # install_dir is set by runtime. is it a directory?
+        self.assertTrue(os.path.isdir(self.aura_config.config.get("install_dir")))
+
+        # calendarurl and importerurl set and valid urls?
+        self.assertTrue(validators.url(self.aura_config.config.get("calendarurl")))
+        self.assertTrue(validators.url(self.aura_config.config.get("importerurl")))
+
+        # is liquidsoap socketdir set and a directory?
+        self.assertTrue(os.path.isdir(self.aura_config.config.get("socketdir")))
+
+        # database settings set?
+        self.assertIsNotNone(self.aura_config.config.get("db_user"))
+        self.assertIsNotNone(self.aura_config.config.get("db_pass"))
+        self.assertIsNotNone(self.aura_config.config.get("db_name"))
+        self.assertIsNotNone(self.aura_config.config.get("db_host"))
+
+class TestSchedule(unittest.TestCase):
+    schedule = None
+
+    def setUp(self):
+        self.schedule = Schedule()
+
+    def test_schedule(self):
+        # select one and check if its not None and a Schedule
+        entry = self.schedule.select_by_id(1)
+        self.assertIsNotNone(entry)
+        self.assertIsInstance(entry, Schedule)
+
+
+class TestScheduleEntry(unittest.TestCase):
+    schedule_entry = None
+
+    def setUp(self):
+        self.schedule_entry = ScheduleEntry()
+
+    def test_schedule_entry(self):
+        # select one playlist and check if its not None, a ScheduleEntry
+        entry = self.schedule_entry.select_playlist(2)
+
+        self.assertIsNotNone(entry)
+        self.assertIsInstance(entry, list)
+        self.assertGreaterEqual(len(entry), 1)
+
+
+class TestTrackService(unittest.TestCase):
+    track_service = None
+
+    def setUp(self):
+        self.track_service = TrackService()
+
+    def test_track_service(self):
+        day = datetime.strptime("19.03.2018", "%d.%m.%Y")
+        entry = self.track_service.select_by_day(day)
+        self.assertIsNotNone(entry)
+        self.assertIsInstance(entry, list)
+
+
+class TestAuraUser(unittest.TestCase):
+    aura_user = None
+
+    def setUp(self):
+        self.aura_user = AuraUser()
+
+    def test_add_user(self):
+        username = "user"
+        password = "password"
+        role = "admin"
+
+        login_cnt = len(self.aura_user.getLogins())
+
+        # insert user
+        key = self.aura_user.insertUser(username, password, role)
+
+        self.assertGreaterEqual(len(self.aura_user.getLogins()), login_cnt)
+
+        # selecting user and check data
+        user = self.aura_user.getUserByKey(key)
+
+        self.assertEqual(user["username"], username)
+        # TODO: no encrypted storage.., but usermgm not really in use
+        self.assertEqual(user["password"], password)
+        self.assertEqual(user["role"], role)
+
+
+class TestLQSComm(unittest.TestCase):
+    comm = None
+
+    def setUp(self):
+        # wosn do passiert?
+        p = AuraConfig().config
+
+        self.comm = LiquidSoapCommunicator(p)
+        self.comm.scheduler = AuraScheduler(p)
+
+        self.comm.init_player()
+
+
+    def test_get_active_channel(self):
+        active_channel = self.comm.get_active_channel()
+
+        print(active_channel)
+
+
+
+if __name__ == '__main__':
+    unittest.main()
\ No newline at end of file