@@ 1,67 @@
+{
+ "_meta": {
+ "hash": {
+ "sha256": "f571171dc12e1cf40d7ceadf4496659cc572eb4953c01d3d1430446cf0d8d20c"
+ },
+ "pipfile-spec": 6,
+ "requires": {
+ "python_version": "3.10"
+ },
+ "sources": [
+ {
+ "name": "pypi",
+ "url": "https://pypi.org/simple",
+ "verify_ssl": true
+ }
+ ]
+ },
+ "default": {
+ "icalendar": {
+ "hashes": [
+ "sha256:35faeae20e58d0fe26e33c04bbfd08b424192d61ae3e1da1d7611ba06ab33570",
+ "sha256:41f2f450707a503365a81c22954be4053779994c8ff01fc11bf4ce33a724092e"
+ ],
+ "markers": "python_version >= '3.7'",
+ "version": "==5.0.1"
+ },
+ "python-dateutil": {
+ "hashes": [
+ "sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86",
+ "sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9"
+ ],
+ "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
+ "version": "==2.8.2"
+ },
+ "pytz": {
+ "hashes": [
+ "sha256:222439474e9c98fced559f1709d89e6c9cbf8d79c794ff3eb9f8800064291427",
+ "sha256:e89512406b793ca39f5971bc999cc538ce125c0e51c27941bef4568b460095e2"
+ ],
+ "version": "==2022.6"
+ },
+ "recurring-ical-events": {
+ "hashes": [
+ "sha256:0aed6a5965c719e50aa2e5f6be896688cdb2c2da7087fae928ad4ac05ff0785f",
+ "sha256:c2dd86ccbb03891ce17f353f79d0a10779199270dbe3cb63b4f8fe387360dea5"
+ ],
+ "index": "pypi",
+ "version": "==1.0.3b0"
+ },
+ "six": {
+ "hashes": [
+ "sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926",
+ "sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"
+ ],
+ "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
+ "version": "==1.16.0"
+ },
+ "x-wr-timezone": {
+ "hashes": [
+ "sha256:c05cb34b9b58a4607a788db086dcae5766728e4b94e0672870dc5593a6e13fe6",
+ "sha256:e438b27b96635f5f712a4fb5dda4c82597a53a412fe834c9fe8409fddb3fc2b1"
+ ],
+ "version": "==0.0.5"
+ }
+ },
+ "develop": {}
+}
@@ 1,158 @@
+from datetime import datetime, timedelta, timezone
+from icalendar import Calendar, vText
+from time import sleep
+import recurring_ical_events
+import urllib.request
+import os
+import subprocess
+import argparse
+
+parser = argparse.ArgumentParser(description='Play audio various web sources using an iCalendar URL as a scheduler')
+parser.add_argument('ical_url', metavar='https://example.com/radio.ics', help='The .ics endpoint where a series of events with audio URLs in their `location` fields can be retrieved.')
+parser.add_argument('--fallback', dest='fallback_url', action='store', help='A fallback audio source URL to play when nothing is on the schedule/queue.')
+
+args = parser.parse_args()
+fallback_url = args.fallback_url
+ical_url = args.ical_url
+
+# convenience function for working with icalendar library's vText format
+def get_text(e, k):
+ return vText.to_ical(e[k]).decode("utf-8")
+
+def get_start(e):
+ return e['DTSTART'].dt.timestamp()
+
+# used as comparator to sort events
+def get_created(e):
+ return e['CREATED'].dt.timestamp()
+
+def is_opaque(e):
+ return get_text(e, 'TRANSP') == 'OPAQUE'
+
+def print_event(e):
+ print(get_text(e, 'SUMMARY'))
+
+# main function; waits for player to finish and then calls itself
+def tuner():
+ url = fallback_url
+ event_to_play = None
+ queue_event_will_play = False
+
+ # pull the .ics from web, parse it with icalendar library, find events that are happening now
+ ical_string = urllib.request.urlopen(ical_url).read()
+ now = datetime.now(timezone.utc).astimezone()
+ calendar = Calendar.from_ical(ical_string)
+ events = recurring_ical_events.of(calendar).between(now, now)
+
+ print("Found", len(events), "events")
+
+ # split events into scheduled shows vs listening queue items; use the TRANSP property, which google cal sets to TRANSPARENT for all-day events (my current solution for adding listening queue items to a cal), to determine the difference
+ # TODO also support using the VTODO feature for listening queue items, which would be the preferred way of doing it with a custom .ics-building (web providers including google cal don't support VTODO)
+ queue_events = []
+ scheduled_events = []
+ for e in events:
+ if get_text(e, "TRANSP") == "TRANSPARENT":
+ queue_events.append(e)
+ else:
+ scheduled_events.append(e)
+
+ if len(scheduled_events) > 0:
+ print("Found", len(scheduled_events), "scheduled event(s):")
+ scheduled_events.sort(key=get_created, reverse=True)
+ list(map(print_event, scheduled_events))
+ print()
+ event_to_play = scheduled_events[0]
+ url = get_text(event_to_play, "LOCATION")
+ print("Selected URL:", url)
+
+ elif len(queue_events) > 0:
+ print("Found no scheduled events but", len(queue_events), "queue entries:")
+ queue_events.sort(key=get_created, reverse=True)
+ list(map(print_event, queue_events))
+ print()
+ event_to_play = queue_events[0]
+ queue_event_will_play = True
+
+ # check for a file that stores last played queue url
+ try:
+ with open('lastplayedqueue', encoding="utf-8") as f:
+ last_played_url = f.readline()
+ print("Found in lastplayedqueue:", last_played_url)
+ if last_played_url != "None":
+ # if it's set, iterate through until it's found in list, at which point select the next
+ for i in range(0, len(queue_events)):
+ if get_text(queue_events[i], "LOCATION") == last_played_url:
+ # wrap around to first in list if match is last item
+ event_to_play = queue_events[(i + 1) % len(queue_events)]
+ except FileNotFoundError:
+ print("Found no lastplayedqueue file")
+ url = get_text(event_to_play, "LOCATION")
+ f = open('lastplayedqueue', 'w', encoding="utf-8")
+ f.write(url)
+ f.close()
+ print("Wrote to lastplayedqueue:", url)
+
+ else:
+ print("Found no scheduled or queue events; using fallback url", fallback_url)
+
+ duration = timedelta(hours=24) # (arbitrary)
+ if not queue_event_will_play:
+ event_end = event_to_play['DTEND'].dt # this errors if checked for an all-day event, which under current approach shouldn't happen
+ duration = event_end - now
+
+ print("Duration =", duration)
+ # overlapping = get events between now and event_end
+ overlapping_events = recurring_ical_events.of(calendar).between(now, now + duration)
+ overlapping_scheduled = list(filter(is_opaque, overlapping_events)) # not interested in queue events here
+ print("Found", len(overlapping_scheduled), "'overlapping' event(s):")
+ list(map(print_event, overlapping_scheduled))
+ print()
+
+ if len(overlapping_scheduled) > 0:
+ overlapping_scheduled.sort(key=get_start)
+ # iterate through overlapping until an event is found where event.created > event_to_play.created (or, if currently event_to_play is an audio queue event, until any scheduled event is found)
+ next_event = overlapping_scheduled[0]
+ if event_to_play and not queue_event_will_play:
+ i = 0
+ while next_event and (get_start(next_event) <= get_start(event_to_play) or get_created(next_event) < get_created(event_to_play)):
+ i += 1
+ if i >= len(overlapping_scheduled):
+ next_event = None
+ else:
+ next_event = overlapping_scheduled[i]
+
+ if next_event:
+ duration = next_event['DTSTART'].dt - now
+ print("Next event:", get_text(next_event, 'SUMMARY'), "at", next_event['DTSTART'].dt.time())
+ print("(Revised current duration to", duration, ")")
+
+ if event_to_play:
+ print()
+ print("~~~ Now playing ~~~")
+ if queue_event_will_play:
+ print("Listening queue:")
+ print(" ", get_text(event_to_play, 'SUMMARY'))
+ print(" ", url)
+ else:
+ print(get_text(event_to_play, 'SUMMARY'))
+ print(url)
+ print("From", now.time(), "to", (now + duration).time())
+ print("~~~~~~~~~~~~~~~~~~~")
+ print()
+
+ if url:
+ # setting timeout in subprocess.run terminates mpv after [duration], which is necessary because for some sources, e.g. bandcamp albums, it ignores --length arg. adding a couple seconds to let the process terminate itself (when mpv does consider --length) feels respectful.
+ grace_period = 2
+ try:
+ subprocess.run(['mpv', '--length=' + str(duration.total_seconds()), url], timeout=(duration.total_seconds() + grace_period))
+ except subprocess.TimeoutExpired:
+ print("Reached timeout of", duration.total_seconds() + grace_period, "seconds")
+
+ # start over
+ tuner()
+
+ else:
+ print("Nothing found in schedule/queue and no fallback URL provided. Exiting.")
+
+tuner()
+