Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

list currently firing alarms in !listreminders #162

Merged
merged 3 commits into from
May 10, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 49 additions & 28 deletions matrix_reminder_bot/bot_commands.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional, Tuple
from typing import List, Optional, Tuple

import arrow
import dateparser
Expand Down Expand Up @@ -366,7 +366,7 @@ async def _silence(self):
reminder_text = " ".join(self.args)
if reminder_text:
# Find the alarm job via its reminder text
alarm_job = ALARMS.get((self.room.room_id, reminder_text.upper()))
alarm_job = ALARMS.get((self.room.room_id, reminder_text.upper())).alarm_job

if alarm_job:
await self._remove_and_silence_alarm(alarm_job, reminder_text)
Expand All @@ -389,14 +389,16 @@ async def _silence(self):
else:
# No reminder text provided. Check if there's a reminder currently firing
# in the room instead then
for alarm_info, job in ALARMS.items():
for alarm_info, reminder in ALARMS.items():
if alarm_info[0] == self.room.room_id:
# Found one!
reminder_text = alarm_info[
1
].capitalize() # normalize the text a bit

await self._remove_and_silence_alarm(job, reminder_text)
await self._remove_and_silence_alarm(
reminder.alarm_job, reminder_text
)
text = f"Alarm '{reminder_text}' silenced."

# Prevent the `else` clause from being triggered
Expand All @@ -422,6 +424,10 @@ async def _list_reminders(self):

Sends a message listing them in the following format, using the alarm clock emoji ⏰ to indicate an alarm:

⏰ Firing Alarms

* [🔁 every <recurring time>;] <start time>; <reminder text>

1️⃣ One-time Reminders

* [⏰] <start time>: <reminder text>
Expand All @@ -440,9 +446,17 @@ async def _list_reminders(self):
"""
output = ""

cron_reminder_lines = []
one_shot_reminder_lines = []
interval_reminder_lines = []
cron_reminder_lines: List = []
one_shot_reminder_lines: List = []
interval_reminder_lines: List = []
firing_alarms_lines: List = []

for alarm in ALARMS.values():
line = "- "
if isinstance(alarm.job.trigger, IntervalTrigger):
line += f"🔁 every {readabledelta(alarm.recurse_timedelta)}; "
line += f'"*{alarm.reminder_text}*"'
firing_alarms_lines.append(line)

# Sort the reminder types
for reminder in REMINDERS.values():
Expand All @@ -460,9 +474,18 @@ async def _list_reminders(self):
# Print the duration before (next) execution
next_execution = reminder.job.next_run_time
next_execution = arrow.get(next_execution)
# One-time reminders
if isinstance(reminder.job.trigger, DateTrigger):
# Just print when the reminder will go off
line += f"{next_execution.humanize()}"

# Repeat reminders
elif isinstance(reminder.job.trigger, IntervalTrigger):
# Print the interval, and when it will next go off
line += f"every {readabledelta(reminder.recurse_timedelta)}; next run {next_execution.humanize()}"

# Cron-based reminders
if isinstance(reminder.job.trigger, CronTrigger):
elif isinstance(reminder.job.trigger, CronTrigger):
# A human-readable cron tab, in addition to the actual tab
human_cron = prettify_cron(reminder.cron_tab)
if human_cron != reminder.cron_tab:
Expand All @@ -471,32 +494,23 @@ async def _list_reminders(self):
line += f"`Every {reminder.cron_tab}`"
line += f"; next run {next_execution.humanize()}"

# One-time reminders
elif isinstance(reminder.job.trigger, DateTrigger):
# Just print when the reminder will go off
line += f"{next_execution.humanize()}"

# Repeat reminders
elif isinstance(reminder.job.trigger, IntervalTrigger):
# Print the interval, and when it will next go off
line += f"every {readabledelta(reminder.recurse_timedelta)}; next run {next_execution.humanize()}"

# Add the reminder's text
line += f'; *"{reminder.reminder_text}"*'

# Output the status of each reminder. We divide up the reminders by type in order
# to show them in separate sections, and display them differently
if isinstance(reminder.job.trigger, CronTrigger):
cron_reminder_lines.append(line)
elif isinstance(reminder.job.trigger, DateTrigger):
if isinstance(reminder.job.trigger, DateTrigger):
one_shot_reminder_lines.append(line)
elif isinstance(reminder.job.trigger, IntervalTrigger):
interval_reminder_lines.append(line)
elif isinstance(reminder.job.trigger, CronTrigger):
cron_reminder_lines.append(line)

if (
not one_shot_reminder_lines
and not cron_reminder_lines
not firing_alarms_lines
and not one_shot_reminder_lines
and not interval_reminder_lines
and not cron_reminder_lines
):
await send_text_to_room(
self.client,
Expand All @@ -505,18 +519,22 @@ async def _list_reminders(self):
)
return

if firing_alarms_lines:
output += "\n\n" + "**⏰ Firing Alarms**" + "\n\n"
output += "\n".join(firing_alarms_lines)

if one_shot_reminder_lines:
output += "\n\n" + "**1️⃣ One-time Reminders**" + "\n\n"
output += "\n".join(one_shot_reminder_lines)

if cron_reminder_lines:
output += "\n\n" + "**📅 Cron Reminders**" + "\n\n"
output += "\n".join(cron_reminder_lines)

if interval_reminder_lines:
output += "\n\n" + "**🔁 Repeating Reminders**" + "\n\n"
output += "\n".join(interval_reminder_lines)

if cron_reminder_lines:
output += "\n\n" + "**📅 Cron Reminders**" + "\n\n"
output += "\n".join(cron_reminder_lines)

await send_text_to_room(self.client, self.room.room_id, output)

@command_syntax("<reminder text>")
Expand All @@ -536,7 +554,10 @@ async def _delete_reminder(self):
# Cancel the reminder and associated alarms
reminder.cancel()

text = "Reminder cancelled."
text = "Reminder"
if reminder.alarm:
text = "Alarm"
text += f' "*{reminder_text}*" cancelled.'
else:
text = f"Unknown reminder '{reminder_text}'."

Expand Down
5 changes: 2 additions & 3 deletions matrix_reminder_bot/reminder.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
from typing import Dict, Optional, Tuple

import pytz
from apscheduler.job import Job
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
Expand Down Expand Up @@ -126,7 +125,7 @@ async def _fire(self):
seconds=int(timedelta_seconds(ALARM_TIMEDELTA)),
),
)
ALARMS[(self.room_id, self.reminder_text.upper())] = self.alarm_job
ALARMS[(self.room_id, self.reminder_text.upper())] = self

# Send the message to the room
await send_text_to_room(
Expand Down Expand Up @@ -203,4 +202,4 @@ def has_target(self) -> bool:
# reminder_text should be accessed and stored as uppercase in order to
# allow for case-insensitive matching when carrying out user actions
REMINDERS: Dict[Tuple[str, str], Reminder] = {}
ALARMS: Dict[Tuple[str, str], Job] = {}
ALARMS: Dict[Tuple[str, str], Reminder] = {}