Skip to content
Merged
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
62 changes: 54 additions & 8 deletions appdaemon/parse.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,56 @@
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, time, timedelta
from datetime import datetime, time, timedelta, tzinfo
from functools import partial
from typing import ClassVar, Literal
from zoneinfo import ZoneInfo

import astral
import pytz
from astral.location import Location


def normalize_tz(tz: tzinfo) -> tzinfo:
"""Convert pytz timezone to ZoneInfo for clean stdlib-compatible handling.

pytz timezones don't behave like normal tzinfo implementations and require
special localize()/normalize() calls. By converting to ZoneInfo at the boundary,
we can use standard replace(tzinfo=...) and astimezone() everywhere else.

Args:
tz: Any tzinfo object (pytz, ZoneInfo, or fixed-offset)

Returns:
ZoneInfo if input was pytz with an IANA zone name, otherwise unchanged
"""
if isinstance(tz, pytz.tzinfo.BaseTzInfo) and tz.zone is not None:
return ZoneInfo(tz.zone)
return tz


def localize_naive(naive_dt: datetime, tz: tzinfo) -> datetime:
"""Interpret a naive datetime as wall-clock time in the given timezone.

This normalizes pytz timezones to ZoneInfo first, so we can use standard
replace(tzinfo=...) semantics instead of pytz's localize().

Args:
naive_dt: A naive datetime (no tzinfo)
tz: The timezone to interpret the datetime in

Returns:
A timezone-aware datetime

Raises:
ValueError: If naive_dt already has tzinfo
"""
if naive_dt.tzinfo is not None:
raise ValueError("expected naive datetime")
tz = normalize_tz(tz)
return naive_dt.replace(tzinfo=tz)


CONVERTERS = {
"hour": lambda v: timedelta(hours=v),
"hr": lambda v: timedelta(hours=v),
Expand Down Expand Up @@ -267,14 +310,16 @@ def _parse(input_string: str) -> ParsedTimeString | time | datetime:
else:
raise ValueError(f"Invalid time string: {input_string}")

tz = normalize_tz(now.tzinfo)

offset = timedelta()
match _parse(time_str):
case time() as parsed_time:
result = datetime.combine(
naive_dt = datetime.combine(
(now + timedelta(days=days_offset)).date(),
parsed_time,
tzinfo=now.tzinfo
)
result = localize_naive(naive_dt, tz)
case datetime() as result:
pass
case Now(offset=offset):
Expand Down Expand Up @@ -338,14 +383,16 @@ def parse_datetime(

assert isinstance(now, datetime) and now.tzinfo is not None, "Now must be a timezone-aware datetime"

tz = normalize_tz(now.tzinfo)

offset = timedelta()
match input_:
case time() as input_time:
result = datetime.combine(
naive_dt = datetime.combine(
(now + timedelta(days=days_offset)).date(),
input_time,
tzinfo=now.tzinfo
)
result = localize_naive(naive_dt, tz)
case datetime() as result:
result += timedelta(days=days_offset)
case str() as time_str:
Expand All @@ -364,10 +411,9 @@ def parse_datetime(

# Make the timezones match for the comparison below
if result.tzinfo is None:
# Just adds the timezone without changing the time values
result = result.replace(tzinfo=now.tzinfo)
result = localize_naive(result, tz)
else:
result = result.astimezone(now.tzinfo)
result = result.astimezone(tz)

# The the days offset is negative, the result can't be forced to today, so set today to False
if days_offset < 0:
Expand Down
Loading