137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""
|
|
This script yields JSON info about the current event + the next one.
|
|
|
|
It depends on the nice ics-query https://github.com/niccokunzmann/ics-query
|
|
"""
|
|
|
|
import os
|
|
import datetime
|
|
import json
|
|
import sys
|
|
from argparse import ArgumentParser
|
|
from pathlib import Path
|
|
from subprocess import check_output
|
|
from typing import Optional
|
|
|
|
import pytz
|
|
from icalendar import Calendar, Event
|
|
|
|
TZ = os.getenv('TZ')
|
|
if TZ is None:
|
|
TZ = check_output('timedatectl show -p Timezone --value'.split()).strip()
|
|
|
|
def parse_dt(s: str) -> datetime.datetime:
|
|
dt = datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%S")
|
|
dt = pytz.timezone(TZ).localize(dt)
|
|
return dt
|
|
|
|
|
|
def parse_minutes(s: str) -> datetime.timedelta:
|
|
return datetime.timedelta(minutes=int(s))
|
|
|
|
|
|
class IcsNow:
|
|
def __init__(self):
|
|
self.parser = ArgumentParser()
|
|
self.parser.add_argument("ics", type=Path)
|
|
self.parser.add_argument(
|
|
"--now",
|
|
default=datetime.datetime.now(pytz.timezone(TZ)),
|
|
type=parse_dt,
|
|
help="Simulate different time. Mostly for debug",
|
|
)
|
|
self.parser.add_argument(
|
|
"--until",
|
|
metavar="DURATION",
|
|
type=parse_minutes,
|
|
default=parse_minutes(60),
|
|
help='How many minute should "next" include?',
|
|
)
|
|
self.parser.add_argument(
|
|
"--next-late",
|
|
metavar="DURATION",
|
|
type=parse_minutes,
|
|
default=parse_minutes(10),
|
|
help='How many minutes can the "next" show be late?',
|
|
)
|
|
self.parser.add_argument(
|
|
"--ics-query",
|
|
type=Path,
|
|
default=Path("/usr/local/bin/ics-query"),
|
|
help="Location of ics-query",
|
|
)
|
|
|
|
def load_from_icsquery(self, cmd, args) -> list[Event]:
|
|
result: bytes = check_output(
|
|
[
|
|
self.args.ics_query,
|
|
cmd,
|
|
"--tz",
|
|
TZ,
|
|
"--component",
|
|
"VEVENT",
|
|
*args,
|
|
self.args.ics,
|
|
]
|
|
)
|
|
if not result:
|
|
return []
|
|
return Calendar.from_ical(result, multiple=True)
|
|
|
|
def pick_most_important(self, events: list[Event]) -> Optional[Event]:
|
|
if not events:
|
|
return None
|
|
# FIXME: sort by priority
|
|
return events[0]
|
|
|
|
def pick_next(self, events: list[Event], current) -> Optional[Event]:
|
|
events = [
|
|
e for e in events
|
|
if e['DTSTART'].dt > (self.args.now - self.args.next_late)
|
|
]
|
|
if not events:
|
|
return None
|
|
events.sort(key=lambda e: e['DTSTART'].dt)
|
|
return events[0]
|
|
|
|
def run(self):
|
|
self.args = self.parser.parse_args()
|
|
ret = {}
|
|
|
|
current = self.pick_most_important(self.load_from_icsquery(
|
|
"at", [self.args.now.strftime("%Y-%m-%dT%H:%M:%S")]
|
|
))
|
|
ret["now"] = self.serialize(current)
|
|
|
|
next_events = self.load_from_icsquery(
|
|
"between",
|
|
[
|
|
self.args.now.strftime("%Y-%m-%dT%H:%M:%S"),
|
|
(self.args.now + self.args.until).strftime("%Y-%m-%dT%H:%M:%S"),
|
|
],
|
|
)
|
|
ret["next"] = self.serialize(self.pick_next(next_events, current))
|
|
json.dump(ret, sys.stdout, indent=2)
|
|
sys.stdout.write("\n")
|
|
|
|
def serialize(self, event: Optional[Event]):
|
|
if event is None:
|
|
return None
|
|
out = {
|
|
"title": event["SUMMARY"],
|
|
}
|
|
for field in ["start", "end"]:
|
|
calfield = f"dt{field}"
|
|
if calfield in event:
|
|
out[field] = event[calfield].dt.isoformat()
|
|
for field in ["uid", "description", "url"]:
|
|
if field in event:
|
|
out[field] = event[field]
|
|
return out
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli = IcsNow()
|
|
cli.run()
|