fixed reconstruction of sun direction from state with a new field validator
This commit is contained in:
@@ -4,7 +4,7 @@ from typing import Annotated, Dict, List, Optional, Self, Union
|
|||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
from astral import SunDirection
|
from astral import SunDirection
|
||||||
from pydantic import BaseModel, BeforeValidator, Field, model_validator
|
from pydantic import BaseModel, BeforeValidator, Field, field_validator, model_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
|
||||||
@@ -47,7 +47,7 @@ class ApplyKwargs(BaseModel):
|
|||||||
class ControllerStateConfig(BaseModel):
|
class ControllerStateConfig(BaseModel):
|
||||||
time: Optional[str | datetime.time | datetime.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[SunDirection] = None
|
||||||
off_duration: Optional[OffDuration] = None
|
off_duration: Optional[OffDuration] = None
|
||||||
scene: dict[str, State] | str = Field(default_factory=dict)
|
scene: dict[str, State] | str = Field(default_factory=dict)
|
||||||
|
|
||||||
@@ -57,6 +57,17 @@ 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
|
||||||
|
|
||||||
|
@field_validator('direction', mode='before')
|
||||||
|
@classmethod
|
||||||
|
def check_sun_dir(cls, val: int | str | SunDirection | None) -> SunDirection:
|
||||||
|
if isinstance(val, str):
|
||||||
|
print(f'Str sun direction: {val}')
|
||||||
|
return str_to_direction(val)
|
||||||
|
elif isinstance(val, int):
|
||||||
|
return SunDirection.SETTING if val < 0 else SunDirection.RISING
|
||||||
|
elif isinstance(val, SunDirection):
|
||||||
|
return val
|
||||||
|
|
||||||
def to_apply_kwargs(self, transition: int = None):
|
def to_apply_kwargs(self, transition: int = None):
|
||||||
return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True)
|
return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True)
|
||||||
|
|
||||||
|
|||||||
@@ -95,6 +95,7 @@ class RoomController(Hass):
|
|||||||
self, sensor_entity_id=motion['sensor'], ref_entity_id=motion['ref_entity']
|
self, sensor_entity_id=motion['sensor'], ref_entity_id=motion['ref_entity']
|
||||||
)
|
)
|
||||||
|
|
||||||
|
state: ControllerStateConfig
|
||||||
for state in sorted(self._room_config.states, key=lambda s: s.time, reverse=True):
|
for state in sorted(self._room_config.states, key=lambda s: s.time, reverse=True):
|
||||||
if isinstance(state.time, datetime.datetime):
|
if isinstance(state.time, datetime.datetime):
|
||||||
t = state.time.time()
|
t = state.time.time()
|
||||||
@@ -184,8 +185,9 @@ class RoomController(Hass):
|
|||||||
try:
|
try:
|
||||||
attrs = self.state_entity.get_state('all')['attributes']
|
attrs = self.state_entity.get_state('all')['attributes']
|
||||||
state = ControllerStateConfig.model_validate(attrs)
|
state = ControllerStateConfig.model_validate(attrs)
|
||||||
except Exception:
|
except Exception as e:
|
||||||
state = ControllerStateConfig()
|
state = ControllerStateConfig()
|
||||||
|
logger.exception(e)
|
||||||
finally:
|
finally:
|
||||||
# self.log(f'Current state: {state.model_dump(exclude_none=True)}', level='DEBUG')
|
# self.log(f'Current state: {state.model_dump(exclude_none=True)}', level='DEBUG')
|
||||||
return state
|
return state
|
||||||
|
|||||||
Reference in New Issue
Block a user