Add config flow to android_ip_webcam (#76222)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Rami Mosleh 2022-08-09 15:08:46 +03:00 committed by GitHub
parent 7d427ddbd4
commit f90d007e73
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1076 additions and 478 deletions

View File

@ -56,7 +56,9 @@ omit =
homeassistant/components/ambient_station/sensor.py
homeassistant/components/amcrest/*
homeassistant/components/ampio/*
homeassistant/components/android_ip_webcam/*
homeassistant/components/android_ip_webcam/binary_sensor.py
homeassistant/components/android_ip_webcam/sensor.py
homeassistant/components/android_ip_webcam/switch.py
homeassistant/components/androidtv/diagnostics.py
homeassistant/components/anel_pwrctrl/switch.py
homeassistant/components/anthemav/media_player.py

View File

@ -72,6 +72,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/amcrest/ @flacjacket
/homeassistant/components/analytics/ @home-assistant/core @ludeeus
/tests/components/analytics/ @home-assistant/core @ludeeus
/homeassistant/components/android_ip_webcam/ @engrbm87
/tests/components/android_ip_webcam/ @engrbm87
/homeassistant/components/androidtv/ @JeffLIrion @ollo69
/tests/components/androidtv/ @JeffLIrion @ollo69
/homeassistant/components/anthemav/ @hyralex

View File

@ -1,13 +1,12 @@
"""Support for Android IP Webcam."""
"""The Android IP Webcam integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from pydroid_ipcam import PyDroidIPCam
import voluptuous as vol
from homeassistant.components.mjpeg import CONF_MJPEG_URL, CONF_STILL_IMAGE_URL
from homeassistant.components.repairs.issue_handler import async_create_issue
from homeassistant.components.repairs.models import IssueSeverity
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
@ -20,168 +19,64 @@ from homeassistant.const import (
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import discovery
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.dt import utcnow
ATTR_AUD_CONNS = "Audio Connections"
ATTR_HOST = "host"
ATTR_VID_CONNS = "Video Connections"
from .const import (
CONF_MOTION_SENSOR,
DEFAULT_NAME,
DEFAULT_PORT,
DEFAULT_TIMEOUT,
DOMAIN,
SCAN_INTERVAL,
SENSORS,
SWITCHES,
)
from .coordinator import AndroidIPCamDataUpdateCoordinator
CONF_MOTION_SENSOR = "motion_sensor"
DATA_IP_WEBCAM = "android_ip_webcam"
DEFAULT_NAME = "IP Webcam"
DEFAULT_PORT = 8080
DEFAULT_TIMEOUT = 10
DOMAIN = "android_ip_webcam"
SCAN_INTERVAL = timedelta(seconds=10)
SIGNAL_UPDATE_DATA = "android_ip_webcam_update"
KEY_MAP = {
"audio_connections": "Audio Connections",
"adet_limit": "Audio Trigger Limit",
"antibanding": "Anti-banding",
"audio_only": "Audio Only",
"battery_level": "Battery Level",
"battery_temp": "Battery Temperature",
"battery_voltage": "Battery Voltage",
"coloreffect": "Color Effect",
"exposure": "Exposure Level",
"exposure_lock": "Exposure Lock",
"ffc": "Front-facing Camera",
"flashmode": "Flash Mode",
"focus": "Focus",
"focus_homing": "Focus Homing",
"focus_region": "Focus Region",
"focusmode": "Focus Mode",
"gps_active": "GPS Active",
"idle": "Idle",
"ip_address": "IPv4 Address",
"ipv6_address": "IPv6 Address",
"ivideon_streaming": "Ivideon Streaming",
"light": "Light Level",
"mirror_flip": "Mirror Flip",
"motion": "Motion",
"motion_active": "Motion Active",
"motion_detect": "Motion Detection",
"motion_event": "Motion Event",
"motion_limit": "Motion Limit",
"night_vision": "Night Vision",
"night_vision_average": "Night Vision Average",
"night_vision_gain": "Night Vision Gain",
"orientation": "Orientation",
"overlay": "Overlay",
"photo_size": "Photo Size",
"pressure": "Pressure",
"proximity": "Proximity",
"quality": "Quality",
"scenemode": "Scene Mode",
"sound": "Sound",
"sound_event": "Sound Event",
"sound_timeout": "Sound Timeout",
"torch": "Torch",
"video_connections": "Video Connections",
"video_chunk_len": "Video Chunk Length",
"video_recording": "Video Recording",
"video_size": "Video Size",
"whitebalance": "White Balance",
"whitebalance_lock": "White Balance Lock",
"zoom": "Zoom",
}
ICON_MAP = {
"audio_connections": "mdi:speaker",
"battery_level": "mdi:battery",
"battery_temp": "mdi:thermometer",
"battery_voltage": "mdi:battery-charging-100",
"exposure_lock": "mdi:camera",
"ffc": "mdi:camera-front-variant",
"focus": "mdi:image-filter-center-focus",
"gps_active": "mdi:crosshairs-gps",
"light": "mdi:flashlight",
"motion": "mdi:run",
"night_vision": "mdi:weather-night",
"overlay": "mdi:monitor",
"pressure": "mdi:gauge",
"proximity": "mdi:map-marker-radius",
"quality": "mdi:quality-high",
"sound": "mdi:speaker",
"sound_event": "mdi:speaker",
"sound_timeout": "mdi:speaker",
"torch": "mdi:white-balance-sunny",
"video_chunk_len": "mdi:video",
"video_connections": "mdi:eye",
"video_recording": "mdi:record-rec",
"whitebalance_lock": "mdi:white-balance-auto",
}
SWITCHES = [
"exposure_lock",
"ffc",
"focus",
"gps_active",
"motion_detect",
"night_vision",
"overlay",
"torch",
"whitebalance_lock",
"video_recording",
PLATFORMS: list[Platform] = [
Platform.BINARY_SENSOR,
Platform.CAMERA,
Platform.SENSOR,
Platform.SWITCH,
]
SENSORS = [
"audio_connections",
"battery_level",
"battery_temp",
"battery_voltage",
"light",
"motion",
"pressure",
"proximity",
"sound",
"video_connections",
]
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(
CONF_TIMEOUT, default=DEFAULT_TIMEOUT
): cv.positive_int,
vol.Optional(
CONF_SCAN_INTERVAL, default=SCAN_INTERVAL
): cv.time_period,
vol.Inclusive(CONF_USERNAME, "authentication"): cv.string,
vol.Inclusive(CONF_PASSWORD, "authentication"): cv.string,
vol.Optional(CONF_SWITCHES): vol.All(
cv.ensure_list, [vol.In(SWITCHES)]
),
vol.Optional(CONF_SENSORS): vol.All(
cv.ensure_list, [vol.In(SENSORS)]
),
vol.Optional(CONF_MOTION_SENSOR): cv.boolean,
}
)
],
)
},
vol.All(
cv.deprecated(DOMAIN),
{
DOMAIN: vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(
CONF_TIMEOUT, default=DEFAULT_TIMEOUT
): cv.positive_int,
vol.Optional(
CONF_SCAN_INTERVAL, default=SCAN_INTERVAL
): cv.time_period,
vol.Inclusive(CONF_USERNAME, "authentication"): cv.string,
vol.Inclusive(CONF_PASSWORD, "authentication"): cv.string,
vol.Optional(CONF_SWITCHES): vol.All(
cv.ensure_list, [vol.In(SWITCHES)]
),
vol.Optional(CONF_SENSORS): vol.All(
cv.ensure_list, [vol.In(SENSORS)]
),
vol.Optional(CONF_MOTION_SENSOR): cv.boolean,
}
)
],
)
},
),
extra=vol.ALLOW_EXTRA,
)
@ -189,165 +84,52 @@ CONFIG_SCHEMA = vol.Schema(
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the IP Webcam component."""
webcams = hass.data[DATA_IP_WEBCAM] = {}
websession = async_get_clientsession(hass)
async def async_setup_ipcamera(cam_config):
"""Set up an IP camera."""
host = cam_config[CONF_HOST]
username: str | None = cam_config.get(CONF_USERNAME)
password: str | None = cam_config.get(CONF_PASSWORD)
name: str = cam_config[CONF_NAME]
interval = cam_config[CONF_SCAN_INTERVAL]
switches = cam_config.get(CONF_SWITCHES)
sensors = cam_config.get(CONF_SENSORS)
motion = cam_config.get(CONF_MOTION_SENSOR)
# Init ip webcam
cam = PyDroidIPCam(
websession,
host,
cam_config[CONF_PORT],
username=username,
password=password,
timeout=cam_config[CONF_TIMEOUT],
ssl=False,
)
if switches is None:
switches = [
setting for setting in cam.enabled_settings if setting in SWITCHES
]
if sensors is None:
sensors = [sensor for sensor in cam.enabled_sensors if sensor in SENSORS]
sensors.extend(["audio_connections", "video_connections"])
if motion is None:
motion = "motion_active" in cam.enabled_sensors
async def async_update_data(now):
"""Update data from IP camera in SCAN_INTERVAL."""
await cam.update()
async_dispatcher_send(hass, SIGNAL_UPDATE_DATA, host)
async_track_point_in_utc_time(hass, async_update_data, utcnow() + interval)
await async_update_data(None)
# Load platforms
webcams[host] = cam
mjpeg_camera = {
CONF_MJPEG_URL: cam.mjpeg_url,
CONF_STILL_IMAGE_URL: cam.image_url,
}
if username and password:
mjpeg_camera.update({CONF_USERNAME: username, CONF_PASSWORD: password})
# Remove incorrect config entry setup via mjpeg platform discovery.
mjpeg_config_entry = next(
(
config_entry
for config_entry in hass.config_entries.async_entries("mjpeg")
if all(
config_entry.options.get(key) == val
for key, val in mjpeg_camera.items()
)
),
None,
)
if mjpeg_config_entry:
await hass.config_entries.async_remove(mjpeg_config_entry.entry_id)
mjpeg_camera[CONF_NAME] = name
if DOMAIN not in config:
return True
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2022.11.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
)
for entry in config[DOMAIN]:
hass.async_create_task(
discovery.async_load_platform(
hass, Platform.CAMERA, DOMAIN, mjpeg_camera, config
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=entry
)
)
if sensors:
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.SENSOR,
DOMAIN,
{CONF_NAME: name, CONF_HOST: host, CONF_SENSORS: sensors},
config,
)
)
if switches:
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.SWITCH,
DOMAIN,
{CONF_NAME: name, CONF_HOST: host, CONF_SWITCHES: switches},
config,
)
)
if motion:
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.BINARY_SENSOR,
DOMAIN,
{CONF_HOST: host, CONF_NAME: name},
config,
)
)
tasks = [async_setup_ipcamera(conf) for conf in config[DOMAIN]]
if tasks:
await asyncio.wait(tasks)
return True
class AndroidIPCamEntity(Entity):
"""The Android device running IP Webcam."""
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Android IP Webcam from a config entry."""
websession = async_get_clientsession(hass)
cam = PyDroidIPCam(
websession,
entry.data[CONF_HOST],
entry.data[CONF_PORT],
username=entry.data.get(CONF_USERNAME),
password=entry.data.get(CONF_PASSWORD),
ssl=False,
)
coordinator = AndroidIPCamDataUpdateCoordinator(hass, entry, cam)
await coordinator.async_config_entry_first_refresh()
def __init__(self, host, ipcam):
"""Initialize the data object."""
self._host = host
self._ipcam = ipcam
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
async def async_added_to_hass(self):
"""Register update dispatcher."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@callback
def async_ipcam_update(host):
"""Update callback."""
if self._host != host:
return
self.async_schedule_update_ha_state(True)
return True
self.async_on_remove(
async_dispatcher_connect(self.hass, SIGNAL_UPDATE_DATA, async_ipcam_update)
)
@property
def should_poll(self):
"""Return True if entity has to be polled for state."""
return False
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
@property
def available(self):
"""Return True if entity is available."""
return self._ipcam.available
@property
def extra_state_attributes(self):
"""Return the state attributes."""
state_attr = {ATTR_HOST: self._host}
if self._ipcam.status_data is None:
return state_attr
state_attr[ATTR_VID_CONNS] = self._ipcam.status_data.get("video_connections")
state_attr[ATTR_AUD_CONNS] = self._ipcam.status_data.get("audio_connections")
return state_attr
return unload_ok

View File

@ -4,46 +4,57 @@ from __future__ import annotations
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import CONF_HOST, CONF_NAME, DATA_IP_WEBCAM, KEY_MAP, AndroidIPCamEntity
from .const import DOMAIN, MOTION_ACTIVE
from .coordinator import AndroidIPCamDataUpdateCoordinator
from .entity import AndroidIPCamBaseEntity
BINARY_SENSOR_DESCRIPTION = BinarySensorEntityDescription(
key="motion_active",
name="Motion active",
device_class=BinarySensorDeviceClass.MOTION,
)
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the IP Webcam binary sensors."""
if discovery_info is None:
return
"""Set up the IP Webcam sensors from config entry."""
host = discovery_info[CONF_HOST]
name = discovery_info[CONF_NAME]
ipcam = hass.data[DATA_IP_WEBCAM][host]
coordinator: AndroidIPCamDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
async_add_entities([IPWebcamBinarySensor(name, host, ipcam, "motion_active")], True)
async_add_entities([IPWebcamBinarySensor(coordinator)])
class IPWebcamBinarySensor(AndroidIPCamEntity, BinarySensorEntity):
class IPWebcamBinarySensor(AndroidIPCamBaseEntity, BinarySensorEntity):
"""Representation of an IP Webcam binary sensor."""
_attr_device_class = BinarySensorDeviceClass.MOTION
def __init__(self, name, host, ipcam, sensor):
def __init__(
self,
coordinator: AndroidIPCamDataUpdateCoordinator,
) -> None:
"""Initialize the binary sensor."""
super().__init__(host, ipcam)
self.entity_description = BINARY_SENSOR_DESCRIPTION
self._attr_unique_id = (
f"{coordinator.config_entry.entry_id}-{BINARY_SENSOR_DESCRIPTION.key}"
)
super().__init__(coordinator)
self._sensor = sensor
self._mapped_name = KEY_MAP.get(self._sensor, self._sensor)
self._attr_name = f"{name} {self._mapped_name}"
self._attr_is_on = None
@property
def available(self) -> bool:
"""Return avaibility if setting is enabled."""
return MOTION_ACTIVE in self.cam.enabled_sensors and super().available
async def async_update(self):
"""Retrieve latest state."""
state, _ = self._ipcam.export_sensor(self._sensor)
self._attr_is_on = state == 1.0
@property
def is_on(self) -> bool:
"""Return if motion is detected."""
return self.cam.export_sensor(MOTION_ACTIVE)[0] == 1.0

View File

@ -2,43 +2,59 @@
from __future__ import annotations
from homeassistant.components.mjpeg import MjpegCamera, filter_urllib3_logging
from homeassistant.const import HTTP_BASIC_AUTHENTICATION
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
HTTP_BASIC_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import DOMAIN
from .coordinator import AndroidIPCamDataUpdateCoordinator
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the IP Webcam camera."""
if discovery_info is None:
return
"""Set up the IP Webcam camera from config entry."""
filter_urllib3_logging()
async_add_entities([IPWebcamCamera(**discovery_info)])
coordinator: AndroidIPCamDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
async_add_entities([IPWebcamCamera(coordinator)])
class IPWebcamCamera(MjpegCamera):
"""Representation of a IP Webcam camera."""
def __init__(
self,
name: str,
mjpeg_url: str,
still_image_url: str,
username: str | None = None,
password: str = "",
) -> None:
_attr_has_entity_name = True
def __init__(self, coordinator: AndroidIPCamDataUpdateCoordinator) -> None:
"""Initialize the camera."""
name = None
# keep imported name until YAML is removed
if CONF_NAME in coordinator.config_entry.data:
name = coordinator.config_entry.data[CONF_NAME]
self._attr_has_entity_name = False
super().__init__(
name=name,
mjpeg_url=mjpeg_url,
still_image_url=still_image_url,
mjpeg_url=coordinator.cam.mjpeg_url,
still_image_url=coordinator.cam.image_url,
authentication=HTTP_BASIC_AUTHENTICATION,
username=username,
password=password,
username=coordinator.config_entry.data.get(CONF_USERNAME),
password=coordinator.config_entry.data.get(CONF_PASSWORD, ""),
)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}-camera"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
name=name or coordinator.config_entry.data[CONF_HOST],
)

View File

@ -0,0 +1,84 @@
"""Config flow for Android IP Webcam integration."""
from __future__ import annotations
from typing import Any
from pydroid_ipcam import PyDroidIPCam
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_TIMEOUT,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DEFAULT_PORT, DOMAIN
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Inclusive(CONF_USERNAME, "authentication"): str,
vol.Inclusive(CONF_PASSWORD, "authentication"): str,
}
)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> bool:
"""Validate the user input allows us to connect."""
websession = async_get_clientsession(hass)
cam = PyDroidIPCam(
websession,
data[CONF_HOST],
data[CONF_PORT],
username=data.get(CONF_USERNAME),
password=data.get(CONF_PASSWORD),
ssl=False,
)
await cam.update()
return cam.available
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Android IP Webcam."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
self._async_abort_entries_match(
{CONF_HOST: user_input[CONF_HOST], CONF_PORT: user_input[CONF_PORT]}
)
# to be removed when YAML import is removed
title = user_input.get(CONF_NAME) or user_input[CONF_HOST]
if await validate_input(self.hass, user_input):
return self.async_create_entry(title=title, data=user_input)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors={"base": "cannot_connect"},
)
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Import a config entry from configuration.yaml."""
import_config.pop(CONF_SCAN_INTERVAL)
import_config.pop(CONF_TIMEOUT)
return await self.async_step_user(import_config)

View File

@ -0,0 +1,41 @@
"""Constants for the Android IP Webcam integration."""
from datetime import timedelta
from typing import Final
DOMAIN: Final = "android_ip_webcam"
DEFAULT_NAME: Final = "IP Webcam"
DEFAULT_PORT: Final = 8080
DEFAULT_TIMEOUT: Final = 10
CONF_MOTION_SENSOR: Final = "motion_sensor"
MOTION_ACTIVE: Final = "motion_active"
SCAN_INTERVAL: Final = timedelta(seconds=10)
SWITCHES = [
"exposure_lock",
"ffc",
"focus",
"gps_active",
"motion_detect",
"night_vision",
"overlay",
"torch",
"whitebalance_lock",
"video_recording",
]
SENSORS = [
"audio_connections",
"battery_level",
"battery_temp",
"battery_voltage",
"light",
"motion",
"pressure",
"proximity",
"sound",
"video_connections",
]

View File

@ -0,0 +1,42 @@
"""Coordinator object for the Android IP Webcam integration."""
from datetime import timedelta
import logging
from pydroid_ipcam import PyDroidIPCam
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class AndroidIPCamDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""Coordinator class for the Android IP Webcam."""
def __init__(
self,
hass: HomeAssistant,
config_entry: ConfigEntry,
cam: PyDroidIPCam,
) -> None:
"""Initialize the Android IP Webcam."""
self.hass = hass
self.config_entry: ConfigEntry = config_entry
self.cam = cam
super().__init__(
self.hass,
_LOGGER,
name=f"{DOMAIN} {config_entry.data[CONF_HOST]}",
update_interval=timedelta(seconds=10),
)
async def _async_update_data(self) -> None:
"""Update Android IP Webcam entities."""
await self.cam.update()
if not self.cam.available:
raise UpdateFailed

View File

@ -0,0 +1,32 @@
"""Base class for Android IP Webcam entities."""
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import AndroidIPCamDataUpdateCoordinator
class AndroidIPCamBaseEntity(CoordinatorEntity[AndroidIPCamDataUpdateCoordinator]):
"""Base class for Android IP Webcam entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: AndroidIPCamDataUpdateCoordinator,
) -> None:
"""Initialize the base entity."""
super().__init__(coordinator)
if CONF_NAME in coordinator.config_entry.data:
# name is legacy imported from YAML config
# this block can be removed when removing import from YAML
self._attr_name = f"{coordinator.config_entry.data[CONF_NAME]} {self.entity_description.name}"
self._attr_has_entity_name = False
self.cam = coordinator.cam
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
name=coordinator.config_entry.data.get(CONF_NAME)
or coordinator.config_entry.data[CONF_HOST],
)

View File

@ -1,8 +1,10 @@
{
"domain": "android_ip_webcam",
"name": "Android IP Webcam",
"config_flow": true,
"dependencies": ["repairs"],
"documentation": "https://www.home-assistant.io/integrations/android_ip_webcam",
"requirements": ["pydroid-ipcam==1.3.1"],
"codeowners": [],
"codeowners": ["@engrbm87"],
"iot_class": "local_polling"
}

View File

@ -1,75 +1,172 @@
"""Support for Android IP Webcam sensors."""
from __future__ import annotations
from homeassistant.components.sensor import SensorEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.icon import icon_for_battery_level
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from collections.abc import Callable
from dataclasses import dataclass
from . import (
CONF_HOST,
CONF_NAME,
CONF_SENSORS,
DATA_IP_WEBCAM,
ICON_MAP,
KEY_MAP,
AndroidIPCamEntity,
from pydroid_ipcam import PyDroidIPCam
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import DOMAIN
from .coordinator import AndroidIPCamDataUpdateCoordinator
from .entity import AndroidIPCamBaseEntity
@dataclass
class AndroidIPWebcamSensorEntityDescriptionMixin:
"""Mixin for required keys."""
value_fn: Callable[[PyDroidIPCam], StateType]
@dataclass
class AndroidIPWebcamSensorEntityDescription(
SensorEntityDescription, AndroidIPWebcamSensorEntityDescriptionMixin
):
"""Entity description class for Android IP Webcam sensors."""
unit_fn: Callable[[PyDroidIPCam], str | None] = lambda _: None
SENSOR_TYPES: tuple[AndroidIPWebcamSensorEntityDescription, ...] = (
AndroidIPWebcamSensorEntityDescription(
key="audio_connections",
name="Audio connections",
icon="mdi:speaker",
state_class=SensorStateClass.TOTAL,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda ipcam: ipcam.status_data.get("audio_connections"),
),
AndroidIPWebcamSensorEntityDescription(
key="battery_level",
name="Battery level",
device_class=SensorDeviceClass.BATTERY,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda ipcam: ipcam.export_sensor("battery_level")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("battery_level")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="battery_temp",
name="Battery temperature",
icon="mdi:thermometer",
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda ipcam: ipcam.export_sensor("battery_temp")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("battery_temp")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="battery_voltage",
name="Battery voltage",
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda ipcam: ipcam.export_sensor("battery_voltage")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("battery_voltage")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="light",
name="Light level",
icon="mdi:flashlight",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda ipcam: ipcam.export_sensor("light")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("light")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="motion",
name="Motion",
icon="mdi:run",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda ipcam: ipcam.export_sensor("motion")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("motion")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="pressure",
name="Pressure",
icon="mdi:gauge",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda ipcam: ipcam.export_sensor("pressure")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("pressure")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="proximity",
name="Proximity",
icon="mdi:map-marker-radius",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda ipcam: ipcam.export_sensor("proximity")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("proximity")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="sound",
name="Sound",
icon="mdi:speaker",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda ipcam: ipcam.export_sensor("sound")[0],
unit_fn=lambda ipcam: ipcam.export_sensor("sound")[1],
),
AndroidIPWebcamSensorEntityDescription(
key="video_connections",
name="Video connections",
icon="mdi:eye",
state_class=SensorStateClass.TOTAL,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda ipcam: ipcam.status_data.get("video_connections"),
),
)
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the IP Webcam Sensor."""
if discovery_info is None:
return
"""Set up the IP Webcam sensors from config entry."""
host = discovery_info[CONF_HOST]
name = discovery_info[CONF_NAME]
sensors = discovery_info[CONF_SENSORS]
ipcam = hass.data[DATA_IP_WEBCAM][host]
all_sensors = []
for sensor in sensors:
all_sensors.append(IPWebcamSensor(name, host, ipcam, sensor))
async_add_entities(all_sensors, True)
coordinator: AndroidIPCamDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
sensor_types = [
sensor
for sensor in SENSOR_TYPES
if sensor.key
in coordinator.cam.enabled_sensors + ["audio_connections", "video_connections"]
]
async_add_entities(
IPWebcamSensor(coordinator, description) for description in sensor_types
)
class IPWebcamSensor(AndroidIPCamEntity, SensorEntity):
class IPWebcamSensor(AndroidIPCamBaseEntity, SensorEntity):
"""Representation of a IP Webcam sensor."""
def __init__(self, name, host, ipcam, sensor):
entity_description: AndroidIPWebcamSensorEntityDescription
def __init__(
self,
coordinator: AndroidIPCamDataUpdateCoordinator,
description: AndroidIPWebcamSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(host, ipcam)
self._sensor = sensor
self._mapped_name = KEY_MAP.get(self._sensor, self._sensor)
self._attr_name = f"{name} {self._mapped_name}"
self._attr_native_value = None
self._attr_native_unit_of_measurement = None
async def async_update(self):
"""Retrieve latest state."""
if self._sensor in ("audio_connections", "video_connections"):
if not self._ipcam.status_data:
return
self._attr_native_value = self._ipcam.status_data.get(self._sensor)
self._attr_native_unit_of_measurement = "Connections"
else:
(
self._attr_native_value,
self._attr_native_unit_of_measurement,
) = self._ipcam.export_sensor(self._sensor)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}-{description.key}"
self.entity_description = description
super().__init__(coordinator)
@property
def icon(self):
"""Return the icon for the sensor."""
if self._sensor == "battery_level" and self._attr_native_value is not None:
return icon_for_battery_level(int(self._attr_native_value))
return ICON_MAP.get(self._sensor, "mdi:eye")
def native_value(self) -> StateType:
"""Return native value of sensor."""
return self.entity_description.value_fn(self.cam)
@property
def native_unit_of_measurement(self) -> str | None:
"""Return native unit of measurement of sensor."""
return self.entity_description.unit_fn(self.cam)

View File

@ -0,0 +1,26 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"issues": {
"deprecated_yaml": {
"title": "The Android IP Webcam YAML configuration is being removed",
"description": "Configuring Android IP Webcam using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the Android IP Webcam YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
}
}
}

View File

@ -1,88 +1,170 @@
"""Support for Android IP Webcam settings."""
from __future__ import annotations
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from collections.abc import Callable
from dataclasses import dataclass
from . import (
CONF_HOST,
CONF_NAME,
CONF_SWITCHES,
DATA_IP_WEBCAM,
ICON_MAP,
KEY_MAP,
AndroidIPCamEntity,
from pydroid_ipcam import PyDroidIPCam
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .coordinator import AndroidIPCamDataUpdateCoordinator
from .entity import AndroidIPCamBaseEntity
@dataclass
class AndroidIPWebcamSwitchEntityDescriptionMixin:
"""Mixin for required keys."""
on_func: Callable[[PyDroidIPCam], None]
off_func: Callable[[PyDroidIPCam], None]
@dataclass
class AndroidIPWebcamSwitchEntityDescription(
SwitchEntityDescription, AndroidIPWebcamSwitchEntityDescriptionMixin
):
"""Entity description class for Android IP Webcam switches."""
SWITCH_TYPES: tuple[AndroidIPWebcamSwitchEntityDescription, ...] = (
AndroidIPWebcamSwitchEntityDescription(
key="exposure_lock",
name="Exposure lock",
icon="mdi:camera",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("exposure_lock", True),
off_func=lambda ipcam: ipcam.change_setting("exposure_lock", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="ffc",
name="Front-facing camera",
icon="mdi:camera-front-variant",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("ffc", True),
off_func=lambda ipcam: ipcam.change_setting("ffc", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="focus",
name="Focus",
icon="mdi:image-filter-center-focus",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.torch(activate=True),
off_func=lambda ipcam: ipcam.torch(activate=False),
),
AndroidIPWebcamSwitchEntityDescription(
key="gps_active",
name="GPS active",
icon="mdi:crosshairs-gps",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("gps_active", True),
off_func=lambda ipcam: ipcam.change_setting("gps_active", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="motion_detect",
name="Motion detection",
icon="mdi:flash",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("motion_detect", True),
off_func=lambda ipcam: ipcam.change_setting("motion_detect", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="night_vision",
name="Night vision",
icon="mdi:weather-night",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("night_vision", True),
off_func=lambda ipcam: ipcam.change_setting("night_vision", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="overlay",
name="Overlay",
icon="mdi:monitor",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("overlay", True),
off_func=lambda ipcam: ipcam.change_setting("overlay", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="torch",
name="Torch",
icon="mdi:white-balance-sunny",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.torch(activate=True),
off_func=lambda ipcam: ipcam.torch(activate=False),
),
AndroidIPWebcamSwitchEntityDescription(
key="whitebalance_lock",
name="White balance lock",
icon="mdi:white-balance-auto",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.change_setting("whitebalance_lock", True),
off_func=lambda ipcam: ipcam.change_setting("whitebalance_lock", False),
),
AndroidIPWebcamSwitchEntityDescription(
key="video_recording",
name="Video recording",
icon="mdi:record-rec",
entity_category=EntityCategory.CONFIG,
on_func=lambda ipcam: ipcam.record(activate=True),
off_func=lambda ipcam: ipcam.record(activate=False),
),
)
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the IP Webcam switch platform."""
if discovery_info is None:
return
"""Set up the IP Webcam switches from config entry."""
host = discovery_info[CONF_HOST]
name = discovery_info[CONF_NAME]
switches = discovery_info[CONF_SWITCHES]
ipcam = hass.data[DATA_IP_WEBCAM][host]
all_switches = []
for setting in switches:
all_switches.append(IPWebcamSettingsSwitch(name, host, ipcam, setting))
async_add_entities(all_switches, True)
coordinator: AndroidIPCamDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
switch_types = [
switch
for switch in SWITCH_TYPES
if switch.key in coordinator.cam.enabled_settings
]
async_add_entities(
[
IPWebcamSettingSwitch(coordinator, description)
for description in switch_types
]
)
class IPWebcamSettingsSwitch(AndroidIPCamEntity, SwitchEntity):
"""An abstract class for an IP Webcam setting."""
class IPWebcamSettingSwitch(AndroidIPCamBaseEntity, SwitchEntity):
"""Representation of a IP Webcam setting."""
def __init__(self, name, host, ipcam, setting):
"""Initialize the settings switch."""
super().__init__(host, ipcam)
entity_description: AndroidIPWebcamSwitchEntityDescription
self._setting = setting
self._mapped_name = KEY_MAP.get(self._setting, self._setting)
self._attr_name = f"{name} {self._mapped_name}"
self._attr_is_on = False
def __init__(
self,
coordinator: AndroidIPCamDataUpdateCoordinator,
description: AndroidIPWebcamSwitchEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(coordinator)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}-{description.key}"
async def async_update(self):
"""Get the updated status of the switch."""
self._attr_is_on = bool(self._ipcam.current_settings.get(self._setting))
@property
def is_on(self) -> bool:
"""Return if settings is on or off."""
return bool(self.cam.current_settings.get(self.entity_description.key))
async def async_turn_on(self, **kwargs):
"""Turn device on."""
if self._setting == "torch":
await self._ipcam.torch(activate=True)
elif self._setting == "focus":
await self._ipcam.focus(activate=True)
elif self._setting == "video_recording":
await self._ipcam.record(record=True)
else:
await self._ipcam.change_setting(self._setting, True)
self._attr_is_on = True
self.async_write_ha_state()
await self.entity_description.on_func(self.cam)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs):
"""Turn device off."""
if self._setting == "torch":
await self._ipcam.torch(activate=False)
elif self._setting == "focus":
await self._ipcam.focus(activate=False)
elif self._setting == "video_recording":
await self._ipcam.record(record=False)
else:
await self._ipcam.change_setting(self._setting, False)
self._attr_is_on = False
self.async_write_ha_state()
@property
def icon(self):
"""Return the icon for the switch."""
return ICON_MAP.get(self._setting, "mdi:flash")
await self.entity_description.off_func(self.cam)
await self.coordinator.async_request_refresh()

View File

@ -0,0 +1,26 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect"
},
"step": {
"user": {
"data": {
"host": "Host",
"password": "Password",
"port": "Port",
"username": "Username"
}
}
}
},
"issues": {
"deprecated_yaml": {
"description": "Configuring Android IP Webcam using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the Android IP Webcam YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue.",
"title": "The Android IP Webcamepush YAML configuration is being removed"
}
}
}

View File

@ -28,6 +28,7 @@ FLOWS = {
"amberelectric",
"ambiclimate",
"ambient_station",
"android_ip_webcam",
"androidtv",
"anthemav",
"apple_tv",

View File

@ -1018,6 +1018,9 @@ pydeconz==103
# homeassistant.components.dexcom
pydexcom==0.2.3
# homeassistant.components.android_ip_webcam
pydroid-ipcam==1.3.1
# homeassistant.components.econet
pyeconet==0.1.15

View File

@ -0,0 +1 @@
"""Tests for the Android IP Webcam integration."""

View File

@ -0,0 +1,25 @@
"""Fixtures for tests."""
from http import HTTPStatus
import pytest
from homeassistant.const import CONTENT_TYPE_JSON
from tests.common import load_fixture
@pytest.fixture
def aioclient_mock_fixture(aioclient_mock) -> None:
"""Fixture to provide a aioclient mocker."""
aioclient_mock.get(
"http://1.1.1.1:8080/status.json?show_avail=1",
text=load_fixture("android_ip_webcam/status_data.json"),
status=HTTPStatus.OK,
headers={"Content-Type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
"http://1.1.1.1:8080/sensors.json",
text=load_fixture("android_ip_webcam/sensor_data.json"),
status=HTTPStatus.OK,
headers={"Content-Type": CONTENT_TYPE_JSON},
)

View File

@ -0,0 +1,57 @@
{
"light": {
"unit": "lx",
"data": [
[1659602753873, [708.0]],
[1659602773708, [991.0]]
]
},
"proximity": { "unit": "cm", "data": [[1659602773709, [5.0]]] },
"motion": {
"unit": "",
"data": [
[1659602752922, [8.0]],
[1659602773709, [12.0]]
]
},
"motion_active": { "unit": "", "data": [[1659602773710, [0.0]]] },
"battery_voltage": {
"unit": "V",
"data": [
[1659602769351, [4.077]],
[1659602771352, [4.077]],
[1659602773353, [4.077]]
]
},
"battery_level": {
"unit": "%",
"data": [
[1659602769351, [87.0]],
[1659602771352, [87.0]],
[1659602773353, [87.0]]
]
},
"battery_temp": {
"unit": "℃",
"data": [
[1659602769351, [33.9]],
[1659602771352, [33.9]],
[1659602773353, [33.9]]
]
},
"sound": {
"unit": "dB",
"data": [
[1659602768710, [181.0]],
[1659602769200, [191.0]],
[1659602769690, [186.0]],
[1659602770181, [186.0]],
[1659602770670, [188.0]],
[1659602771145, [192.0]],
[1659602771635, [192.0]],
[1659602772125, [179.0]],
[1659602772615, [186.0]],
[1659602773261, [178.0]]
]
}
}

View File

@ -0,0 +1,62 @@
{
"video_connections": 0,
"audio_connections": 0,
"video_status": { "result": "status", "enabled": "True", "mode": "none" },
"curvals": {
"orientation": "landscape",
"idle": "off",
"audio_only": "off",
"overlay": "off",
"quality": "49",
"focus_homing": "off",
"ip_address": "192.168.3.88",
"motion_limit": "250",
"adet_limit": "200",
"night_vision": "off",
"night_vision_average": "2",
"night_vision_gain": "1.0",
" ": "off",
"motion_detect": "on",
"motion_display": "off",
"video_chunk_len": "60",
"gps_active": "off",
"video_size": "1920x1080",
"mirror_flip": "none",
"ffc": "off",
"rtsp_video_formats": "",
"rtsp_audio_formats": "",
"video_connections": "0",
"audio_connections": "0",
"ivideon_streaming": "off",
"zoom": "100",
"crop_x": "50",
"crop_y": "50",
"coloreffect": "none",
"scenemode": "auto",
"focusmode": "continuous-video",
"whitebalance": "auto",
"flashmode": "off",
"antibanding": "off",
"torch": "off",
"focus_distance": "0.0",
"focal_length": "4.25",
"aperture": "1.7",
"filter_density": "0.0",
"exposure_ns": "9384",
"frame_duration": "33333333",
"iso": "100",
"manual_sensor": "off",
"photo_size": "1920x1080",
"photo_rotation": "-1"
},
"idle": ["on", "off"],
"audio_only": ["on", "off"],
"overlay": ["on", "off"],
"focus_homing": ["on", "off"],
"night_vision": ["on", "off"],
"motion_detect": ["on", "off"],
"motion_display": ["on", "off"],
"gps_active": ["on", "off"],
"ffc": ["on", "off"],
"torch": ["on", "off"]
}

View File

@ -0,0 +1,122 @@
"""Test the Android IP Webcam config flow."""
from datetime import timedelta
from unittest.mock import patch
import aiohttp
from homeassistant import config_entries
from homeassistant.components.android_ip_webcam.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .test_init import MOCK_CONFIG_DATA
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_form(hass: HomeAssistant, aioclient_mock_fixture) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
with patch(
"homeassistant.components.android_ip_webcam.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"port": 8080,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "1.1.1.1"
assert result2["data"] == {
"host": "1.1.1.1",
"port": 8080,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_flow_success(hass: HomeAssistant, aioclient_mock_fixture) -> None:
"""Test a successful import of yaml."""
with patch(
"homeassistant.components.android_ip_webcam.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"name": "IP Webcam",
"host": "1.1.1.1",
"port": 8080,
"timeout": 10,
"scan_interval": timedelta(seconds=30),
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "IP Webcam"
assert result2["data"] == {
"name": "IP Webcam",
"host": "1.1.1.1",
"port": 8080,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_device_already_configured(
hass: HomeAssistant, aioclient_mock_fixture
) -> None:
"""Test aborting if the device is already configured."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"port": 8080,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"
async def test_form_cannot_connect(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
aioclient_mock.get(
"http://1.1.1.1:8080/status.json?show_avail=1",
exc=aiohttp.ClientError,
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}

View File

@ -0,0 +1,82 @@
"""Tests for the Android IP Webcam integration."""
from collections.abc import Awaitable
from typing import Callable
import aiohttp
from homeassistant.components.android_ip_webcam.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.components.repairs import get_repairs
from tests.test_util.aiohttp import AiohttpClientMocker
MOCK_CONFIG_DATA = {
"name": "IP Webcam",
"host": "1.1.1.1",
"port": 8080,
}
async def test_setup(
hass: HomeAssistant,
aioclient_mock_fixture,
hass_ws_client: Callable[
[HomeAssistant], Awaitable[aiohttp.ClientWebSocketResponse]
],
) -> None:
"""Test integration failed due to an error."""
assert await async_setup_component(hass, DOMAIN, {DOMAIN: [MOCK_CONFIG_DATA]})
assert hass.config_entries.async_entries(DOMAIN)
issues = await get_repairs(hass, hass_ws_client)
assert len(issues) == 1
assert issues[0]["issue_id"] == "deprecated_yaml"
async def test_successful_config_entry(
hass: HomeAssistant, aioclient_mock_fixture
) -> None:
"""Test settings up integration from config entry."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state == ConfigEntryState.LOADED
async def test_setup_failed(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test integration failed due to an error."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
aioclient_mock.get(
"http://1.1.1.1:8080/status.json?show_avail=1",
exc=aiohttp.ClientError,
)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state == ConfigEntryState.SETUP_RETRY
async def test_unload_entry(hass: HomeAssistant, aioclient_mock_fixture) -> None:
"""Test removing integration."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
assert entry.entry_id not in hass.data[DOMAIN]