Files
changedetection.io/changedetectionio/time_handler.py
dgtlmoonandClaude Opus 5.5 ab06c89c73 Watch time_schedule_limit - Typed, frozen pydantic model with shared defaults, halves per-watch memory
Every watch carried its own copy of the 7-day schedule (15 nested dicts, ~30 strings),
about 3.7KiB of a ~7.6KiB watch, even though nearly every schedule is the untouched
default. At 40,000 watches that is ~146MB.

time_schedule_limit is now a slots + frozen pydantic dataclass (model/TimeScheduleLimit.py)
built on a small generic base (model/shared_defaults.py):

- Anything equal to the default is the one shared default instance, recursively, so an
  untouched day or the whole default schedule costs nothing per watch.
- watch.json only stores what differs from the default (exclude_defaults), an enabled
  schedule with default days is {"enabled": true}, a disabled custom week is kept.
- Frozen, so the shared default can never be mutated in place, assign a new value.
- Every write path (loading, edit form, API, clone) converts in watch_base __setitem__/update.
- Blank/None fields fall back per field, one empty value no longer loses the whole week.
- hours/minutes are ints, in the model, the form and the API spec (no upper bound,
  multi-day windows are supported).
- The scheduler converts the global schedule once per pass instead of reading it per watch.

update_34 rewrites watch.json/tag.json compact and converts the global schedule's
"24"/"00" strings to ints.

Measured: 7.6KiB -> 3.9KiB per watch (296MiB -> 150MiB at 40k watches).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 12:00:22 +02:00

145 lines
5.2 KiB
Python

import os
from functools import lru_cache
from enum import IntEnum
def default_timezone_name(configured=None):
"""
Resolve the scheduler's default IANA timezone name. Never returns None/''.
`scheduler_timezone_default` is initialised to None in model/App.py, so the
key always EXISTS with a None value. That makes the common idiom
settings['application'].get('scheduler_timezone_default', os.getenv('TZ', 'UTC'))
silently return None - dict.get() only falls back when the key is absent,
not when its value is None. The None then reaches is_within_schedule(),
which does `tz_name.strip()` and raises AttributeError.
Args:
configured: the stored setting value (may be None or empty)
Returns:
str: a non-empty timezone name, falling back to $TZ then 'UTC'
"""
return (configured or os.getenv('TZ') or 'UTC').strip() or 'UTC'
class Weekday(IntEnum):
"""Enumeration for days of the week."""
Monday = 0
Tuesday = 1
Wednesday = 2
Thursday = 3
Friday = 4
Saturday = 5
Sunday = 6
def am_i_inside_time(
day_of_week: str,
time_str: str,
timezone_str: str,
duration: int = 15,
) -> bool:
"""
Determines if the current time falls within a specified time range.
Parameters:
day_of_week (str): The day of the week (e.g., 'Monday').
time_str (str): The start time in 'HH:MM' format.
timezone_str (str): The timezone identifier (e.g., 'Europe/Berlin').
duration (int, optional): The duration of the time range in minutes. Default is 15.
Returns:
bool: True if the current time is within the time range, False otherwise.
"""
import arrow
# Parse the target day of the week
try:
target_weekday = Weekday[day_of_week.capitalize()]
except KeyError:
raise ValueError(f"Invalid day_of_week: '{day_of_week}'. Must be a valid weekday name.")
# Parse the start time
try:
hour, minute = map(int, time_str.split(':'))
if not (0 <= hour <= 23 and 0 <= minute <= 59):
raise ValueError
except (ValueError, AttributeError):
raise ValueError(f"Invalid time_str: '{time_str}'. Must be in 'HH:MM' format.")
# Get the current time in the specified timezone
try:
now_tz = arrow.now(timezone_str.strip())
except Exception as e:
raise ValueError(f"Invalid timezone_str: '{timezone_str}'. Must be a valid timezone identifier.")
# Check if the current day matches the target day or overlaps due to duration
current_weekday = now_tz.weekday()
# Create start datetime for today in target timezone
start_datetime_tz = now_tz.replace(hour=hour, minute=minute, second=0, microsecond=0)
# Handle previous day's overlap
if target_weekday == (current_weekday - 1) % 7:
# Calculate start and end times for the overlap from the previous day
start_datetime_tz = start_datetime_tz.shift(days=-1)
end_datetime_tz = start_datetime_tz.shift(minutes=duration)
if start_datetime_tz <= now_tz <= end_datetime_tz:
return True
# Handle current day's range
if target_weekday == current_weekday:
end_datetime_tz = start_datetime_tz.shift(minutes=duration)
if start_datetime_tz <= now_tz <= end_datetime_tz:
return True
# Handle next day's overlap
if target_weekday == (current_weekday + 1) % 7:
end_datetime_tz = start_datetime_tz.shift(minutes=duration)
if now_tz < start_datetime_tz and now_tz.shift(days=1) <= end_datetime_tz:
return True
return False
def is_within_schedule(time_schedule_limit, default_tz="UTC"):
"""
Check if the current time is within a scheduled time window.
Parameters:
time_schedule_limit (TimeScheduleLimit|dict): Schedule configuration with timezone, day settings, etc.
A watch/tag holds the typed model, global settings still hold a dict.
default_tz (str): Default timezone to use if not specified. Default is 'UTC'.
Returns:
bool: True if current time is within the schedule, False otherwise.
"""
import arrow
from changedetectionio.model.TimeScheduleLimit import as_time_schedule_limit
time_schedule_limit = as_time_schedule_limit(time_schedule_limit)
if time_schedule_limit.enabled:
# Get the timezone the time schedule is in, so we know what day it is there
tz_name = time_schedule_limit.timezone
if not tz_name:
tz_name = default_tz
# Get current day name in the target timezone
now_day_name_in_tz = arrow.now(tz_name.strip()).format('dddd')
selected_day_schedule = time_schedule_limit.day(now_day_name_in_tz)
if not selected_day_schedule.enabled:
return False
duration = selected_day_schedule.duration
selected_day_run_duration_m = duration.hours * 60 + duration.minutes
is_valid = am_i_inside_time(day_of_week=now_day_name_in_tz,
time_str=selected_day_schedule.start_time,
timezone_str=tz_name,
duration=selected_day_run_duration_m)
return is_valid
return False