fixed default off duration
This commit is contained in:
@@ -13,10 +13,19 @@ from appdaemon.plugins.mqtt.mqttapi import Mqtt
|
|||||||
from astral import SunDirection
|
from astral import SunDirection
|
||||||
|
|
||||||
|
|
||||||
|
def str_to_timedelta(input_str: str) -> timedelta:
|
||||||
|
try:
|
||||||
|
hours, minutes, seconds = map(int, input_str.split(':'))
|
||||||
|
return timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
||||||
|
except Exception:
|
||||||
|
return timedelta()
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class RoomState:
|
class RoomState:
|
||||||
scene: Dict[str, Dict[str, str | int]]
|
scene: Dict[str, Dict[str, str | int]]
|
||||||
off_duration: timedelta = field(default_factory=timedelta)
|
# off_duration: timedelta = field(default_factory=timedelta)
|
||||||
|
off_duration: timedelta = None
|
||||||
time: time = None
|
time: time = None
|
||||||
time_fmt: List[str] = field(default_factory=lambda : ['%H:%M:%S', '%I:%M:%S %p'], repr=False)
|
time_fmt: List[str] = field(default_factory=lambda : ['%H:%M:%S', '%I:%M:%S %p'], repr=False)
|
||||||
elevation: int | float = None
|
elevation: int | float = None
|
||||||
@@ -45,11 +54,7 @@ class RoomState:
|
|||||||
self.elevation = float(self.elevation)
|
self.elevation = float(self.elevation)
|
||||||
|
|
||||||
if isinstance(self.off_duration, str):
|
if isinstance(self.off_duration, str):
|
||||||
try:
|
self.off_duration = str_to_timedelta(self.off_duration)
|
||||||
hours, minutes, seconds = map(int, self.off_duration.split(':'))
|
|
||||||
self.off_duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
|
||||||
except Exception:
|
|
||||||
self.off_duration = timedelta()
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_json(cls, json_input):
|
def from_json(cls, json_input):
|
||||||
@@ -61,13 +66,21 @@ class RoomConfig:
|
|||||||
states: List[RoomState]
|
states: List[RoomState]
|
||||||
off_duration: timedelta = field(default_factory=timedelta)
|
off_duration: timedelta = field(default_factory=timedelta)
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if isinstance(self.off_duration, str):
|
||||||
|
self.off_duration = str_to_timedelta(self.off_duration)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_app_config(cls, app_cfg: Dict[str, Dict]):
|
def from_app_config(cls, app_cfg: Dict[str, Dict]):
|
||||||
if 'class' in app_cfg:
|
if 'class' in app_cfg:
|
||||||
app_cfg.pop('class')
|
app_cfg.pop('class')
|
||||||
if 'module' in app_cfg:
|
if 'module' in app_cfg:
|
||||||
app_cfg.pop('module')
|
app_cfg.pop('module')
|
||||||
return cls(states=[RoomState.from_json(s) for s in app_cfg['states']])
|
self = cls(states=[RoomState.from_json(s) for s in app_cfg['states']])
|
||||||
|
for state in self.states:
|
||||||
|
if state.off_duration is None:
|
||||||
|
state.off_duration = self.off_duration
|
||||||
|
return self
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_yaml(cls, yaml_path: Path, app_name: str):
|
def from_yaml(cls, yaml_path: Path, app_name: str):
|
||||||
@@ -180,16 +193,16 @@ class RoomController(Hass, Mqtt):
|
|||||||
# self.log(f'Getting state for datetime: {now.strftime("%I:%M:%S %p")}')
|
# self.log(f'Getting state for datetime: {now.strftime("%I:%M:%S %p")}')
|
||||||
time = time or (await self.get_now()).time()
|
time = time or (await self.get_now()).time()
|
||||||
time_fmt = "%I:%M:%S %p"
|
time_fmt = "%I:%M:%S %p"
|
||||||
self.log(f'Getting state before: {time.strftime(time_fmt)}')
|
# self.log(f'Getting state before: {time.strftime(time_fmt)}')
|
||||||
for state in self.states:
|
for state in self.states:
|
||||||
time_str = state.time.strftime(time_fmt)
|
time_str = state.time.strftime(time_fmt)
|
||||||
if state.time <= time:
|
if state.time <= time:
|
||||||
self.log(f'Selected state from {time_str}')
|
# self.log(f'Selected state from {time_str}')
|
||||||
return state
|
return state
|
||||||
|
# else:
|
||||||
|
# self.log(f'Not {time_str}')
|
||||||
else:
|
else:
|
||||||
self.log(f'Not {time_str}')
|
# self.log(f'Defaulting to first state')
|
||||||
else:
|
|
||||||
self.log(f'Defaulting to first state')
|
|
||||||
return self.states[0]
|
return self.states[0]
|
||||||
|
|
||||||
async def current_scene(self, time: time = None) -> Dict[str, Dict[str, str | int]]:
|
async def current_scene(self, time: time = None) -> Dict[str, Dict[str, str | int]]:
|
||||||
|
|||||||
Reference in New Issue
Block a user