Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implement pydantic for sockets #435

Merged
merged 8 commits into from
Feb 26, 2022
Merged
Show file tree
Hide file tree
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
44 changes: 44 additions & 0 deletions mypy.ini
Original file line number Diff line number Diff line change
Expand Up @@ -146,3 +146,47 @@ disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true

[mypy-pytradfri.device.socket]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true

[mypy-pytradfri.device.socket_control]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true

[mypy-pytradfri.device.air_purifier]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true

[mypy-pytradfri.device.air_purifier_control]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
3 changes: 2 additions & 1 deletion pytradfri/device/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from .blind_control import BlindControl, BlindResponse
from .light_control import LightControl
from .signal_repeater_control import SignalRepeaterControl
from .socket import SocketResponse
from .socket_control import SocketControl


Expand Down Expand Up @@ -60,7 +61,7 @@ class DeviceResponse(ApiResourceResponse):
signal_repeater_control: Optional[List[Dict[str, Any]]] = Field(
alias=ROOT_SIGNAL_REPEATER
)
socket_control: Optional[List[Dict[str, Any]]] = Field(alias=ATTR_SWITCH_PLUG)
socket_control: Optional[List[SocketResponse]] = Field(alias=ATTR_SWITCH_PLUG)


class Device(ApiResource):
Expand Down
12 changes: 6 additions & 6 deletions pytradfri/device/air_purifier_control.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,36 +29,36 @@ def air_purifiers(self) -> list[AirPurifier]:
"""Return air purifier objects of the air purifier control."""
return [AirPurifier(self._device, i) for i in range(len(self.raw))]

def turn_off(self, *, index=0) -> Command:
def turn_off(self, *, index: int = 0) -> Command:
"""Turn the device off."""
return self.set_value({ATTR_AIR_PURIFIER_MODE: 0}, index=index)

def turn_on_auto_mode(self, *, index=0) -> Command:
def turn_on_auto_mode(self, *, index: int = 0) -> Command:
"""Turn on auto mode."""
return self.set_value(
{ATTR_AIR_PURIFIER_MODE: ATTR_AIR_PURIFIER_MODE_AUTO}, index=index
)

def set_fan_speed(self, mode: int, *, index=0) -> Command:
def set_fan_speed(self, mode: int, *, index: int = 0) -> Command:
"""Set the fan speed of the purifier."""
self._value_validate(mode, RANGE_AIR_PURIFIER, "Air Purifier mode")
return self.set_value({ATTR_AIR_PURIFIER_MODE: mode}, index=index)

def set_controls_locked(self, locked: bool, *, index=0) -> Command:
def set_controls_locked(self, locked: bool, *, index: int = 0) -> Command:
"""Set physical controls locked of the air purifier."""

return self.set_value(
{ATTR_AIR_PURIFIER_CONTROLS_LOCKED: 1 if locked else 0}, index=index
)

def set_leds_off(self, leds_off: bool, *, index=0) -> Command:
def set_leds_off(self, leds_off: bool, *, index: int = 0) -> Command:
"""Set led's off/on of the air purifier."""

return self.set_value(
{ATTR_AIR_PURIFIER_LEDS_OFF: 1 if leds_off else 0}, index=index
)

def set_value(self, value: dict[str, bool | int], *, index=0) -> Command:
def set_value(self, value: dict[str, bool | int], *, index: int = 0) -> Command:
"""Set values on air purifier control.

Returns a Command.
Expand Down
28 changes: 21 additions & 7 deletions pytradfri/device/socket.py
Original file line number Diff line number Diff line change
@@ -1,32 +1,46 @@
"""Represent a socket."""
MartinHjelmare marked this conversation as resolved.
Show resolved Hide resolved
from __future__ import annotations

from typing import Any
from typing import TYPE_CHECKING

from ..const import ATTR_DEVICE_STATE
from pydantic import BaseModel, Field

from ..const import ATTR_DEVICE_STATE, ATTR_ID


class SocketResponse(BaseModel):
"""Represent API response for a blind."""

id: int = Field(alias=ATTR_ID)
state: int = Field(alias=ATTR_DEVICE_STATE)


if TYPE_CHECKING:
# avoid cyclic import at runtime.
from . import Device


class Socket:
"""Represent a socket."""

def __init__(self, device, index):
def __init__(self, device: Device, index: int) -> None:
"""Create object of class."""
self.device = device
self.index = index

@property
def state(self):
def state(self) -> bool:
"""State."""
return self.raw.get(ATTR_DEVICE_STATE) == 1
return self.raw.state == 1

@property
def raw(self) -> dict[str, Any]:
def raw(self) -> SocketResponse:
"""Return raw data that it represents."""
socket_control_response = self.device.raw.socket_control
assert socket_control_response is not None
return socket_control_response[self.index]

def __repr__(self):
def __repr__(self) -> str:
"""Return representation of class object."""
state = "on" if self.state else "off"
return f"<Socket #{self.index} - name: {self.device.name}, state: {state}>"
14 changes: 6 additions & 8 deletions pytradfri/device/socket_control.py
Original file line number Diff line number Diff line change
@@ -1,34 +1,32 @@
"""Class to control the sockets."""
from __future__ import annotations

from typing import Any

from ..command import Command
from ..const import ATTR_DEVICE_STATE, ATTR_SWITCH_PLUG
from .base_controller import BaseController
from .socket import Socket
from .socket import Socket, SocketResponse


class SocketControl(BaseController):
"""Class to control the sockets."""

@property
def raw(self) -> list[dict[str, Any]]:
def raw(self) -> list[SocketResponse]:
"""Return raw data that it represents."""
socket_control_response = self._device.raw.socket_control
assert socket_control_response is not None
return socket_control_response

@property
def sockets(self):
def sockets(self) -> list[Socket]:
"""Return socket objects of the socket control."""
return [Socket(self._device, i) for i in range(len(self.raw))]

def set_state(self, state, *, index=0):
def set_state(self, state: int, *, index: int = 0) -> Command:
"""Set state of a socket."""
return self.set_values({ATTR_DEVICE_STATE: int(state)}, index=index)
return self.set_values({ATTR_DEVICE_STATE: state}, index=index)

def set_values(self, values, *, index=0):
def set_values(self, values: dict[str, int], *, index: int = 0) -> Command:
"""Set values on socket control.

Returns a Command.
Expand Down
12 changes: 8 additions & 4 deletions tests/test_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -336,15 +336,19 @@ def test_socket_value_setting(function_name, comment, test_input, expected_resul

def test_socket_state_off():
"""Test socket off."""
socket = Device(deepcopy(OUTLET)).socket_control.sockets[0]
socket.raw[ATTR_DEVICE_STATE] = 0
socket_response = deepcopy(OUTLET)
socket_response[ATTR_SWITCH_PLUG][0][ATTR_DEVICE_STATE] = 0

socket = Device(socket_response).socket_control.sockets[0]
assert socket.state is False


def test_socket_state_on():
"""Test socket on."""
socket = Device(deepcopy(OUTLET)).socket_control.sockets[0]
socket.raw[ATTR_DEVICE_STATE] = 1
socket_response = deepcopy(OUTLET)
socket_response[ATTR_SWITCH_PLUG][0][ATTR_DEVICE_STATE] = 1

socket = Device(socket_response).socket_control.sockets[0]
assert socket.state is True


Expand Down