Compare commits
4 Commits
9788b83d82
...
b9d8a339d2
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b9d8a339d2 | ||
|
|
9fe46133da | ||
|
|
f35172fde1 | ||
|
|
6b79aca0be |
@@ -22,7 +22,8 @@
|
||||
".*.exe" = ""
|
||||
"firefox" = ""
|
||||
"galaxyclient.exe" = ""
|
||||
"Gimp-.*" = ""
|
||||
"\\.?gimp-.*" = ""
|
||||
"F?imv.*" = ""
|
||||
"KiCad" = ""
|
||||
"kitty" = ""
|
||||
"libreoffice-calc" = ""
|
||||
@@ -39,7 +40,7 @@
|
||||
"Spotify" = ""
|
||||
"steam" = ""
|
||||
"thunar" = ""
|
||||
"thunderbird" = ""
|
||||
"thunderbird" = ""
|
||||
"Tor Browser" = ""
|
||||
"vesktop" = ""
|
||||
"virt-manager" = ""
|
||||
|
||||
@@ -349,7 +349,7 @@ in {
|
||||
"SUPER SHIFT, F1, movetoworkspace, 100"
|
||||
# "SUPER, F2,"
|
||||
# "SUPER, F3, toggleopaque"
|
||||
"SUPER, F4, exec, rm /tmp/caldav_event_cache.json && notify-send Saved event deleted!"
|
||||
"SUPER, F4, exec, rm /tmp/caldav_event_cache.json && notify-send 'Cleared Saved Event!' ''"
|
||||
"SUPER, F5, exec, nx_gcal_event force-lookup"
|
||||
"SUPER SHIFT, F5, exec, nx_gcal_event reauthenticate"
|
||||
"SUPER, F6, exec, ${terminal-exec}'htop'"
|
||||
|
||||
@@ -47,104 +47,113 @@ in {
|
||||
'';})
|
||||
(writers.writePython3Bin "caldav_event" {
|
||||
libraries = with pkgs.python3Packages; [ caldav ics pytz ];
|
||||
flakeIgnore = [ "E302" "E305""E501" ];
|
||||
flakeIgnore = [ "E302" "E305""E501" "E261" ];
|
||||
} /* python */ ''
|
||||
import os
|
||||
import json
|
||||
from caldav import DAVClient
|
||||
from datetime import datetime, timezone
|
||||
import json
|
||||
from ics import Calendar
|
||||
from pytz import UTC
|
||||
|
||||
def get_password(password_file):
|
||||
with open(password_file, "r") as file:
|
||||
return file.read().strip()
|
||||
|
||||
def datetime_converter(obj):
|
||||
if isinstance(obj, datetime):
|
||||
return obj.isoformat()
|
||||
return obj
|
||||
|
||||
def datetime_parser(dct):
|
||||
for key, value in dct.items():
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
dct[key] = datetime.fromisoformat(value)
|
||||
except ValueError:
|
||||
pass
|
||||
return dct
|
||||
|
||||
def load_cache(cache_file):
|
||||
if os.path.exists(cache_file):
|
||||
with open(cache_file, "r") as file:
|
||||
return json.load(file)
|
||||
return json.load(file, object_hook=datetime_parser)
|
||||
return None
|
||||
|
||||
def save_cache(cache_file, data):
|
||||
with open(cache_file, "w") as file:
|
||||
json.dump(data, file)
|
||||
json.dump(data, file, default=datetime_converter)
|
||||
|
||||
def get_ongoing_and_next_event(url, username, password):
|
||||
|
||||
def get_ongoing_or_next_event(url, username, password):
|
||||
now = datetime.now(timezone.utc)
|
||||
ongoing_events = []
|
||||
upcoming_events = []
|
||||
|
||||
try:
|
||||
client = DAVClient(url, username=username, password=password)
|
||||
principal = client.principal()
|
||||
calendars = principal.calendars()
|
||||
|
||||
next_event_dict = {
|
||||
'event_name': "fake",
|
||||
'event_begin': datetime(9000, 1, 1, tzinfo=UTC), # in the year 9000
|
||||
'event_end': datetime(9000, 1, 1, 8, tzinfo=UTC),
|
||||
}
|
||||
|
||||
for calendar in calendars:
|
||||
events = calendar.events()
|
||||
for event in events:
|
||||
ical_data = event.data
|
||||
calendar_parsed = Calendar(ical_data)
|
||||
for event in calendar.events():
|
||||
calendar_parsed = Calendar(event.data)
|
||||
for ics_event in calendar_parsed.events:
|
||||
event_dict = {}
|
||||
event_dict['event_name'] = ics_event.name or "(No Title)"
|
||||
event_dict['event_begin'] = ics_event.begin.astimezone(timezone.utc)
|
||||
event_dict['event_end'] = ics_event.end.astimezone(timezone.utc)
|
||||
|
||||
for event in calendar_parsed.events:
|
||||
event_name = event.name or "(No Title)"
|
||||
start_time = event.begin.astimezone(timezone.utc)
|
||||
end_time = event.end.astimezone(timezone.utc)
|
||||
if event_dict['event_begin'] <= now and now <= event_dict['event_end']:
|
||||
return event_dict
|
||||
elif event_dict['event_begin'] >= now and next_event_dict['event_begin'] > event_dict['event_begin']:
|
||||
next_event_dict = event_dict
|
||||
return next_event_dict
|
||||
|
||||
if start_time <= now <= end_time:
|
||||
ongoing_events.append((event_name, start_time.timestamp(), end_time.timestamp()))
|
||||
elif start_time > now:
|
||||
upcoming_events.append((event_name, start_time.timestamp(), end_time.timestamp()))
|
||||
except Exception as e:
|
||||
print(f"Error accessing {url}: {e}")
|
||||
|
||||
upcoming_events.sort(key=lambda x: x[1]) # Sort by start time
|
||||
return ongoing_events, upcoming_events[0] if upcoming_events else None
|
||||
return None
|
||||
|
||||
if __name__ == "__main__":
|
||||
password_file = "${config.sops.secrets."nx2site/radicale/password".path}" # Path to password file
|
||||
password_file = "/home/nx2/.config/sops-nix/secrets/nx2site/radicale/password" # Path to password file
|
||||
cache_file = "/tmp/caldav_event_cache.json" # Path to cache file
|
||||
url = "https://dav.${domain}/"
|
||||
username = "${user}"
|
||||
password = get_password(password_file)
|
||||
|
||||
cache = load_cache(cache_file)
|
||||
event_dict = load_cache(cache_file)
|
||||
now = datetime.now(timezone.utc).timestamp()
|
||||
|
||||
if cache and cache.get("next_event_start") and now < cache["next_event_start"]:
|
||||
ongoing_events = cache.get("ongoing_events", [])
|
||||
next_event = (cache["next_event_name"], cache["next_event_start"], cache["next_event_end"]) if "next_event_name" in cache else None
|
||||
else:
|
||||
ongoing_events, next_event = get_ongoing_and_next_event(url, username, password)
|
||||
|
||||
if event_dict is None or event_dict['event_begin'].timestamp() <= now and now < event_dict['event_end'].timestamp():
|
||||
event_dict = get_ongoing_or_next_event(url, username, password)
|
||||
if event_dict is None:
|
||||
print("No upcoming events found.")
|
||||
exit(0)
|
||||
cache_data = {
|
||||
"ongoing_events": ongoing_events,
|
||||
"next_event_name": next_event[0] if next_event else None,
|
||||
"next_event_start": next_event[1] if next_event else None,
|
||||
"next_event_end": next_event[2] if next_event else None
|
||||
"event_name": event_dict['event_name'] if event_dict is not None else None,
|
||||
"event_begin": event_dict['event_begin'] if event_dict is not None else None,
|
||||
"event_end": event_dict['event_end'] if event_dict is not None else None
|
||||
}
|
||||
save_cache(cache_file, cache_data)
|
||||
|
||||
if ongoing_events:
|
||||
for event_name, start_time, end_time in ongoing_events:
|
||||
time_remaining = end_time - now
|
||||
if event_dict:
|
||||
event_start = event_dict['event_begin'].timestamp()
|
||||
event_end = event_dict['event_end'].timestamp()
|
||||
|
||||
if event_start <= now <= event_end:
|
||||
time_remaining = event_end - now
|
||||
hours, rem = divmod(int(time_remaining), 3600)
|
||||
minutes, _ = divmod(rem, 60)
|
||||
|
||||
if hours == 0:
|
||||
print(f"{event_name} {minutes} minute{'s ' if minutes > 1 else ' '}left")
|
||||
print(f"{event_dict['event_name']} ends in {hours} hour{'s ' if hours != 1 else ' '}and {minutes} minute{'s ' if minutes != 1 else ' '}")
|
||||
else:
|
||||
print(f"{event_name} {hours} hour{'s ' if hours > 1 else ' '}and {minutes} minute{'s ' if minutes > 1 else ' '}left")
|
||||
else:
|
||||
if next_event:
|
||||
event_name, start_time, end_time = next_event
|
||||
time_until_start = start_time - now
|
||||
time_until_start = event_start - now
|
||||
hours, rem = divmod(int(time_until_start), 3600)
|
||||
minutes, _ = divmod(rem, 60)
|
||||
|
||||
if hours == 0:
|
||||
print(f"'{event_name}' starts in {minutes} minute{'s ' if minutes > 1 else ' '}")
|
||||
else:
|
||||
print(f"'{event_name}' starts in {hours} hour{'s ' if hours > 1 else ' '}and {minutes} minute{'s ' if minutes > 1 else ' '}")
|
||||
print(f"{event_dict['event_name']} starts in {hours} hour{'s ' if hours != 1 else ' '}and {minutes} minute{'s ' if minutes != 1 else ' '}")
|
||||
else:
|
||||
print("No upcoming events found.")
|
||||
'')
|
||||
|
||||
@@ -46,6 +46,7 @@
|
||||
{ on = [ "g" "C" ]; run = "cd ~/.cache"; desc = "Go to the .cache directory"; }
|
||||
{ on = [ "g" "m" ]; run = "cd ~/media"; desc = "Go to the media (udiskie mount) directory"; }
|
||||
{ on = [ "g" "v" ]; run = "cd ~/Videos"; desc = "Go to the Videos directory"; }
|
||||
{ on = [ "g" "t" ]; run = "cd /tmp"; desc = "Go to the /tmp directory"; }
|
||||
{ on = [ "g" "d" ]; run = "cd ~/Downloads"; desc = "Go to the downloads directory"; }
|
||||
{ on = [ "g" "D" ]; run = "cd ~/Documents"; desc = "Go to the Documents directory"; }
|
||||
{ on = [ "g" "r" ]; run = "cd /"; desc = "Go to the root (/) directory"; }
|
||||
@@ -195,10 +196,11 @@
|
||||
opener = {
|
||||
"edit" = [
|
||||
{ run = ''hx "$@"''; desc = "helix"; block = true; }
|
||||
{ run = ''codium "$@"''; desc = "helix"; orphan = true; }
|
||||
{ run = ''codium "$@"''; desc = "code"; orphan = true; }
|
||||
];
|
||||
"play" = [
|
||||
{ run = ''mpv "$@"''; }
|
||||
{ run = ''mpv "$@"''; desc = "mpv"; orphan = true; }
|
||||
{ run = ''mpv --vf=negate "$@"''; desc = "mpv inverted"; orphan = true; }
|
||||
{ run = ''mediainfo "$1"; echo "Press enter to exit"; read''; block = true; desc = "Show mediainfo"; }
|
||||
];
|
||||
"archive" = [
|
||||
@@ -257,6 +259,7 @@
|
||||
{ mime = "text/htm"; use = [ "edit" "browser" ]; }
|
||||
{ mime = "text/x-python"; use = "python"; }
|
||||
{ mime = "text/*"; use = "edit"; }
|
||||
{ mime = "text"; use = "edit"; }
|
||||
|
||||
{ mine = "inode/x-empty"; use = "edit"; }
|
||||
{ mine = "inode/directory"; use = "edit"; }
|
||||
|
||||
Reference in New Issue
Block a user