jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-24 05:37:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Split out Axis config data (#112825)

Browse Source 
Split out config data
This commit is contained in:
Robert Svensson 2024-03-09 19:00:25 +01:00 committed by GitHub
parent 03e4a20cdf
commit 020b656f51
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 94 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -94,13 +94,13 @@ class AxisBinarySensor(AxisEventEntity, BinarySensorEntity):
self.cancel_scheduled_update()
self.cancel_scheduled_update = None
if self.is_on or self.hub.option_trigger_time == 0:
if self.is_on or self.hub.config.trigger_time == 0:
self.async_write_ha_state()
return
self.cancel_scheduled_update = async_call_later(
self.hass,
timedelta(seconds=self.hub.option_trigger_time),
timedelta(seconds=self.hub.config.trigger_time),
scheduled_update,
)

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

@ -51,8 +51,8 @@ class AxisCamera(AxisEntity, MjpegCamera):
MjpegCamera.__init__(
self,
username=hub.username,
password=hub.password,
username=hub.config.username,
password=hub.config.password,
mjpeg_url=self.mjpeg_source,
still_image_url=self.image_source,
authentication=HTTP_DIGEST_AUTHENTICATION,
@ -76,27 +76,27 @@ class AxisCamera(AxisEntity, MjpegCamera):
"""
image_options = self.generate_options(skip_stream_profile=True)
self._still_image_url = (
f"http://{self.hub.host}:{self.hub.port}/axis-cgi"
f"http://{self.hub.config.host}:{self.hub.config.port}/axis-cgi"
f"/jpg/image.cgi{image_options}"
)
mjpeg_options = self.generate_options()
self._mjpeg_url = (
f"http://{self.hub.host}:{self.hub.port}/axis-cgi"
f"http://{self.hub.config.host}:{self.hub.config.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.hub.username}:{self.hub.password}"
f"@{self.hub.host}/axis-media/media.amp{stream_options}"
f"rtsp://{self.hub.config.username}:{self.hub.config.password}"
f"@{self.hub.config.host}/axis-media/media.amp{stream_options}"
)
self.hub.additional_diagnostics["camera_sources"] = {
"Image": self._still_image_url,
"MJPEG": self._mjpeg_url,
"Stream": (
f"rtsp://user:pass@{self.hub.host}/axis-media"
f"rtsp://user:pass@{self.hub.config.host}/axis-media"
f"/media.amp{stream_options}"
),
}
@ -126,12 +126,12 @@ class AxisCamera(AxisEntity, MjpegCamera):
if (
not skip_stream_profile
and self.hub.option_stream_profile != DEFAULT_STREAM_PROFILE
and self.hub.config.stream_profile != DEFAULT_STREAM_PROFILE
):
options_dict["streamprofile"] = self.hub.option_stream_profile
options_dict["streamprofile"] = self.hub.config.stream_profile
if self.hub.option_video_source != DEFAULT_VIDEO_SOURCE:
options_dict["camera"] = self.hub.option_video_source
if self.hub.config.video_source != DEFAULT_VIDEO_SOURCE:
options_dict["camera"] = self.hub.config.video_source
if not options_dict:
return ""

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

@ -272,7 +272,7 @@ class AxisOptionsFlowHandler(OptionsFlowWithConfigEntry):
schema[
vol.Optional(
CONF_STREAM_PROFILE, default=self.hub.option_stream_profile
CONF_STREAM_PROFILE, default=self.hub.config.stream_profile
)
] = vol.In(stream_profiles)
@ -291,7 +291,7 @@ class AxisOptionsFlowHandler(OptionsFlowWithConfigEntry):
video_sources[int(idx) + 1] = video_source.name
schema[
vol.Optional(CONF_VIDEO_SOURCE, default=self.hub.option_video_source)
vol.Optional(CONF_VIDEO_SOURCE, default=self.hub.config.video_source)
] = vol.In(video_sources)
return self.async_show_form(

66
homeassistant/components/axis/hub/config.py Normal file
View File

@ -0,0 +1,66 @@
"""Axis network device abstraction."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Self
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_MODEL,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_TRIGGER_TIME,
CONF_USERNAME,
)
from ..const import (
CONF_STREAM_PROFILE,
CONF_VIDEO_SOURCE,
DEFAULT_STREAM_PROFILE,
DEFAULT_TRIGGER_TIME,
DEFAULT_VIDEO_SOURCE,
)
@dataclass
class AxisConfig:
"""Represent a Axis config entry."""
entry: ConfigEntry
host: str
port: int
username: str
password: str
model: str
name: str
# Options
stream_profile: str
"""Option defining what stream profile camera platform should use."""
trigger_time: int
"""Option defining minimum number of seconds to keep trigger high."""
video_source: str
"""Option defining what video source camera platform should use."""
@classmethod
def from_config_entry(cls, config_entry: ConfigEntry) -> Self:
"""Create object from config entry."""
config = config_entry.data
options = config_entry.options
return cls(
entry=config_entry,
host=config[CONF_HOST],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
port=config[CONF_PORT],
model=config[CONF_MODEL],
name=config[CONF_NAME],
stream_profile=options.get(CONF_STREAM_PROFILE, DEFAULT_STREAM_PROFILE),
trigger_time=options.get(CONF_TRIGGER_TIME, DEFAULT_TRIGGER_TIME),
video_source=options.get(CONF_VIDEO_SOURCE, DEFAULT_VIDEO_SOURCE),
)

90
homeassistant/components/axis/hub/hub.py
View File

@ -14,30 +14,14 @@ from homeassistant.components import mqtt
from homeassistant.components.mqtt import DOMAIN as MQTT_DOMAIN
from homeassistant.components.mqtt.models import ReceiveMessage
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_MODEL,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_TRIGGER_TIME,
CONF_USERNAME,
)
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.setup import async_when_setup
from ..const import (
ATTR_MANUFACTURER,
CONF_STREAM_PROFILE,
CONF_VIDEO_SOURCE,
DEFAULT_STREAM_PROFILE,
DEFAULT_TRIGGER_TIME,
DEFAULT_VIDEO_SOURCE,
DOMAIN as AXIS_DOMAIN,
)
from ..const import ATTR_MANUFACTURER, DOMAIN as AXIS_DOMAIN
from .config import AxisConfig
class AxisHub:
@ -48,7 +32,7 @@ class AxisHub:
) -> None:
"""Initialize the device."""
self.hass = hass
self.config_entry = config_entry
self.config = AxisConfig.from_config_entry(config_entry)
self.api = api
self.available = True
@ -64,65 +48,10 @@ class AxisHub:
hub: AxisHub = hass.data[AXIS_DOMAIN][config_entry.entry_id]
return hub
@property
def host(self) -> str:
"""Return the host address of this device."""
host: str = self.config_entry.data[CONF_HOST]
return host
@property
def port(self) -> int:
"""Return the HTTP port of this device."""
port: int = self.config_entry.data[CONF_PORT]
return port
@property
def username(self) -> str:
"""Return the username of this device."""
username: str = self.config_entry.data[CONF_USERNAME]
return username
@property
def password(self) -> str:
"""Return the password of this device."""
password: str = self.config_entry.data[CONF_PASSWORD]
return password
@property
def model(self) -> str:
"""Return the model of this device."""
model: str = self.config_entry.data[CONF_MODEL]
return model
@property
def name(self) -> str:
"""Return the name of this device."""
name: str = self.config_entry.data[CONF_NAME]
return name
@property
def unique_id(self) -> str | None:
"""Return the unique ID (serial number) of this device."""
return self.config_entry.unique_id
# Options
@property
def option_stream_profile(self) -> str:
"""Config entry option defining what stream profile camera platform should use."""
return self.config_entry.options.get(
CONF_STREAM_PROFILE, DEFAULT_STREAM_PROFILE
)
@property
def option_trigger_time(self) -> int:
"""Config entry option defining minimum number of seconds to keep trigger high."""
return self.config_entry.options.get(CONF_TRIGGER_TIME, DEFAULT_TRIGGER_TIME)
@property
def option_video_source(self) -> str:
"""Config entry option defining what video source camera platform should use."""
return self.config_entry.options.get(CONF_VIDEO_SOURCE, DEFAULT_VIDEO_SOURCE)
return self.config.entry.unique_id
# Signals
@ -161,20 +90,21 @@ class AxisHub:
cannot be used with weak references.
"""
hub = AxisHub.get_hub(hass, config_entry)
hub.api.config.host = hub.host
hub.config = AxisConfig.from_config_entry(config_entry)
hub.api.config.host = hub.config.host
async_dispatcher_send(hass, hub.signal_new_address)
async def async_update_device_registry(self) -> None:
"""Update device registry."""
device_registry = dr.async_get(self.hass)
device_registry.async_get_or_create(
config_entry_id=self.config_entry.entry_id,
config_entry_id=self.config.entry.entry_id,
configuration_url=self.api.config.url,
connections={(CONNECTION_NETWORK_MAC, self.unique_id)}, # type: ignore[arg-type]
identifiers={(AXIS_DOMAIN, self.unique_id)}, # type: ignore[arg-type]
manufacturer=ATTR_MANUFACTURER,
model=f"{self.model} {self.product_type}",
name=self.name,
model=f"{self.config.model} {self.product_type}",
name=self.config.name,
sw_version=self.fw_version,
)
@ -186,7 +116,7 @@ class AxisHub:
# This means the user has too low privileges
return
if status.status.state == ClientState.ACTIVE:
self.config_entry.async_on_unload(
self.config.entry.async_on_unload(
await mqtt.async_subscribe(
hass, f"{self.api.vapix.serial_number}/#", self.mqtt_message
)

6
tests/components/axis/test_hub.py
View File

@ -63,9 +63,9 @@ async def test_device_setup(
assert forward_entry_setup.mock_calls[2][1][1] == "light"
assert forward_entry_setup.mock_calls[3][1][1] == "switch"
assert hub.host == config[CONF_HOST]
assert hub.model == config[CONF_MODEL]
assert hub.name == config[CONF_NAME]
assert hub.config.host == config[CONF_HOST]
assert hub.config.model == config[CONF_MODEL]
assert hub.config.name == config[CONF_NAME]
assert hub.unique_id == FORMATTED_MAC
device_entry = device_registry.async_get_device(