From 5ee5f19255d89620f45471e22b45adc39dc4e952 Mon Sep 17 00:00:00 2001 From: "G. Gibson" Date: Fri, 2 Oct 2026 19:19:44 -0700 Subject: [PATCH] Added scheduler.rb --- scheduler.rb | 265 +++++++++++++++++++++++++++++++++++++++++++++++++++ station.liq | 28 ++++-- 2 files changed, 287 insertions(+), 6 deletions(-) create mode 100755 scheduler.rb diff --git a/scheduler.rb b/scheduler.rb new file mode 100755 index 0000000..cb01902 --- /dev/null +++ b/scheduler.rb @@ -0,0 +1,265 @@ +#!/usr/bin/env ruby +# scheduler.rb - Clock-driven show/stream scheduler for the radio automation. +# +# Designed to run every minute from cron as the liquidsoap user: +# * * * * * cd /srv/radio && ./run_radio.sh scheduler.rb >> /mnt/storage/radio/logs/scheduler_cron.log 2>&1 +# +# Reads schedule.txt (cron-style: min hour dom mon dow TYPE TARGET [RUNLENGTH]), +# matches entries against the current time, and writes the resolved URI into +# /queues/current_queue.txt so station.liq picks it up within ~2s. +# +# Resolution rules: +# stream -> the URL is written verbatim. RUNLENGTH (seconds) is carried in a +# comment line; liquidsoap plays the live stream until the next +# scheduled event overwrites the file. +# show -> look up the slug in subscriptions.db, find the next unplayed +# episode in played.db (prefer local_path, fall back to url), +# mark it played, and write that path/url. +# +# Idempotent: a fire marker prevents re-firing the same entry twice in one +# minute if cron double-runs or the script is invoked manually. + +require 'json' +require 'sequel' +require 'fileutils' +require 'time' + +SCRIPT_DIR = File.expand_path(File.dirname(__FILE__)) +CONFIG_PATH = File.join(SCRIPT_DIR, "config.json") +SCHEDULE_FILE = File.join(SCRIPT_DIR, "schedule.txt") +QUEUE_DIR = nil # derived below +QUEUE_FILE = nil # derived below +FIRE_MARKER_DIR = nil + +CFG = JSON.parse(File.read(CONFIG_PATH)) +STORAGE = CFG["storage"] +STATE_DIR = File.join(STORAGE, "state") +SUBS_DB = File.join(STATE_DIR, "subscriptions.db") +PLAYED_DB = File.join(STATE_DIR, "played.db") +LOG_DIR = File.join(STORAGE, "logs") + +QUEUE_DIR = File.join(STORAGE, "queues") +QUEUE_FILE = File.join(QUEUE_DIR, "current_queue.txt") +FIRE_MARKER_DIR = File.join(STATE_DIR, "fire_markers") + +FileUtils.mkdir_p(QUEUE_DIR) +FileUtils.mkdir_p(FIRE_MARKER_DIR) + +def log(msg) + puts "#{Time.now.strftime('%Y-%m-%d %H:%M:%S')} #{msg}" +end + +# --- Cron field matching ------------------------------------------------------ +# Fields: min (0-59), hour (0-23), dom (1-31), mon (1-12), dow (0-6, Sun=0). +# Supports: "*", "N", "A-B", "*/S", and comma lists of any of those. + +def field_matches?(field, value) + return true if field == "*" + + field.split(",").any? do |part| + if (m = part.match(/\A\*\/(\d+)\z/)) + value % m[1].to_i == 0 + elsif (m = part.match(/\A(\d+)-(\d+)\z/)) + lo, hi = m[1].to_i, m[2].to_i + value >= lo && value <= hi + elsif part.match?(/\A\d+\z/) + part.to_i == value + else + false + end + end +end + +def entry_matches_now?(entry, now) + min_f, hour_f, dom_f, mon_f, dow_f = entry[:fields] + field_matches?(min_f, now.min) && + field_matches?(hour_f, now.hour) && + field_matches?(dom_f, now.day) && + field_matches?(mon_f, now.month) && + field_matches?(dow_f, now.wday) +end + +# --- Schedule parsing --------------------------------------------------------- +# Format: min hour dom mon dow TYPE TARGET [RUNLENGTH_SECONDS] +# Lines starting with # and blank lines are ignored. + +def parse_schedule(path) + entries = [] + File.readlines(path).each do |raw| + line = raw.strip + next if line.empty? || line.start_with?("#") + + parts = line.split(/\s+/) + next if parts.length < 6 + + fields = parts[0..4] + type = parts[5].downcase + target = parts[6] + runlength = parts[7]&.to_i + + unless %w[show stream].include?(type) + log("WARN: unknown TYPE '#{type}' in line: #{line}") + next + end + + entries << { + fields: fields, + type: type, + target: target, + runlength: runlength, + raw: line + } + end + entries +end + +# --- Fire markers (idempotency) ----------------------------------------------- +# Marker filename encodes date + entry index so each entry fires at most once +# per calendar minute. Old markers (>1 day) are pruned on each run. + +def fire_marker_key(entry_index, now) + "#{now.strftime('%Y%m%d%H%M')}_#{entry_index}" +end + +def already_fired?(key) + File.exist?(File.join(FIRE_MARKER_DIR, key)) +end + +def mark_fired(key) + File.write(File.join(FIRE_MARKER_DIR, key), Time.now.iso8601) +end + +def prune_old_markers(now) + cutoff = now - 86_400 + Dir.glob(File.join(FIRE_MARKER_DIR, "*")).each do |f| + begin + File.delete(f) if File.mtime(f) < cutoff + rescue => e + log("WARN: could not prune marker #{f}: #{e.message}") + end + end +end + +# --- Database helpers ---------------------------------------------------------- + +def open_subs_db + Sequel.sqlite(SUBS_DB, database_options: { timeout: 5 }) +end + +def open_played_db + Sequel.sqlite(PLAYED_DB, database_options: { timeout: 5 }) +end + +# Resolve a show slug to the next unplayed episode's playable path/url. +# Returns a String (local path or remote URL) or nil if nothing is available. +def resolve_show(slug) + subs_db = open_subs_db + played_db = open_played_db + + show = subs_db[:shows].where(slug: slug).first + if show.nil? + log("WARN: show slug '#{slug}' not found in subscriptions.db") + return nil + end + + ep = played_db[:episodes] + .where(show_guid: show[:guid], played: 0) + .order(Sequel.desc(:created_at)) + .first + + if ep.nil? + log("INFO: no unplayed episodes for show '#{slug}' (#{show[:title]})") + return nil + end + + playable = ep[:local_path] || ep[:url] + if playable.nil? || playable.to_s.strip.empty? + log("WARN: episode #{ep[:guid]} for '#{slug}' has neither local_path nor url") + return nil + end + + # Mark as played atomically before returning, so a crash mid-fire doesn't + # cause a double-play on the next run. + played_db[:episodes].where(guid: ep[:guid]).update(played: 1) + log("RESOLVED: show '#{slug}' -> #{playable} (episode: #{ep[:title]})") + playable +ensure + subs_db&.disconnect + played_db&.disconnect +end + +# --- Queue writer -------------------------------------------------------------- +# Writes the URI (and optionally a runlength hint) atomically: write to a temp +# file in the same directory, then rename over the target. This prevents +# station.liq from reading a partially-written file. + +def write_queue(uri, runlength_sec = nil) + tmp = "#{QUEUE_FILE}.tmp" + content = uri + if runlength_sec && runlength_sec > 0 + content += "\n# runlength=#{runlength_sec}" + end + File.write(tmp, content) + File.rename(tmp, QUEUE_FILE) + log("QUEUED: #{uri}#{runlength_sec ? " (runlength=#{runlength_sec}s)" : ''}") +end + +# --- Main ---------------------------------------------------------------------- + +def main + now = Time.now + log("--- scheduler tick #{now.strftime('%Y-%m-%d %H:%M:%S')} ---") + + unless File.exist?(SCHEDULE_FILE) + log("ERROR: schedule.txt not found at #{SCHEDULE_FILE}") + return + end + + prune_old_markers(now) + entries = parse_schedule(SCHEDULE_FILE) + log("Loaded #{entries.length} schedule entries") + + fired_any = false + entries.each_with_index do |entry, idx| + unless entry_matches_now?(entry, now) + next + end + + key = fire_marker_key(idx, now) + if already_fired?(key) + log("SKIP (already fired this minute): #{entry[:raw]}") + next + end + + log("MATCH: #{entry[:raw]}") + + case entry[:type] + when "stream" + write_queue(entry[:target], entry[:runlength]) + mark_fired(key) + fired_any = true + + when "show" + uri = resolve_show(entry[:target]) + if uri + write_queue(uri) + mark_fired(key) + fired_any = true + else + # No playable episode right now; don't mark fired so we retry next + # minute (fetch_podcasts may populate new episodes shortly). + log("DEFERRED: no playable episode yet for '#{entry[:target]}', will retry next minute") + end + end + end + + log(fired_any ? "--- tick complete (fired) ---" : "--- tick complete (no matches) ---") +end + +begin + main +rescue => e + log("ERROR: #{e.class}: #{e.message}") + log(e.backtrace.first(5).join("\n")) + exit 1 +end diff --git a/station.liq b/station.liq index 2aa7420..453e47b 100644 --- a/station.liq +++ b/station.liq @@ -14,7 +14,10 @@ let json.parse ({ storage, icecast = { host, port, mount, source_username, sourc = file.contents("config.json") storage_base = storage -music_dir = "#{storage}/music" +music_dir = "#{storage}/music" +jingles_dir = "#{storage}/jingles" +announcements_dir = "#{storage}/announcements" +podcasts_dir = "#{storage}/podcasts" queue_dir = "#{storage}/queues" icecast_host = host @@ -25,19 +28,32 @@ icecast_pass = source_password print("Config loaded: storage=#{storage_base}, icecast=#{icecast_host}:#{icecast_port}#{icecast_mount}") -# --- Background music playlist (loops forever by default) --- -background_music = playlist(music_dir) +# --- Background source: rotate among the featured directories, with 0.5 s +# crossfades between tracks. Each playlist(dir) reads its directory +# recursively, so audio in sub-directories is included automatically. +music_pl = playlist(music_dir) +jingles_pl = playlist(jingles_dir) +announcements_pl = playlist(announcements_dir) +podcasts_pl = playlist(podcasts_dir) + +# Rotate among the four directories; transition_length bounds each handoff. +rotating_bg = rotate(transition_length=0.5, [music_pl, jingles_pl, announcements_pl, podcasts_pl]) + +# Crossfade 0.5 s between consecutive tracks of the rotated stream. +background_music = crossfade(fade_in=0.5, fade_out=0.5, rotating_bg) # --- Show queue source (reads current_queue.txt each tick) --- # In 2.2.x the callback must return a request (type ()->request), not null. -# When the queue is empty we play a short synthesized tone so the source -# stays up without needing an external file. +# When the queue is empty we return a NON-resolving URI ("noshow:"). Liquidsoap +# fails to resolve it, so this source goes DOWN -- which is exactly what lets +# fallback() drop through to the rotating background music below. (A resolvable +# sentinel like synth: would keep the show source up forever, masking the music.) def read_queue() qf = "#{queue_dir}/current_queue.txt" if file.exists(qf) and file.contents(qf) != "" then request.create(file.contents(qf)) else - request.create("synth:shape=sine,frequency=440.,duration=1.") + request.create("noshow:") end end