Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • aura/engine
  • hermannschwaerzler/engine
  • sumpfralle/aura-engine
3 results
Show changes
Showing
with 4784 additions and 0 deletions
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
if not icecast_vorbis_metadata then
source := add(normalize=false, [amplify(0.00001, noise()), !source])
end
if bitrate == 24 or bitrate == 32 or bitrate == 48 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=-0.1, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=-0.1, channels = 1), mean(!source)))
end
elsif bitrate == 64 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0, channels = 1), mean(!source)))
end
elsif bitrate == 96 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.2, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.2, channels = 1), mean(!source)))
end
elsif bitrate == 128 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.4, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.4, channels = 1), mean(!source)))
end
elsif bitrate == 160 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.5, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.5, channels = 1), mean(!source)))
end
elsif bitrate == 192 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.6, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.6, channels = 1), mean(!source)))
end
elsif bitrate == 224 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.7, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.7, channels = 1), mean(!source)))
end
elsif bitrate == 256 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.8, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.8, channels = 1), mean(!source)))
end
elsif bitrate == 320 then
if stereo then
ignore(output_icecast_stereo(%vorbis(quality=0.9, channels = 2), !source))
else
ignore(output_icecast_mono(%vorbis(quality=0.9, channels = 1), mean(!source)))
end
end
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
if bitrate == 24 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 24, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 24, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 32 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 32, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 32, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 48 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 48, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 48, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 64 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 64, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 64, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 96 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 96, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 96, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 128 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 128, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 128, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 160 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 160, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 160, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 192 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 192, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 192, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 224 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 224, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 224, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 256 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 256, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 256, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
elsif bitrate == 320 then
if stereo then
ignore(output_icecast_stereo(%opus(bitrate = 320, channels = 2, signal="music", application="audio", complexity=10, vbr="constrained"), !source))
else
ignore(output_icecast_mono(%opus(bitrate = 320, channels = 1, signal="music", application="audio", complexity=10, vbr="constrained"), mean(!source)))
end
end
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#################
# Read INI File #
#################
debug = false
def read_ini(file)
# read ini file
settings_file_content = get_process_lines("cat "^file )
# one entry stored as ["setting_name", "setting"]
settings_map = list.map(string.split(separator="="), settings_file_content)
def filter_pair(setting_pair) =
if debug then
print(" +++ IN FILTER_PAIR +++")
end
# get head of settings_pair
setting_name = list.hd(default="", setting_pair)
if debug then
print(" -- setting_name:")
print(setting_name)
end
# settings in ini are with '"', so read them with high comma
setting_unfiltered = list.nth(default="", setting_pair, 1)
if debug then
print(" -- setting_unfiltered:")
print(setting_unfiltered)
end
# filter high comma out. why the hell comes an array ["1", setting] returned?
# the filter patterns are perl regex
setting = string.extract(pattern='"(.*)"', setting_unfiltered)
if debug then
print(" -- setting ( after string.extract):")
print(setting['1'])
end
filtered_pair = [(setting_name, setting['1'])]
if debug then
print(" -- filter_pair returning: --")
print(filtered_pair)
end
# return filtered_pair
filtered_pair
end
def filter_map(filled_list, next_element) =
# if debug then
# print(" +++ IN FILTER_MAP +++")
# print(" .. length of filled_list: ")
# print(list.length(filled_list))
# print(" .. next_element")
# print(next_element)
# end
# the normal case: settingname and its setting
if list.length(next_element) >= 2 then
if debug then
print(" ===> LENGTH list to insert in settings_list is equal TWO! <===")
print(" -- next_element")
print(next_element)
end
setting_pair = filter_pair(next_element)
# add settings_pair to liquidsoap list
#list.add(setting_pair, filled_list)
list.append(setting_pair, filled_list)
else
if list.length(next_element) >= 1 then
if debug then
print(" ===> LENGTH of list to insert in settings_list is equal ONE! <===")
print(" -- next_element")
print(next_element)
end
#
list.append([(list.hd(default="",next_element), "")], filled_list)
else
if debug then
print(" ===> LENGTH of list to insert in settings_list is greater then two or less than one <===")
print(" -- next_element")
print(next_element)
end
# return the list as is
filled_list
end
end
end
# get install dir
pwd = get_process_lines("pwd") # returns an array
absolute_path = list.hd(default="", pwd) # grabs head of array (pwd only returns one line which is our desired result)
install_dir = '"' ^ path.dirname(path.dirname(absolute_path)) ^ '"' # pre and append " for filter_pair function. otherwise this setting would be lost through regex and fold function below would be imho much more difficult to read
# add installdir to the loaded settings from ini file
#settings_map = list.add(["install_dir", install_dir], settings_map)
# list.fold param description:
# f(a,b,c)
# a) a function which is fold
# b) the list which is to be filled (optional with already containing elements)
# c) from this list every element is taken, function applied and inserted into b
# fold function without pre and appended "
settings = list.fold(filter_map, [("install_dir", install_dir)], settings_map)
#settings = list.fold(filter_map, [], settings_map)
if debug then
print(settings)
end
settings
end
if debug then
ini = read_ini("/etc/aura/engine.ini")
print(ini)
print(list.assoc(default="", "install_dir", ini))
end
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
def icy_update(v) =
# Parse the argument
l = string.split(separator=",",v)
def split(l,v) =
v = string.split(separator="=",v)
if list.length(v) >= 2 then
list.append(l,[(list.nth(v,0,default=""),list.nth(v,1,default=""))])
else
l
end
end
meta = list.fold(split,[],l)
# Update metadata
if s0_enable == true then
icy.update_metadata(
mount=s0_mount,
user=s0_user,
password=s0_pass,
host=s0_host,
port=s0_port,
meta
)
end
if s1_enable == true then
icy.update_metadata(
mount=s1_mount,
user=s1_user,
password=s1_pass,
host=s1_host,
port=s1_port,
meta
)
end
if s2_enable == true then
icy.update_metadata(
mount=s2_mount,
user=s2_user,
password=s2_pass,
host=s2_host,
port=s2_port,
meta
)
end
if s3_enable == true then
icy.update_metadata(
mount=s3_mount,
user=s3_user,
password=s3_pass,
host=s3_host,
port=s3_port,
meta
)
end
if s4_enable == true then
icy.update_metadata(
mount=s4_mount,
user=s4_user,
password=s4_pass,
host=s4_host,
port=s4_port,
meta
)
end
"Done!"
end
server.register("update", namespace="metadata",
description="Update metadata",
usage="update title=foo,album=bar, ...",
icy_update)
# shutdown server function
#server.register(namespace='server',
# description="shutdown server",
# usage="stop",
# "stop",
# fun(x,y) -> shutdown )
#
# to reduce complexity of lqs => query 'mixer.inputs' over socket and parse it in python
#server.register(namespace="auraengine",
# "enabled_lineins",
# fun (s) -> begin
# log("auraengine.enabled_lineins")
# "0:#{!linein_0_enabled}, 1:#{!linein_1_enabled}, 2:#{!linein_2_enabled}, 3:#{!linein_3_enabled}, 4:#{!linein_4_enabled}"
# end
#)
#server.register(namespace="auraengine",
# "enabled_lineouts",
# fun(s) -> begin
# log("auraengine.enabled_lineouts")
# "0:#{!lineout_0_enabled}, 1:#{!lineout_1_enabled}, 2:#{!lineout_2_enabled}, 3:#{!lineout_3_enabled}, 4:#{!lineout_4_enabled}"
# end
#)
# are outgoing streams connected?
server.register(namespace="auraengine",
description="returns if outgoing streams are connected",
usage="out_streams_connected",
"out_streams_connected",
fun (s) -> begin
log("streams.connection_status")
"0:#{!s0_connected}, 1:#{!s1_connected}, 2:#{!s2_connected}, 3:#{!s3_connected}, 4:#{!s4_connected}"
end
)
# return a state of the inputs/outputs of the soundserver as JSON
server.register(namespace = "auraengine",
description="returns enabled lineouts/lineins, connected outgoing streams, and recorder. Also returns fallbacksettings.",
usage="state",
"state",
fun(s) -> begin
log("auraengine.state")
ret = '{'
ret = ret^'"streams": {'
ret = ret^'"stream_0": {"enabled": #{s0_enable}, "connected": #{!s0_connected}},'
ret = ret^'"stream_1": {"enabled": #{s1_enable}, "connected": #{!s1_connected}},'
ret = ret^'"stream_2": {"enabled": #{s2_enable}, "connected": #{!s2_connected}},'
ret = ret^'"stream_3": {"enabled": #{s3_enable}, "connected": #{!s3_connected}},'
ret = ret^'"stream_4": {"enabled": #{s4_enable}, "connected": #{!s4_connected}}'
ret = ret^'},'
ret = ret^'"recorder": {'
ret = ret^'"recorder_0": {"enabled": #{r0_enable}, "recording": #{!r0_is_recording}},'
ret = ret^'"recorder_1": {"enabled": #{r1_enable}, "recording": #{!r1_is_recording}},'
ret = ret^'"recorder_2": {"enabled": #{r2_enable}, "recording": #{!r2_is_recording}},'
ret = ret^'"recorder_3": {"enabled": #{r3_enable}, "recording": #{!r3_is_recording}},'
ret = ret^'"recorder_4": {"enabled": #{r4_enable}, "recording": #{!r4_is_recording}}'
ret = ret^'},'
ret = ret^'"linein": {'
ret = ret^'"linein_0": {"enabled": #{a0_in != ""}},'
ret = ret^'"linein_1": {"enabled": #{a1_in != ""}},'
ret = ret^'"linein_2": {"enabled": #{a2_in != ""}},'
ret = ret^'"linein_3": {"enabled": #{a3_in != ""}},'
ret = ret^'"linein_4": {"enabled": #{a4_in != ""}}'
ret = ret^'},'
ret = ret^'"lineout": {'
ret = ret^'"lineout_0": {"enabled": #{a0_out != ""}},'
ret = ret^'"lineout_1": {"enabled": #{a1_out != ""}},'
ret = ret^'"lineout_2": {"enabled": #{a2_out != ""}},'
ret = ret^'"lineout_3": {"enabled": #{a3_out != ""}},'
ret = ret^'"lineout_4": {"enabled": #{a4_out != ""}}'
ret = ret^'}'
ret = ret^'}'
ret
# outgoing streams enabled?
#ret = "stream_0_enabled:#{!s0_enable}, stream_1_enabled:#{!s1_enable}, stream_2_enabled:#{!s2_enable}, stream_3_enabled:#{!s3_enable}, stream_4_enabled:#{!s4_enable}, "
# outgoing recorder enabled
#ret = ret^"recorder_0_enabled:#{r0_enable}, recorder_1_enabled:#{r1_enable}, recorder_2_enabled:#{r2_enable}, recorder_3_enabled:#{r3_enable}, recorder_4_enabled:#{r4_enable}, "
#ret = ret^"linein_0_enabled:#{a0_in != ''}, linein_1_enabled:#{a1_in != ''}, linein_2_enabled:#{a2_in != ''}, linein_3_enabled:#{a3_in != ''}, linein_4_enabled:#{a4_in != ''}, "
#ret = ret^"lineout_0_enabled:#{a0_out != ''}, lineout_1_enabled:#{a1_out != ''}, lineout_2_enabled:#{a2_out != ''}, lineout_3_enabled:#{a3_out != ''}, lineout_4_enabled:#{a4_out != ''}, "
#ret = ret^"fallback_max_blank:#{fallback_max_blank}, fallback_min_noise:#{fallback_min_noise}, fallback_threshold:#{fallback_threshold}"
end
)
def fadeTo(source_number) =
if source_number == "" then
print(source_number)
"Usage: mixer.fadeto <source nb> #{source_number}"
else
r = server.execute("mixer.select #{source_number} true")
print(r)
"Donee!"
end
end
# enable fadeTo for the mixer
server.register(namespace = "mixer",
description = "is fading from one mixer input to another",
usage = "fadeto <source number>",
"fadeto",
fadeTo
)
ignore(fade_in_time)
ignore(fade_out_time)
# Activate a source by selecting it and setting the volume to 100 (or vice versa)
def activate(p) =
params=string.split(separator=" ", p)
if list.length(params) < 2 then
print(p)
"Usage: mixer.activate <source nb> <true|false>"
else
source_number = list.nth(default="0", params, 0)
source_enable = list.nth(default="false", params, 1)
if source_enable == "true" then
r = server.execute("mixer.select #{source_number} true")
print(r)
r = server.execute("mixer.volume #{source_number} 100")
print(r)
else
r = server.execute("mixer.volume #{source_number} 0")
print(r)
r = server.execute("mixer.select #{source_number} false")
print(r)
end
"Done!"
end
end
server.register(namespace = "mixer",
description = "is selecting a source and setting the volume to 100",
usage = "activate <source nb> <true|false>",
"activate",
activate
)
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# READ INI FILE
%include "readini.liq"
default_config = "../../config/engine.ini"
config = list.hd(default=default_config, get_process_lines("ls /etc/aura/engine.ini"))
log("Config file used: #{config}")
ini = read_ini(config)
engine_config_folder = string.split(separator="/engine.ini", config)
engine_config_folder = list.nth(default="../../config/", engine_config_folder, 0)
# ALLOW LIQUIDSOAP RUN AS ROOT
lqs_allow_root = list.assoc(default="false", "liquidsoap_as_root", ini)
if lqs_allow_root == "true" then
set("init.allow_root", true)
end
# BASICS
set("console.colorize","always")
# TELNET SETTINGS
set("server.telnet", true)
set("server.telnet.bind_addr", "0.0.0.0")
set("server.telnet.port", 1234)
# LOG FILE SETTINGS
log_dir = list.assoc(default="", "logdir", ini)
set("log.file.path", "#{log_dir}/engine-lqs.log")
# SOCKET SETTINGS
set("server.socket", true)
set("server.socket.path", "./<script>.sock")
#set("request.grace_time",2.)
# SOUND CARD SETTINGS
#print(ini)
#a0_in = ini['input_device_0'][0]
a0_in = list.assoc(default="", "input_device_0", ini)
a1_in = list.assoc(default="", "input_device_1", ini)
a2_in = list.assoc(default="", "input_device_2", ini)
a3_in = list.assoc(default="", "input_device_3", ini)
a4_in = list.assoc(default="", "input_device_4", ini)
a0_out = list.assoc(default="", "output_device_0", ini)
a1_out = list.assoc(default="", "output_device_1", ini)
a2_out = list.assoc(default="", "output_device_2", ini)
a3_out = list.assoc(default="", "output_device_3", ini)
a4_out = list.assoc(default="", "output_device_4", ini)
input_stream_buffer = float_of_string(list.assoc(default="3.0", "input_stream_buffer", ini))
# FALLBACK SETTINGS
fallback_station_playlist_path = "#{engine_config_folder}/playlists/station-fallback-playlist.m3u"
fallback_station_playlist_path = list.assoc(default=fallback_station_playlist_path, "fallback_music_playlist", ini)
fallback_station_dir = list.assoc(default="/var/audio/station", "fallback_music_folder", ini)
fallback_station_dir_reload = int_of_string(list.assoc(default="300", "fallback_music_folder_reload", ini))
fallback_max_blank = float_of_string(list.assoc(default="", "fallback_max_blank", ini))
fallback_min_noise = float_of_string(list.assoc(default="", "fallback_min_noise", ini))
fallback_threshold = float_of_string(list.assoc(default="", "fallback_threshold", ini))
# FADING SETTINGS
fade_in_time = list.assoc(default="", "fade_in_time", ini) #int_of_string(list.assoc(default="", "fade_in_time", ini))
fade_out_time = list.assoc(default="", "fade_out_time", ini) #int_of_string(list.assoc(default="", "fade_out_time", ini))
# RECORDER SETTINGS
#rec_0_filetype = list.assoc(default="", "rec_0_filetype", ini)
#rec_1_filetype = list.assoc(default="", "rec_1_filetype", ini)
#rec_2_filetype = list.assoc(default="", "rec_2_filetype", ini)
#rec_3_filetype = list.assoc(default="", "rec_3_filetype", ini)
#rec_4_filetype = list.assoc(default="", "rec_4_filetype", ini)
# ALSA / pulse settings
soundsystem = list.assoc(default="", "soundsystem", ini)
use_alsa = soundsystem == "alsa"
use_jack = soundsystem == "jack"
if use_alsa then
frame_duration = float_of_string(list.assoc(default="", "frame_duration", ini))
frame_size = int_of_string(list.assoc(default="", "frame_size", ini))
alsa_buffer = int_of_string(list.assoc(default="", "alsa_buffer", ini))
alsa_buffer_length = int_of_string(list.assoc(default="", "alsa_buffer_length", ini))
alsa_periods = int_of_string(list.assoc(default="", "alsa_periods", ini))
if frame_duration > 0.0 then
print("setting frame.duration to #{frame_duration}s")
set("frame.duration", frame_duration)
end
if frame_size > 0 then
print("setting frame.size to #{frame_size}")
set("frame.audio.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
print("**************************************************************************************")
print(" AURA ENGINE - LIQUIDSOAP SETTINGS ")
print("**************************************************************************************")
print(" Engine Configuration Folder: #{engine_config_folder}")
print(" Station Fallback Playlist: #{fallback_station_playlist_path}")
print(" Station Fallback Directory: #{fallback_station_dir}")
print("**************************************************************************************")
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# Crossfade between tracks,
# taking the respective volume levels
# into account in the choice of the
# transition.
# @category Source / Track Processing
# @param ~start_next Crossing duration, if any.
# @param ~fade_in Fade-in duration, if any.
# @param ~fade_out Fade-out duration, if any.
# @param ~width Width of the volume analysis window.
# @param ~conservative Always prepare for
# a premature end-of-track.
# @param s The input source.
def crossfade (~start_next=5.,~fade_in=3.,
~fade_out=3., ~width=2.,
~conservative=false,s)
high = -20.
medium = -32.
margin = 4.
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="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
# If A and B and not too loud and close,
# fully cross-fade them.
a <= medium and
b <= medium and
abs(a - b) <= margin
then
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("Transition: crossed, fade-out.")
add(fade.out(sa),sb)
elsif
# Do not fade if it's already very low.
b >= a + margin and a <= medium and b <= high
then
log("Transition: crossed, no fade-out.")
add(sa,sb)
elsif
# Opposite as the previous one.
a >= b + margin and b >= medium and a <= high
then
log("Transition: crossed, fade-in.")
add(sa,fade.in(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: just sequencing.")
sequence([sa, sb])
end
end
cross(width=width, duration=start_next,
conservative=conservative,
transition,s)
end
# Custom crossfade to deal with jingles.
# def smarter_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="smarter_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)
# smart_crossfade(duration=start_next, fade_in=fade_in, fade_out=fade_out, width=width, 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(max_blank=fallback_max_blank, min_noise=fallback_min_noise, threshold=fallback_threshold, source)
else
source
end
# Tell the system when a new track is played
def do_meta(meta) =
filename = meta["filename"]
# artist = meta["artist"]
# title = meta["title"]
system('#{list.assoc(default="", "install_dir", ini)}/guru.py --on_play "#{filename}"')
end
source = on_metadata(do_meta, source)
log("channel created")
# Finally apply a smart crossfading
#smarter_crossfade(source)
crossfade(source)
end
def create_dynamic_playlist(next)
request.create(list.hd(default="", next))
end
def create_playlist() =
log("requesting next song for PLAYLIST")
result = get_process_lines('#{list.assoc(default="", "install_dir", ini)}/guru.py --get-next-file-for "playlist" --quiet')
create_dynamic_playlist(result)
end
def create_station_fallback() =
result = get_process_lines('#{list.assoc(default="", "install_dir", ini)}/guru.py --get-next-file-for station --quiet')
log("next song for STATION fallback is: #{result}")
create_dynamic_playlist(result)
end
def create_show_fallback() =
result = get_process_lines('#{list.assoc(default="", "install_dir", ini)}/guru.py --get-next-file-for show --quiet')
log("next song for SHOW fallback is: #{result}")
create_dynamic_playlist(result)
end
def create_timeslot_fallback() =
result = get_process_lines('#{list.assoc(default="", "install_dir", ini)}/guru.py --get-next-file-for timeslot --quiet')
log("next song for TIMESLOT fallback is: #{result}")
create_dynamic_playlist(result)
end
# create fallbacks
timeslot_fallback = fallback_create(skip=true, "timeslot_fallback", create_timeslot_fallback)
station_fallback = fallback_create(skip=true, "station_fallback", create_station_fallback)
show_fallback = fallback_create(skip=true, "show_fallback", create_show_fallback)
\ No newline at end of file
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# %include "readini.liq"
# ini = read_ini("/etc/aura/engine.ini")
# %include "settings.liq"
# TELNET SETTINGS
# set("server.telnet", true)
# set("server.telnet.bind_addr", "0.0.0.0")
# set("server.telnet.port", 2345)
inst = if argv(1) != "" then string_of(argv(1)) else 'record' end
instance = ref inst
audiobase = if !instance == 'record' then list.assoc(default="", "audiobase", ini) else list.assoc(default="", "altaudiobase", ini) end
rec_filetype = list.assoc(default="", "rec_filetype", ini)
filenamepattern = ref audiobase^"/%Y-%m-%d/%Y-%m-%d-%H-%M.flac"
# 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
is_record_active = ref false
# Stop dump - wir definieren die Funktion, die stop_f ausführt
def stop_dump() =
f = !stop_f
f ()
end
def on_start()
recordingfile := list.hd(default="", get_process_lines("date +#{!filenamepattern}"))
end
# Wav header fixen und ggf. die Aufzeichnung beenden
def on_close(filename)
# es darf wieder aufgenommen werden
is_record_active := false
# if list.assoc(default="", "rec_filetype", ini) == 'wav'
# # Korrekten WAV-Header schreiben
# system("qwavheaderdump -F #{filename}")
# Naechsten Dateinamen vormerken
recordingfile := list.hd(default="", get_process_lines("date +#{!filenamepattern}"))
end
# Der input wie oben definiert
def get_input()
output_source
# input.alsa()
end
def get_output()
input = get_input()
d = int_of_string(list.assoc(default="", "rec_duration", ini))
if rec_filetype == 'flac' then
log("output file type is FLAC")
output.file(
id="recorder",
%flac(samplerate=44100, channels=2, compression=5, bits_per_sample=16),
perm = 0o664,
on_start=on_start,
!filenamepattern,
on_close=on_close,
reopen_when={ if !instance == 'record' then int_of_float(gettimeofday()/60.) mod 30 == 0 else false end },
input
)
else
# record in WAV
log("output file type is WAV")
output.file(
id="recorder",
%wav(stereo=true, channels=2, samplesize=16, header=true),
perm = 0o664,
on_start=on_start,
!filenamepattern,
on_close=on_close,
reopen_when={ if !instance == 'record' then int_of_float(gettimeofday()/60.) mod d == 0 else false end },
input
)
end
end
# Funktion gibt Auskunft welches File aktuell ist und wieviel Prozent bereits aufgenommen werden
def currecording()
curfile = !recordingfile
if curfile != "" then
percent_done = default="", get_process_lines("echo $(($(stat -c%s "^curfile^")/3174777))"))
"#{curfile}, #{percent_done}%"
else
"Nothing is being recorded now"
end
end
#Funktion zum Start der Aufzeichnung
def start_dump() =
log('start dump')
# don't record twice is_record_active
if !is_record_active == false then
is_record_active := true
log('starting to record')
record = get_output()
log('record defined')
# 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
else
log("recorder already active")
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"))
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# define file name pattern
filenamepattern = ref audiobase^"/%Y-%m-%d/%Y-%m-%d-%H-%M.flac"
# Der aktuelle Dateiname für die Aufnahme
recordingfile = ref ""
# bewahre uns davor, dass zweimal gleichzeitig die gleiche Date aufgenommen wird
is_record_active = ref false
#wir definieren eine Referenz für die Stop-Funktion, die aber bisher noch nichts tun kann
stop_f = ref (fun () -> ())
def start_wav_output(recfile, filenamepattern, recorder_number)
duration = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_duration", ini))
channels = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_channels", ini))
samplesize = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_samplesize", ini))
ignore(duration)
ignore(channels)
ignore(samplesize)
# def on_start()
# recfile := list.hd(default="", get_process_lines("date +#{!filenamepattern}"))
# end
# def on_close(filename)
# recordingfile := list.hd(default="", get_process_lines("date +#{!filenamepattern}"))
# end
print(channels)
print(samplesize)
#out_wav = output.file(id="recorder", perm = 0o664, on_start=on_start, on_close=on_close, reopen_when={ int_of_float(gettimeofday()/60.) mod duration == 0 })
#out_wav(%wav(stereo=true, channels=2, samplesize=8, header=true, !filenamepattern, output_source)
#ignore(out_wav)
output.dummy(id="wav_dummy_recording", blank())
# if channels == 2 then
# output.file(id="recorder", %wav(stereo=true, channels=2, samplesize=8, header=true), perm = 0o664, on_start=on_start, on_close=on_close, reopen_when={ int_of_float(gettimeofday()/60.) mod duration == 0 }, !filenamepattern, audio_to_stereo(output_source))
## out_wav(output_source)
# else
# output.file(id="recorder", %wav(stereo=true, channels=1, samplesize=8, header=true), perm = 0o664, on_start=on_start, on_close=on_close, reopen_when={ int_of_float(gettimeofday()/60.) mod duration == 0 }, !filenamepattern, output_source)
# out_wav(output_source)
# end
#if channels == 2 then
# if samplesize < 12 then
# ignore(out_wav(%wav(stereo=true, channels=2, samplesize=8, header=true), !filenamepattern, output_source))
# else
# ignore(out_wav(%wav(stereo=true, channels=2, samplesize=16, header=true), !filenamepattern, output_source))
# end
#else
# if samplesize < 12 then
# ignore(out_wav(%wav(stereo=true, channels=1, samplesize=8, header=true), !filenamepattern, output_source))
# else
# ignore(out_wav(%wav(stereo=true, channels=1, samplesize=16, header=true), !filenamepattern, output_source))
# end
#end
end
def start_flac_output(recorder_number, filenamepattern, duration)
# duration = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_samplerate", ini))
# samplerate = list.assoc(default="", "rec_"^recorder_number^"_samplerate", ini)
channels = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_channels", ini))
# compression = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_compression", ini))
# bits_per_sample = int_of_string(list.assoc(default="", "rec_"^recorder_number^"_bits_per_sample", ini))
#output.file(id="recorder", %flac(samplerate=44100, channels=1, compression=7, bits_per_sample=32), perm = 0o664, on_start=on_start, !filenamepattern, on_close=on_close, reopen_when={ int_of_float(gettimeofday()/60.) mod duration == 0 }, output_source)
recfile = ref ''
def on_start()
recfile := list.hd(default="", get_process_lines("date +#{filenamepattern}"))
end
def on_close(filename)
recfile := list.hd(default="", get_process_lines("date +#{filenamepattern}"))
end
# dumbass liquidsoap cannot handle one output definition for mono and stereo
output_filesystem_mono = output.file(id="recorder", perm = 0o664, on_start=on_start, on_close=on_close, reopen_when={ int_of_float(gettimeofday()/60.) mod duration == 0 })
output_filesystem_stereo = output.file(id="recorder", perm = 0o664, on_start=on_start, on_close=on_close, reopen_when={ int_of_float(gettimeofday()/60.) mod duration == 0 })
ignore(output_filesystem_mono)
ignore(output_filesystem_stereo)
if channels == 2 then
output.dummy(id="flac_dummy_recording_stereo", blank())
# output_stereof(%flac(samplerate=44100, channels=1, compression=1, bits_per_sample=16), !filenamepattern, output_source)
else
output.dummy(id="flac_dummy_recording_mono", blank())
# output_icecast_monof(%flac(samplerate=44100, channels=1, compression=1, bits_per_sample=16), !filenamepattern, output_source)
end
end
def enable_stop_function(record)
# 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
def set_recorder_output(rec_filetype, recorder_number)
# flac output
if rec_filetype == 'flac' then
log("output file type is FLAC")
record = start_flac_output(recorder_number)
enable_stop_function(record)
# WAV output
else
log("output file type is WAV")
record = start_wav_output(recorder_number)
enable_stop_function(record)
end
end
# shows current file and how many bytes were written so far
def currecording()
curfile = !recordingfile
if curfile != "" then
bytes_written = list.hd(default="", get_process_lines("echo $(($(stat -c%s "^curfile^")))"))
"#{curfile}, #{bytes_written}B"
else
""
end
end
\ No newline at end of file
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
def get_icecast_mp3_stream(number)
stream_bitrate = int_of_string(list.assoc(default="", "stream_#{number}_bitrate", ini))
stream_mountpoint = list.assoc(default="", "stream_#{number}_mountpoint", ini)
stream_host = list.assoc(default="", "stream_#{number}_host", ini)
stream_port = int_of_string(list.assoc(default="", "stream_#{number}_port", ini))
stream_name = list.assoc(default="", "stream_#{number}_name", ini)
stream_url = list.assoc(default="", "stream_#{number}_url", ini)
stream_genre = list.assoc(default="", "stream_#{number}_genre", ini)
stream_description = list.assoc(default="", "stream_#{number}_description", ini)
stream_user = list.assoc(default="", "stream_#{number}_user", ini)
stream_password = list.assoc(default="", "stream_#{number}_password", ini)
if stream_bitrate == 24 then
icecast_stream = output.icecast(%mp3(bitrate = 24, samplerate = 22050), mount=stream_mountpoint, host=stream_host, port=stream_port, name=stream_name, url=stream_url, genre=stream_genre, description=stream_description, user=stream_user, password=stream_password, icy_metadata="true", fallible=true, buffer(output_source))
clock.assign_new(id="stream_#{number}",
[icecast_stream]
)
icecast_stream
else
icecast_stream = output.icecast(%mp3(samplerate = 44100), mount=stream_mountpoint, host=stream_host, port=stream_port, name=stream_name, url=stream_url, genre=stream_genre, description=stream_description, user=stream_user, password=stream_password, icy_metadata="true", fallible=true, buffer(output_source))
clock.assign_new(id="stream_#{number}",
[icecast_stream]
)
icecast_stream
end
end
def get_icecast_ogg_stream(number)
stream_quality = float_of_string(list.assoc(default="", "stream_#{number}_quality", ini))
stream_mountpoint = list.assoc(default="", "stream_#{number}_mountpoint", ini)
stream_host = list.assoc(default="", "stream_#{number}_host", ini)
stream_port = int_of_string(default="", list.assoc("stream_#{number}_port", ini))
stream_name = list.assoc(default="", "stream_#{number}_name", ini)
stream_url = list.assoc(default="", "stream_#{number}_url", ini)
stream_genre = list.assoc(default="", "stream_#{number}_genre", ini)
stream_description = list.assoc(default="", "stream_#{number}_description", ini)
stream_user = list.assoc(default="", "stream_#{number}_user", ini)
stream_password = list.assoc(default="", "stream_#{number}_password", ini)
if stream_quality >= 0.5 then
icecast_stream = output.icecast(%vorbis(quality = 1.0), mount=stream_mountpoint, host=stream_host, port=stream_port, name=stream_name, url=stream_url, genre=stream_genre, description=stream_description, user=stream_user, password=stream_password, icy_metadata="true", fallible=true, buffer(output_source))
clock.assign_new(id="stream_#{number}",
[icecast_stream]
)
icecast_stream
else
icecast_stream = output.icecast(%vorbis(quality = 0.1), mount=stream_mountpoint, host=stream_host, port=stream_port, name=stream_name, url=stream_url, genre=stream_genre, description=stream_description, user=stream_user, password=stream_password, icy_metadata="true", fallible=true, buffer(output_source))
clock.assign_new(id="stream_#{number}",
[icecast_stream]
)
icecast_stream
end
end
def get_harbour_mp3_stream(number)
stream_bitrate = int_of_string(list.assoc(default="", "stream_#{number}_bitrate", ini))
stream_user = list.assoc(default="", "stream_#{number}_user", ini)
stream_password = list.assoc(default="", "stream_#{number}_password", ini)
stream_port = int_of_string(list.assoc(default="", "stream_#{number}_port", ini))
stream_url = list.assoc(default="", "stream_#{number}_url", ini)
stream_mountpoint = list.assoc(default="", "stream_#{number}_mountpoint", ini)
if stream_bitrate == 24 then
harbourstream = output.harbor(%mp3(bitrate = 24, samplerate = 22050), user=stream_user, password=stream_password, id="stream", port=stream_port, url=stream_url, mount=stream_mountpoint, icy_metadata="true", fallible=true, buffer(output_source))
clock.assign_new(id="stream_#{number}",
[harbourstream]
)
harbourstream
else
harbourstream = output.harbor(%mp3, user=stream_user, password=stream_password, id="stream", port=stream_port, url=stream_url, mount=stream_mountpoint, icy_metadata="true", fallible=true, buffer(output_source))
clock.assign_new(id="stream_#{number}",
[harbourstream]
)
harbourstream
end
end
def get_stream(number)
stream = list.assoc(default="", "stream_#{number}", ini)
stream_type = list.assoc(default="", "stream_#{number}_type", ini)
stream_format = list.assoc(default="", "stream_#{number}_format", ini)
# is stream enabled?
if stream == "y" then
log("activating stream #{number}")
if stream_type == "icecast" then
log("its an icecast stream")
if stream_format == "mp3" then
log("filled with mp3")
get_icecast_mp3_stream(number)
elsif stream_format == "vorbis" then
log("filled with ogg")
get_icecast_ogg_stream(number)
else
output.dummy(id="no_valid_stream_format_DUMMY_ICECAST", blank())
end
elsif stream_type == "harbor" then
log("its an harbor stream")
if stream_format == "mp3" then
get_harbour_mp3_stream(number)
else
output.dummy(id="no_valid_stream_format_DUMMY_HARBOUR", blank())
end
else
output.dummy(id="no_valid_stream_type_DUMMY", blank())
end
else
output.dummy(id="no_stream_enabled_DUMMY", blank())
end
end
def set_streams()
stream_0 = get_stream(0)
#stream_1 = get_stream(1)
#stream_2 = get_stream(2)
#stream_3 = get_stream(3)
#stream_4 = get_stream(4)
end
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import datetime
import urllib
import logging
import json
import threading
import platform
import requests
from enum import Enum
from socket import socket, AF_INET, SOCK_DGRAM, SOL_SOCKET, SO_BROADCAST
import meta
from src.base.config import AuraConfig
from src.base.utils import SimpleUtil as SU
from src.base.mail import AuraMailer
# Exceptions
class EngineMalfunctionException(Exception):
pass
# Status Codes
class MonitorResponseCode(Enum):
OK = "OK"
INVALID_STATE = "INVALID-STATE"
class AuraMonitor:
"""
Engine Monitoring is in charge of:
- Checking the overall status of all components and external API endpoints
- Checking the vital parts, which are minimal requirements for running the engine
- Sending a heartbeat to a defined server via socket
"""
logger = None
engine = None
mailer = None
status = None
already_invalid = None
engine_id = None
heartbeat_server = None
heartbeat_port = None
heartbeat_frequency = None
heartbeat_socket = None
heartbeat_running = None
def __init__(self, engine):
"""
Initialize Monitoring
"""
self.logger = logging.getLogger("AuraEngine")
self.config = AuraConfig.config()
self.engine = engine
self.mailer = AuraMailer(self.config)
self.status = dict()
self.status["engine"] = dict()
self.status["lqs"] = dict()
self.status["api"] = dict()
self.status["api"]["steering"] = dict()
self.status["api"]["tank"] = dict()
self.status["api"]["engine"] = dict()
self.already_invalid = False
# Register as an engine plugin
self.engine.plugins["monitor"] = self
# Heartbeat settings
self.heartbeat_running = False
self.heartbeat_server = self.config.get("heartbeat_server")
self.heartbeat_port = self.config.get("heartbeat_port")
self.heartbeat_frequency = self.config.get("heartbeat_frequency")
self.heartbeat_socket = socket(AF_INET, SOCK_DGRAM)
self.engine_id = self.get_engine_id()
#
# EVENTS
#
def on_boot(self):
"""
Called when the engine is booting.
"""
# Start Monitoring
is_valid = self.has_valid_status(False)
status = self.get_status()
self.logger.info("Status Monitor:\n%s" % json.dumps(status, indent=4))
if not is_valid:
self.logger.info("Engine Status: " + SU.red(status["engine"]["status"]))
self.post_health(status, False)
raise EngineMalfunctionException
else:
self.logger.info("Engine Status: " + SU.green("[OK]"))
self.post_health(status, True)
def on_sick(self, data):
"""
Called when the engine is in some unhealthy state.
"""
self.post_health(data, False)
def on_resurrect(self, data):
"""
Called when the engine turned healthy again after being sick.
"""
self.post_health(data, True)
#
# PUBLIC METHODS
#
def get_status(self):
"""
Retrieves the current monitoring status.
"""
return self.status
def has_valid_status(self, update_vitality_only):
"""
Checks if the current status is valid to run engine. By default it
does not request new status information, rather using the cached one.
To request new data either call `get_status()` before or use the
`update_vital` parameter.
Args:
update_vitality_only (Boolean): Refreshes only the vital parts required for the heartbeat
"""
is_valid = False
if update_vitality_only:
self.update_vitality_status()
else:
self.update_status()
try:
if self.status["lqs"]["active"] \
and self.status["lqs"]["mixer"]["in_filesystem_0"] \
and self.status["audio_source"]["exists"]:
self.status["engine"]["status"] = MonitorResponseCode.OK.value
is_valid = True
else:
self.status["engine"]["status"] = MonitorResponseCode.INVALID_STATE.value
except Exception as e:
self.logger.error("Exception while validating engine status: " + str(e))
self.status["engine"]["status"] = MonitorResponseCode.INVALID_STATE.value
return is_valid
#
# PRIVATE METHODS
#
def post_health(self, data, is_healthy):
"""
Post unhealthy state info to Engine API.
"""
body = dict()
body["log_time"] = datetime.datetime.now()
body["is_healthy"] = is_healthy
body["details"] = json.dumps(data, default=str)
json_data = json.dumps(body, default=str)
url = self.config.get("api_engine_store_health")
url = url.replace("${ENGINE_NUMBER}", str(self.config.get("api_engine_number")))
headers = {'content-type': 'application/json'}
r = requests.post(url, data=json_data, headers=headers)
if r.status_code == 204:
self.logger.info("Successfully posted healthy=%s state to Engine API!" % is_healthy)
else:
self.logger.error("HTTP %s | Error while pushing health state to Engine API: %s" % (r.status_code, str(r.json())))
def update_status(self):
"""
Requests the current status of all components
"""
self.status["engine"]["version"] = meta.__version__
self.engine.player.connector.enable_transaction()
self.status["lqs"]["version"] = self.engine.version()
self.status["lqs"]["uptime"] = self.engine.uptime()
self.status["lqs"]["io"] = self.get_io_state()
self.status["lqs"]["mixer"] = self.engine.player.mixer.mixer_status()
self.status["lqs"]["mixer_fallback"] = self.engine.player.mixer_fallback.mixer_status()
self.engine.player.connector.disable_transaction()
self.status["api"]["steering"]["url"] = self.config.get("api_steering_status")
self.status["api"]["steering"]["available"] = self.validate_url_connection(self.config.get("api_steering_status"))
self.status["api"]["tank"]["url"] = self.config.get("api_tank_status")
self.status["api"]["tank"]["available"] = self.validate_url_connection(self.config.get("api_tank_status"))
self.status["api"]["tank"]["status"] = self.get_url_response(self.config.get("api_tank_status"))
self.status["api"]["engine"]["url"] = self.config.get("api_engine_status")
self.status["api"]["engine"]["available"] = self.validate_url_connection(self.config.get("api_engine_status"))
self.update_vitality_status()
def update_vitality_status(self):
"""
Refreshes the vital status info which are required for the engine to survive.
"""
self.engine.player.connector.enable_transaction()
self.status["lqs"]["active"] = self.engine.is_connected()
self.engine.player.connector.disable_transaction()
self.status["audio_source"] = self.validate_directory(self.config.get("audio_source_folder"))
# After first update start the Heartbeat Monitor
if not self.heartbeat_running:
self.heartbeat_running = True
if self.config.get("heartbeat_frequency") > 0:
self.heartbeat()
def heartbeat(self):
"""
Every `heartbeat_frequency` seconds the current vitality status is checked. If it's okay,
a heartbeat is sent to the configured server.
"""
if self.has_valid_status(True):
self.heartbeat_socket.sendto(str.encode("OK"), (self.heartbeat_server, self.heartbeat_port))
# Engine resurrected into normal state
if self.already_invalid:
self.already_invalid = False
status = json.dumps(self.get_status())
self.logger.info(SU.green("OK - Engine turned back into some healthy state!")+"\n"+str(status))
self.mailer.send_admin_mail( \
"OK - Engine turned back into some HEALTHY STATE!", \
"Things seem fine again at '%s':\n\n%s" % (self.engine_id, status))
# Route call of event via event dispatcher to provide ability for additional hooks
self.engine.event_dispatcher.on_resurrect(status)
else:
# Engine turned into invalid state
if not self.already_invalid:
self.already_invalid = True
status = json.dumps(self.get_status())
self.logger.critical(SU.red("Engine turned into some INVALID STATE!")+"\n"+str(status))
self.mailer.send_admin_mail( \
"ERROR - Engine turned into some INVALID STATE!", \
"There's an issue with Aura Engine '%s':\n\n%s" % (self.engine_id, status))
# Route call of event via event dispatcher to provide ability for additional hooks
self.engine.event_dispatcher.on_sick(status)
threading.Timer(self.config.get("heartbeat_frequency"), self.heartbeat).start()
def get_io_state(self):
"""
Retrieves all input and outputs provided by the engine.
"""
ios = self.engine.engine_state()
try:
ios = ios.replace('"connected":', '"connected": ""')
ios = json.loads(ios, strict=False)
return ios
except Exception as e:
self.logger.warn("Got invalid JSON from Liquidsoap - " + str(e))
return MonitorResponseCode.INVALID_STATE.value
def validate_url_connection(self, url):
"""
Checks if connection to passed URL is successful.
"""
try:
request = urllib.request.Request(url)
response = urllib.request.urlopen(request)
response.read()
except Exception:
return False
return True
def validate_directory(self, dir_path):
"""
Checks if a given directory is existing and holds content
"""
status = dict()
status["path"] = dir_path
status["exists"] = os.path.exists(dir_path) and os.path.isdir(dir_path)
status["has_content"] = False
if status["exists"]:
status["has_content"] = any([True for _ in os.scandir(dir_path)])
return status
def get_url_response(self, url):
"""
Fetches JSON data from the given URL.
Args:
url (String): The API endpoint to call
Returns:
(dict[]): A Python object representing the JSON structure
"""
data = None
try:
request = urllib.request.Request(url)
response = urllib.request.urlopen(request)
data = response.read()
return json.loads(data, strict=False)
except (urllib.error.URLError, IOError, ValueError) as e:
self.logger.error("Error while connecting to URL '%s' - %s" % (url, e))
return MonitorResponseCode.INVALID_STATE.value
def get_engine_id(self):
"""
Retrieves a String identifier consisting of IP and Hostname to differentiate
the engine in mails and status broadcasts.
"""
host = platform.node()
return "%s (%s)" % (self.get_ip(), host)
def get_ip(self):
"""
Returns the IP of the Engine instance.
"""
try:
s = socket(AF_INET, SOCK_DGRAM)
s.setsockopt(SOL_SOCKET, SO_BROADCAST, 1)
s.connect(('<broadcast>', 0))
return s.getsockname()[0]
except:
self.logger.critical(SU.red("Error while accessing network via <broadcast>!"))
return "<UNKNOWN NETWORK>"
\ No newline at end of file
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
import logging
import requests
from collections import deque
from datetime import datetime, timedelta
from src.base.config import AuraConfig
from src.base.utils import SimpleUtil as SU
from src.core.resources import ResourceClass
from src.core.resources import ResourceUtil
from src.scheduling.fallback import FallbackType
class TrackServiceHandler():
"""
Sends the trackservice entry and studio clock information to the `engine-api` REST endpoint.
"""
logger = None
config = None
engine = None
playlog = None
def __init__(self, engine):
"""
Initialize.
"""
self.logger = logging.getLogger("AuraEngine")
self.config = AuraConfig.config()
self.engine = engine
self.playlog = Playlog(engine)
def on_timeslot(self, timeslot=None):
"""
Some new timeslot has just started.
"""
if timeslot:
self.logger.info(f"Active timeslot used for trackservice '{timeslot}'")
self.playlog.set_timeslot(timeslot)
def on_queue(self, entries):
"""
Items have been queued. They are stored to the local playlog, allowing later
matching and retrieval to augment meta-information.
Args:
entries ([PlaylistEntry]): The entries which got queued
"""
for entry in entries:
self.playlog.add(entry)
def on_play(self, entry):
"""
Some `PlaylistEntry` started playing. This is likely only a LIVE or STREAM entry.
"""
content_class = ResourceUtil.get_content_class(entry.get_content_type())
if content_class == ResourceClass.FILE:
# Files are handled by "on_metadata" called via Liquidsoap
return
diff = (entry.entry_start_actual - entry.entry_start).total_seconds()
self.logger.info("There's a difference of %s seconds between planned and actual start of the entry" % diff)
data = {}
data["track_start"] = entry.entry_start_actual
if entry.meta_data:
data["track_artist"] = entry.meta_data.artist
data["track_album"] = entry.meta_data.album
data["track_title"] = entry.meta_data.title
data["track_duration"] = entry.duration
data["track_num"] = entry.entry_num
data["track_type"] = content_class.numeric
data["playlist_id"] = entry.playlist.playlist_id
data["schedule_id"] = entry.playlist.schedule.schedule_id
data["show_id"] = entry.playlist.schedule.show_id
data["show_name"] = entry.playlist.schedule.show_name
data["log_source"] = self.config.get("api_engine_number")
self.store_trackservice(data)
self.store_clock_info(data)
def on_metadata(self, meta):
"""
Some metadata update was sent from Liquidsoap.
"""
data = {}
data["track_start"] = meta.get("on_air")
data["track_artist"] = meta.get("artist")
data["track_album"] = meta.get("album")
data["track_title"] = meta.get("title")
data["track_type"] = ResourceClass.FILE.numeric
#lqs_source = meta["source"]
if "duration" in meta:
duration = float(meta.get("duration"))
data["track_duration"] = int(duration)
else:
data["track_duration"] = 0
entry = self.playlog.resolve_entry(meta["filename"])
if entry:
# This is a playlog according to the scheduled playlist (normal or fallback)
data["track_num"] = entry.entry_num
data["playlist_id"] = entry.playlist.playlist_id
data["schedule_id"] = entry.playlist.schedule.schedule_id
data["show_id"] = entry.playlist.schedule.show_id
data["show_name"] = entry.playlist.schedule.show_name
else:
# This is a fallback playlog which wasn't scheduled actually (e.g. station fallback)
(past, timeslot, next) = self.playlog.get_timeslots()
if timeslot:
data = {**data, **timeslot}
data["log_source"] = self.config.get("api_engine_number")
data = SU.clean_dictionary(data)
self.store_trackservice(data)
self.store_clock_info(data)
def store_trackservice(self, data):
"""
Posts the given `PlaylistEntry` to the Engine API Playlog.
"""
data = SU.clean_dictionary(data)
self.logger.info("Posting playlog to Engine API...")
url = self.config.get("api_engine_store_playlog")
headers = {'content-type': 'application/json'}
body = json.dumps(data, indent=4, sort_keys=True, default=str)
response = requests.post(url, data=body, headers=headers)
if response.status_code != 204 or response.status_code != 204:
msg = f"Error while posting playlog to Engine API: {response.reason} (Error {response.status_code})\n"
self.logger.info(SU.red(msg) + response.content.decode("utf-8"))
def store_clock_info(self, data):
"""
Posts the current and next show information to the Engine API.
"""
current_playlist = None
if self.engine.scheduler:
current_playlist = self.engine.scheduler.get_active_playlist()
(past_timeslot, current_timeslot, next_timeslot) = self.playlog.get_timeslots()
data = dict()
data["engine_source"] = self.config.get("api_engine_number")
if current_playlist:
data["current_playlist"] = dict()
data["current_playlist"]["playlist_id"] = current_playlist.playlist_id
data["current_playlist"]["entries"] = []
for e in current_playlist.entries:
entry = dict()
entry["track_start"] = e.entry_start
if e.meta_data:
entry["track_artist"] = e.meta_data.artist
entry["track_album"] = e.meta_data.album
entry["track_title"] = e.meta_data.title
entry["track_num"] = e.entry_num
entry["track_duration"] = e.duration
content_class = ResourceUtil.get_content_class(e.get_content_type)
entry["track_type"] = content_class.numeric
entry = SU.clean_dictionary(entry)
data["current_playlist"]["entries"].append(entry)
if current_timeslot:
data["current_schedule"] = current_timeslot
if next_timeslot:
data["next_schedule"] = next_timeslot
data = SU.clean_dictionary(data)
self.logger.info("Posting clock info update to Engine API...")
url = self.config.get("api_engine_store_clock")
headers = {'content-type': 'application/json'}
body = json.dumps(data, indent=4, sort_keys=True, default=str)
response = requests.put(url, data=body, headers=headers)
if response.status_code != 204 or response.status_code != 204:
msg = f"Error while posting clock-info to Engine API: {response.reason} (Error {response.status_code})\n"
self.logger.info(SU.red(msg) + response.content.decode("utf-8"))
class Playlog:
"""
Playlog keeps a short history of currently playing entries. It stores the recent
active entries to a local cache `history` being able to manage concurrently playing entries.
It also is in charge of resolving relevant meta information of the currently playing entry for
the TrackService handler.
The records are stored in pre-formatted dictionary structure, allowing easy serialization when
posting them to the Engine API.
"""
config = None
logger = None
engine = None
history = None
previous_timeslot = None
current_timeslot = None
next_timeslot = None
def __init__(self, engine):
"""
Constructor
"""
self.config = AuraConfig.config()
self.logger = logging.getLogger("AuraEngine")
self.engine = engine
self.history = deque([None, None, None])
self.current_timeslot = {}
self.init_timeslot(None)
def init_timeslot(self, next_timeslot=None):
"""
Initializes the timeslot.
"""
data = {}
self.assign_fallback_playlist(data, None)
data["schedule_id"] = -1
data["show_id"] = -1
data["show_name"] = ""
if self.previous_timeslot:
data["schedule_start"] = self.previous_timeslot.get("schedule_end")
else:
data["schedule_start"] = datetime.now()
if next_timeslot:
data["schedule_end"] = next_timeslot.schedule_start
else:
# Fake the end, because the timeslot is actually not existing
data["schedule_end"] = datetime.now() + timedelta(hours=1)
self.current_timeslot = data
def set_timeslot(self, timeslot):
"""
Sets the current timeslot and proper default values if no timeslot is available.
Any previous timeslot is stored to `self.previous_timeslot` and the following one
to `self.next_timeslot`.
This method is protect by overwritting by multiple calls with the same timeslot.
Args:
timeslot (Timeslot): The current timeslot
"""
if timeslot and self.previous_timeslot:
if self.previous_timeslot.get("schedule_start") == timeslot.schedule_start:
return # Avoid overwrite by multiple calls in a row
data = {}
next_timeslot = self.engine.scheduler.get_next_timeslots(1)
if next_timeslot:
next_timeslot = next_timeslot[0]
else:
next_timeslot = None
# A valid timeslot from the scheduler is available
if timeslot:
self.assign_fallback_playlist(data, timeslot)
data["schedule_id"] = timeslot.schedule_id
data["schedule_start"] = timeslot.schedule_start
data["schedule_end"] = timeslot.schedule_end
data["show_id"] = timeslot.show_id
data["show_name"] = timeslot.show_name
data = SU.clean_dictionary(data)
# Any previous (fake) timeslots should get the proper end now
if not self.previous_timeslot:
self.current_timeslot["schedule_end"] = timeslot.schedule_start
self.previous_timeslot = self.current_timeslot
self.current_timeslot = data
# Defaults for a not existing timeslot
else:
self.init_timeslot(next_timeslot)
# A valid following timeslot is available
if next_timeslot:
ns = {}
self.assign_fallback_playlist(ns, next_timeslot)
ns["schedule_id"] = next_timeslot.schedule_id
ns["schedule_start"] = next_timeslot.schedule_start
ns["schedule_end"] = next_timeslot.schedule_end
ns["show_id"] = next_timeslot.show_id
ns["show_name"] = next_timeslot.show_name
ns["playlist_id"] = next_timeslot.playlist_id
ns = SU.clean_dictionary(ns)
self.next_timeslot = ns
def assign_fallback_playlist(self, data, timeslot):
"""
Assigns fallback info to the given timeslot.
Args:
data ({}): The dictionary holding the (virtual) timeslot
timeslot (Timeslot): The actual timeslot object to retrieve fallback info from
"""
fallback_type = None
playlist = None
if timeslot:
fallback_type, playlist = self.engine.scheduler.fallback.resolve_playlist(timeslot)
if playlist:
data["playlist_id"] = playlist.playlist_id
else:
data["playlist_id"] = -1
if fallback_type:
data["fallback_type"] = fallback_type.id
else:
data["fallback_type"] = FallbackType.STATION.id
def get_timeslots(self):
"""
Retrieves all available timeslots for the past, future and the current one.
Returns:
({}, {}, {})
"""
return (self.previous_timeslot, self.current_timeslot, self.next_timeslot)
def add(self, entry):
"""
Saves the currently preloaded [`Entry`] to the local cache.
"""
self.history.pop()
self.history.appendleft(entry)
def get_recent_entries(self):
"""
Retrieves the currently playing [`Entry`] from the local cache.
"""
return self.history
def resolve_entry(self, uri):
"""
Retrieves the playlog matching the provied file URI.
Args:
path (String): The URI of the resource
"""
result = None
entries = self.get_recent_entries()
if not entries:
return None
for entry in entries:
if entry:
entry_source = entry.source
if entry.get_content_type() in ResourceClass.FILE.types:
base_dir = self.config.get("audio_source_folder")
extension = self.config.get("audio_source_extension")
entry_source = ResourceUtil.uri_to_filepath(base_dir, entry.source, extension)
if entry_source == uri:
self.logger.info("Resolved '%s' entry '%s' for URI '%s'" % (entry.get_content_type(), entry, uri))
result = entry
break
if not result:
msg = "Found no entry in the recent history which matches the given source '%s'" % (uri)
self.logger.critical(SU.red(msg))
return result
def print_entry_history(self):
"""
Prints all recents entries of the history.
"""
msg = "Active entry history:\n"
for entries in self.history:
msg += "["
for e in entries:
msg += "\n" + str(e)
msg += "]"
self.logger.info(msg)
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program 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 Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import threading
import queue
import logging
from datetime import datetime
from src.base.utils import SimpleUtil as SU
from src.scheduling.models import Schedule, Playlist, PlaylistEntry, PlaylistEntryMetaData
from src.scheduling.calender_fetcher import CalendarFetcher
class AuraCalendarService(threading.Thread):
"""
The `AuraCalendarService` retrieves all current schedules and related
playlists including audio files from the configured API endpoints and
stores it in the local database.
To perform the API queries it utilizes the CalendarFetcher class.
Attributes:
#FIXME Review attributes not needed.
"""
queue = None
config = None
logger = None
fetched_schedule_data = None
calendar_fetcher = None
stop_event = None
def __init__(self, config):
"""
Initializes the class.
Args:
config (AuraConfig): The configuration
"""
threading.Thread.__init__(self)
self.config = config
self.logger = logging.getLogger("AuraEngine")
self.queue = queue.Queue()
self.stop_event = threading.Event()
self.calendar_fetcher = CalendarFetcher(config)
def get_queue(self):
"""
Retrieves the queue of fetched data.
"""
return self.queue
def run(self):
"""
Fetch calendar data and store it in the database. Also handles local deletion of remotely
deleted schedules.
Returns
Schedule ([]): An arrar of retrieved schedules passed via `self.queue`
"""
result = []
now_unix = SU.timestamp()
scheduling_window_start = self.config.get("scheduling_window_start")
try:
fetched_schedule_data = self.calendar_fetcher.fetch()
self.logger.debug("Schedule data fetched from API: " + str(fetched_schedule_data))
# If nothing is fetched, return
if not fetched_schedule_data:
self.queue.put("fetching_aborted Nothing fetched")
return
# Check if existing schedules have been deleted
local_schedules = Schedule.select_programme(datetime.now())
for local_schedule in local_schedules:
# Only allow deletion of schedules which are deleted before the start of the scheduling window
if local_schedule.start_unix > now_unix:
if (local_schedule.start_unix - scheduling_window_start) > now_unix:
# Filter the local schedule from the fetched ones
existing_schedule = list(filter(lambda new_schedule: \
new_schedule["schedule_id"] == local_schedule.schedule_id, fetched_schedule_data))
if existing_schedule:
# self.logger.debug("Schedule #%s is still existing remotely!" % (local_schedule.schedule_id))
pass
else:
self.logger.info("Schedule #%s has been deleted remotely, hence also delete it locally [%s]" % \
(local_schedule.schedule_id, str(local_schedule)))
local_schedule.delete(commit=True)
self.logger.info("Deleted local schedule #%s from database" % local_schedule.schedule_id)
else:
msg = "Schedule #%s has been deleted remotely. Since the scheduling window has already started, it won't be deleted locally." % \
local_schedule.schedule_id
self.logger.error(SU.red(msg))
# Process fetched schedules
for schedule in fetched_schedule_data:
# Check schedule for validity
if "start" not in schedule:
self.logger.warning("No 'start' of schedule given. Skipping the schedule: %s " % str(schedule))
continue
if "end" not in schedule:
self.logger.warning("No 'end' of schedule given. Skipping the schedule: %s " % str(schedule))
continue
# Store the schedule
schedule_db = self.store_schedule(schedule)
# Store playlists to play
self.store_playlist(schedule_db, schedule_db.playlist_id, schedule["playlist"])
if schedule_db.schedule_fallback_id:
self.store_playlist(schedule_db, schedule_db.schedule_fallback_id, schedule["schedule_fallback"])
if schedule_db.show_fallback_id:
self.store_playlist(schedule_db, schedule_db.show_fallback_id, schedule["show_fallback"])
if schedule_db.station_fallback_id:
self.store_playlist(schedule_db, schedule_db.station_fallback_id, schedule["station_fallback"])
# self.store_playlist(schedule_db, schedule_db.playlist_id, schedule["playlist"], PlaylistType.DEFAULT.id)
# if schedule_db.schedule_fallback_id:
# self.store_playlist(schedule_db, schedule_db.schedule_fallback_id, schedule["schedule_fallback"], PlaylistType.TIMESLOT.id)
# if schedule_db.show_fallback_id:
# self.store_playlist(schedule_db, schedule_db.show_fallback_id, schedule["show_fallback"], PlaylistType.SHOW.id)
# if schedule_db.station_fallback_id:
# self.store_playlist(schedule_db, schedule_db.station_fallback_id, schedule["station_fallback"], PlaylistType.STATION.id)
result.append(schedule_db)
# Release the mutex
self.queue.put(result)
except Exception as e:
# Release the mutex
self.logger.warning("Fetching aborted due to: %s" % str(e), e)
self.queue.put("fetching_aborted " + str(e))
# terminate the thread
return
def store_schedule(self, schedule):
"""
Stores the given schedule to the database.
Args:
schedule (Schedule): The schedule
"""
schedule_db = Schedule.select_show_on_datetime(schedule["start"])
havetoadd = False
if not schedule_db:
self.logger.debug("no schedule with given schedule id in database => create new")
schedule_db = Schedule()
havetoadd = True
schedule_db.show_id = schedule["show_id"]
schedule_db.schedule_id = schedule["schedule_id"]
schedule_db.schedule_start = schedule["start"]
schedule_db.schedule_end = schedule["end"]
schedule_db.show_name = schedule["show_name"]
schedule_db.show_hosts = schedule["show_hosts"]
schedule_db.is_repetition = schedule["is_repetition"]
schedule_db.funding_category = schedule["show_fundingcategory"]
schedule_db.languages = schedule["show_languages"]
schedule_db.type = schedule["show_type"]
schedule_db.category = schedule["show_categories"]
schedule_db.topic = schedule["show_topics"]
schedule_db.musicfocus = schedule["show_musicfocus"]
schedule_db.playlist_id = schedule["playlist_id"]
schedule_db.schedule_fallback_id = schedule["schedule_fallback_id"]
schedule_db.show_fallback_id = schedule["show_fallback_id"]
schedule_db.station_fallback_id = schedule["station_fallback_id"]
schedule_db.store(add=havetoadd, commit=True)
return schedule_db
# def store_playlist(self, schedule_db, playlist_id, fetched_playlist, fallbackplaylist_type=0):
def store_playlist(self, schedule_db, playlist_id, fetched_playlist):
"""
Stores the Playlist to the database.
"""
if not playlist_id or not fetched_playlist:
self.logger.debug(f"Playlist ID#{playlist_id} is not available!")
# self.logger.debug("Playlist type %s with ID '%s' is not available!" % (fallbackplaylist_type, playlist_id))
return
playlist_db = Playlist.select_playlist_for_schedule(schedule_db.schedule_start, playlist_id)
havetoadd = False
if not playlist_db:
playlist_db = Playlist()
havetoadd = True
self.logger.debug("Storing playlist %d for schedule (%s)" % (playlist_id, str(schedule_db)))
playlist_db.playlist_id = playlist_id
playlist_db.schedule_start = schedule_db.schedule_start
playlist_db.show_name = schedule_db.show_name
# playlist_db.fallback_type = fallbackplaylist_type
if "entries" in fetched_playlist:
playlist_db.entry_count = len(fetched_playlist["entries"])
else:
playlist_db.entry_count = 0
playlist_db.store(havetoadd, commit=True)
if playlist_db.entry_count > 0:
self.store_playlist_entries(schedule_db, playlist_db, fetched_playlist)
return playlist_db
def store_playlist_entries(self, schedule_db, playlist_db, fetched_playlist):
"""
Stores the playlist entries to the database.
"""
entry_num = 0
time_marker = playlist_db.start_unix
self.expand_entry_duration(schedule_db, fetched_playlist)
self.delete_orphaned_entries(playlist_db, fetched_playlist)
for entry in fetched_playlist["entries"]:
entry_db = PlaylistEntry.select_playlistentry_for_playlist(playlist_db.artificial_id, entry_num)
havetoadd = False
if not entry_db:
entry_db = PlaylistEntry()
havetoadd = True
entry_db.entry_start = datetime.fromtimestamp(time_marker)
entry_db.artificial_playlist_id = playlist_db.artificial_id
entry_db.entry_num = entry_num
entry_db.duration = SU.nano_to_seconds(entry["duration"])
if "uri" in entry:
# FIXME Refactor mix of uri/filename/file/source
entry_db.uri = entry["uri"]
entry_db.source = entry["uri"]
if "filename" in entry:
entry_db.source = entry["filename"]
entry_db.store(havetoadd, commit=True)
if "file" in entry:
self.store_playlist_entry_metadata(entry_db, entry["file"]["metadata"])
entry_num = entry_num + 1
time_marker += entry_db.duration
def delete_orphaned_entries(self, playlist_db, fetched_playlist):
"""
Deletes all playlist entries which are beyond the current playlist's `entry_count`.
Such entries might be existing due to a remotely changed playlist, which now has
less entries than before.
"""
new_last_idx = len(fetched_playlist["entries"])
existing_last_idx = PlaylistEntry.count_entries(playlist_db.artificial_id)-1
if existing_last_idx < new_last_idx:
return
for entry_num in range(new_last_idx, existing_last_idx+1, 1):
PlaylistEntry.delete_entry(playlist_db.artificial_id, entry_num)
self.logger.info(SU.yellow("Deleted playlist entry %s:%s" % (playlist_db.artificial_id, entry_num)))
entry_num += 1
def expand_entry_duration(self, schedule_db, fetched_playlist):
"""
If some playlist entry doesn't have a duration assigned, its duration is expanded to the
remaining duration of the playlist (= schedule duration minus playlist entries with duration).
If there's more than one entry without duration, such entries are removed from the playlist.
"""
total_seconds = (schedule_db.schedule_end - schedule_db.schedule_start).total_seconds()
total_duration = SU.seconds_to_nano(total_seconds)
actual_duration = 0
missing_duration = []
idx = 0
for entry in fetched_playlist["entries"]:
if not "duration" in entry:
missing_duration.append(idx)
else:
actual_duration += entry["duration"]
idx += 1
if len(missing_duration) == 1:
fetched_playlist["entries"][missing_duration[0]]["duration"] = total_duration - actual_duration
self.logger.info("Expanded duration of playlist entry #%s:%s" % (fetched_playlist["id"], missing_duration[0]))
elif len(missing_duration) > 1:
# This case should actually never happen, as TANK doesn't allow more than one entry w/o duration anymore
for i in reversed(missing_duration[1:-1]):
self.logger.error(SU.red("Deleted Playlist Entry without duration: %s" % \
str(fetched_playlist["entries"][i])))
del fetched_playlist["entries"][i]
def store_playlist_entry_metadata(self, entry_db, metadata):
"""
Stores the meta-data for a PlaylistEntry.
"""
metadata_db = PlaylistEntryMetaData.select_metadata_for_entry(entry_db.artificial_id)
havetoadd = False
if not metadata_db:
metadata_db = PlaylistEntryMetaData()
havetoadd = True
metadata_db.artificial_entry_id = entry_db.artificial_id
if "artist" in metadata:
metadata_db.artist = metadata["artist"]
else:
metadata_db.artist = ""
if "album" in metadata:
metadata_db.album = metadata["album"]
else:
metadata_db.album = ""
if "title" in metadata:
metadata_db.title = metadata["title"]
else:
metadata_db.title = ""
metadata_db.store(havetoadd, commit=True)
def stop(self):
self.stop_event.set()
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program 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 Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
import requests
from datetime import datetime
from src.base.utils import SimpleUtil as SU
class CalendarFetcher:
"""
Fetches the schedules, playlists and playlist entries as JSON
via the API endpoints of Steering and Tank.
"""
config = None
logging = None
has_already_fetched = False
fetched_schedule_data = None
# Config for API Endpoints
steering_calendar_url = None
tank_playlist_url = None
tank_session = None
tank_secret = None
def __init__(self, config):
"""
Constructor
Args:
config (AuraConfig): Holds the engine configuration
"""
self.config = config
self.logger = logging.getLogger("AuraEngine")
self.steering_calendar_url = self.config.get("api_steering_calendar")
self.tank_playlist_url = self.config.get("api_tank_playlist")
self.tank_session = self.config.get("api_tank_session")
self.tank_secret = self.config.get("api_tank_secret")
#
# PUBLIC METHODS
#
def fetch(self):
"""
Retrieve all required data from the API.
"""
return_data = []
self.logger.debug("Fetching schedules from STEERING")
self.fetched_schedule_data = self.fetch_schedule_data()
if not self.fetched_schedule_data:
self.logger.critical(SU.red("No schedules fetched from API!"))
return None
self.logger.debug("Fetching playlists from TANK")
self.fetch_playlists()
try:
for schedule in self.fetched_schedule_data:
# Skip schedule if no start or end is given
if "start" not in schedule:
self.logger.warning("No start of schedule given. Skipping schedule: " + str(schedule))
schedule = None
if "end" not in schedule:
self.logger.warning("No end of schedule given. Skipping schedule: " + str(schedule))
schedule = None
if "playlist" not in schedule \
and "show_fallback" not in schedule \
and "schedule_fallback" not in schedule \
and "station_fallback" not in schedule:
self.logger.warning("No playlist for schedule given. Skipping schedule: " + str(schedule))
schedule = None
if schedule:
return_data.append(schedule)
except TypeError:
self.logger.error(SU.red("Nothing fetched ..."))
self.fetched_schedule_data = None
return None
return return_data
#
# PRIVATE METHODS
#
def fetch_schedule_data(self):
"""
Fetches schedule data from Steering.
Returns:
([Schedule]): An array of schedules
"""
schedule = None
headers = { "content-type": "application/json" }
try:
self.logger.debug("Fetch schedules from Steering API...")
response = requests.get(self.steering_calendar_url, data=None, headers=headers)
if not response.status_code == 200:
self.logger.critical(SU.red("HTTP Status: %s | Schedules could not be fetched! Response: %s" % \
(str(response.status_code), response.text)))
return None
schedule = response.json()
except Exception as e:
self.logger.critical(SU.red("Error while requesting schedules from Steering!"), e)
if not schedule:
self.logger.error(SU.red("Got no schedule via Playout API (Steering)!"))
return None
return self.remove_unnecessary_data(schedule)
def fetch_playlists(self):
"""
Fetches all playlists including fallback playlists for every schedule.
This method used the class member `fetched_schedule_data`` to iterate
over and extend schedule data.
"""
# store fetched entries => do not have to fetch playlist_id more than once
fetched_entries=[]
try:
for schedule in self.fetched_schedule_data:
# Get IDs of playlists
playlist_id = self.get_playlist_id(schedule, "playlist_id")
schedule_fallback_id = self.get_playlist_id(schedule, "schedule_fallback_id")
show_fallback_id = self.get_playlist_id(schedule, "show_fallback_id")
station_fallback_id = self.get_playlist_id(schedule, "station_fallback_id")
# Retrieve playlist and the fallback playlists for every schedule.
# If a playlist (like station_fallback) is already fetched, it is not fetched again but reused
schedule["playlist"] = self.fetch_playlist(playlist_id, fetched_entries)
schedule["schedule_fallback"] = self.fetch_playlist(schedule_fallback_id, fetched_entries)
schedule["show_fallback"] = self.fetch_playlist(show_fallback_id, fetched_entries)
schedule["station_fallback"] = self.fetch_playlist(station_fallback_id, fetched_entries)
except Exception as e:
self.logger.error("Error while fetching playlists from API endpoints: " + str(e), e)
def fetch_playlist(self, playlist_id, fetched_playlists):
"""
Fetches the playlist for a given schedule.
Args:
id_name (String): The type of playlist to fetch (e.g. normal vs. fallback)
fetched_playlists ([]): Previously fetched playlists to avoid re-fetching
Returns:
(Playlist): Playlist of type `id_name`
"""
if not playlist_id:
return None
playlist = None
url = self.tank_playlist_url.replace("${ID}", playlist_id)
headers = {
"Authorization": "Bearer %s:%s" % (self.tank_session, self.tank_secret),
"content-type": "application/json"
}
# If playlist is already fetched in this round, use the existing one
for playlist in fetched_playlists:
if playlist["id"] == playlist_id:
self.logger.debug("Playlist #%s already fetched" % playlist_id)
return playlist
try:
self.logger.debug("Fetch playlist from Tank API...")
response = requests.get(url, data=None, headers=headers)
if not response.status_code == 200:
self.logger.critical(SU.red("HTTP Status: %s | Playlist #%s could not be fetched or is not available! Response: %s" % \
(str(response.status_code), str(playlist_id), response.text)))
return None
playlist = response.json()
except Exception as e:
self.logger.critical(SU.red("Error while requesting playlist #%s from Tank" % str(playlist_id)), e)
return None
fetched_playlists.append(playlist)
return playlist
def get_playlist_id(self, schedule, id_name):
"""
Extracts the playlist ID for a given playlist (fallback) type.
"""
playlist_id = str(schedule[id_name])
if not playlist_id or playlist_id == "None":
self.logger.debug("No value defined for '%s' in schedule '#%s'" % (id_name, schedule["schedule_id"]))
return None
return playlist_id
def remove_unnecessary_data(self, schedule):
"""
Removes all schedules which are not relevant for
further processing.
"""
count_before = len(schedule)
schedule = self.remove_data_more_than_24h_in_the_future(schedule)
schedule = self.remove_data_in_the_past(schedule)
count_after = len(schedule)
self.logger.debug("Removed %d unnecessary schedules from response. Entries left: %d" % ((count_before - count_after), count_after))
return schedule
def remove_data_more_than_24h_in_the_future(self, schedules):
"""
Removes entries 24h in the future and 12 hours in the past.
Note: This might influence resuming (in case of a crash)
single schedules which are longer than 12 hours long.
Think e.g. live broadcasts.
"""
items = []
now = SU.timestamp()
now_plus_24hours = now + (12*60*60)
now_minus_12hours = now - (12*60*60)
for s in schedules:
start_time = datetime.strptime(s["start"], "%Y-%m-%dT%H:%M:%S")
start_time = SU.timestamp(start_time)
if start_time <= now_plus_24hours and start_time >= now_minus_12hours:
items.append(s)
return items
def remove_data_in_the_past(self, schedules):
"""
Removes all schedules from the past, except the one which is
currently playing.
"""
items = []
now = SU.timestamp()
for s in schedules:
start_time = datetime.strptime(s["start"], "%Y-%m-%dT%H:%M:%S")
start_time = SU.timestamp(start_time)
end_time = datetime.strptime(s["end"], "%Y-%m-%dT%H:%M:%S")
end_time = SU.timestamp(end_time)
# Append all elements in the future
if start_time >= now:
items.append(s)
# Append the one which is playing now
elif start_time < now < end_time:
items.append(s)
return items
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program 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 Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
from enum import Enum
from datetime import timedelta
from src.base.config import AuraConfig
from src.base.utils import SimpleUtil as SU
from src.base.mail import AuraMailer
from src.core.resources import ResourceClass
from src.core.channels import Channel
from src.core.control import EngineExecutor
class FallbackType(Enum):
"""
Types of playlists.
"""
NONE = { "id": 0, "name": "default", "lqs_sources": [ Channel.QUEUE_A, Channel.QUEUE_A] } # No fallback active, default playout
SCHEDULE = { "id": 1, "name": "schedule", "lqs_sources": [ Channel.FALLBACK_QUEUE_A, Channel.FALLBACK_QUEUE_B]} # The first played when some default playlist fails
SHOW = { "id": 2, "name": "show", "lqs_sources": [ "station_folder", "station_playlist"]} # The second played when the timeslot fallback fails
STATION = { "id": 3, "name": "station", "lqs_sources": [ "station_folder", "station_playlist"] } # The last played when everything else fails
@property
def id(self):
return self.value["id"]
def __str__(self):
return str(self.value["name"])
class FallbackManager:
"""
Handles all types of fallbacks in case there is an outage
for the regular radio programme.
Attributes:
config (AuraConfig): The engine configuration
logger (AuraLogger): The logger
mail (AuraMailer): Mail service
scheduler (AuraScheduler): The scheduler
"""
config = None
logger = None
mailer = None
scheduler = None
def __init__(self, scheduler):
"""
Constructor
Args:
"""
self.config = AuraConfig.config()
self.logger = logging.getLogger("AuraEngine")
self.mailer = AuraMailer(self.config)
self.scheduler = scheduler
#
# METHODS
#
def queue_fallback_playlist(self, timeslot):
"""
Evaluates the scheduled fallback and queues it using a timed thread.
"""
(fallback_type, playlist) = self.get_fallback_playlist(timeslot)
if playlist:
self.logger.info(f"Resolved {fallback_type.value} fallback")
return FallbackCommand(timeslot, playlist.entries)
else:
msg = f"There is no schedule- or show-fallback defined for timeslot#{timeslot.schedule_id}. "
msg += f"The station fallback will be used automatically."
self.logger.info(msg)
def resolve_playlist(self, timeslot):
"""
Retrieves the currently planned (fallback) playlist. If a normal playlist is available,
this one is returned. In case of station fallback no playlist is returned.
Args:
timeslot (Schedule)
Returns:
(FallbackType, Playlist)
"""
planned_playlist = None
fallback_type = None
if self.validate_playlist(timeslot, "playlist"):
planned_playlist = timeslot.get_playlist()
fallback_type = FallbackType.NONE
else:
(fallback_type, planned_playlist) = self.get_fallback_playlist(timeslot)
return (fallback_type, planned_playlist)
def get_fallback_playlist(self, timeslot):
"""
Retrieves the playlist to be used in a fallback scenario.
Args:
timeslot (Schedule)
Returns:
(Playlist)
"""
playlist = None
fallback_type = FallbackType.STATION
if self.validate_playlist(timeslot, "schedule_fallback"):
playlist = timeslot.schedule_fallback[0]
fallback_type = FallbackType.SCHEDULE
elif self.validate_playlist(timeslot, "show_fallback"):
playlist = timeslot.show_fallback[0]
fallback_type = FallbackType.SHOW
return (fallback_type, playlist)
def validate_playlist(self, timeslot, playlist_type):
"""
Checks if a playlist is valid for play-out.
Following checks are done for all playlists:
- has one or more entries
Fallback playlists should either:
- have filesystem entries only
- reference a recording of a previous playout of a show (also filesystem)
Otherwise, if a fallback playlist contains Live or Stream entries,
the exact playout behaviour can hardly be predicted.
"""
playlist = getattr(timeslot, playlist_type)
if playlist \
and isinstance(playlist, list) \
and playlist[0].entries \
and len(playlist[0].entries) > 0:
# Default playlist
if playlist_type == "playlist":
return True
# Fallback playlist
elif playlist[0].entries:
is_fs_only = True
for entry in playlist[0].entries:
if entry.get_content_type() not in ResourceClass.FILE.types:
self.logger.error(SU.red("Fallback playlist of type '%s' contains not only file-system entries! \
Skipping fallback level..." % playlist_type))
is_fs_only = False
break
return is_fs_only
return False
class FallbackCommand(EngineExecutor):
"""
Command composition for executing timed scheduling and unscheduling of fallback playlists.
Based on the `timeslot.start_date` and `timeslot.end_date` two `EngineExecutor commands
are created.
"""
def __init__(self, timeslot, entries):
"""
Constructor
Args:
timeslot (Timeslot): The timeslot any fallback entries should be scheduled for
entries (List): List of entries to be scheduled as fallback
"""
from src.core.engine import Engine
def do_play(entries):
self.logger.info(SU.cyan(f"=== start_fallback_playlist('{entries}') ==="))
Engine.get_instance().player.start_fallback_playlist(entries)
def do_stop():
self.logger.info(SU.cyan("=== stop_fallback_playlist() ==="))
Engine.get_instance().player.stop_fallback_playlist()
# Start fade-out 50% before the end of the timeslot for a more smooth transition
end_time_offset = int(float(AuraConfig.config().get("fade_out_time")) / 2 * 1000 * -1)
end_time = SU.timestamp(timeslot.schedule_end + timedelta(milliseconds=end_time_offset))
self.logger.info(f"Starting fade-out of scheduled fallback with an offset of {end_time_offset} milliseconds at {end_time}")
child_timer = EngineExecutor("FALLBACK", None, end_time, do_stop, None)
super().__init__("FALLBACK", child_timer, timeslot.start_unix, do_play, entries)
\ No newline at end of file
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import sys
import time
import logging
import datetime
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import orm
from sqlalchemy import BigInteger, Boolean, Column, DateTime, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.hybrid import hybrid_property
from src.base.config import AuraConfig
from src.base.utils import SimpleUtil
from src.core.resources import ResourceUtil
# Init Config
config = AuraConfig()
# Initialize DB Model and session
engine = sa.create_engine(config.get_database_uri())
Base = declarative_base()
Base.metadata.bind = engine
class DB():
session = orm.scoped_session(orm.sessionmaker())(bind=engine)
Model = Base
class AuraDatabaseModel():
"""
AuraDataBaseModel.
Holding all tables and relationships for the engine.
"""
logger = None
def __init__(self):
"""
Constructor.
"""
self.logger = logging.getLogger("AuraEngine")
def store(self, add=False, commit=False):
"""
Store to the database
"""
if add:
DB.session.add(self)
else:
DB.session.merge(self)
if commit:
DB.session.commit()
def delete(self, commit=False):
"""
Delete from the database
"""
DB.session.delete(self)
if commit:
DB.session.commit()
def refresh(self):
"""
Refreshes the currect record
"""
DB.session.expire(self)
DB.session.refresh(self)
def _asdict(self):
return self.__dict__
@staticmethod
def recreate_db(systemexit = False):
"""
Re-creates the database for developments purposes.
"""
manualschedule = Schedule()
manualschedule.schedule_id = 0
manualschedule.show_name = "Manual Show"
Base.metadata.drop_all()
Base.metadata.create_all()
# self.logger.debug("inserting manual scheduling possibility and fallback trackservice schedule")
# DB.session.add(manualschedule)
# db.session.add(fallback_trackservice_schedule)
# self.logger.debug("all created. commiting...")
DB.session.commit()
if systemexit:
sys.exit(0)
#
# SCHEDULES & PLAYLISTS
#
class Schedule(DB.Model, AuraDatabaseModel):
"""
One specific Schedule for a show on a timeslot.
Holding references to playlists and fallback-playlists.
"""
__tablename__ = 'schedule'
# Primary keys
id = Column(Integer, primary_key=True, autoincrement=True)
schedule_start = Column(DateTime, unique=True, index=True)
schedule_end = Column(DateTime, unique=True, index=True)
schedule_id = Column(Integer, unique=True)
show_id = Column(Integer)
show_name = Column(String(256))
show_hosts = Column(String(256))
funding_category = Column(String(256))
comment = Column(String(512))
languages = Column(String(256))
type = Column(String(256))
category = Column(String(256))
topic = Column(String(256))
musicfocus = Column(String(256))
is_repetition = Column(Boolean())
playlist_id = Column(Integer) #, ForeignKey("playlist.playlist_id"))
schedule_fallback_id = Column(Integer)
show_fallback_id = Column(Integer)
station_fallback_id = Column(Integer)
fadeouttimer = None # Used to fade-out the schedule, even when entries are longer
playlist = relationship("Playlist",
primaryjoin="and_(Schedule.schedule_start==Playlist.schedule_start, Schedule.playlist_id==Playlist.playlist_id, Schedule.show_name==Playlist.show_name)",
back_populates="schedule")
schedule_fallback = relationship("Playlist",
primaryjoin="and_(Schedule.schedule_start==Playlist.schedule_start, Schedule.schedule_fallback_id==Playlist.playlist_id, Schedule.show_name==Playlist.show_name)",
back_populates="schedule")
show_fallback = relationship("Playlist",
primaryjoin="and_(Schedule.schedule_start==Playlist.schedule_start, Schedule.show_fallback_id==Playlist.playlist_id, Schedule.show_name==Playlist.show_name)",
back_populates="schedule")
station_fallback = relationship("Playlist",
primaryjoin="and_(Schedule.schedule_start==Playlist.schedule_start, Schedule.station_fallback_id==Playlist.playlist_id, Schedule.show_name==Playlist.show_name)",
back_populates="schedule")
@staticmethod
def select_show_on_datetime(date_time):
return DB.session.query(Schedule).filter(Schedule.schedule_start == date_time).first()
@staticmethod
def select_programme(date_from=datetime.date.today()):
"""
Select all schedules starting from `date_from` or from today if no
parameter is passed.
Args:
date_from (datetime): Select schedules from this date and time on
Returns:
([Schedule]): List of schedules
"""
schedules = DB.session.query(Schedule).\
filter(Schedule.schedule_start >= date_from).\
order_by(Schedule.schedule_start).all()
return schedules
@staticmethod
def select_upcoming(n):
"""
Selects the (`n`) upcoming schedules.
"""
now = datetime.datetime.now()
DB.session.commit() # Required since independend session is used.
schedules = DB.session.query(Schedule).\
filter(Schedule.schedule_start > str(now)).\
order_by(Schedule.schedule_start.asc()).limit(n).all()
return schedules
def get_playlist(self):
"""
Returns the assigned playlist.
"""
# TODO Refactor to avoid storing array of playlists.
if self.playlist and self.playlist[0]:
return self.playlist[0]
return None
def has_queued_entries(self):
"""
Checks if entries of this timeslot have been queued at the engine.
"""
#TODO Make logic more transparent
if hasattr(self, "queued_entries"):
if len(self.queued_entries) > 0:
return True
return False
@hybrid_property
def start_unix(self):
"""
Start time of the schedule in UNIX time.
"""
return time.mktime(self.schedule_start.timetuple())
@hybrid_property
def end_unix(self):
"""
End time of the schedule in UNIX time.
"""
return time.mktime(self.schedule_end.timetuple())
def as_dict(self):
"""
Returns the schedule as a dictionary for serialization.
"""
playlist = self.playlist
return {
"schedule_id": self.schedule_id,
"schedule_start": self.schedule_start.isoformat(),
"schedule_end": self.schedule_end.isoformat(),
"topic": self.topic,
"musicfocus": self.musicfocus,
"funding_category": self.funding_category,
"is_repetition": self.is_repetition,
"category": self.category,
"languages": self.languages,
"comment": self.comment,
"playlist_id": self.playlist_id,
"schedule_fallback_id": self.schedule_fallback_id,
"show_fallback_id": self.show_fallback_id,
"station_fallback_id": self.station_fallback_id,
"show": {
"name": self.show_name,
"host": self.show_hosts
},
"playlist": playlist
}
def __str__(self):
"""
String representation of the object.
"""
time_start = SimpleUtil.fmt_time(self.start_unix)
time_end = SimpleUtil.fmt_time(self.end_unix)
return "ID#%s [Show: %s, ShowID: %s | %s - %s ]" % (str(self.schedule_id), self.show_name, str(self.show_id), time_start, time_end)
class Playlist(DB.Model, AuraDatabaseModel):
"""
The playlist containing playlist entries.
"""
__tablename__ = 'playlist'
# pk,fk
artificial_id = Column(Integer, primary_key=True)
schedule_start = Column(DateTime, ForeignKey("schedule.schedule_start"))
# relationships
schedule = relationship("Schedule", uselist=False, back_populates="playlist")
entries = relationship("PlaylistEntry", back_populates="playlist")
# data
playlist_id = Column(Integer, autoincrement=False) # , ForeignKey("schedule.playlist_id"))
show_name = Column(String(256))
entry_count = Column(Integer)
@staticmethod
def select_all():
"""
Fetches all entries
"""
all_entries = DB.session.query(Playlist).filter(Playlist.fallback_type == 0).all()
cnt = 0
for entry in all_entries:
entry.programme_index = cnt
cnt = cnt + 1
return all_entries
@staticmethod
def select_playlist_for_schedule(start_date, playlist_id):
"""
Retrieves the playlist for the given schedule identified by `start_date` and `playlist_id`
Args:
start_date (datetime): Date and time when the playlist is scheduled
playlist_id (Integer): The ID of the playlist
Returns:
(Playlist): The playlist, if existing for schedule
Raises:
Exception: In case there a inconsistent database state, such es multiple playlists for given date/time.
"""
playlist = None
playlists = DB.session.query(Playlist).filter(Playlist.schedule_start == start_date).all()
# FIXME There are unknown issues with the native SQL query by ID
# playlists = DB.session.query(Playlist).filter(Playlist.schedule_start == datetime and Playlist.playlist_id == playlist_id).all()
for p in playlists:
if p.playlist_id == playlist_id:
playlist = p
return playlist
@staticmethod
def select_playlist(playlist_id):
"""
Retrieves all paylists for that given playlist ID.
Args:
playlist_id (Integer): The ID of the playlist
Returns:
(Array<Playlist>): An array holding the playlists
"""
return DB.session.query(Playlist).filter(Playlist.playlist_id == playlist_id).order_by(Playlist.schedule_start).all()
@staticmethod
def is_empty():
"""
Checks if the given is empty
"""
try:
return not DB.session.query(Playlist).one_or_none()
except sa.orm.exc.MultipleResultsFound:
return False
@hybrid_property
def start_unix(self):
"""
Start time of the playlist in UNIX time.
"""
return time.mktime(self.schedule_start.timetuple())
@hybrid_property
def end_unix(self):
"""
End time of the playlist in UNIX time.
"""
return time.mktime(self.schedule_start.timetuple()) + self.duration
@hybrid_property
def duration(self):
"""
Returns the total length of the playlist in seconds.
Returns:
(Integer): Length in seconds
"""
total = 0
for entry in self.entries:
total += entry.duration
return total
def as_dict(self):
"""
Returns the playlist as a dictionary for serialization.
"""
entries = []
for e in self.entries:
entries.append(e.as_dict())
playlist = {
"playlist_id": self.playlist_id,
"fallback_type": self.fallback_type,
"entry_count": self.entry_count,
"entries": entries
}
return playlist
def __str__(self):
"""
String representation of the object.
"""
time_start = SimpleUtil.fmt_time(self.start_unix)
time_end = SimpleUtil.fmt_time(self.end_unix)
return "ID#%s [items: %s | %s - %s]" % (str(self.playlist_id), str(self.entry_count), str(time_start), str(time_end))
class PlaylistEntry(DB.Model, AuraDatabaseModel):
"""
Playlist entries are the individual items of a playlist such as audio files.
"""
__tablename__ = 'playlist_entry'
# primary keys
artificial_id = Column(Integer, primary_key=True)
# foreign keys
artificial_playlist_id = Column(Integer, ForeignKey("playlist.artificial_id"))
entry_num = Column(Integer) # , primary_key=True)
uri = Column(String(1024))
duration = Column(BigInteger)
source = Column(String(1024))
entry_start = Column(DateTime)
entry_start_actual = None # Assigned when the entry is actually played
channel = None # Assigned when entry is actually played
queue_state = None # Assigned when entry is about to be queued
status = None # Assigned when state changes
switchtimer = None
loadtimer = None
fadeouttimer = None
# relationships
playlist = relationship("Playlist", uselist=False, back_populates="entries")
meta_data = relationship("PlaylistEntryMetaData", uselist=False, back_populates="entry")
@staticmethod
def select_playlistentry_for_playlist(artificial_playlist_id, entry_num):
"""
Selects one entry identified by `playlist_id` and `entry_num`.
"""
return DB.session.query(PlaylistEntry).filter(PlaylistEntry.artificial_playlist_id == artificial_playlist_id, PlaylistEntry.entry_num == entry_num).first()
@staticmethod
def delete_entry(artificial_playlist_id, entry_num):
"""
Deletes the playlist entry and associated metadata.
"""
entry = PlaylistEntry.select_playlistentry_for_playlist(artificial_playlist_id, entry_num)
metadata = PlaylistEntryMetaData.select_metadata_for_entry(entry.artificial_id)
metadata.delete()
entry.delete()
DB.session.commit()
@staticmethod
def count_entries(artificial_playlist_id):
"""
Returns the count of all entries.
"""
result = DB.session.query(PlaylistEntry).filter(PlaylistEntry.artificial_playlist_id == artificial_playlist_id).count()
return result
@hybrid_property
def entry_end(self):
return self.entry_start + datetime.timedelta(seconds=self.duration)
@hybrid_property
def start_unix(self):
return time.mktime(self.entry_start.timetuple())
@hybrid_property
def end_unix(self):
return time.mktime(self.entry_end.timetuple())
@hybrid_property
def volume(self):
return 100 # FIXME Make DB Column
def get_content_type(self):
return ResourceUtil.get_content_type(self.uri)
def get_prev_entries(self):
"""
Retrieves all previous entries as part of the current entry's playlist.
Returns:
(List): List of PlaylistEntry
"""
prev_entries = []
for entry in self.playlist.entries:
if entry.entry_start < self.entry_start:
prev_entries.append(entry)
return prev_entries
def get_next_entries(self, schedule_sensitive=True):
"""
Retrieves all following entries as part of the current entry's playlist.
Args:
schedule_sensitive (Boolean): If `True` entries which start after \
the end of the schedule are excluded
Returns:
(List): List of PlaylistEntry
"""
next_entries = []
for entry in self.playlist.entries:
if entry.entry_start > self.entry_start:
if schedule_sensitive:
if entry.entry_start < self.playlist.schedule.schedule_end:
next_entries.append(entry)
else:
next_entries.append(entry)
return next_entries
def as_dict(self):
"""
Returns the entry as a dictionary for serialization.
"""
if self.meta_data:
return {
"id": self.artificial_id,
"duration": self.duration,
"artist": self.meta_data.artist,
"album": self.meta_data.album,
"title": self.meta_data.title
}
return None
def __str__(self):
"""
String representation of the object.
"""
time_start = SimpleUtil.fmt_time(self.start_unix)
time_end = SimpleUtil.fmt_time(self.end_unix)
track = self.source[-25:]
return "PlaylistEntry #%s [%s - %s | %ssec | Source: ...%s]" % (str(self.artificial_id), time_start, time_end, self.duration, track)
class PlaylistEntryMetaData(DB.Model, AuraDatabaseModel):
"""
Metadata for a playlist entry such as the artist and track name.
"""
__tablename__ = "playlist_entry_metadata"
artificial_id = Column(Integer, primary_key=True)
artificial_entry_id = Column(Integer, ForeignKey("playlist_entry.artificial_id"))
artist = Column(String(256))
title = Column(String(256))
album = Column(String(256))
entry = relationship("PlaylistEntry", uselist=False, back_populates="meta_data")
@staticmethod
def select_metadata_for_entry(artificial_playlistentry_id):
return DB.session.query(PlaylistEntryMetaData).filter(PlaylistEntryMetaData.artificial_entry_id == artificial_playlistentry_id).first()
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program 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 Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
import threading
import time
import sqlalchemy
from enum import Enum
from operator import attrgetter
from datetime import datetime, timedelta
from src.base.config import AuraConfig
from src.base.utils import SimpleUtil as SU
from src.scheduling.models import AuraDatabaseModel, Schedule, Playlist
from src.base.exceptions import NoActiveScheduleException, LoadSourceException
from src.core.control import EngineExecutor
from src.core.engine import Engine
from src.core.channels import ChannelType, TransitionType, EntryPlayState
from src.core.resources import ResourceClass, ResourceUtil
from src.scheduling.calendar import AuraCalendarService
from src.scheduling.fallback import FallbackManager
class EntryQueueState(Enum):
"""
Types of playlist entry behaviours.
"""
OKAY = "ok"
CUT = "cut"
OUT_OF_SCHEDULE = "oos"
class AuraScheduler(threading.Thread):
"""
Aura Scheduler Class
- Retrieves data from Steering and Tank
- Executes engine actions in an automated fashion
Attributes:
config (AuraConfig): Holds the Engine Configuration
logger: The logger
exit_event(threading.Event): Used to exit the thread if requested
engine: Virtual mixer
last_successful_fetch (datetime): Stores the last time a fetch from Steering/Tank was successful
programme: The current radio programme to be played as defined in the local engine database
active_entry(Show, Track): This is a Tuple consisting of the currently played `Show` and `Track`
message_timer(List<threading.Timer>): The timer queue of sound-system commands for playlists/entries to be played
"""
config = None
logger = None
engine = None
exit_event = None
is_initialized = None
is_initialized = None
last_successful_fetch = None
programme = None
message_timer = []
fallback = None
def __init__(self, engine):
"""
Constructor
Args:
config (AuraConfig): Reads the engine configuration
engine (Engine): The engine to play the schedule on
func_on_init (Function): The function to be called when the scheduler is initialized
"""
self.config = AuraConfig.config()
self.logger = logging.getLogger("AuraEngine")
AuraScheduler.init_database()
self.fallback = FallbackManager(self)
self.engine = engine
self.engine.scheduler = self
self.is_soundsytem_init = False
# Scheduler Initialization
self.is_initialized = False
self.is_engine_ready = False
# Init scheduling thread
threading.Thread.__init__(self)
self.exit_event = threading.Event()
self.start()
def run(self):
"""
Called when thread is started via `start()`. It does the following:
1. `self.fetch_new_programme()` periodically from the API depending on the `fetching_frequency` defined in the engine configuration.
2. Loads the latest programme from the database and sets the instance state `self.programme` with current schedules.
3. Queues all schedules of the programme, if the soundssystem is ready to accept commands.
On every cycle the configuration file is reloaded, to allow modifications while running the engine.
"""
while not self.exit_event.is_set():
try:
self.config.load_config()
seconds_to_wait = int(self.config.get("fetching_frequency"))
self.logger.info(SU.cyan(f"== start fetching new schedules (every {seconds_to_wait} seconds) =="))
# Load some stuff from the API in any case
self.fetch_new_programme()
# Called upon first boot only
if self.is_engine_ready:
if not self.is_initialized:
# Queue the start items
self.is_initialized = True
self.on_scheduler_ready()
# Queue all the other ones
self.queue_programme()
except Exception as e:
self.logger.critical(SU.red(f"Unhandled error while fetching & scheduling new programme! ({str(e)})"), e)
# Keep on working anyway
self.clean_timer_queue()
self.print_timer_queue()
# FIXME better location for call
if self.engine.event_dispatcher:
current_timeslot = self.get_active_schedule()
self.engine.event_dispatcher.on_timeslot(current_timeslot)
EngineExecutor.log_commands()
self.exit_event.wait(seconds_to_wait)
#
# PUBLIC METHODS
#
def on_ready(self):
"""
Called when the engine is ready.
"""
self.is_engine_ready = True
def on_scheduler_ready(self):
"""
Called when the scheduler is ready.
"""
self.logger.info(self.get_ascii_programme())
try:
self.play_active_entry()
self.queue_startup_entries()
except NoActiveScheduleException:
# That's not good, but keep on working...
pass
def play_active_entry(self):
"""
Plays the entry scheduled for the very current moment and forwards to the scheduled position in time.
Usually called when the Engine boots.
Raises:
(NoActiveScheduleException): If there's no schedule in the programme, within the scheduling window
"""
sleep_offset = 10
active_schedule = self.get_active_schedule()
# Schedule any available fallback playlist
if active_schedule:
self.fallback.queue_fallback_playlist(active_schedule)
# Queue the fade-out of the schedule
if not active_schedule.fadeouttimer:
self.queue_end_of_schedule(active_schedule, True)
active_entry = self.get_active_entry()
if not active_entry:
raise NoActiveScheduleException
# In case of a file-system source, we need to fast-foward to the current marker as per schedule
if active_entry.get_content_type() in ResourceClass.FILE.types:
# Calculate the seconds we have to fast-forward
now_unix = Engine.engine_time()
seconds_to_seek = now_unix - active_entry.start_unix
# If the seek exceeds the length of the current track,
# there's no need to do anything - the scheduler takes care of the rest
if (seconds_to_seek + sleep_offset) > active_entry.duration:
self.logger.info("The FFWD [>>] range exceeds the length of the entry. Drink some tea and wait for the sound of the next entry.")
else:
# Pre-roll and play active entry
self.engine.player.preload(active_entry)
self.engine.player.play(active_entry, TransitionType.FADE)
# Fast-forward to the scheduled position
if seconds_to_seek > 0:
# Without plenty of timeout (10s) the seek doesn't work
def async_cue_seek(seconds_to_seek):
seconds_to_seek += sleep_offset
time.sleep(sleep_offset)
self.logger.info("Going to fast-forward %s seconds" % seconds_to_seek)
response = self.engine.player.queue_seek(active_entry.channel, seconds_to_seek)
self.logger.info("Sound-system seek response: " + response)
thread = threading.Thread(target = async_cue_seek, args = (seconds_to_seek,))
thread.start()
elif active_entry.get_content_type() in ResourceClass.STREAM.types \
or active_entry.get_content_type() in ResourceClass.LIVE.types:
# Pre-roll and play active entry
self.engine.player.preload(active_entry)
self.engine.player.play(active_entry, TransitionType.FADE)
else:
self.logger.critical("Unknown Entry Type: %s" % active_entry)
def get_active_entry(self):
"""
Retrieves the current `PlaylistEntry` which should be played as per programme.
Returns:
(PlaylistEntry): The track which is (or should) currently being played
"""
now_unix = Engine.engine_time()
# Load programme if necessary
if not self.programme:
self.load_programme_from_db()
# Check for current schedule
current_schedule = self.get_active_schedule()
if not current_schedule:
self.logger.warning(SU.red("There's no active timeslot"))
return None
# Check for scheduled playlist
current_playlist = current_schedule.get_playlist()
if not current_playlist:
msg = "There's no playlist assigned to the current timeslot. Most likely a fallback will make things okay again."
self.logger.warning(SU.red(msg))
return None
# Iterate over playlist entries and store the current one
current_entry = None
for entry in current_playlist.entries:
if entry.start_unix <= now_unix and now_unix <= entry.end_unix:
current_entry = entry
break
if not current_entry:
# Nothing playing ... fallback will kick-in
msg = "There's no entry scheduled for playlist '%s' at %s" % (str(current_playlist), SU.fmt_time(now_unix))
self.logger.warning(SU.red(msg))
return None
return current_entry
def get_active_schedule(self):
"""
Retrieves the schedule currently to be played.
Returns:
(Schedule): The current schedule
"""
current_schedule = None
now_unix = Engine.engine_time()
# Iterate over all schedules and find the one to be played right now
if self.programme:
for schedule in self.programme:
if schedule.start_unix <= now_unix and now_unix < schedule.end_unix:
current_schedule = schedule
break
return current_schedule
def get_next_timeslots(self, max_count=0):
"""
Retrieves the timeslots to be played after the current one.
Args:
max_count (Integer): Maximum of timeslots to return, if `0` all exitsing ones are returned
Returns:
([Timeslot]): The next timeslots
"""
now_unix = Engine.engine_time()
next_schedules = []
for schedule in self.programme:
if schedule.start_unix > now_unix:
if (len(next_schedules) < max_count) or max_count == 0:
next_schedules.append(schedule)
else:
break
return next_schedules
def get_active_playlist(self):
"""
Retrieves the currently playing playlist.
Returns:
(Playlist): The resolved playlist
"""
schedule = self.get_active_schedule()
if schedule:
return schedule.get_playlist()
return None
def print_timer_queue(self):
"""
Prints the current timer queue i.e. playlists in the queue to be played.
"""
message_queue = ""
messages = sorted(self.message_timer, key=attrgetter('diff'))
if not messages:
self.logger.warning(SU.red("There's nothing in the Timer Queue!"))
else:
for msg in messages:
message_queue += str(msg)+"\n"
self.logger.info("Timer queue: \n" + message_queue)
def clean_timer_queue(self):
"""
Removes inactive timers from the queue.
"""
len_before = len(self.message_timer)
self.message_timer[:] = [m for m in self.message_timer if m.is_alive()]
len_after = len(self.message_timer)
self.logger.debug("Removed %s finished timer objects from queue" % (len_before - len_after))
def get_ascii_programme(self):
"""
Creates a printable version of the current programme (playlists and entries as per schedule)
Returns:
(String): An ASCII representation of the programme
"""
active_schedule = self.get_active_schedule()
s = "\n\n SCHEDULED NOW:"
s += "\n┌──────────────────────────────────────────────────────────────────────────────────────────────────────"
if active_schedule:
planned_playlist = None
if active_schedule.playlist:
planned_playlist = active_schedule.playlist[0]
(fallback_type, resolved_playlist) = self.fallback.resolve_playlist(active_schedule)
s += "\n│ Playing timeslot %s " % active_schedule
if planned_playlist:
if resolved_playlist and resolved_playlist.playlist_id != planned_playlist.playlist_id:
s += "\n│ └── Playlist %s " % planned_playlist
s += "\n"
s += SU.red("↑↑↑ That's the originally planned playlist.") + ("Instead playing the fallback playlist below ↓↓↓")
if resolved_playlist:
if not planned_playlist:
s += "\n"
s += SU.red("No playlist assigned to timeslot. Instead playing the `%s` playlist below ↓↓↓" % SU.cyan(str(fallback_type)))
s += "\n│ └── Playlist %s " % resolved_playlist
active_entry = self.get_active_entry()
# Finished entries
for entry in resolved_playlist.entries:
if active_entry == entry:
break
else:
s += self.build_entry_string("\n│ └── ", entry, True)
# Entry currently being played
if active_entry:
s += "\n│ └── Entry %s | %s " % \
(str(active_entry.entry_num+1), SU.green("PLAYING > "+str(active_entry)))
# Open entries for current playlist
rest_of_playlist = active_entry.get_next_entries(False)
entries = self.preprocess_entries(rest_of_playlist, False)
s += self.build_playlist_string(entries)
else:
s += "\n│ └── %s" % (SU.red("No active playlist. There should be at least some fallback playlist running..."))
else:
s += "\n│ Nothing. "
s += "\n└──────────────────────────────────────────────────────────────────────────────────────────────────────"
s += "\n SCHEDULED NEXT:"
s += "\n┌──────────────────────────────────────────────────────────────────────────────────────────────────────"
next_schedules = self.get_next_timeslots()
if not next_schedules:
s += "\n│ Nothing. "
else:
for schedule in next_schedules:
(fallback_type, resolved_playlist) = self.fallback.resolve_playlist(schedule)
if resolved_playlist:
s += "\n│ Queued timeslot %s " % schedule
s += "\n│ └── Playlist %s (Type: %s)" % (resolved_playlist, SU.cyan(str(fallback_type)))
if resolved_playlist.end_unix > schedule.end_unix:
s += "\n│ %s! " % \
(SU.red("↑↑↑ Playlist #%s ends after timeslot #%s!" % (resolved_playlist.playlist_id, schedule.schedule_id)))
entries = self.preprocess_entries(resolved_playlist.entries, False)
s += self.build_playlist_string(entries)
s += "\n└──────────────────────────────────────────────────────────────────────────────────────────────────────\n\n"
return s
def build_playlist_string(self, entries):
"""
Returns a stringified list of entries
"""
s = ""
is_out_of_schedule = False
for entry in entries:
if entry.queue_state == EntryQueueState.OUT_OF_SCHEDULE and not is_out_of_schedule:
s += "\n│ %s" % \
SU.red("↓↓↓ These entries won't be played because they are out of schedule.")
is_out_of_schedule = True
s += self.build_entry_string("\n│ └── ", entry, is_out_of_schedule)
return s
def build_entry_string(self, prefix, entry, strike):
"""
Returns an stringified entry.
"""
s = ""
if entry.queue_state == EntryQueueState.CUT:
s = "\n│ %s" % SU.red("↓↓↓ This entry is going to be cut.")
if strike:
entry_str = SU.strike(entry)
else:
entry_str = str(entry)
entry_line = "%sEntry %s | %s" % (prefix, str(entry.entry_num+1), entry_str)
return s + entry_line
#
# PRIVATE METHODS
#
def filter_scheduling_window(self, schedules):
"""
Ignore schedules which are beyond the scheduling window. The end of the scheduling window
is defined by the config option `scheduling_window_end`. This value defines the seconds
minus the actual start time of the schedule.
"""
now_unix = Engine.engine_time()
len_before = len(schedules)
window_start = self.config.get("scheduling_window_start")
window_end = self.config.get("scheduling_window_end")
schedules = list(filter(lambda s: (s.start_unix - window_end) > now_unix and (s.start_unix - window_start) < now_unix, schedules))
len_after = len(schedules)
self.logger.info("For now, skipped %s future schedule(s) which are out of the scheduling window (-%ss <-> -%ss)" % ((len_before - len_after), window_start, window_end))
return schedules
def is_schedule_in_window(self, schedule):
"""
Checks if the schedule is within the scheduling window.
"""
now_unix = Engine.engine_time()
window_start = self.config.get("scheduling_window_start")
window_end = self.config.get("scheduling_window_end")
if schedule.start_unix - window_start < now_unix and \
schedule.start_unix - window_end > now_unix:
return True
return False
def queue_programme(self):
"""
Queues the current programme (playlists as per schedule) by creating
timed commands to the sound-system to enable the individual tracks of playlists.
"""
# Get a clean set of the timeslots within the scheduling window
schedules = self.get_next_timeslots()
schedules = self.filter_scheduling_window(schedules)
# Queue the schedules, their playlists and entries
if schedules:
for next_schedule in schedules:
# Schedule any available fallback playlist
self.fallback.queue_fallback_playlist(next_schedule)
if next_schedule.playlist:
self.queue_playlist_entries(next_schedule, next_schedule.get_playlist().entries, False, True)
# Queue the fade-out of the schedule
if not next_schedule.fadeouttimer:
self.queue_end_of_schedule(next_schedule, True)
self.logger.info(SU.green("Finished queuing programme."))
def queue_startup_entries(self):
"""
Queues all entries after the one currently playing upon startup. Don't use
this method in any other scenario, as it doesn't respect the scheduling window.
"""
current_schedule = self.get_active_schedule()
# Queue the (rest of the) currently playing schedule upon startup
if current_schedule:
current_playlist = current_schedule.get_playlist()
if current_playlist:
active_entry = self.get_active_entry()
# Finished entries
for entry in current_playlist.entries:
if active_entry == entry:
break
# Entry currently being played
if active_entry:
# Queue open entries for current playlist
rest_of_playlist = active_entry.get_next_entries(True)
self.queue_playlist_entries(current_schedule, rest_of_playlist, False, True)
# Store them for later reference
current_schedule.queued_entries = [active_entry]
if rest_of_playlist:
current_schedule.queued_entries.append(rest_of_playlist)
def queue_playlist_entries(self, schedule, entries, fade_in, fade_out):
"""
Creates sound-system player commands for all playlist items to be executed at the scheduled time.
Since each scheduled playlist can consist of multiple entry types such as *file*, *live*,
and *stream*, the play-out of the schedule is actually a bit more complex. Before any playlist
entries of the schedule can be turned into sound, they need to be grouped, queued and pre-loaded.
1. First, all entries are aggregated when they hold filesystem entries.
Given you have a playlist with 10 entries, the first 4 are consisting of files, the next two
of a a stream and a live source. The last 4 are files again. These entries are now
aggregated into 4 groups: one for the files, one for the stream, one for the live entry
and another one for files. For each group a timer for executing the next step is created.
2. Now, the playlist entries are going to be "pre-loaded". This means that filesystem
entries are queued and pre-loaded and entries which are based on audio streams are buffered.
This is required to allow a seamless play-out, when its time to do so (in the next stage).
Due to their nature, playlist entries which hold live audio sources are not affected by
this stage at all.
Args:
schedule (Schedule): The schedule this entries belong to
entries ([PlaylistEntry]): The playlist entries to be scheduled for playout
fade_in (Boolean): Fade-in at the beginning of the set of entries
fade_out (Boolean): Fade-out at the end of the set of entries
Returns:
(String): Formatted string to display playlist entries in log
"""
entry_groups = []
entry_groups.append([])
previous_entry = None
index = 0
# Mark entries which start after the end of their schedule or are cut
clean_entries = self.preprocess_entries(entries, True)
# Group/aggregate all filesystem entries, allowing them to be queued at once
for entry in clean_entries:
if previous_entry == None or \
(previous_entry != None and \
previous_entry.get_content_type() == entry.get_content_type() and \
entry.get_content_type() in ResourceClass.FILE.types):
entry_groups[index].append(entry)
else:
index += 1
entry_groups.append([])
entry_groups[index].append(entry)
previous_entry = entry
self.logger.info("Built %s entry group(s)" % len(entry_groups))
# Schedule function calls
do_queue_schedule_end = False
if len(clean_entries) > 0 and len(entry_groups) > 0:
for entries in entry_groups:
if not isinstance(entries, list):
raise ValueError("Invalid Entry Group: %s" % str(entries))
# Create timers for each entry group
self.set_entries_timer(entries, fade_in, fade_out)
# Store them for later reference
schedule.queued_entries = clean_entries
else:
self.logger.warn(SU.red("Nothing to schedule ..."))
def set_entries_timer(self, entries, fade_in, fade_out):
"""
Creates timer for loading and playing one or multiple entries. Existing timers are
updated.
Args:
entries ([]): List of multiple filesystem entries, or a single entry of other types
"""
play_timer = self.is_something_planned_at_time(entries[0].start_unix)
now_unix = Engine.engine_time()
diff = entries[0].start_unix - now_unix
# Play function to be called by timer
def do_play(entries):
self.logger.info(SU.cyan("=== play('%s') ===" % ResourceUtil.get_entries_string(entries)))
transition_type = TransitionType.INSTANT
if fade_in:
transition_type = TransitionType.FADE
if entries[-1].status != EntryPlayState.READY:
self.logger.critical(SU.red("PLAY: For some reason the entry/entries are not yet ready to be played (Entries: %s)" % ResourceUtil.get_entries_string(entries)))
# Let 'em play anyway ...
self.engine.player.play(entries[0], transition_type)
self.logger.info(self.get_ascii_programme())
if play_timer:
# Check if the Playlist IDs are different
if self.have_entries_changed(play_timer, entries):
# If not, stop and remove the old timer, create a new one
self.stop_timer(play_timer)
else:
# If the playlist entries do not differ => reuse the old timer and do nothing
self.logger.debug("Playlist Entry %s is already scheduled - no new timer created." % ResourceUtil.get_entries_string(entries))
return
# If nothing is planned at given time, create a new timer
(entries[0].switchtimer, entries[0].loadtimer) = self.create_timer(diff, do_play, entries, switcher=True)
def have_entries_changed(self, timer, new_entries):
"""
Checks if the new entries and playlists are matching the existing queued ones,
or if they should be updated.
Args:
timer (CallFunctionTimer): The timer holding queued entries
new_entries ([PlaylistEntry]): The possibly updated entries
Returns:
(Boolean): `True` if it has changed
"""
old_entries = timer.entries
if old_entries[0].playlist and new_entries[0].playlist:
if old_entries[0].playlist.playlist_id != new_entries[0].playlist.playlist_id:
return True
if len(old_entries) != len(new_entries):
return True
for old_entry, new_entry in zip(old_entries, new_entries):
if old_entry.source != new_entry.source:
return True
return False
def preprocess_entries(self, entries, cut_oos):
"""
Analyses and marks entries which are going to be cut or excluded.
Args:
entries ([PlaylistEntry]): The playlist entries to be scheduled for playout
cut_oos (Boolean): If `True` entries which are 'out of schedule' are not returned
Returns:
([PlaylistEntry]): The list of processed playlist entries
"""
clean_entries = []
for entry in entries:
if entry.entry_start >= entry.playlist.schedule.schedule_end:
msg = "Filtered entry (%s) after end-of schedule (%s) ... SKIPPED" % (entry, entry.playlist.schedule)
self.logger.warning(SU.red(msg))
entry.queue_state = EntryQueueState.OUT_OF_SCHEDULE
elif entry.end_unix > entry.playlist.schedule.end_unix:
entry.queue_state = EntryQueueState.CUT
else:
entry.queue_state = EntryQueueState.OKAY
if not entry.queue_state == EntryQueueState.OUT_OF_SCHEDULE or not cut_oos:
clean_entries.append(entry)
return clean_entries
def queue_end_of_schedule(self, schedule, fade_out):
"""
Queues a engine action to stop/fade-out the given schedule.
Args:
schedule (PlaylistEntry): The schedule
fade_out (Boolean): If the schedule should be faded-out
"""
schedule_end = schedule.schedule_end
schedule_end_unix = schedule.end_unix
now_unix = Engine.engine_time()
fade_out_time = 0
# Stop function to be called when schedule ends
def do_stop(schedule):
if schedule.has_queued_entries():
last_entry = schedule.queued_entries[-1]
self.logger.info(SU.cyan("=== stop('%s') ===" % str(last_entry.playlist.schedule)))
transition_type = TransitionType.INSTANT
if fade_out:
transition_type = TransitionType.FADE
self.engine.player.stop(last_entry, transition_type)
if fade_out == True:
fade_out_time = int(round(float(self.config.get("fade_out_time")))) #TODO Use float
# Stop any existing fade-out timer
if schedule.fadeouttimer:
schedule.fadeouttimer.cancel()
self.message_timer.remove(schedule.fadeouttimer)
# Create timer to fade-out
start_fade_out = schedule_end_unix - now_unix - fade_out_time
# last_entry = schedule.queued_entries[-1]
schedule.fadeouttimer = self.create_timer(start_fade_out, do_stop, schedule, fadeout=True)
self.logger.info("Fading out schedule in %s seconds at %s | Schedule: %s" % (str(start_fade_out), str(schedule_end), schedule))
def fetch_new_programme(self):
"""
Fetch the latest programme from `AuraCalendarService` which stores it to the database.
After that, the programme is in turn loaded from the database and stored in `self.programme`.
"""
# Fetch programme from API endpoints
self.logger.debug("Trying to fetch new programe from API endpoints...")
acs = AuraCalendarService(self.config)
queue = acs.get_queue()
acs.start() # start fetching thread
response = queue.get() # wait for the end
self.logger.debug("... Programme fetch via API done!")
# Reset last successful fetch state
lsf = self.last_successful_fetch
self.last_successful_fetch = None
if response is None:
msg = SU.red("Trying to load programme from Engine Database, because AuraCalendarService returned an empty response.")
self.logger.warning(msg)
elif type(response) is list:
self.programme = response
if self.programme is not None and len(self.programme) > 0:
self.last_successful_fetch = datetime.now()
self.logger.info(SU.green("Finished fetching current programme from API"))
if len(self.programme) == 0:
self.logger.critical("Programme fetched from Steering/Tank has no entries!")
elif response.startswith("fetching_aborted"):
msg = SU.red("Trying to load programme from database only, because fetching was being aborted from AuraCalendarService! Reason: ")
self.logger.warning(msg + response[16:])
else:
msg = SU.red("Trying to load programme from database only, because of an unknown response from AuraCalendarService: " + response)
self.logger.warning(msg)
# Always load latest programme from the database
self.last_successful_fetch = lsf
self.load_programme_from_db()
self.logger.info(SU.green("Finished loading current programme from database (%s schedules)" % str(len(self.programme))))
for schedule in self.programme:
self.logger.debug("\tSchedule %s with Playlist %s" % (str(schedule), str(schedule.playlist)))
def load_programme_from_db(self):
"""
Loads the programme from Engine's database and enables
them via `self.enable_entries(..)`. After that, the
current message queue is printed to the console.
"""
self.programme = Schedule.select_programme()
if not self.programme:
self.logger.critical(SU.red("Could not load programme from database. We are in big trouble my friend!"))
return
def is_something_planned_at_time(self, given_time):
"""
Checks for existing timers at the given time.
"""
for t in self.message_timer:
if t.fadein or t.switcher:
if t.entries[0].start_unix == given_time:
return t
return False
def create_timer(self, diff, func, param, fadein=False, fadeout=False, switcher=False):
"""
Creates a new timer for timed execution of mixer commands.
Args:
diff (Integer): The difference in seconds from now, when the call should happen
func (Function): The function to call
param ([]): A schedule or list of entries
Returns:
(CallFunctionTimer, CallFunctionTimer): In case of a "switch" command, the switch and pre-roll timer is returned
(CallFunctionTimer): In all other cases only the timer for the command is returned
"""
if not fadein and not fadeout and not switcher or fadein and fadeout or fadein and switcher or fadeout and switcher:
raise ValueError("You have to call me with either fadein=true, fadeout=true or switcher=True")
if not isinstance(param, list) and not isinstance(param, Schedule):
raise ValueError("No list of entries nor schedule passed!")
t = CallFunctionTimer(diff=diff, func=func, param=param, fadein=fadein, fadeout=fadeout, switcher=switcher)
self.message_timer.append(t)
t.start()
if switcher:
# Pre-roll function to be called by timer
def do_preload(entries):
try:
if entries[0].get_content_type() in ResourceClass.FILE.types:
self.logger.info(SU.cyan("=== preload_group('%s') ===" % ResourceUtil.get_entries_string(entries)))
self.engine.player.preload_group(entries, ChannelType.QUEUE)
else:
self.logger.info(SU.cyan("=== preload('%s') ===" % ResourceUtil.get_entries_string(entries)))
self.engine.player.preload(entries[0])
except LoadSourceException as e:
self.logger.critical(SU.red("Could not pre-roll entries %s" % ResourceUtil.get_entries_string(entries)), e)
if entries[-1].status != EntryPlayState.READY:
self.logger.critical(SU.red("Entries didn't reach 'ready' state during pre-rolling (Entries: %s)" % ResourceUtil.get_entries_string(entries)))
loader_diff = diff - self.config.get("preload_offset")
loader = CallFunctionTimer(diff=loader_diff, func=do_preload, param=param, fadein=fadein, fadeout=fadeout, switcher=False, loader=True)
self.message_timer.append(loader)
loader.start()
return (t, loader)
else:
return t
def stop_timer(self, timer):
"""
Stops the given timer.
Args:
timer (Timer): The timer to stop.
"""
timer.cancel()
count = 1
for entry in timer.entries:
if entry.loadtimer is not None:
entry.loadtimer.cancel()
self.message_timer.remove(entry.loadtimer)
count += 1
# if timer.entries[0].fadeintimer is not None:
# timer.entries[0].fadeintimer.cancel()
# self.message_timer.remove(timer.entries[0].fadeintimer)
# count += 1
# if entry.fadeouttimer is not None:
# entry.fadeouttimer.cancel()
# self.message_timer.remove(entry.fadeouttimer)
# count += 1
# Remove it from message queue
self.message_timer.remove(timer)
self.logger.info("Stopped %s timers for: %s" % (str(count), ResourceUtil.get_entries_string(timer.entries)))
# FIXME Move to adequate module
@staticmethod
def init_database():
"""
Initializes the database.
Raises:
sqlalchemy.exc.ProgrammingError: In case the DB model is invalid
"""
if AuraConfig.config().get("recreate_db") is not None:
AuraDatabaseModel.recreate_db(systemexit=True)
# Check if tables exists, if not create them
try:
Playlist.is_empty()
except sqlalchemy.exc.ProgrammingError as e:
errcode = e.orig.args[0]
if errcode == 1146: # Error for no such table
model = AuraDatabaseModel()
model.recreate_db()
else:
raise
def terminate(self):
"""
Called when thread is stopped or a signal to terminate is received.
"""
self.exit_event.set()
self.logger.info("Shutting down scheduler ...")
# ------------------------------------------------------------------------------------------ #
class CallFunctionTimer(threading.Timer):
logger = None
param = None
entries = None
diff = None
dt = None
fadein = False
fadeout = False
switcher = False
loader = False
def __init__(self, diff=None, func=None, param=None, fadein=False, fadeout=False, switcher=False, loader=False):
self.logger = logging.getLogger("AuraEngine")
self.logger.debug("Executing engine command '%s' in %s seconds..." % (str(func.__name__), str(diff)))
threading.Timer.__init__(self, diff, func, (param,))
if not fadein and not fadeout and not switcher and not loader \
or fadein and fadeout \
or fadein and switcher \
or fadeout and switcher:
raise Exception("You have to create me with either fadein=True, fadeout=True or switcher=True")
self.diff = diff
self.dt = datetime.now() + timedelta(seconds=diff)
self.func = func
self.param = param
self.entries = param # TODO Refactor since param can hold [entries] or a schedule, depending on the timer type
self.fadein = fadein
self.fadeout = fadeout
self.switcher = switcher
self.loader = loader
def __str__(self):
"""
String represenation of the timer.
"""
status = "Timer (Alive: %s)" % self.is_alive()
status += " starting at " + str(self.dt)
if self.fadein:
return status + " fading in entries '" + ResourceUtil.get_entries_string(self.entries)
elif self.fadeout:
return status + " fading out schedule '" + str(self.param)
elif self.switcher:
return status + " switching to entries '" + ResourceUtil.get_entries_string(self.entries)
elif self.loader:
return status + " pre-rolling entries '" + ResourceUtil.get_entries_string(self.entries)
else:
return "CORRUPTED CallFunctionTimer around! How can that be?"
; supervisor config file
[unix_http_server]
file=/opt/aura/engine/tmp/supervisor.sock ; (the path to the socket file)
chmod=0700 ; sockef file mode (default 0700)
chown=engineuser:engineuser
[supervisord]
logfile=/var/log/supervisor/supervisord.log ; (main log file;default $CWD/supervisord.log)
pidfile=/opt/aura/engine/tmp/supervisord.pid ; (supervisord pidfile;default supervisord.pid)
childlogdir=/var/log/supervisor ; ('AUTO' child log dir, default $TEMP)
; the below section must remain in the config file for RPC
; (supervisorctl/web interface) to work, additional interfaces may be
; added by defining them in separate rpcinterface: sections
[rpcinterface:supervisor]
supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface
[supervisorctl]
serverurl=unix:///opt/aura/engine/tmp/supervisor.sock ; use a unix:// URL for a unix socket
; The [include] section can just contain the "files" setting. This
; setting can list multiple files (separated by whitespace or
; newlines). It can also contain wildcards. The filenames are
; interpreted as relative to this file. Included files *cannot*
; include files themselves.
[include]
; files = /etc/supervisor/conf.d/*.conf
files = /opt/aura/engine/config/supervisor/*.conf