Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Streamline naming in the Axis integration (#112044) 

* Rename device.py to hub.py

* Rename AxisNetworkDevice to AxisHub

* Move hub.py into hub package

* Rename get_axis_device to get_axis_api

* Split out get_axis_api to its own module

* Rename device object to hub

* Rename device to api in config flow

* Convenience method to get hub
pull/111883/head
Robert Svensson 2024-03-02 17:32:51 +01:00 committed by GitHub
parent 196089e8b7
commit ece5587e1f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 183 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
homeassistant/components/axis/__init__.py
View File

@ -7,8 +7,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from .const import DOMAIN as AXIS_DOMAIN, PLATFORMS
from .device import AxisNetworkDevice, get_axis_device
from .errors import AuthenticationRequired, CannotConnect
from .hub import AxisHub, get_axis_api
_LOGGER = logging.getLogger(__name__)
@ -18,21 +18,21 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
hass.data.setdefault(AXIS_DOMAIN, {})
try:
api = await get_axis_device(hass, config_entry.data)
api = await get_axis_api(hass, config_entry.data)
except CannotConnect as err:
raise ConfigEntryNotReady from err
except AuthenticationRequired as err:
raise ConfigEntryAuthFailed from err
device = AxisNetworkDevice(hass, config_entry, api)
hass.data[AXIS_DOMAIN][config_entry.entry_id] = device
await device.async_update_device_registry()
hub = AxisHub(hass, config_entry, api)
hass.data[AXIS_DOMAIN][config_entry.entry_id] = hub
await hub.async_update_device_registry()
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
device.async_setup_events()
hub.async_setup_events()
config_entry.add_update_listener(device.async_new_address_callback)
config_entry.add_update_listener(hub.async_new_address_callback)
config_entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, device.shutdown)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, hub.shutdown)
)
return True
@ -40,8 +40,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload Axis device config entry."""
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN].pop(config_entry.entry_id)
return await device.async_reset()
hub: AxisHub = hass.data[AXIS_DOMAIN].pop(config_entry.entry_id)
return await hub.async_reset()
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:

35
homeassistant/components/axis/binary_sensor.py
View File

@ -19,9 +19,8 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
from .entity import AxisEventEntity
from .hub import AxisHub
DEVICE_CLASS = {
EventGroup.INPUT: BinarySensorDeviceClass.CONNECTIVITY,
@ -52,14 +51,14 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a Axis binary sensor."""
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.entry_id]
hub = AxisHub.get_hub(hass, config_entry)
@callback
def async_create_entity(event: Event) -> None:
"""Create Axis binary sensor entity."""
async_add_entities([AxisBinarySensor(event, device)])
async_add_entities([AxisBinarySensor(event, hub)])
device.api.event.subscribe(
hub.api.event.subscribe(
async_create_entity,
topic_filter=EVENT_TOPICS,
operation_filter=EventOperation.INITIALIZED,
@ -69,9 +68,9 @@ async def async_setup_entry(
class AxisBinarySensor(AxisEventEntity, BinarySensorEntity):
"""Representation of a binary Axis event."""
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, hub: AxisHub) -> None:
"""Initialize the Axis binary sensor."""
super().__init__(event, device)
super().__init__(event, hub)
self.cancel_scheduled_update: Callable[[], None] | None = None
self._attr_device_class = DEVICE_CLASS.get(event.group)
@ -94,13 +93,13 @@ class AxisBinarySensor(AxisEventEntity, BinarySensorEntity):
self.cancel_scheduled_update()
self.cancel_scheduled_update = None
if self.is_on or self.device.option_trigger_time == 0:
if self.is_on or self.hub.option_trigger_time == 0:
self.async_write_ha_state()
return
self.cancel_scheduled_update = async_call_later(
self.hass,
timedelta(seconds=self.device.option_trigger_time),
timedelta(seconds=self.hub.option_trigger_time),
scheduled_update,
)
@ -109,21 +108,21 @@ class AxisBinarySensor(AxisEventEntity, BinarySensorEntity):
"""Set binary sensor name."""
if (
event.group == EventGroup.INPUT
and event.id in self.device.api.vapix.ports
and self.device.api.vapix.ports[event.id].name
and event.id in self.hub.api.vapix.ports
and self.hub.api.vapix.ports[event.id].name
):
self._attr_name = self.device.api.vapix.ports[event.id].name
self._attr_name = self.hub.api.vapix.ports[event.id].name
elif event.group == EventGroup.MOTION:
event_data: FenceGuardHandler | LoiteringGuardHandler | MotionGuardHandler | Vmd4Handler | None = None
if event.topic_base == EventTopic.FENCE_GUARD:
event_data = self.device.api.vapix.fence_guard
event_data = self.hub.api.vapix.fence_guard
elif event.topic_base == EventTopic.LOITERING_GUARD:
event_data = self.device.api.vapix.loitering_guard
event_data = self.hub.api.vapix.loitering_guard
elif event.topic_base == EventTopic.MOTION_GUARD:
event_data = self.device.api.vapix.motion_guard
event_data = self.hub.api.vapix.motion_guard
elif event.topic_base == EventTopic.MOTION_DETECTION_4:
event_data = self.device.api.vapix.vmd4
event_data = self.hub.api.vapix.vmd4
if (
event_data
and event_data.initialized
@ -137,8 +136,8 @@ class AxisBinarySensor(AxisEventEntity, BinarySensorEntity):
if (
event.topic_base == EventTopic.OBJECT_ANALYTICS
and self.device.api.vapix.object_analytics.initialized
and (scenarios := self.device.api.vapix.object_analytics["0"].scenarios)
and self.hub.api.vapix.object_analytics.initialized
and (scenarios := self.hub.api.vapix.object_analytics["0"].scenarios)
):
for scenario_id, scenario in scenarios.items():
device_id = scenario.devices[0]["id"]

42
homeassistant/components/axis/camera.py
View File

@ -9,9 +9,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DEFAULT_STREAM_PROFILE, DEFAULT_VIDEO_SOURCE, DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
from .const import DEFAULT_STREAM_PROFILE, DEFAULT_VIDEO_SOURCE
from .entity import AxisEntity
from .hub import AxisHub
async def async_setup_entry(
@ -22,15 +22,15 @@ async def async_setup_entry(
"""Set up the Axis camera video stream."""
filter_urllib3_logging()
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.entry_id]
hub = AxisHub.get_hub(hass, config_entry)
if (
not (prop := device.api.vapix.params.property_handler.get("0"))
not (prop := hub.api.vapix.params.property_handler.get("0"))
or not prop.image_format
):
return
async_add_entities([AxisCamera(device)])
async_add_entities([AxisCamera(hub)])
class AxisCamera(AxisEntity, MjpegCamera):
@ -42,27 +42,27 @@ class AxisCamera(AxisEntity, MjpegCamera):
_mjpeg_url: str
_stream_source: str
def __init__(self, device: AxisNetworkDevice) -> None:
def __init__(self, hub: AxisHub) -> None:
"""Initialize Axis Communications camera component."""
AxisEntity.__init__(self, device)
AxisEntity.__init__(self, hub)
self._generate_sources()
MjpegCamera.__init__(
self,
username=device.username,
password=device.password,
username=hub.username,
password=hub.password,
mjpeg_url=self.mjpeg_source,
still_image_url=self.image_source,
authentication=HTTP_DIGEST_AUTHENTICATION,
unique_id=f"{device.unique_id}-camera",
unique_id=f"{hub.unique_id}-camera",
)
async def async_added_to_hass(self) -> None:
"""Subscribe camera events."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, self.device.signal_new_address, self._generate_sources
self.hass, self.hub.signal_new_address, self._generate_sources
)
)
@ -75,27 +75,27 @@ class AxisCamera(AxisEntity, MjpegCamera):
"""
image_options = self.generate_options(skip_stream_profile=True)
self._still_image_url = (
f"http://{self.device.host}:{self.device.port}/axis-cgi"
f"http://{self.hub.host}:{self.hub.port}/axis-cgi"
f"/jpg/image.cgi{image_options}"
)
mjpeg_options = self.generate_options()
self._mjpeg_url = (
f"http://{self.device.host}:{self.device.port}/axis-cgi"
f"http://{self.hub.host}:{self.hub.port}/axis-cgi"
f"/mjpg/video.cgi{mjpeg_options}"
)
stream_options = self.generate_options(add_video_codec_h264=True)
self._stream_source = (
f"rtsp://{self.device.username}:{self.device.password}"
f"@{self.device.host}/axis-media/media.amp{stream_options}"
f"rtsp://{self.hub.username}:{self.hub.password}"
f"@{self.hub.host}/axis-media/media.amp{stream_options}"
)
self.device.additional_diagnostics["camera_sources"] = {
self.hub.additional_diagnostics["camera_sources"] = {
"Image": self._still_image_url,
"MJPEG": self._mjpeg_url,
"Stream": (
f"rtsp://user:pass@{self.device.host}/axis-media"
f"rtsp://user:pass@{self.hub.host}/axis-media"
f"/media.amp{stream_options}"
),
}
@ -125,12 +125,12 @@ class AxisCamera(AxisEntity, MjpegCamera):
if (
not skip_stream_profile
and self.device.option_stream_profile != DEFAULT_STREAM_PROFILE
and self.hub.option_stream_profile != DEFAULT_STREAM_PROFILE
):
options_dict["streamprofile"] = self.device.option_stream_profile
options_dict["streamprofile"] = self.hub.option_stream_profile
if self.device.option_video_source != DEFAULT_VIDEO_SOURCE:
options_dict["camera"] = self.device.option_video_source
if self.hub.option_video_source != DEFAULT_VIDEO_SOURCE:
options_dict["camera"] = self.hub.option_video_source
if not options_dict:
return ""

18
homeassistant/components/axis/config_flow.py
View File

@ -37,8 +37,8 @@ from .const import (
DEFAULT_VIDEO_SOURCE,
DOMAIN as AXIS_DOMAIN,
)
from .device import AxisNetworkDevice, get_axis_device
from .errors import AuthenticationRequired, CannotConnect
from .hub import AxisHub, get_axis_api
AXIS_OUI = {"00:40:8c", "ac:cc:8e", "b8:a4:4f"}
DEFAULT_PORT = 80
@ -71,9 +71,9 @@ class AxisFlowHandler(ConfigFlow, domain=AXIS_DOMAIN):
if user_input is not None:
try:
device = await get_axis_device(self.hass, MappingProxyType(user_input))
api = await get_axis_api(self.hass, MappingProxyType(user_input))
serial = device.vapix.serial_number
serial = api.vapix.serial_number
await self.async_set_unique_id(format_mac(serial))
self._abort_if_unique_id_configured(
@ -90,7 +90,7 @@ class AxisFlowHandler(ConfigFlow, domain=AXIS_DOMAIN):
CONF_PORT: user_input[CONF_PORT],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_MODEL: device.vapix.product_number,
CONF_MODEL: api.vapix.product_number,
}
return await self._create_entry(serial)
@ -238,13 +238,13 @@ class AxisFlowHandler(ConfigFlow, domain=AXIS_DOMAIN):
class AxisOptionsFlowHandler(OptionsFlowWithConfigEntry):
"""Handle Axis device options."""
device: AxisNetworkDevice
hub: AxisHub
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the Axis device options."""
self.device = self.hass.data[AXIS_DOMAIN][self.config_entry.entry_id]
self.hub = AxisHub.get_hub(self.hass, self.config_entry)
return await self.async_step_configure_stream()
async def async_step_configure_stream(
@ -257,7 +257,7 @@ class AxisOptionsFlowHandler(OptionsFlowWithConfigEntry):
schema = {}
vapix = self.device.api.vapix
vapix = self.hub.api.vapix
# Stream profiles
@ -271,7 +271,7 @@ class AxisOptionsFlowHandler(OptionsFlowWithConfigEntry):
schema[
vol.Optional(
CONF_STREAM_PROFILE, default=self.device.option_stream_profile
CONF_STREAM_PROFILE, default=self.hub.option_stream_profile
)
] = vol.In(stream_profiles)
@ -290,7 +290,7 @@ class AxisOptionsFlowHandler(OptionsFlowWithConfigEntry):
video_sources[int(idx) + 1] = video_source.name
schema[
vol.Optional(CONF_VIDEO_SOURCE, default=self.device.option_video_source)
vol.Optional(CONF_VIDEO_SOURCE, default=self.hub.option_video_source)
] = vol.In(video_sources)
return self.async_show_form(

19
homeassistant/components/axis/diagnostics.py
View File

@ -8,8 +8,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MAC, CONF_PASSWORD, CONF_UNIQUE_ID, CONF_USERNAME
from homeassistant.core import HomeAssistant
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
from .hub import AxisHub
REDACT_CONFIG = {CONF_MAC, CONF_PASSWORD, CONF_UNIQUE_ID, CONF_USERNAME}
REDACT_BASIC_DEVICE_INFO = {"SerialNumber", "SocSerialNumber"}
@ -20,26 +19,26 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: ConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.entry_id]
diag: dict[str, Any] = device.additional_diagnostics.copy()
hub = AxisHub.get_hub(hass, config_entry)
diag: dict[str, Any] = hub.additional_diagnostics.copy()
diag["config"] = async_redact_data(config_entry.as_dict(), REDACT_CONFIG)
if device.api.vapix.api_discovery:
if hub.api.vapix.api_discovery:
diag["api_discovery"] = [
{"id": api.id, "name": api.name, "version": api.version}
for api in device.api.vapix.api_discovery.values()
for api in hub.api.vapix.api_discovery.values()
]
if device.api.vapix.basic_device_info:
if hub.api.vapix.basic_device_info:
diag["basic_device_info"] = async_redact_data(
device.api.vapix.basic_device_info["0"],
hub.api.vapix.basic_device_info["0"],
REDACT_BASIC_DEVICE_INFO,
)
if device.api.vapix.params:
if hub.api.vapix.params:
diag["params"] = async_redact_data(
device.api.vapix.params.items(),
hub.api.vapix.params.items(),
REDACT_VAPIX_PARAMS,
)

22
homeassistant/components/axis/entity.py
View File

@ -10,7 +10,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
from .hub import AxisHub
TOPIC_TO_EVENT_TYPE = {
EventTopic.DAY_NIGHT_VISION: "DayNight",
@ -37,13 +37,13 @@ class AxisEntity(Entity):
_attr_has_entity_name = True
def __init__(self, device: AxisNetworkDevice) -> None:
def __init__(self, hub: AxisHub) -> None:
"""Initialize the Axis event."""
self.device = device
self.hub = hub
self._attr_device_info = DeviceInfo(
identifiers={(AXIS_DOMAIN, device.unique_id)}, # type: ignore[arg-type]
serial_number=device.unique_id,
identifiers={(AXIS_DOMAIN, hub.unique_id)}, # type: ignore[arg-type]
serial_number=hub.unique_id,
)
async def async_added_to_hass(self) -> None:
@ -51,7 +51,7 @@ class AxisEntity(Entity):
self.async_on_remove(
async_dispatcher_connect(
self.hass,
self.device.signal_reachable,
self.hub.signal_reachable,
self.async_signal_reachable_callback,
)
)
@ -59,7 +59,7 @@ class AxisEntity(Entity):
@callback
def async_signal_reachable_callback(self) -> None:
"""Call when device connection state change."""
self._attr_available = self.device.available
self._attr_available = self.hub.available
self.async_write_ha_state()
@ -68,16 +68,16 @@ class AxisEventEntity(AxisEntity):
_attr_should_poll = False
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, hub: AxisHub) -> None:
"""Initialize the Axis event."""
super().__init__(device)
super().__init__(hub)
self._event_id = event.id
self._event_topic = event.topic_base
self._event_type = TOPIC_TO_EVENT_TYPE[event.topic_base]
self._attr_name = f"{self._event_type} {event.id}"
self._attr_unique_id = f"{device.unique_id}-{event.topic}-{event.id}"
self._attr_unique_id = f"{hub.unique_id}-{event.topic}-{event.id}"
self._attr_device_class = event.group.value
@ -90,7 +90,7 @@ class AxisEventEntity(AxisEntity):
"""Subscribe sensors events."""
await super().async_added_to_hass()
self.async_on_remove(
self.device.api.event.subscribe(
self.hub.api.event.subscribe(
self.async_event_callback,
id_filter=self._event_id,
topic_filter=self._event_topic,

4
homeassistant/components/axis/hub/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""Internal functionality not part of HA infrastructure."""
from .api import get_axis_api # noqa: F401
from .hub import AxisHub # noqa: F401

53
homeassistant/components/axis/hub/api.py Normal file
View File

@ -0,0 +1,53 @@
"""Axis network device abstraction."""
from asyncio import timeout
from types import MappingProxyType
from typing import Any
import axis
from axis.configuration import Configuration
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.httpx_client import get_async_client
from ..const import LOGGER
from ..errors import AuthenticationRequired, CannotConnect
async def get_axis_api(
hass: HomeAssistant,
config: MappingProxyType[str, Any],
) -> axis.AxisDevice:
"""Create a Axis device API."""
session = get_async_client(hass, verify_ssl=False)
device = axis.AxisDevice(
Configuration(
session,
config[CONF_HOST],
port=config[CONF_PORT],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
)
)
try:
async with timeout(30):
await device.vapix.initialize()
return device
except axis.Unauthorized as err:
LOGGER.warning(
"Connected to device at %s but not registered", config[CONF_HOST]
)
raise AuthenticationRequired from err
except (TimeoutError, axis.RequestError) as err:
LOGGER.error("Error connecting to the Axis device at %s", config[CONF_HOST])
raise CannotConnect from err
except axis.AxisException as err:
LOGGER.exception("Unknown Axis communication error occurred")
raise AuthenticationRequired from err

65
homeassistant/components/axis/device.py → homeassistant/components/axis/hub/hub.py
View File

@ -1,11 +1,10 @@
"""Axis network device abstraction."""
from asyncio import timeout
from types import MappingProxyType
from __future__ import annotations
from typing import Any
import axis
from axis.configuration import Configuration
from axis.errors import Unauthorized
from axis.stream_manager import Signal, State
from axis.vapix.interfaces.mqtt import mqtt_json_to_event
@ -28,10 +27,9 @@ from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.setup import async_when_setup
from .const import (
from ..const import (
ATTR_MANUFACTURER,
CONF_EVENTS,
CONF_STREAM_PROFILE,
@ -41,13 +39,11 @@ from .const import (
DEFAULT_TRIGGER_TIME,
DEFAULT_VIDEO_SOURCE,
DOMAIN as AXIS_DOMAIN,
LOGGER,
PLATFORMS,
)
from .errors import AuthenticationRequired, CannotConnect
class AxisNetworkDevice:
class AxisHub:
"""Manages a Axis device."""
def __init__(
@ -64,6 +60,13 @@ class AxisNetworkDevice:
self.additional_diagnostics: dict[str, Any] = {}
@callback
@staticmethod
def get_hub(hass: HomeAssistant, config_entry: ConfigEntry) -> AxisHub:
"""Get Axis hub from config entry."""
hub: AxisHub = hass.data[AXIS_DOMAIN][config_entry.entry_id]
return hub
@property
def host(self) -> str:
"""Return the host address of this device."""
@ -157,7 +160,7 @@ class AxisNetworkDevice:
@staticmethod
async def async_new_address_callback(
hass: HomeAssistant, entry: ConfigEntry
hass: HomeAssistant, config_entry: ConfigEntry
) -> None:
"""Handle signals of device getting new address.
@ -165,9 +168,9 @@ class AxisNetworkDevice:
This is a static method because a class method (bound method),
cannot be used with weak references.
"""
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][entry.entry_id]
device.api.config.host = device.host
async_dispatcher_send(hass, device.signal_new_address)
hub = AxisHub.get_hub(hass, config_entry)
hub.api.config.host = hub.host
async_dispatcher_send(hass, hub.signal_new_address)
async def async_update_device_registry(self) -> None:
"""Update device registry."""
@ -237,41 +240,3 @@ class AxisNetworkDevice:
return await self.hass.config_entries.async_unload_platforms(
self.config_entry, PLATFORMS
)
async def get_axis_device(
hass: HomeAssistant,
config: MappingProxyType[str, Any],
) -> axis.AxisDevice:
"""Create a Axis device."""
session = get_async_client(hass, verify_ssl=False)
device = axis.AxisDevice(
Configuration(
session,
config[CONF_HOST],
port=config[CONF_PORT],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
)
)
try:
async with timeout(30):
await device.vapix.initialize()
return device
except axis.Unauthorized as err:
LOGGER.warning(
"Connected to device at %s but not registered", config[CONF_HOST]
)
raise AuthenticationRequired from err
except (TimeoutError, axis.RequestError) as err:
LOGGER.error("Error connecting to the Axis device at %s", config[CONF_HOST])
raise CannotConnect from err
except axis.AxisException as err:
LOGGER.exception("Unknown Axis communication error occurred")
raise AuthenticationRequired from err

36
homeassistant/components/axis/light.py
View File

@ -8,9 +8,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
from .entity import AxisEventEntity
from .hub import AxisHub
async def async_setup_entry(
@ -19,20 +18,17 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a Axis light."""
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.entry_id]
hub = AxisHub.get_hub(hass, config_entry)
if (
device.api.vapix.light_control is None
or len(device.api.vapix.light_control) == 0
):
if hub.api.vapix.light_control is None or len(hub.api.vapix.light_control) == 0:
return
@callback
def async_create_entity(event: Event) -> None:
"""Create Axis light entity."""
async_add_entities([AxisLight(event, device)])
async_add_entities([AxisLight(event, hub)])
device.api.event.subscribe(
hub.api.event.subscribe(
async_create_entity,
topic_filter=EventTopic.LIGHT_STATUS,
operation_filter=EventOperation.INITIALIZED,
@ -44,16 +40,16 @@ class AxisLight(AxisEventEntity, LightEntity):
_attr_should_poll = True
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, hub: AxisHub) -> None:
"""Initialize the Axis light."""
super().__init__(event, device)
super().__init__(event, hub)
self._light_id = f"led{event.id}"
self.current_intensity = 0
self.max_intensity = 0
light_type = device.api.vapix.light_control[self._light_id].light_type
light_type = hub.api.vapix.light_control[self._light_id].light_type
self._attr_name = f"{light_type} {self._event_type} {event.id}"
self._attr_is_on = event.is_tripped
@ -65,13 +61,11 @@ class AxisLight(AxisEventEntity, LightEntity):
await super().async_added_to_hass()
current_intensity = (
await self.device.api.vapix.light_control.get_current_intensity(
self._light_id
)
await self.hub.api.vapix.light_control.get_current_intensity(self._light_id)
)
self.current_intensity = current_intensity
max_intensity = await self.device.api.vapix.light_control.get_valid_intensity(
max_intensity = await self.hub.api.vapix.light_control.get_valid_intensity(
self._light_id
)
self.max_intensity = max_intensity.high
@ -90,24 +84,22 @@ class AxisLight(AxisEventEntity, LightEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on light."""
if not self.is_on:
await self.device.api.vapix.light_control.activate_light(self._light_id)
await self.hub.api.vapix.light_control.activate_light(self._light_id)
if ATTR_BRIGHTNESS in kwargs:
intensity = int((kwargs[ATTR_BRIGHTNESS] / 255) * self.max_intensity)
await self.device.api.vapix.light_control.set_manual_intensity(
await self.hub.api.vapix.light_control.set_manual_intensity(
self._light_id, intensity
)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off light."""
if self.is_on:
await self.device.api.vapix.light_control.deactivate_light(self._light_id)
await self.hub.api.vapix.light_control.deactivate_light(self._light_id)
async def async_update(self) -> None:
"""Update brightness."""
current_intensity = (
await self.device.api.vapix.light_control.get_current_intensity(
self._light_id
)
await self.hub.api.vapix.light_control.get_current_intensity(self._light_id)
)
self.current_intensity = current_intensity

21
homeassistant/components/axis/switch.py
View File

@ -8,9 +8,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
from .entity import AxisEventEntity
from .hub import AxisHub
async def async_setup_entry(
@ -19,14 +18,14 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a Axis switch."""
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.entry_id]
hub = AxisHub.get_hub(hass, config_entry)
@callback
def async_create_entity(event: Event) -> None:
"""Create Axis switch entity."""
async_add_entities([AxisSwitch(event, device)])
async_add_entities([AxisSwitch(event, hub)])
device.api.event.subscribe(
hub.api.event.subscribe(
async_create_entity,
topic_filter=EventTopic.RELAY,
operation_filter=EventOperation.INITIALIZED,
@ -36,11 +35,11 @@ async def async_setup_entry(
class AxisSwitch(AxisEventEntity, SwitchEntity):
"""Representation of a Axis switch."""
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, hub: AxisHub) -> None:
"""Initialize the Axis switch."""
super().__init__(event, device)
if event.id and device.api.vapix.ports[event.id].name:
self._attr_name = device.api.vapix.ports[event.id].name
super().__init__(event, hub)
if event.id and hub.api.vapix.ports[event.id].name:
self._attr_name = hub.api.vapix.ports[event.id].name
self._attr_is_on = event.is_tripped
@callback
@ -51,8 +50,8 @@ class AxisSwitch(AxisEventEntity, SwitchEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on switch."""
await self.device.api.vapix.ports.close(self._event_id)
await self.hub.api.vapix.ports.close(self._event_id)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off switch."""
await self.device.api.vapix.ports.open(self._event_id)
await self.hub.api.vapix.ports.open(self._event_id)

4
tests/components/axis/test_config_flow.py
View File

@ -124,7 +124,7 @@ async def test_flow_fails_faulty_credentials(hass: HomeAssistant) -> None:
assert result["step_id"] == "user"
with patch(
"homeassistant.components.axis.config_flow.get_axis_device",
"homeassistant.components.axis.config_flow.get_axis_api",
side_effect=config_flow.AuthenticationRequired,
):
result = await hass.config_entries.flow.async_configure(
@ -150,7 +150,7 @@ async def test_flow_fails_cannot_connect(hass: HomeAssistant) -> None:
assert result["step_id"] == "user"
with patch(
"homeassistant.components.axis.config_flow.get_axis_device",
"homeassistant.components.axis.config_flow.get_axis_api",
side_effect=config_flow.CannotConnect,
):
result = await hass.config_entries.flow.async_configure(

14
tests/components/axis/test_device.py
View File

@ -182,7 +182,7 @@ async def test_device_not_accessible(
hass: HomeAssistant, config_entry, setup_default_vapix_requests
) -> None:
"""Failed setup schedules a retry of setup."""
with patch.object(axis, "get_axis_device", side_effect=axis.errors.CannotConnect):
with patch.object(axis, "get_axis_api", side_effect=axis.errors.CannotConnect):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[AXIS_DOMAIN] == {}
@ -193,7 +193,7 @@ async def test_device_trigger_reauth_flow(
) -> None:
"""Failed authentication trigger a reauthentication flow."""
with patch.object(
axis, "get_axis_device", side_effect=axis.errors.AuthenticationRequired
axis, "get_axis_api", side_effect=axis.errors.AuthenticationRequired
), patch.object(hass.config_entries.flow, "async_init") as mock_flow_init:
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@ -205,7 +205,7 @@ async def test_device_unknown_error(
hass: HomeAssistant, config_entry, setup_default_vapix_requests
) -> None:
"""Unknown errors are handled."""
with patch.object(axis, "get_axis_device", side_effect=Exception):
with patch.object(axis, "get_axis_api", side_effect=Exception):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[AXIS_DOMAIN] == {}
@ -217,7 +217,7 @@ async def test_shutdown(config) -> None:
entry = Mock()
entry.data = config
axis_device = axis.device.AxisNetworkDevice(hass, entry, Mock())
axis_device = axis.hub.AxisHub(hass, entry, Mock())
await axis_device.shutdown(None)
@ -229,7 +229,7 @@ async def test_get_device_fails(hass: HomeAssistant, config) -> None:
with patch(
"axis.vapix.vapix.Vapix.initialize", side_effect=axislib.Unauthorized
), pytest.raises(axis.errors.AuthenticationRequired):
await axis.device.get_axis_device(hass, config)
await axis.hub.get_axis_api(hass, config)
async def test_get_device_device_unavailable(hass: HomeAssistant, config) -> None:
@ -237,7 +237,7 @@ async def test_get_device_device_unavailable(hass: HomeAssistant, config) -> Non
with patch(
"axis.vapix.vapix.Vapix.request", side_effect=axislib.RequestError
), pytest.raises(axis.errors.CannotConnect):
await axis.device.get_axis_device(hass, config)
await axis.hub.get_axis_api(hass, config)
async def test_get_device_unknown_error(hass: HomeAssistant, config) -> None:
@ -245,4 +245,4 @@ async def test_get_device_unknown_error(hass: HomeAssistant, config) -> None:
with patch(
"axis.vapix.vapix.Vapix.request", side_effect=axislib.AxisException
), pytest.raises(axis.errors.AuthenticationRequired):
await axis.device.get_axis_device(hass, config)
await axis.hub.get_axis_api(hass, config)

2
tests/components/axis/test_init.py
View File

@ -18,7 +18,7 @@ async def test_setup_entry_fails(hass: HomeAssistant, config_entry) -> None:
mock_device = Mock()
mock_device.async_setup = AsyncMock(return_value=False)
with patch.object(axis, "AxisNetworkDevice") as mock_device_class:
with patch.object(axis, "AxisHub") as mock_device_class:
mock_device_class.return_value = mock_device
assert not await hass.config_entries.async_setup(config_entry.entry_id)