350 lines
14 KiB
Python
350 lines
14 KiB
Python
import copy
|
|
import logging
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from queue import Empty, Queue
|
|
from subprocess import DEVNULL, PIPE, run
|
|
from threading import Thread, Timer
|
|
from typing import Any, Dict, Union
|
|
|
|
from libretime_api_client.v1 import ApiClient as LegacyClient
|
|
from libretime_api_client.v2 import ApiClient
|
|
from requests import RequestException
|
|
|
|
from ..config import CACHE_DIR, POLL_INTERVAL, Config
|
|
from ..liquidsoap.client import LiquidsoapClient
|
|
from ..liquidsoap.models import Info, MessageFormatKind, StreamPreferences, StreamState
|
|
from .events import Events, FileEvent, FileEvents
|
|
from .liquidsoap import Liquidsoap
|
|
from .schedule import get_schedule, receive_schedule
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
here = Path(__file__).parent
|
|
|
|
|
|
# pylint: disable=too-many-instance-attributes
|
|
class PypoFetch(Thread):
|
|
name = "fetch"
|
|
daemon = True
|
|
|
|
# pylint: disable=too-many-arguments
|
|
def __init__(
|
|
self,
|
|
fetch_queue: "Queue[Dict[str, Any]]",
|
|
push_queue: "Queue[Events]",
|
|
file_queue: "Queue[FileEvents]",
|
|
liq_client: LiquidsoapClient,
|
|
liquidsoap: Liquidsoap,
|
|
config: Config,
|
|
api_client: ApiClient,
|
|
legacy_client: LegacyClient,
|
|
):
|
|
Thread.__init__(self)
|
|
|
|
self.api_client = api_client
|
|
self.legacy_client = legacy_client
|
|
self.fetch_queue = fetch_queue
|
|
self.push_queue = push_queue
|
|
self.media_prepare_queue = file_queue
|
|
self.last_update_schedule_timestamp = time.time()
|
|
self.config = config
|
|
self.listener_timeout = POLL_INTERVAL
|
|
|
|
self.liq_client = liq_client
|
|
self.liquidsoap = liquidsoap
|
|
|
|
self.cache_dir = CACHE_DIR
|
|
logger.debug("Cache dir %s", self.cache_dir)
|
|
|
|
self.schedule_data: Events = {}
|
|
logger.info("PypoFetch: init complete")
|
|
|
|
# Handle a message from RabbitMQ, put it into our yucky global var.
|
|
# Hopefully there is a better way to do this.
|
|
|
|
def handle_message(self, message: Dict[str, Any]) -> None:
|
|
try:
|
|
command = message["event_type"]
|
|
logger.debug("handling event %s: %s", command, message)
|
|
|
|
if command == "update_schedule":
|
|
self.schedule_data = receive_schedule(message["schedule"]["media"])
|
|
self.process_schedule(self.schedule_data)
|
|
elif command == "reset_liquidsoap_bootstrap":
|
|
self.set_bootstrap_variables()
|
|
elif command == "update_stream_format":
|
|
logger.info("Updating stream format...")
|
|
self.update_liquidsoap_stream_format(message["stream_format"])
|
|
elif command == "update_message_offline":
|
|
logger.info("Updating message offline...")
|
|
self.update_liquidsoap_message_offline(message["message_offline"])
|
|
elif command == "update_station_name":
|
|
logger.info("Updating station name...")
|
|
self.update_liquidsoap_station_name(message["station_name"])
|
|
elif command == "update_transition_fade":
|
|
logger.info("Updating transition_fade...")
|
|
self.update_liquidsoap_transition_fade(message["transition_fade"])
|
|
elif command == "switch_source":
|
|
logger.info("switch_on_source show command received...")
|
|
self.liquidsoap.telnet_liquidsoap.switch_source(
|
|
message["sourcename"], message["status"]
|
|
)
|
|
elif command == "disconnect_source":
|
|
logger.info("disconnect_on_source show command received...")
|
|
self.liquidsoap.telnet_liquidsoap.disconnect_source(
|
|
message["sourcename"]
|
|
)
|
|
else:
|
|
logger.info("Unknown command: %s", command)
|
|
|
|
# update timeout value
|
|
if command == "update_schedule":
|
|
self.listener_timeout = POLL_INTERVAL
|
|
else:
|
|
self.listener_timeout = max(
|
|
self.last_update_schedule_timestamp - time.time() + POLL_INTERVAL,
|
|
0,
|
|
)
|
|
logger.info("New timeout: %s", self.listener_timeout)
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception(exception)
|
|
|
|
# Initialize Liquidsoap environment
|
|
def set_bootstrap_variables(self) -> None:
|
|
logger.debug("Getting information needed on bootstrap from Airtime")
|
|
try:
|
|
info = Info(**self.api_client.get_info().json())
|
|
preferences = StreamPreferences(
|
|
**self.api_client.get_stream_preferences().json()
|
|
)
|
|
state = StreamState(**self.api_client.get_stream_state().json())
|
|
|
|
except RequestException as exception:
|
|
logger.exception("Unable to get stream settings: %s", exception)
|
|
return
|
|
|
|
logger.debug("info: %s", info)
|
|
logger.debug("preferences: %s", preferences)
|
|
logger.debug("state: %s", state)
|
|
|
|
try:
|
|
self.liquidsoap.liq_client.settings_update(
|
|
station_name=info.station_name,
|
|
message_format=preferences.message_format,
|
|
message_offline=preferences.message_offline,
|
|
input_fade_transition=preferences.input_fade_transition,
|
|
)
|
|
|
|
self.liquidsoap.liq_client.source_switch_status(
|
|
name="master_dj",
|
|
streaming=state.input_main_streaming,
|
|
)
|
|
self.liquidsoap.liq_client.source_switch_status(
|
|
name="live_dj",
|
|
streaming=state.input_show_streaming,
|
|
)
|
|
self.liquidsoap.liq_client.source_switch_status(
|
|
name="scheduled_play",
|
|
streaming=state.schedule_streaming,
|
|
)
|
|
|
|
except OSError as exception:
|
|
logger.exception(exception)
|
|
|
|
self.liquidsoap.clear_all_queues()
|
|
self.liquidsoap.clear_queue_tracker()
|
|
|
|
def update_liquidsoap_stream_format(
|
|
self,
|
|
stream_format: Union[MessageFormatKind, int],
|
|
) -> None:
|
|
try:
|
|
self.liq_client.settings_update(message_format=stream_format)
|
|
except OSError as exception:
|
|
logger.exception(exception)
|
|
|
|
def update_liquidsoap_message_offline(self, message_offline: str) -> None:
|
|
try:
|
|
self.liq_client.settings_update(message_offline=message_offline)
|
|
except OSError as exception:
|
|
logger.exception(exception)
|
|
|
|
def update_liquidsoap_transition_fade(self, fade: float) -> None:
|
|
try:
|
|
self.liq_client.settings_update(input_fade_transition=fade)
|
|
except OSError as exception:
|
|
logger.exception(exception)
|
|
|
|
def update_liquidsoap_station_name(self, station_name: str) -> None:
|
|
try:
|
|
self.liq_client.settings_update(station_name=station_name)
|
|
except OSError as exception:
|
|
logger.exception(exception)
|
|
|
|
# Process the schedule
|
|
# - Reads the scheduled entries of a given range (actual time +/- "prepare_ahead" /
|
|
# "cache_for")
|
|
# - Saves a serialized file of the schedule
|
|
# - playlists are prepared. (brought to liquidsoap format) and, if not mounted via
|
|
# nsf, files are copied to the cache dir
|
|
# (Folder-structure: cache/YYYY-MM-DD-hh-mm-ss)
|
|
# - runs the cleanup routine, to get rid of unused cached files
|
|
|
|
def process_schedule(self, events: Events) -> None:
|
|
self.last_update_schedule_timestamp = time.time()
|
|
logger.debug(events)
|
|
file_events: FileEvents = {}
|
|
all_events: Events = {}
|
|
|
|
# Download all the media and put playlists in liquidsoap "annotate" format
|
|
try:
|
|
for key in events:
|
|
item = events[key]
|
|
if isinstance(item, FileEvent):
|
|
file_events[key] = item
|
|
all_events[key] = item
|
|
|
|
self.media_prepare_queue.put(copy.copy(file_events))
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception(exception)
|
|
|
|
# Send the data to pypo-push
|
|
logger.debug("Pushing to pypo-push")
|
|
self.push_queue.put(all_events)
|
|
|
|
# cleanup
|
|
try:
|
|
self.cache_cleanup(events)
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception(exception)
|
|
|
|
def is_file_opened(self, path: str) -> bool:
|
|
result = run(["lsof", "--", path], stdout=PIPE, stderr=DEVNULL, check=False)
|
|
return bool(result.stdout)
|
|
|
|
def cache_cleanup(self, events: Events):
|
|
"""
|
|
Get list of all files in the cache dir and remove them if they aren't being used
|
|
anymore.
|
|
Input dict() media, lists all files that are scheduled or currently playing. Not
|
|
being in this dict() means the file is safe to remove.
|
|
"""
|
|
cached_file_set = set(os.listdir(self.cache_dir))
|
|
scheduled_file_set = set()
|
|
|
|
for key in events:
|
|
item = events[key]
|
|
if isinstance(item, FileEvent):
|
|
scheduled_file_set.add(item.local_filepath.name)
|
|
|
|
expired_files = cached_file_set - scheduled_file_set
|
|
|
|
logger.debug("Files to remove %s", str(expired_files))
|
|
for expired_file in expired_files:
|
|
try:
|
|
expired_filepath = os.path.join(self.cache_dir, expired_file)
|
|
logger.debug("Removing %s", expired_filepath)
|
|
|
|
# check if this file is opened (sometimes Liquidsoap is still
|
|
# playing the file due to our knowledge of the track length
|
|
# being incorrect!)
|
|
if not self.is_file_opened(expired_filepath):
|
|
os.remove(expired_filepath)
|
|
logger.info("File '%s' removed", expired_filepath)
|
|
else:
|
|
logger.info("File '%s' not removed. Still busy!", expired_filepath)
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception(
|
|
"Problem removing file '%s': %s", expired_file, exception
|
|
)
|
|
|
|
def manual_schedule_fetch(self) -> bool:
|
|
try:
|
|
self.schedule_data = get_schedule(self.api_client)
|
|
logger.debug("Received event from API client: %s", self.schedule_data)
|
|
self.process_schedule(self.schedule_data)
|
|
return True
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception("Unable to fetch schedule: %s", exception)
|
|
return False
|
|
|
|
def persistent_manual_schedule_fetch(self, max_attempts=1) -> bool:
|
|
success = False
|
|
num_attempts = 0
|
|
while not success and num_attempts < max_attempts:
|
|
success = self.manual_schedule_fetch()
|
|
num_attempts += 1
|
|
|
|
return success
|
|
|
|
# This function makes a request to Airtime to see if we need to
|
|
# push metadata to TuneIn. We have to do this because TuneIn turns
|
|
# off metadata if it does not receive a request every 5 minutes.
|
|
def update_metadata_on_tunein(self):
|
|
self.legacy_client.update_metadata_on_tunein()
|
|
Timer(120, self.update_metadata_on_tunein).start()
|
|
|
|
def main(self):
|
|
# Make sure all Liquidsoap queues are empty. This is important in the
|
|
# case where we've just restarted the pypo scheduler, but Liquidsoap still
|
|
# is playing tracks. In this case let's just restart everything from scratch
|
|
# so that we can repopulate our dictionary that keeps track of what
|
|
# Liquidsoap is playing much more easily.
|
|
self.liquidsoap.clear_all_queues()
|
|
|
|
self.set_bootstrap_variables()
|
|
|
|
self.update_metadata_on_tunein()
|
|
|
|
# Bootstrap: since we are just starting up, we need to grab the
|
|
# most recent schedule. After that we fetch the schedule every 8
|
|
# minutes or wait for schedule updates to get pushed.
|
|
success = self.persistent_manual_schedule_fetch(max_attempts=5)
|
|
|
|
if success:
|
|
logger.info("Bootstrap schedule received: %s", self.schedule_data)
|
|
|
|
loops = 1
|
|
while True:
|
|
logger.info("Loop #%s", loops)
|
|
manual_fetch_needed = False
|
|
try:
|
|
# our simple_queue.get() requires a timeout, in which case we
|
|
# fetch the Airtime schedule manually. It is important to fetch
|
|
# the schedule periodically because if we didn't, we would only
|
|
# get schedule updates via RabbitMq if the user was constantly
|
|
# using the Airtime interface.
|
|
|
|
# If the user is not using the interface, RabbitMq messages are not
|
|
# sent, and we will have very stale (or non-existent!) data about the
|
|
# schedule.
|
|
|
|
# Currently we are checking every POLL_INTERVAL seconds
|
|
|
|
message = self.fetch_queue.get(
|
|
block=True, timeout=self.listener_timeout
|
|
)
|
|
manual_fetch_needed = False
|
|
self.handle_message(message)
|
|
except Empty:
|
|
logger.info("Queue timeout. Fetching schedule manually")
|
|
manual_fetch_needed = True
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception(exception)
|
|
|
|
try:
|
|
if manual_fetch_needed:
|
|
self.persistent_manual_schedule_fetch(max_attempts=5)
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
logger.exception("Failed to manually fetch the schedule: %s", exception)
|
|
|
|
loops += 1
|
|
|
|
def run(self):
|
|
"""
|
|
Entry point of the thread
|
|
"""
|
|
self.main()
|