Skip to content

Commit

Permalink
construct schedule schemas via factory
Browse files Browse the repository at this point in the history
  • Loading branch information
@zxdavb
zxdavb committed Oct 30, 2024
1 parent 3c03c85 commit b7d9225
Showing 1 changed file with 147 additions and 106 deletions.
253 changes: 147 additions & 106 deletions src/evohomeasync2/schema/schedule.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from __future__ import annotations

from collections.abc import Callable
from typing import Any, Final

import voluptuous as vol
Expand All @@ -20,123 +21,148 @@
SZ_SWITCHPOINTS,
SZ_TIME_OF_DAY,
)
from .helpers import pascal_case
from .helpers import do_nothing, pascal_case, snake_to_camel
from .typedefs import _EvoDictT, _EvoListT

_ScheduleT = dict[str, dict[str, Any]]


#
# These are returned from vendor's API (GET)...
SCH_GET_SWITCHPOINT_DHW: Final = vol.Schema( # TODO: checkme
{
vol.Required(SZ_DHW_STATE): vol.Any(SZ_ON, SZ_OFF),
vol.Required(SZ_TIME_OF_DAY): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_SWITCHPOINT_ZONE: Final = vol.Schema(
{
vol.Optional(SZ_COOL_SETPOINT): float, # an extrapolation
vol.Required(SZ_HEAT_SETPOINT): vol.All(float, vol.Range(min=5, max=35)),
vol.Required(SZ_TIME_OF_DAY): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_DAY_OF_WEEK_DHW: Final = vol.Schema(
{
vol.Required(SZ_DAY_OF_WEEK): vol.In(DAYS_OF_WEEK),
vol.Required(SZ_SWITCHPOINTS): [SCH_GET_SWITCHPOINT_DHW],
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_DAY_OF_WEEK_ZONE: Final = vol.Schema(
{
vol.Required(SZ_DAY_OF_WEEK): vol.In(DAYS_OF_WEEK),
vol.Required(SZ_SWITCHPOINTS): [SCH_GET_SWITCHPOINT_ZONE],
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_SCHEDULE_DHW: Final = vol.Schema(
{
vol.Required(SZ_DAILY_SCHEDULES): [SCH_GET_DAY_OF_WEEK_DHW],
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_SCHEDULE_ZONE: Final = vol.Schema(
{
vol.Required(SZ_DAILY_SCHEDULES): [SCH_GET_DAY_OF_WEEK_ZONE],
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_SCHEDULE: Final = vol.Schema( # PUT /{self.TYPE}/{self.id}/schedule
vol.Any(SCH_GET_SCHEDULE_DHW, SCH_GET_SCHEDULE_ZONE),
extra=vol.PREVENT_EXTRA,
)
def _factory_get_schedule_dhw(fnc: Callable[[str], str] = do_nothing) -> vol.Schema:
"""Factory for the DHW schedule schema."""

SCH_GET_SWITCHPOINT_DHW: Final = vol.Schema( # TODO: checkme
{
vol.Required(fnc(SZ_DHW_STATE)): vol.Any(SZ_ON, SZ_OFF),
vol.Required(fnc(SZ_TIME_OF_DAY)): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_DAY_OF_WEEK_DHW: Final = vol.Schema(
{
vol.Required(fnc(SZ_DAY_OF_WEEK)): vol.In(DAYS_OF_WEEK),
vol.Required(fnc(SZ_SWITCHPOINTS)): [SCH_GET_SWITCHPOINT_DHW],
},
extra=vol.PREVENT_EXTRA,
)

return vol.Schema(
{
vol.Required(fnc(SZ_DAILY_SCHEDULES)): [SCH_GET_DAY_OF_WEEK_DHW],
},
extra=vol.PREVENT_EXTRA,
)


def _factory_get_schedule_zone(fnc: Callable[[str], str] = do_nothing) -> vol.Schema:
"""Factory for the zone schedule schema."""

SCH_GET_SWITCHPOINT_ZONE: Final = vol.Schema(
{
vol.Optional(fnc(SZ_COOL_SETPOINT)): float, # an extrapolation
vol.Required(fnc(SZ_HEAT_SETPOINT)): vol.All(
float, vol.Range(min=5, max=35)
),
vol.Required(fnc(SZ_TIME_OF_DAY)): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_GET_DAY_OF_WEEK_ZONE: Final = vol.Schema(
{
vol.Required(fnc(SZ_DAY_OF_WEEK)): vol.In(DAYS_OF_WEEK),
vol.Required(fnc(SZ_SWITCHPOINTS)): [SCH_GET_SWITCHPOINT_ZONE],
},
extra=vol.PREVENT_EXTRA,
)

return vol.Schema(
{
vol.Required(fnc(SZ_DAILY_SCHEDULES)): [SCH_GET_DAY_OF_WEEK_ZONE],
},
extra=vol.PREVENT_EXTRA,
)


def _factory_get_schedule(fnc: Callable[[str], str] = do_nothing) -> vol.Schema:
"""Factory for the schedule schema."""

return vol.Schema(
vol.Any(SCH_GET_SCHEDULE_DHW, SCH_GET_SCHEDULE_ZONE),
extra=vol.PREVENT_EXTRA,
)


#
# These are as to be provided to the vendor's API (PUT)...
# This is after modified by evohome-client (PUT), an evohome-client anachronism?
SCH_PUT_SWITCHPOINT_DHW: Final = vol.Schema( # TODO: checkme
{
vol.Required(pascal_case(SZ_DHW_STATE)): vol.Any(SZ_ON, SZ_OFF),
vol.Required(pascal_case(SZ_TIME_OF_DAY)): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_SWITCHPOINT_ZONE: Final = vol.Schema(
{ # NOTE: SZ_HEAT_SETPOINT is not .capitalized()
vol.Required(SZ_HEAT_SETPOINT): vol.All(float, vol.Range(min=5, max=35)),
vol.Required(pascal_case(SZ_TIME_OF_DAY)): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_DAY_OF_WEEK_DHW: Final = vol.Schema(
{
vol.Required(pascal_case(SZ_DAY_OF_WEEK)): vol.All(
int, vol.Range(min=0, max=6)
), # 0 is Monday
vol.Required(pascal_case(SZ_SWITCHPOINTS)): [SCH_PUT_SWITCHPOINT_DHW],
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_DAY_OF_WEEK_ZONE: Final = vol.Schema(
{
vol.Required(pascal_case(SZ_DAY_OF_WEEK)): vol.All(
int, vol.Range(min=0, max=6)
), # 0 is Monday
vol.Required(pascal_case(SZ_SWITCHPOINTS)): [SCH_PUT_SWITCHPOINT_ZONE],
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_SCHEDULE_DHW: Final = vol.Schema(
{
vol.Required(pascal_case(SZ_DAILY_SCHEDULES)): [SCH_PUT_DAY_OF_WEEK_DHW],
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_SCHEDULE_ZONE: Final = vol.Schema(
{
vol.Required(pascal_case(SZ_DAILY_SCHEDULES)): [SCH_PUT_DAY_OF_WEEK_ZONE],
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_SCHEDULE: Final = vol.Schema( # PUT /{self.TYPE}/{self.id}/schedule
vol.Any(SCH_PUT_SCHEDULE_DHW, SCH_PUT_SCHEDULE_ZONE),
extra=vol.PREVENT_EXTRA,
)
def _factory_put_schedule_dhw(fnc: Callable[[str], str] = do_nothing) -> vol.Schema:
"""Factory for the zone schedule schema."""

SCH_PUT_SWITCHPOINT_DHW: Final = vol.Schema( # TODO: checkme
{
vol.Required(fnc(SZ_DHW_STATE)): vol.Any(SZ_ON, SZ_OFF),
vol.Required(fnc(SZ_TIME_OF_DAY)): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_DAY_OF_WEEK_DHW: Final = vol.Schema(
{
vol.Required(fnc(SZ_DAY_OF_WEEK)): vol.All(
int, vol.Range(min=0, max=6)
), # 0 is Monday
vol.Required(fnc(SZ_SWITCHPOINTS)): [SCH_PUT_SWITCHPOINT_DHW],
},
extra=vol.PREVENT_EXTRA,
)

return vol.Schema(
{
vol.Required(fnc(SZ_DAILY_SCHEDULES)): [SCH_PUT_DAY_OF_WEEK_DHW],
},
extra=vol.PREVENT_EXTRA,
)


def _factory_put_schedule_zone(fnc: Callable[[str], str] = do_nothing) -> vol.Schema:
"""Factory for the zone schedule schema."""

SCH_PUT_SWITCHPOINT_ZONE: Final = vol.Schema(
{ # NOTE: SZ_HEAT_SETPOINT is not .capitalized()
vol.Required(SZ_HEAT_SETPOINT): vol.All(float, vol.Range(min=5, max=35)),
vol.Required(fnc(SZ_TIME_OF_DAY)): vol.Datetime(format="%H:%M:00"),
},
extra=vol.PREVENT_EXTRA,
)

SCH_PUT_DAY_OF_WEEK_ZONE: Final = vol.Schema(
{
vol.Required(fnc(SZ_DAY_OF_WEEK)): vol.All(
int, vol.Range(min=0, max=6)
), # 0 is Monday
vol.Required(fnc(SZ_SWITCHPOINTS)): [SCH_PUT_SWITCHPOINT_ZONE],
},
extra=vol.PREVENT_EXTRA,
)

return vol.Schema(
{
vol.Required(fnc(SZ_DAILY_SCHEDULES)): [SCH_PUT_DAY_OF_WEEK_ZONE],
},
extra=vol.PREVENT_EXTRA,
)


def _factory_put_schedule(fnc: Callable[[str], str] = do_nothing) -> vol.Schema:
"""Factory for the schedule schema."""

return vol.Schema(
vol.Any(_factory_put_schedule_dhw(fnc), _factory_put_schedule_zone(fnc)),
extra=vol.PREVENT_EXTRA,
)


#
Expand Down Expand Up @@ -207,3 +233,18 @@ def convert_to_get_schedule(schedule: _EvoDictT) -> _EvoDictT:
get_schedule[SZ_DAILY_SCHEDULES].append(get_day_schedule)

return get_schedule


SCH_GET_SCHEDULE_DHW: Final = _factory_get_schedule_dhw(snake_to_camel)

SCH_GET_SCHEDULE_ZONE: Final = _factory_get_schedule_zone(snake_to_camel)

SCH_PUT_SCHEDULE_DHW: Final = _factory_put_schedule_dhw(pascal_case)

SCH_PUT_SCHEDULE_ZONE: Final = _factory_put_schedule_zone(pascal_case)

# GET /{self.TYPE}/{self.id}/schedule
SCH_GET_SCHEDULE: Final = _factory_get_schedule(snake_to_camel)

# PUT /{self.TYPE}/{self.id}/schedule
SCH_PUT_SCHEDULE: Final = _factory_put_schedule(pascal_case)

0 comments on commit b7d9225

Please sign in to comment.