mirror of
https://github.com/home-assistant/core.git
synced 2025-07-18 18:57:06 +00:00
Remove broken Arlo component (#70330)
* Remove broken Arlo component This component has been fully broken since early 2019 because of authentication changes from Arlo. * Update requirements
This commit is contained in:
parent
95e38b16f9
commit
6c391ecc59
@ -73,7 +73,6 @@ omit =
|
||||
homeassistant/components/arest/binary_sensor.py
|
||||
homeassistant/components/arest/sensor.py
|
||||
homeassistant/components/arest/switch.py
|
||||
homeassistant/components/arlo/*
|
||||
homeassistant/components/arris_tg2492lg/*
|
||||
homeassistant/components/aruba/device_tracker.py
|
||||
homeassistant/components/arwn/sensor.py
|
||||
|
@ -1,93 +0,0 @@
|
||||
"""Support for Netgear Arlo IP cameras."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
|
||||
from pyarlo import PyArlo
|
||||
from requests.exceptions import ConnectTimeout, HTTPError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import persistent_notification
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_SCAN_INTERVAL, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.dispatcher import dispatcher_send
|
||||
from homeassistant.helpers.event import track_time_interval
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTRIBUTION = "Data provided by arlo.netgear.com"
|
||||
|
||||
DATA_ARLO = "data_arlo"
|
||||
DEFAULT_BRAND = "Netgear Arlo"
|
||||
DOMAIN = "arlo"
|
||||
|
||||
NOTIFICATION_ID = "arlo_notification"
|
||||
NOTIFICATION_TITLE = "Arlo Component Setup"
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=60)
|
||||
|
||||
SIGNAL_UPDATE_ARLO = "arlo_update"
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Optional(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL): cv.time_period,
|
||||
}
|
||||
)
|
||||
},
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up an Arlo component."""
|
||||
conf = config[DOMAIN]
|
||||
username = conf[CONF_USERNAME]
|
||||
password = conf[CONF_PASSWORD]
|
||||
scan_interval = conf[CONF_SCAN_INTERVAL]
|
||||
|
||||
try:
|
||||
|
||||
arlo = PyArlo(username, password, preload=False)
|
||||
if not arlo.is_connected:
|
||||
return False
|
||||
|
||||
# assign refresh period to base station thread
|
||||
arlo_base_station = next((station for station in arlo.base_stations), None)
|
||||
|
||||
if arlo_base_station is not None:
|
||||
arlo_base_station.refresh_rate = scan_interval.total_seconds()
|
||||
elif not arlo.cameras:
|
||||
_LOGGER.error("No Arlo camera or base station available")
|
||||
return False
|
||||
|
||||
hass.data[DATA_ARLO] = arlo
|
||||
|
||||
except (ConnectTimeout, HTTPError) as ex:
|
||||
_LOGGER.error("Unable to connect to Netgear Arlo: %s", str(ex))
|
||||
persistent_notification.create(
|
||||
hass,
|
||||
f"Error: {ex}<br />You will need to restart hass after fixing.",
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID,
|
||||
)
|
||||
return False
|
||||
|
||||
def hub_refresh(_: ServiceCall | datetime) -> None:
|
||||
"""Call ArloHub to refresh information."""
|
||||
_LOGGER.debug("Updating Arlo Hub component")
|
||||
hass.data[DATA_ARLO].update(update_cameras=True, update_base_station=True)
|
||||
dispatcher_send(hass, SIGNAL_UPDATE_ARLO)
|
||||
|
||||
# register service
|
||||
hass.services.register(DOMAIN, "update", hub_refresh)
|
||||
|
||||
# register scan interval for ArloHub
|
||||
track_time_interval(hass, hub_refresh, scan_interval)
|
||||
return True
|
@ -1,142 +0,0 @@
|
||||
"""Support for Arlo Alarm Control Panels."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.alarm_control_panel import (
|
||||
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
|
||||
AlarmControlPanelEntity,
|
||||
AlarmControlPanelEntityFeature,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ATTRIBUTION,
|
||||
ATTR_DEVICE_ID,
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_ARMED_NIGHT,
|
||||
STATE_ALARM_DISARMED,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import ATTRIBUTION, DATA_ARLO, SIGNAL_UPDATE_ARLO
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ARMED = "armed"
|
||||
|
||||
CONF_HOME_MODE_NAME = "home_mode_name"
|
||||
CONF_AWAY_MODE_NAME = "away_mode_name"
|
||||
CONF_NIGHT_MODE_NAME = "night_mode_name"
|
||||
|
||||
ICON = "mdi:security"
|
||||
|
||||
PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_HOME_MODE_NAME, default=ARMED): cv.string,
|
||||
vol.Optional(CONF_AWAY_MODE_NAME, default=ARMED): cv.string,
|
||||
vol.Optional(CONF_NIGHT_MODE_NAME, default=ARMED): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the Arlo Alarm Control Panels."""
|
||||
arlo = hass.data[DATA_ARLO]
|
||||
|
||||
if not arlo.base_stations:
|
||||
return
|
||||
|
||||
home_mode_name = config[CONF_HOME_MODE_NAME]
|
||||
away_mode_name = config[CONF_AWAY_MODE_NAME]
|
||||
night_mode_name = config[CONF_NIGHT_MODE_NAME]
|
||||
base_stations = []
|
||||
for base_station in arlo.base_stations:
|
||||
base_stations.append(
|
||||
ArloBaseStation(
|
||||
base_station, home_mode_name, away_mode_name, night_mode_name
|
||||
)
|
||||
)
|
||||
add_entities(base_stations, True)
|
||||
|
||||
|
||||
class ArloBaseStation(AlarmControlPanelEntity):
|
||||
"""Representation of an Arlo Alarm Control Panel."""
|
||||
|
||||
_attr_supported_features = (
|
||||
AlarmControlPanelEntityFeature.ARM_HOME
|
||||
| AlarmControlPanelEntityFeature.ARM_AWAY
|
||||
| AlarmControlPanelEntityFeature.ARM_NIGHT
|
||||
)
|
||||
_attr_icon = ICON
|
||||
|
||||
def __init__(self, data, home_mode_name, away_mode_name, night_mode_name):
|
||||
"""Initialize the alarm control panel."""
|
||||
self._base_station = data
|
||||
self._home_mode_name = home_mode_name
|
||||
self._away_mode_name = away_mode_name
|
||||
self._night_mode_name = night_mode_name
|
||||
self._attr_name = data.name
|
||||
self._attr_extra_state_attributes = {
|
||||
ATTR_ATTRIBUTION: ATTRIBUTION,
|
||||
ATTR_DEVICE_ID: data.device_id,
|
||||
}
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register callbacks."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIGNAL_UPDATE_ARLO, self._update_callback
|
||||
)
|
||||
)
|
||||
|
||||
@callback
|
||||
def _update_callback(self):
|
||||
"""Call update method."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
def update(self):
|
||||
"""Update the state of the device."""
|
||||
_LOGGER.debug("Updating Arlo Alarm Control Panel %s", self.name)
|
||||
mode = self._base_station.mode
|
||||
self._attr_state = self._get_state_from_mode(mode) if mode else None
|
||||
|
||||
def alarm_disarm(self, code=None):
|
||||
"""Send disarm command."""
|
||||
self._base_station.mode = STATE_ALARM_DISARMED
|
||||
|
||||
def alarm_arm_away(self, code=None):
|
||||
"""Send arm away command. Uses custom mode."""
|
||||
self._base_station.mode = self._away_mode_name
|
||||
|
||||
def alarm_arm_home(self, code=None):
|
||||
"""Send arm home command. Uses custom mode."""
|
||||
self._base_station.mode = self._home_mode_name
|
||||
|
||||
def alarm_arm_night(self, code=None):
|
||||
"""Send arm night command. Uses custom mode."""
|
||||
self._base_station.mode = self._night_mode_name
|
||||
|
||||
def _get_state_from_mode(self, mode):
|
||||
"""Convert Arlo mode to Home Assistant state."""
|
||||
if mode == ARMED:
|
||||
return STATE_ALARM_ARMED_AWAY
|
||||
if mode == STATE_ALARM_DISARMED:
|
||||
return STATE_ALARM_DISARMED
|
||||
if mode == self._home_mode_name:
|
||||
return STATE_ALARM_ARMED_HOME
|
||||
if mode == self._away_mode_name:
|
||||
return STATE_ALARM_ARMED_AWAY
|
||||
if mode == self._night_mode_name:
|
||||
return STATE_ALARM_ARMED_NIGHT
|
||||
return mode
|
@ -1,173 +0,0 @@
|
||||
"""Support for Netgear Arlo IP cameras."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from haffmpeg.camera import CameraMjpeg
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera
|
||||
from homeassistant.components.ffmpeg import get_ffmpeg_manager
|
||||
from homeassistant.const import ATTR_BATTERY_LEVEL
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_aiohttp_proxy_stream
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import DATA_ARLO, DEFAULT_BRAND, SIGNAL_UPDATE_ARLO
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ARLO_MODE_ARMED = "armed"
|
||||
ARLO_MODE_DISARMED = "disarmed"
|
||||
|
||||
ATTR_BRIGHTNESS = "brightness"
|
||||
ATTR_FLIPPED = "flipped"
|
||||
ATTR_MIRRORED = "mirrored"
|
||||
ATTR_MOTION = "motion_detection_sensitivity"
|
||||
ATTR_POWERSAVE = "power_save_mode"
|
||||
ATTR_SIGNAL_STRENGTH = "signal_strength"
|
||||
ATTR_UNSEEN_VIDEOS = "unseen_videos"
|
||||
ATTR_LAST_REFRESH = "last_refresh"
|
||||
|
||||
CONF_FFMPEG_ARGUMENTS = "ffmpeg_arguments"
|
||||
DEFAULT_ARGUMENTS = "-pred 1"
|
||||
|
||||
POWERSAVE_MODE_MAPPING = {1: "best_battery_life", 2: "optimized", 3: "best_video"}
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{vol.Optional(CONF_FFMPEG_ARGUMENTS, default=DEFAULT_ARGUMENTS): cv.string}
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up an Arlo IP Camera."""
|
||||
arlo = hass.data[DATA_ARLO]
|
||||
|
||||
cameras = []
|
||||
for camera in arlo.cameras:
|
||||
cameras.append(ArloCam(hass, camera, config))
|
||||
|
||||
add_entities(cameras)
|
||||
|
||||
|
||||
class ArloCam(Camera):
|
||||
"""An implementation of a Netgear Arlo IP camera."""
|
||||
|
||||
def __init__(self, hass, camera, device_info):
|
||||
"""Initialize an Arlo camera."""
|
||||
super().__init__()
|
||||
self._camera = camera
|
||||
self._attr_name = camera.name
|
||||
self._motion_status = False
|
||||
self._ffmpeg = get_ffmpeg_manager(hass)
|
||||
self._ffmpeg_arguments = device_info.get(CONF_FFMPEG_ARGUMENTS)
|
||||
self._last_refresh = None
|
||||
self.attrs = {}
|
||||
|
||||
def camera_image(
|
||||
self, width: int | None = None, height: int | None = None
|
||||
) -> bytes | None:
|
||||
"""Return a still image response from the camera."""
|
||||
return self._camera.last_image_from_cache
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register callbacks."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIGNAL_UPDATE_ARLO, self.async_write_ha_state
|
||||
)
|
||||
)
|
||||
|
||||
async def handle_async_mjpeg_stream(self, request):
|
||||
"""Generate an HTTP MJPEG stream from the camera."""
|
||||
video = await self.hass.async_add_executor_job(
|
||||
getattr, self._camera, "last_video"
|
||||
)
|
||||
|
||||
if not video:
|
||||
error_msg = (
|
||||
f"Video not found for {self.name}. "
|
||||
f"Is it older than {self._camera.min_days_vdo_cache} days?"
|
||||
)
|
||||
_LOGGER.error(error_msg)
|
||||
return
|
||||
|
||||
stream = CameraMjpeg(self._ffmpeg.binary)
|
||||
await stream.open_camera(video.video_url, extra_cmd=self._ffmpeg_arguments)
|
||||
|
||||
try:
|
||||
stream_reader = await stream.get_reader()
|
||||
return await async_aiohttp_proxy_stream(
|
||||
self.hass,
|
||||
request,
|
||||
stream_reader,
|
||||
self._ffmpeg.ffmpeg_stream_content_type,
|
||||
)
|
||||
finally:
|
||||
await stream.close()
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
return {
|
||||
name: value
|
||||
for name, value in (
|
||||
(ATTR_BATTERY_LEVEL, self._camera.battery_level),
|
||||
(ATTR_BRIGHTNESS, self._camera.brightness),
|
||||
(ATTR_FLIPPED, self._camera.flip_state),
|
||||
(ATTR_MIRRORED, self._camera.mirror_state),
|
||||
(ATTR_MOTION, self._camera.motion_detection_sensitivity),
|
||||
(
|
||||
ATTR_POWERSAVE,
|
||||
POWERSAVE_MODE_MAPPING.get(self._camera.powersave_mode),
|
||||
),
|
||||
(ATTR_SIGNAL_STRENGTH, self._camera.signal_strength),
|
||||
(ATTR_UNSEEN_VIDEOS, self._camera.unseen_videos),
|
||||
)
|
||||
if value is not None
|
||||
}
|
||||
|
||||
@property
|
||||
def model(self):
|
||||
"""Return the camera model."""
|
||||
return self._camera.model_id
|
||||
|
||||
@property
|
||||
def brand(self):
|
||||
"""Return the camera brand."""
|
||||
return DEFAULT_BRAND
|
||||
|
||||
@property
|
||||
def motion_detection_enabled(self):
|
||||
"""Return the camera motion detection status."""
|
||||
return self._motion_status
|
||||
|
||||
def set_base_station_mode(self, mode):
|
||||
"""Set the mode in the base station."""
|
||||
# Get the list of base stations identified by library
|
||||
|
||||
# Some Arlo cameras does not have base station
|
||||
# So check if there is base station detected first
|
||||
# if yes, then choose the primary base station
|
||||
# Set the mode on the chosen base station
|
||||
if base_stations := self.hass.data[DATA_ARLO].base_stations:
|
||||
primary_base_station = base_stations[0]
|
||||
primary_base_station.mode = mode
|
||||
|
||||
def enable_motion_detection(self):
|
||||
"""Enable the Motion detection in base station (Arm)."""
|
||||
self._motion_status = True
|
||||
self.set_base_station_mode(ARLO_MODE_ARMED)
|
||||
|
||||
def disable_motion_detection(self):
|
||||
"""Disable the motion detection in base station (Disarm)."""
|
||||
self._motion_status = False
|
||||
self.set_base_station_mode(ARLO_MODE_DISARMED)
|
@ -1,10 +0,0 @@
|
||||
{
|
||||
"domain": "arlo",
|
||||
"name": "Arlo",
|
||||
"documentation": "https://www.home-assistant.io/integrations/arlo",
|
||||
"requirements": ["pyarlo==0.2.4"],
|
||||
"dependencies": ["ffmpeg"],
|
||||
"codeowners": [],
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["pyarlo", "sseclient_py"]
|
||||
}
|
@ -1,225 +0,0 @@
|
||||
"""Sensor support for Netgear Arlo IP cameras."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
PLATFORM_SCHEMA,
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
CONCENTRATION_PARTS_PER_MILLION,
|
||||
CONF_MONITORED_CONDITIONS,
|
||||
PERCENTAGE,
|
||||
TEMP_CELSIUS,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.icon import icon_for_battery_level
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import ATTRIBUTION, DATA_ARLO, DEFAULT_BRAND, SIGNAL_UPDATE_ARLO
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
|
||||
SensorEntityDescription(
|
||||
key="last_capture",
|
||||
name="Last",
|
||||
icon="mdi:run-fast",
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="total_cameras",
|
||||
name="Arlo Cameras",
|
||||
icon="mdi:video",
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="captured_today",
|
||||
name="Captured Today",
|
||||
icon="mdi:file-video",
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="battery_level",
|
||||
name="Battery Level",
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
device_class=SensorDeviceClass.BATTERY,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="signal_strength",
|
||||
name="Signal Strength",
|
||||
icon="mdi:signal",
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="temperature",
|
||||
name="Temperature",
|
||||
native_unit_of_measurement=TEMP_CELSIUS,
|
||||
device_class=SensorDeviceClass.TEMPERATURE,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="humidity",
|
||||
name="Humidity",
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
device_class=SensorDeviceClass.HUMIDITY,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="air_quality",
|
||||
name="Air Quality",
|
||||
native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
|
||||
icon="mdi:biohazard",
|
||||
),
|
||||
)
|
||||
|
||||
SENSOR_KEYS = [desc.key for desc in SENSOR_TYPES]
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_MONITORED_CONDITIONS, default=SENSOR_KEYS): vol.All(
|
||||
cv.ensure_list, [vol.In(SENSOR_KEYS)]
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up an Arlo IP sensor."""
|
||||
if not (arlo := hass.data.get(DATA_ARLO)):
|
||||
return
|
||||
|
||||
sensors = []
|
||||
for sensor_original in SENSOR_TYPES:
|
||||
if sensor_original.key not in config[CONF_MONITORED_CONDITIONS]:
|
||||
continue
|
||||
sensor_entry = replace(sensor_original)
|
||||
if sensor_entry.key == "total_cameras":
|
||||
sensors.append(ArloSensor(arlo, sensor_entry))
|
||||
else:
|
||||
for camera in arlo.cameras:
|
||||
if sensor_entry.key in ("temperature", "humidity", "air_quality"):
|
||||
continue
|
||||
|
||||
sensor_entry.name = f"{sensor_entry.name} {camera.name}"
|
||||
sensors.append(ArloSensor(camera, sensor_entry))
|
||||
|
||||
for base_station in arlo.base_stations:
|
||||
if (
|
||||
sensor_entry.key in ("temperature", "humidity", "air_quality")
|
||||
and base_station.model_id == "ABC1000"
|
||||
):
|
||||
sensor_entry.name = f"{sensor_entry.name} {base_station.name}"
|
||||
sensors.append(ArloSensor(base_station, sensor_entry))
|
||||
|
||||
add_entities(sensors, True)
|
||||
|
||||
|
||||
class ArloSensor(SensorEntity):
|
||||
"""An implementation of a Netgear Arlo IP sensor."""
|
||||
|
||||
_attr_attribution = ATTRIBUTION
|
||||
|
||||
def __init__(self, device, sensor_entry):
|
||||
"""Initialize an Arlo sensor."""
|
||||
self.entity_description = sensor_entry
|
||||
self._data = device
|
||||
self._state = None
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register callbacks."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIGNAL_UPDATE_ARLO, self._update_callback
|
||||
)
|
||||
)
|
||||
|
||||
@callback
|
||||
def _update_callback(self):
|
||||
"""Call update method."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Icon to use in the frontend, if any."""
|
||||
if self.entity_description.key == "battery_level" and self._state is not None:
|
||||
return icon_for_battery_level(
|
||||
battery_level=int(self._state), charging=False
|
||||
)
|
||||
return self.entity_description.icon
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data and updates the state."""
|
||||
_LOGGER.debug("Updating Arlo sensor %s", self.name)
|
||||
if self.entity_description.key == "total_cameras":
|
||||
self._state = len(self._data.cameras)
|
||||
|
||||
elif self.entity_description.key == "captured_today":
|
||||
self._state = len(self._data.captured_today)
|
||||
|
||||
elif self.entity_description.key == "last_capture":
|
||||
try:
|
||||
video = self._data.last_video
|
||||
self._state = video.created_at_pretty("%m-%d-%Y %H:%M:%S")
|
||||
except (AttributeError, IndexError):
|
||||
error_msg = (
|
||||
f"Video not found for {self.name}. "
|
||||
f"Older than {self._data.min_days_vdo_cache} days?"
|
||||
)
|
||||
_LOGGER.debug(error_msg)
|
||||
self._state = None
|
||||
|
||||
elif self.entity_description.key == "battery_level":
|
||||
try:
|
||||
self._state = self._data.battery_level
|
||||
except TypeError:
|
||||
self._state = None
|
||||
|
||||
elif self.entity_description.key == "signal_strength":
|
||||
try:
|
||||
self._state = self._data.signal_strength
|
||||
except TypeError:
|
||||
self._state = None
|
||||
|
||||
elif self.entity_description.key == "temperature":
|
||||
try:
|
||||
self._state = self._data.ambient_temperature
|
||||
except TypeError:
|
||||
self._state = None
|
||||
|
||||
elif self.entity_description.key == "humidity":
|
||||
try:
|
||||
self._state = self._data.ambient_humidity
|
||||
except TypeError:
|
||||
self._state = None
|
||||
|
||||
elif self.entity_description.key == "air_quality":
|
||||
try:
|
||||
self._state = self._data.ambient_air_quality
|
||||
except TypeError:
|
||||
self._state = None
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the device state attributes."""
|
||||
attrs = {}
|
||||
|
||||
attrs["brand"] = DEFAULT_BRAND
|
||||
|
||||
if self.entity_description.key != "total_cameras":
|
||||
attrs["model"] = self._data.model_id
|
||||
|
||||
return attrs
|
@ -1,5 +0,0 @@
|
||||
# Describes the format for available arlo services
|
||||
|
||||
update:
|
||||
name: Update
|
||||
description: Update the state for all cameras and the base station.
|
@ -1350,9 +1350,6 @@ pyairvisual==5.0.9
|
||||
# homeassistant.components.almond
|
||||
pyalmond==0.0.2
|
||||
|
||||
# homeassistant.components.arlo
|
||||
pyarlo==0.2.4
|
||||
|
||||
# homeassistant.components.atag
|
||||
pyatag==0.3.5.3
|
||||
|
||||
|
@ -904,9 +904,6 @@ pyairvisual==5.0.9
|
||||
# homeassistant.components.almond
|
||||
pyalmond==0.0.2
|
||||
|
||||
# homeassistant.components.arlo
|
||||
pyarlo==0.2.4
|
||||
|
||||
# homeassistant.components.atag
|
||||
pyatag==0.3.5.3
|
||||
|
||||
|
@ -1 +0,0 @@
|
||||
"""Tests for the Arlo integration."""
|
@ -1,233 +0,0 @@
|
||||
"""The tests for the Netgear Arlo sensors."""
|
||||
from collections import namedtuple
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.arlo import DATA_ARLO, sensor as arlo
|
||||
from homeassistant.components.arlo.sensor import SENSOR_TYPES
|
||||
from homeassistant.components.sensor import SensorDeviceClass
|
||||
from homeassistant.const import PERCENTAGE
|
||||
|
||||
|
||||
def _get_named_tuple(input_dict):
|
||||
return namedtuple("Struct", input_dict.keys())(*input_dict.values())
|
||||
|
||||
|
||||
def _get_sensor(hass, name="Last", sensor_type="last_capture", data=None):
|
||||
if data is None:
|
||||
data = {}
|
||||
sensor_entry = next(
|
||||
sensor_entry for sensor_entry in SENSOR_TYPES if sensor_entry.key == sensor_type
|
||||
)
|
||||
sensor_entry.name = name
|
||||
sensor = arlo.ArloSensor(data, sensor_entry)
|
||||
sensor.hass = hass
|
||||
return sensor
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def default_sensor(hass):
|
||||
"""Create an ArloSensor with default values."""
|
||||
return _get_sensor(hass)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def battery_sensor(hass):
|
||||
"""Create an ArloSensor with battery data."""
|
||||
data = _get_named_tuple({"battery_level": 50})
|
||||
return _get_sensor(hass, "Battery Level", "battery_level", data)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def temperature_sensor(hass):
|
||||
"""Create a temperature ArloSensor."""
|
||||
return _get_sensor(hass, "Temperature", "temperature")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def humidity_sensor(hass):
|
||||
"""Create a humidity ArloSensor."""
|
||||
return _get_sensor(hass, "Humidity", "humidity")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def cameras_sensor(hass):
|
||||
"""Create a total cameras ArloSensor."""
|
||||
data = _get_named_tuple({"cameras": [0, 0]})
|
||||
return _get_sensor(hass, "Arlo Cameras", "total_cameras", data)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def captured_sensor(hass):
|
||||
"""Create a captured today ArloSensor."""
|
||||
data = _get_named_tuple({"captured_today": [0, 0, 0, 0, 0]})
|
||||
return _get_sensor(hass, "Captured Today", "captured_today", data)
|
||||
|
||||
|
||||
class PlatformSetupFixture:
|
||||
"""Fixture for testing platform setup call to add_entities()."""
|
||||
|
||||
def __init__(self):
|
||||
"""Instantiate the platform setup fixture."""
|
||||
self.sensors = None
|
||||
self.update = False
|
||||
|
||||
def add_entities(self, sensors, update):
|
||||
"""Mock method for adding devices."""
|
||||
self.sensors = sensors
|
||||
self.update = update
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def platform_setup():
|
||||
"""Create an instance of the PlatformSetupFixture class."""
|
||||
return PlatformSetupFixture()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_dispatch():
|
||||
"""Mock the dispatcher connect method."""
|
||||
target = "homeassistant.components.arlo.sensor.async_dispatcher_connect"
|
||||
with patch(target) as _mock:
|
||||
yield _mock
|
||||
|
||||
|
||||
def test_setup_with_no_data(platform_setup, hass):
|
||||
"""Test setup_platform with no data."""
|
||||
arlo.setup_platform(hass, None, platform_setup.add_entities)
|
||||
assert platform_setup.sensors is None
|
||||
assert not platform_setup.update
|
||||
|
||||
|
||||
def test_setup_with_valid_data(platform_setup, hass):
|
||||
"""Test setup_platform with valid data."""
|
||||
config = {
|
||||
"monitored_conditions": [
|
||||
"last_capture",
|
||||
"total_cameras",
|
||||
"captured_today",
|
||||
"battery_level",
|
||||
"signal_strength",
|
||||
"temperature",
|
||||
"humidity",
|
||||
"air_quality",
|
||||
]
|
||||
}
|
||||
|
||||
hass.data[DATA_ARLO] = _get_named_tuple(
|
||||
{
|
||||
"cameras": [_get_named_tuple({"name": "Camera", "model_id": "ABC1000"})],
|
||||
"base_stations": [
|
||||
_get_named_tuple({"name": "Base Station", "model_id": "ABC1000"})
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
arlo.setup_platform(hass, config, platform_setup.add_entities)
|
||||
assert len(platform_setup.sensors) == 8
|
||||
assert platform_setup.update
|
||||
|
||||
|
||||
def test_sensor_name(default_sensor):
|
||||
"""Test the name property."""
|
||||
assert default_sensor.name == "Last"
|
||||
|
||||
|
||||
async def test_async_added_to_hass(default_sensor, mock_dispatch):
|
||||
"""Test dispatcher called when added."""
|
||||
await default_sensor.async_added_to_hass()
|
||||
assert len(mock_dispatch.mock_calls) == 1
|
||||
kall = mock_dispatch.call_args
|
||||
args, kwargs = kall
|
||||
assert len(args) == 3
|
||||
assert args[0] == default_sensor.hass
|
||||
assert args[1] == "arlo_update"
|
||||
assert not kwargs
|
||||
|
||||
|
||||
def test_sensor_state_default(default_sensor):
|
||||
"""Test the state property."""
|
||||
assert default_sensor.state is None
|
||||
|
||||
|
||||
def test_sensor_device_class__battery(battery_sensor):
|
||||
"""Test the battery device_class."""
|
||||
assert battery_sensor.device_class == SensorDeviceClass.BATTERY
|
||||
|
||||
|
||||
def test_sensor_device_class(temperature_sensor):
|
||||
"""Test the device_class property."""
|
||||
assert temperature_sensor.device_class == SensorDeviceClass.TEMPERATURE
|
||||
|
||||
|
||||
def test_unit_of_measure(default_sensor, battery_sensor):
|
||||
"""Test the unit_of_measurement property."""
|
||||
assert default_sensor.unit_of_measurement is None
|
||||
assert battery_sensor.unit_of_measurement == PERCENTAGE
|
||||
|
||||
|
||||
def test_device_class(default_sensor, temperature_sensor, humidity_sensor):
|
||||
"""Test the device_class property."""
|
||||
assert default_sensor.device_class is None
|
||||
assert temperature_sensor.device_class == SensorDeviceClass.TEMPERATURE
|
||||
assert humidity_sensor.device_class == SensorDeviceClass.HUMIDITY
|
||||
|
||||
|
||||
def test_attribution(default_sensor, temperature_sensor, humidity_sensor):
|
||||
"""Test the device_class property."""
|
||||
assert default_sensor.attribution == "Data provided by arlo.netgear.com"
|
||||
assert temperature_sensor.attribution == "Data provided by arlo.netgear.com"
|
||||
assert humidity_sensor.attribution == "Data provided by arlo.netgear.com"
|
||||
|
||||
|
||||
def test_update_total_cameras(cameras_sensor):
|
||||
"""Test update method for total_cameras sensor type."""
|
||||
cameras_sensor.update()
|
||||
assert cameras_sensor.state == 2
|
||||
|
||||
|
||||
def test_update_captured_today(captured_sensor):
|
||||
"""Test update method for captured_today sensor type."""
|
||||
captured_sensor.update()
|
||||
assert captured_sensor.state == 5
|
||||
|
||||
|
||||
def _test_attributes(hass, sensor_type):
|
||||
data = _get_named_tuple({"model_id": "TEST123"})
|
||||
sensor = _get_sensor(hass, "test", sensor_type, data)
|
||||
attrs = sensor.extra_state_attributes
|
||||
assert attrs.get("brand") == "Netgear Arlo"
|
||||
assert attrs.get("model") == "TEST123"
|
||||
|
||||
|
||||
def test_state_attributes(hass):
|
||||
"""Test attributes for camera sensor types."""
|
||||
_test_attributes(hass, "battery_level")
|
||||
_test_attributes(hass, "signal_strength")
|
||||
_test_attributes(hass, "temperature")
|
||||
_test_attributes(hass, "humidity")
|
||||
_test_attributes(hass, "air_quality")
|
||||
|
||||
|
||||
def test_attributes_total_cameras(cameras_sensor):
|
||||
"""Test attributes for total cameras sensor type."""
|
||||
attrs = cameras_sensor.extra_state_attributes
|
||||
assert attrs.get("brand") == "Netgear Arlo"
|
||||
assert attrs.get("model") is None
|
||||
|
||||
|
||||
def _test_update(hass, sensor_type, key, value):
|
||||
data = _get_named_tuple({key: value})
|
||||
sensor = _get_sensor(hass, "test", sensor_type, data)
|
||||
sensor.update()
|
||||
assert sensor.state == value
|
||||
|
||||
|
||||
def test_update(hass):
|
||||
"""Test update method for direct transcription sensor types."""
|
||||
_test_update(hass, "battery_level", "battery_level", 100)
|
||||
_test_update(hass, "signal_strength", "signal_strength", 100)
|
||||
_test_update(hass, "temperature", "ambient_temperature", 21.4)
|
||||
_test_update(hass, "humidity", "ambient_humidity", 45.1)
|
||||
_test_update(hass, "air_quality", "ambient_air_quality", 14.2)
|
Loading…
x
Reference in New Issue
Block a user