Use APIRequestProtocol

This commit is contained in:
Martin Hjelmare 2022-06-20 15:35:27 +02:00
parent 00ab475d0f
commit b21f69ab7b
7 changed files with 23 additions and 27 deletions

View File

@ -2,11 +2,10 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from datetime import timedelta from datetime import timedelta
from typing import Any, cast from typing import cast
from pytradfri.command import Command from pytradfri.api.aiocoap_api import APIRequestProtocol
from pytradfri.device import Device from pytradfri.device import Device
from pytradfri.error import RequestError from pytradfri.error import RequestError
from pytradfri.resource import ApiResource from pytradfri.resource import ApiResource
@ -30,7 +29,7 @@ class TradfriDeviceDataUpdateCoordinator(DataUpdateCoordinator[Device]):
hass: HomeAssistant, hass: HomeAssistant,
*, *,
config_entry: ConfigEntry, config_entry: ConfigEntry,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
device: Device, device: Device,
) -> None: ) -> None:
"""Initialize device coordinator.""" """Initialize device coordinator."""

View File

@ -2,10 +2,9 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any, cast from typing import Any, cast
from pytradfri.command import Command from pytradfri.api.aiocoap_api import APIRequestProtocol
from homeassistant.components.cover import ATTR_POSITION, CoverEntity from homeassistant.components.cover import ATTR_POSITION, CoverEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -25,7 +24,7 @@ async def async_setup_entry(
"""Load Tradfri covers based on a config entry.""" """Load Tradfri covers based on a config entry."""
gateway_id = config_entry.data[CONF_GATEWAY_ID] gateway_id = config_entry.data[CONF_GATEWAY_ID]
coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR]
api: Callable[[Command | list[Command]], Awaitable[Any]] = coordinator_data[KEY_API] api: APIRequestProtocol = coordinator_data[KEY_API]
async_add_entities( async_add_entities(
TradfriCover( TradfriCover(
@ -46,7 +45,7 @@ class TradfriCover(TradfriBaseEntity, CoverEntity):
def __init__( def __init__(
self, self,
device_coordinator: TradfriDeviceDataUpdateCoordinator, device_coordinator: TradfriDeviceDataUpdateCoordinator,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
gateway_id: str, gateway_id: str,
) -> None: ) -> None:
"""Initialize a switch.""" """Initialize a switch."""

View File

@ -3,10 +3,11 @@
from __future__ import annotations from __future__ import annotations
from abc import abstractmethod from abc import abstractmethod
from collections.abc import Awaitable, Callable, Coroutine from collections.abc import Callable, Coroutine
from functools import wraps from functools import wraps
from typing import Any, cast from typing import Any, cast
from pytradfri.api.aiocoap_api import APIRequestProtocol
from pytradfri.command import Command from pytradfri.command import Command
from pytradfri.device import Device from pytradfri.device import Device
from pytradfri.error import RequestError from pytradfri.error import RequestError
@ -20,7 +21,7 @@ from .coordinator import TradfriDeviceDataUpdateCoordinator
def handle_error( def handle_error(
func: Callable[[Command | list[Command]], Awaitable[Any]], func: APIRequestProtocol,
) -> Callable[[Command | list[Command]], Coroutine[Any, Any, None]]: ) -> Callable[[Command | list[Command]], Coroutine[Any, Any, None]]:
"""Handle tradfri api call error.""" """Handle tradfri api call error."""
@ -44,7 +45,7 @@ class TradfriBaseEntity(CoordinatorEntity[TradfriDeviceDataUpdateCoordinator]):
self, self,
device_coordinator: TradfriDeviceDataUpdateCoordinator, device_coordinator: TradfriDeviceDataUpdateCoordinator,
gateway_id: str, gateway_id: str,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
) -> None: ) -> None:
"""Initialize a device.""" """Initialize a device."""
super().__init__(device_coordinator) super().__init__(device_coordinator)

View File

@ -2,10 +2,9 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any, cast from typing import Any, cast
from pytradfri.command import Command from pytradfri.api.aiocoap_api import APIRequestProtocol
from homeassistant.components.fan import FanEntity, FanEntityFeature from homeassistant.components.fan import FanEntity, FanEntityFeature
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -38,7 +37,7 @@ async def async_setup_entry(
"""Load Tradfri switches based on a config entry.""" """Load Tradfri switches based on a config entry."""
gateway_id = config_entry.data[CONF_GATEWAY_ID] gateway_id = config_entry.data[CONF_GATEWAY_ID]
coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR]
api: Callable[[Command | list[Command]], Awaitable[Any]] = coordinator_data[KEY_API] api: APIRequestProtocol = coordinator_data[KEY_API]
async_add_entities( async_add_entities(
TradfriAirPurifierFan( TradfriAirPurifierFan(
@ -73,7 +72,7 @@ class TradfriAirPurifierFan(TradfriBaseEntity, FanEntity):
def __init__( def __init__(
self, self,
device_coordinator: TradfriDeviceDataUpdateCoordinator, device_coordinator: TradfriDeviceDataUpdateCoordinator,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
gateway_id: str, gateway_id: str,
) -> None: ) -> None:
"""Initialize a switch.""" """Initialize a switch."""

View File

@ -2,10 +2,9 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any, cast from typing import Any, cast
from pytradfri.command import Command from pytradfri.api.aiocoap_api import APIRequestProtocol
from homeassistant.components.light import ( from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_BRIGHTNESS,
@ -35,7 +34,7 @@ async def async_setup_entry(
"""Load Tradfri lights based on a config entry.""" """Load Tradfri lights based on a config entry."""
gateway_id = config_entry.data[CONF_GATEWAY_ID] gateway_id = config_entry.data[CONF_GATEWAY_ID]
coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR]
api: Callable[[Command | list[Command]], Awaitable[Any]] = coordinator_data[KEY_API] api: APIRequestProtocol = coordinator_data[KEY_API]
async_add_entities( async_add_entities(
TradfriLight( TradfriLight(
@ -58,7 +57,7 @@ class TradfriLight(TradfriBaseEntity, LightEntity):
def __init__( def __init__(
self, self,
device_coordinator: TradfriDeviceDataUpdateCoordinator, device_coordinator: TradfriDeviceDataUpdateCoordinator,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
gateway_id: str, gateway_id: str,
) -> None: ) -> None:
"""Initialize a Light.""" """Initialize a Light."""

View File

@ -2,11 +2,11 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, cast from typing import Any, cast
from pytradfri.command import Command from pytradfri.api.aiocoap_api import APIRequestProtocol
from pytradfri.device import Device from pytradfri.device import Device
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
@ -133,7 +133,7 @@ async def async_setup_entry(
"""Set up a Tradfri config entry.""" """Set up a Tradfri config entry."""
gateway_id = config_entry.data[CONF_GATEWAY_ID] gateway_id = config_entry.data[CONF_GATEWAY_ID]
coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR]
api: Callable[[Command | list[Command]], Awaitable[Any]] = coordinator_data[KEY_API] api: APIRequestProtocol = coordinator_data[KEY_API]
entities: list[TradfriSensor] = [] entities: list[TradfriSensor] = []
@ -178,7 +178,7 @@ class TradfriSensor(TradfriBaseEntity, SensorEntity):
def __init__( def __init__(
self, self,
device_coordinator: TradfriDeviceDataUpdateCoordinator, device_coordinator: TradfriDeviceDataUpdateCoordinator,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
gateway_id: str, gateway_id: str,
description: TradfriSensorEntityDescription, description: TradfriSensorEntityDescription,
) -> None: ) -> None:

View File

@ -2,10 +2,9 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any, cast from typing import Any, cast
from pytradfri.command import Command from pytradfri.api.aiocoap_api import APIRequestProtocol
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -25,7 +24,7 @@ async def async_setup_entry(
"""Load Tradfri switches based on a config entry.""" """Load Tradfri switches based on a config entry."""
gateway_id = config_entry.data[CONF_GATEWAY_ID] gateway_id = config_entry.data[CONF_GATEWAY_ID]
coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] coordinator_data = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR]
api: Callable[[Command | list[Command]], Awaitable[Any]] = coordinator_data[KEY_API] api: APIRequestProtocol = coordinator_data[KEY_API]
async_add_entities( async_add_entities(
TradfriSwitch( TradfriSwitch(
@ -46,7 +45,7 @@ class TradfriSwitch(TradfriBaseEntity, SwitchEntity):
def __init__( def __init__(
self, self,
device_coordinator: TradfriDeviceDataUpdateCoordinator, device_coordinator: TradfriDeviceDataUpdateCoordinator,
api: Callable[[Command | list[Command]], Awaitable[Any]], api: APIRequestProtocol,
gateway_id: str, gateway_id: str,
) -> None: ) -> None:
"""Initialize a switch.""" """Initialize a switch."""