changed datetime import
This commit is contained in:
@@ -1,21 +1,21 @@
|
|||||||
from datetime import datetime, time, timedelta
|
import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Annotated, Dict, List, Optional, Self
|
from typing import Annotated, Dict, List, Optional, Self
|
||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
from astral import SunDirection
|
from astral import SunDirection
|
||||||
from pydantic import BaseModel, BeforeValidator, Field, root_validator
|
from pydantic import BaseModel, BeforeValidator, Field, model_validator, root_validator
|
||||||
from pydantic_core import PydanticCustomError
|
from pydantic_core import PydanticCustomError
|
||||||
from rich.console import Console, ConsoleOptions, RenderResult
|
from rich.console import Console, ConsoleOptions, RenderResult
|
||||||
from rich.table import Column, Table
|
from rich.table import Column, Table
|
||||||
|
|
||||||
|
|
||||||
def str_to_timedelta(input_str: str) -> timedelta:
|
def str_to_timedelta(input_str: str) -> datetime.timedelta:
|
||||||
try:
|
try:
|
||||||
hours, minutes, seconds = map(int, input_str.split(':'))
|
hours, minutes, seconds = map(int, input_str.split(':'))
|
||||||
return timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
return datetime.timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
||||||
except Exception:
|
except Exception:
|
||||||
return timedelta()
|
return datetime.timedelta()
|
||||||
|
|
||||||
|
|
||||||
def str_to_direction(input_str: str) -> SunDirection:
|
def str_to_direction(input_str: str) -> SunDirection:
|
||||||
@@ -27,7 +27,7 @@ def str_to_direction(input_str: str) -> SunDirection:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
OffDuration = Annotated[timedelta, BeforeValidator(str_to_timedelta)]
|
OffDuration = Annotated[datetime.timedelta, BeforeValidator(str_to_timedelta)]
|
||||||
|
|
||||||
|
|
||||||
class State(BaseModel):
|
class State(BaseModel):
|
||||||
@@ -45,13 +45,13 @@ class ApplyKwargs(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
class ControllerStateConfig(BaseModel):
|
class ControllerStateConfig(BaseModel):
|
||||||
time: Optional[str | datetime] = None
|
time: Optional[str | datetime.time | datetime.datetime] = None
|
||||||
elevation: Optional[float] = None
|
elevation: Optional[float] = None
|
||||||
direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = None
|
direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = None
|
||||||
off_duration: Optional[OffDuration] = None
|
off_duration: Optional[OffDuration] = None
|
||||||
scene: dict[str, State] | str
|
scene: dict[str, State] | str = Field(default_factory=dict)
|
||||||
|
|
||||||
@root_validator(pre=True)
|
@model_validator(mode='before')
|
||||||
def check_args(cls, values):
|
def check_args(cls, values):
|
||||||
time, elevation = values.get('time'), values.get('elevation')
|
time, elevation = values.get('time'), values.get('elevation')
|
||||||
if time is not None and elevation is not None:
|
if time is not None and elevation is not None:
|
||||||
@@ -60,8 +60,8 @@ class ControllerStateConfig(BaseModel):
|
|||||||
raise PydanticCustomError('no_sun_dir', 'Needs sun direction with elevation')
|
raise PydanticCustomError('no_sun_dir', 'Needs sun direction with elevation')
|
||||||
return values
|
return values
|
||||||
|
|
||||||
def to_apply_kwargs(self, **kwargs):
|
def to_apply_kwargs(self, transition: int = None):
|
||||||
return ApplyKwargs(entities=self.scene, **kwargs).model_dump(exclude_none=True)
|
return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True)
|
||||||
|
|
||||||
|
|
||||||
class RoomControllerConfig(BaseModel):
|
class RoomControllerConfig(BaseModel):
|
||||||
@@ -102,11 +102,11 @@ class RoomControllerConfig(BaseModel):
|
|||||||
def sort_states(self):
|
def sort_states(self):
|
||||||
"""Should only be called after all the times have been resolved"""
|
"""Should only be called after all the times have been resolved"""
|
||||||
assert all(
|
assert all(
|
||||||
isinstance(state.time, time) for state in self.states
|
isinstance(state.time, datetime.time) for state in self.states
|
||||||
), 'Times have not all been resolved yet'
|
), 'Times have not all been resolved yet'
|
||||||
self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
|
self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
|
||||||
|
|
||||||
def current_state(self, now: time) -> ControllerStateConfig:
|
def current_state(self, now: datetime.time) -> ControllerStateConfig:
|
||||||
self.sort_states()
|
self.sort_states()
|
||||||
for state in self.states:
|
for state in self.states:
|
||||||
if state.time <= now:
|
if state.time <= now:
|
||||||
@@ -114,11 +114,11 @@ class RoomControllerConfig(BaseModel):
|
|||||||
else:
|
else:
|
||||||
return self.states[0]
|
return self.states[0]
|
||||||
|
|
||||||
def current_scene(self, now: time) -> Dict:
|
def current_scene(self, now: datetime.time) -> Dict:
|
||||||
state = self.current_state(now)
|
state = self.current_state(now)
|
||||||
return state.scene
|
return state.scene
|
||||||
|
|
||||||
def current_off_duration(self, now: time) -> timedelta:
|
def current_off_duration(self, now: datetime.time) -> datetime.timedelta:
|
||||||
state = self.current_state(now)
|
state = self.current_state(now)
|
||||||
if state.off_duration is None:
|
if state.off_duration is None:
|
||||||
if self.off_duration is None:
|
if self.off_duration is None:
|
||||||
|
|||||||
Reference in New Issue
Block a user