jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove deprecated logi_circle integration (#123727)

Browse Source
This commit is contained in:
G Johansson 2024-08-12 21:31:10 +02:00 committed by GitHub
parent ff0a44cc12
commit 05c4b1a6a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 0 additions and 1350 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -826,8 +826,6 @@ build.json @home-assistant/supervisor
/tests/components/logbook/ @home-assistant/core /tests/components/logbook/ @home-assistant/core
/homeassistant/components/logger/ @home-assistant/core /homeassistant/components/logger/ @home-assistant/core
/tests/components/logger/ @home-assistant/core /tests/components/logger/ @home-assistant/core
/homeassistant/components/logi_circle/ @evanjd
/tests/components/logi_circle/ @evanjd
/homeassistant/components/london_underground/ @jpbede /homeassistant/components/london_underground/ @jpbede
/tests/components/london_underground/ @jpbede /tests/components/london_underground/ @jpbede
/homeassistant/components/lookin/ @ANMalko @bdraco /homeassistant/components/lookin/ @ANMalko @bdraco

271
homeassistant/components/logi_circle/__init__.py
View File

@ -1,271 +0,0 @@
"""Support for Logi Circle devices."""
import asyncio
from aiohttp.client_exceptions import ClientResponseError
from logi_circle import LogiCircle
from logi_circle.exception import AuthorizationFailed
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import persistent_notification
from homeassistant.components.camera import ATTR_FILENAME
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_MODE,
CONF_API_KEY,
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
CONF_MONITORED_CONDITIONS,
CONF_SENSORS,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import ConfigType
from . import config_flow
from .const import (
CONF_REDIRECT_URI,
DATA_LOGI,
DEFAULT_CACHEDB,
DOMAIN,
LED_MODE_KEY,
RECORDING_MODE_KEY,
SIGNAL_LOGI_CIRCLE_RECONFIGURE,
SIGNAL_LOGI_CIRCLE_RECORD,
SIGNAL_LOGI_CIRCLE_SNAPSHOT,
)
from .sensor import SENSOR_TYPES
NOTIFICATION_ID = "logi_circle_notification"
NOTIFICATION_TITLE = "Logi Circle Setup"
_TIMEOUT = 15 # seconds
SERVICE_SET_CONFIG = "set_config"
SERVICE_LIVESTREAM_SNAPSHOT = "livestream_snapshot"
SERVICE_LIVESTREAM_RECORD = "livestream_record"
ATTR_VALUE = "value"
ATTR_DURATION = "duration"
PLATFORMS = [Platform.CAMERA, Platform.SENSOR]
SENSOR_KEYS = [desc.key for desc in SENSOR_TYPES]
SENSOR_SCHEMA = vol.Schema(
{
vol.Optional(CONF_MONITORED_CONDITIONS, default=SENSOR_KEYS): vol.All(
cv.ensure_list, [vol.In(SENSOR_KEYS)]
)
}
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_CLIENT_ID): cv.string,
vol.Required(CONF_CLIENT_SECRET): cv.string,
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_REDIRECT_URI): cv.string,
vol.Optional(CONF_SENSORS, default={}): SENSOR_SCHEMA,
}
)
},
extra=vol.ALLOW_EXTRA,
)
LOGI_CIRCLE_SERVICE_SET_CONFIG = vol.Schema(
{
vol.Optional(ATTR_ENTITY_ID): cv.comp_entity_ids,
vol.Required(ATTR_MODE): vol.In([LED_MODE_KEY, RECORDING_MODE_KEY]),
vol.Required(ATTR_VALUE): cv.boolean,
}
)
LOGI_CIRCLE_SERVICE_SNAPSHOT = vol.Schema(
{
vol.Optional(ATTR_ENTITY_ID): cv.comp_entity_ids,
vol.Required(ATTR_FILENAME): cv.template,
}
)
LOGI_CIRCLE_SERVICE_RECORD = vol.Schema(
{
vol.Optional(ATTR_ENTITY_ID): cv.comp_entity_ids,
vol.Required(ATTR_FILENAME): cv.template,
vol.Required(ATTR_DURATION): cv.positive_int,
}
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up configured Logi Circle component."""
if DOMAIN not in config:
return True
conf = config[DOMAIN]
config_flow.register_flow_implementation(
hass,
DOMAIN,
client_id=conf[CONF_CLIENT_ID],
client_secret=conf[CONF_CLIENT_SECRET],
api_key=conf[CONF_API_KEY],
redirect_uri=conf[CONF_REDIRECT_URI],
sensors=conf[CONF_SENSORS],
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Logi Circle from a config entry."""
ir.async_create_issue(
hass,
DOMAIN,
DOMAIN,
breaks_in_ha_version="2024.9.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="integration_removed",
translation_placeholders={
"entries": "/config/integrations/integration/logi_circle",
},
)
logi_circle = LogiCircle(
client_id=entry.data[CONF_CLIENT_ID],
client_secret=entry.data[CONF_CLIENT_SECRET],
api_key=entry.data[CONF_API_KEY],
redirect_uri=entry.data[CONF_REDIRECT_URI],
cache_file=hass.config.path(DEFAULT_CACHEDB),
)
if not logi_circle.authorized:
persistent_notification.create(
hass,
(
"Error: The cached access tokens are missing from"
f" {DEFAULT_CACHEDB}.<br />Please unload then re-add the Logi Circle"
" integration to resolve."
),
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID,
)
return False
try:
async with asyncio.timeout(_TIMEOUT):
# Ensure the cameras property returns the same Camera objects for
# all devices. Performs implicit login and session validation.
await logi_circle.synchronize_cameras()
except AuthorizationFailed:
persistent_notification.create(
hass,
(
"Error: Failed to obtain an access token from the cached "
"refresh token.<br />"
"Token may have expired or been revoked.<br />"
"Please unload then re-add the Logi Circle integration to resolve"
),
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID,
)
return False
except TimeoutError:
# The TimeoutError exception object returns nothing when casted to a
# string, so we'll handle it separately.
err = f"{_TIMEOUT}s timeout exceeded when connecting to Logi Circle API"
persistent_notification.create(
hass,
f"Error: {err}<br />You will need to restart hass after fixing.",
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID,
)
return False
except ClientResponseError as 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
hass.data[DATA_LOGI] = logi_circle
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
async def service_handler(service: ServiceCall) -> None:
"""Dispatch service calls to target entities."""
params = dict(service.data)
if service.service == SERVICE_SET_CONFIG:
async_dispatcher_send(hass, SIGNAL_LOGI_CIRCLE_RECONFIGURE, params)
if service.service == SERVICE_LIVESTREAM_SNAPSHOT:
async_dispatcher_send(hass, SIGNAL_LOGI_CIRCLE_SNAPSHOT, params)
if service.service == SERVICE_LIVESTREAM_RECORD:
async_dispatcher_send(hass, SIGNAL_LOGI_CIRCLE_RECORD, params)
hass.services.async_register(
DOMAIN,
SERVICE_SET_CONFIG,
service_handler,
schema=LOGI_CIRCLE_SERVICE_SET_CONFIG,
)
hass.services.async_register(
DOMAIN,
SERVICE_LIVESTREAM_SNAPSHOT,
service_handler,
schema=LOGI_CIRCLE_SERVICE_SNAPSHOT,
)
hass.services.async_register(
DOMAIN,
SERVICE_LIVESTREAM_RECORD,
service_handler,
schema=LOGI_CIRCLE_SERVICE_RECORD,
)
async def shut_down(event=None):
"""Close Logi Circle aiohttp session."""
await logi_circle.auth_provider.close()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, shut_down)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if all(
config_entry.state is config_entries.ConfigEntryState.NOT_LOADED
for config_entry in hass.config_entries.async_entries(DOMAIN)
if config_entry.entry_id != entry.entry_id
):
ir.async_delete_issue(hass, DOMAIN, DOMAIN)
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
logi_circle = hass.data.pop(DATA_LOGI)
# Tell API wrapper to close all aiohttp sessions, invalidate WS connections
# and clear all locally cached tokens
await logi_circle.auth_provider.clear_authorization()
return unload_ok

200
homeassistant/components/logi_circle/camera.py
View File

@ -1,200 +0,0 @@
"""Support to the Logi Circle cameras."""
from __future__ import annotations
from datetime import timedelta
import logging
from homeassistant.components.camera import Camera, CameraEntityFeature
from homeassistant.components.ffmpeg import get_ffmpeg_manager
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_BATTERY_CHARGING,
ATTR_BATTERY_LEVEL,
ATTR_ENTITY_ID,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import (
ATTRIBUTION,
DEVICE_BRAND,
DOMAIN as LOGI_CIRCLE_DOMAIN,
LED_MODE_KEY,
RECORDING_MODE_KEY,
SIGNAL_LOGI_CIRCLE_RECONFIGURE,
SIGNAL_LOGI_CIRCLE_RECORD,
SIGNAL_LOGI_CIRCLE_SNAPSHOT,
)
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=60)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up a Logi Circle Camera. Obsolete."""
_LOGGER.warning("Logi Circle no longer works with camera platform configuration")
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Logi Circle Camera based on a config entry."""
devices = await hass.data[LOGI_CIRCLE_DOMAIN].cameras
ffmpeg = get_ffmpeg_manager(hass)
cameras = [LogiCam(device, ffmpeg) for device in devices]
async_add_entities(cameras, True)
class LogiCam(Camera):
"""An implementation of a Logi Circle camera."""
_attr_attribution = ATTRIBUTION
_attr_should_poll = True # Cameras default to False
_attr_supported_features = CameraEntityFeature.ON_OFF
_attr_has_entity_name = True
_attr_name = None
def __init__(self, camera, ffmpeg):
"""Initialize Logi Circle camera."""
super().__init__()
self._camera = camera
self._has_battery = camera.supports_feature("battery_level")
self._ffmpeg = ffmpeg
self._listeners = []
self._attr_unique_id = camera.mac_address
self._attr_device_info = DeviceInfo(
identifiers={(LOGI_CIRCLE_DOMAIN, camera.id)},
manufacturer=DEVICE_BRAND,
model=camera.model_name,
name=camera.name,
sw_version=camera.firmware,
)
async def async_added_to_hass(self) -> None:
"""Connect camera methods to signals."""
def _dispatch_proxy(method):
"""Expand parameters & filter entity IDs."""
async def _call(params):
entity_ids = params.get(ATTR_ENTITY_ID)
filtered_params = {
k: v for k, v in params.items() if k != ATTR_ENTITY_ID
}
if entity_ids is None or self.entity_id in entity_ids:
await method(**filtered_params)
return _call
self._listeners.extend(
[
async_dispatcher_connect(
self.hass,
SIGNAL_LOGI_CIRCLE_RECONFIGURE,
_dispatch_proxy(self.set_config),
),
async_dispatcher_connect(
self.hass,
SIGNAL_LOGI_CIRCLE_SNAPSHOT,
_dispatch_proxy(self.livestream_snapshot),
),
async_dispatcher_connect(
self.hass,
SIGNAL_LOGI_CIRCLE_RECORD,
_dispatch_proxy(self.download_livestream),
),
]
)
async def async_will_remove_from_hass(self) -> None:
"""Disconnect dispatcher listeners when removed."""
for detach in self._listeners:
detach()
@property
def extra_state_attributes(self):
"""Return the state attributes."""
state = {
"battery_saving_mode": (
STATE_ON if self._camera.battery_saving else STATE_OFF
),
"microphone_gain": self._camera.microphone_gain,
}
# Add battery attributes if camera is battery-powered
if self._has_battery:
state[ATTR_BATTERY_CHARGING] = self._camera.charging
state[ATTR_BATTERY_LEVEL] = self._camera.battery_level
return state
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image from the camera."""
return await self._camera.live_stream.download_jpeg()
async def async_turn_off(self) -> None:
"""Disable streaming mode for this camera."""
await self._camera.set_config("streaming", False)
async def async_turn_on(self) -> None:
"""Enable streaming mode for this camera."""
await self._camera.set_config("streaming", True)
async def set_config(self, mode, value):
"""Set an configuration property for the target camera."""
if mode == LED_MODE_KEY:
await self._camera.set_config("led", value)
if mode == RECORDING_MODE_KEY:
await self._camera.set_config("recording_disabled", not value)
async def download_livestream(self, filename, duration):
"""Download a recording from the camera's livestream."""
# Render filename from template.
stream_file = filename.async_render(variables={ATTR_ENTITY_ID: self.entity_id})
# Respect configured allowed paths.
if not self.hass.config.is_allowed_path(stream_file):
_LOGGER.error("Can't write %s, no access to path!", stream_file)
return
await self._camera.live_stream.download_rtsp(
filename=stream_file,
duration=timedelta(seconds=duration),
ffmpeg_bin=self._ffmpeg.binary,
)
async def livestream_snapshot(self, filename):
"""Download a still frame from the camera's livestream."""
# Render filename from template.
snapshot_file = filename.async_render(
variables={ATTR_ENTITY_ID: self.entity_id}
)
# Respect configured allowed paths.
if not self.hass.config.is_allowed_path(snapshot_file):
_LOGGER.error("Can't write %s, no access to path!", snapshot_file)
return
await self._camera.live_stream.download_jpeg(
filename=snapshot_file, refresh=True
)
async def async_update(self) -> None:
"""Update camera entity and refresh attributes."""
await self._camera.update()

206
homeassistant/components/logi_circle/config_flow.py
View File

@ -1,206 +0,0 @@
"""Config flow to configure Logi Circle component."""
import asyncio
from collections import OrderedDict
from http import HTTPStatus
from logi_circle import LogiCircle
from logi_circle.exception import AuthorizationFailed
import voluptuous as vol
from homeassistant.components.http import KEY_HASS, HomeAssistantView
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import (
CONF_API_KEY,
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
CONF_SENSORS,
)
from homeassistant.core import callback
from .const import CONF_REDIRECT_URI, DEFAULT_CACHEDB, DOMAIN
_TIMEOUT = 15 # seconds
DATA_FLOW_IMPL = "logi_circle_flow_implementation"
EXTERNAL_ERRORS = "logi_errors"
AUTH_CALLBACK_PATH = "/api/logi_circle"
AUTH_CALLBACK_NAME = "api:logi_circle"
@callback
def register_flow_implementation(
hass, domain, client_id, client_secret, api_key, redirect_uri, sensors
):
"""Register a flow implementation.
domain: Domain of the component responsible for the implementation.
client_id: Client ID.
client_secret: Client secret.
api_key: API key issued by Logitech.
redirect_uri: Auth callback redirect URI.
sensors: Sensor config.
"""
if DATA_FLOW_IMPL not in hass.data:
hass.data[DATA_FLOW_IMPL] = OrderedDict()
hass.data[DATA_FLOW_IMPL][domain] = {
CONF_CLIENT_ID: client_id,
CONF_CLIENT_SECRET: client_secret,
CONF_API_KEY: api_key,
CONF_REDIRECT_URI: redirect_uri,
CONF_SENSORS: sensors,
EXTERNAL_ERRORS: None,
}
class LogiCircleFlowHandler(ConfigFlow, domain=DOMAIN):
"""Config flow for Logi Circle component."""
VERSION = 1
def __init__(self) -> None:
"""Initialize flow."""
self.flow_impl = None
async def async_step_import(self, user_input=None):
"""Handle external yaml configuration."""
self._async_abort_entries_match()
self.flow_impl = DOMAIN
return await self.async_step_auth()
async def async_step_user(self, user_input=None):
"""Handle a flow start."""
flows = self.hass.data.get(DATA_FLOW_IMPL, {})
self._async_abort_entries_match()
if not flows:
return self.async_abort(reason="missing_configuration")
if len(flows) == 1:
self.flow_impl = list(flows)[0]
return await self.async_step_auth()
if user_input is not None:
self.flow_impl = user_input["flow_impl"]
return await self.async_step_auth()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required("flow_impl"): vol.In(list(flows))}),
)
async def async_step_auth(self, user_input=None):
"""Create an entry for auth."""
if self._async_current_entries():
return self.async_abort(reason="external_setup")
external_error = self.hass.data[DATA_FLOW_IMPL][DOMAIN][EXTERNAL_ERRORS]
errors = {}
if external_error:
# Handle error from another flow
errors["base"] = external_error
self.hass.data[DATA_FLOW_IMPL][DOMAIN][EXTERNAL_ERRORS] = None
elif user_input is not None:
errors["base"] = "follow_link"
url = self._get_authorization_url()
return self.async_show_form(
step_id="auth",
description_placeholders={"authorization_url": url},
errors=errors,
)
def _get_authorization_url(self):
"""Create temporary Circle session and generate authorization url."""
flow = self.hass.data[DATA_FLOW_IMPL][self.flow_impl]
client_id = flow[CONF_CLIENT_ID]
client_secret = flow[CONF_CLIENT_SECRET]
api_key = flow[CONF_API_KEY]
redirect_uri = flow[CONF_REDIRECT_URI]
logi_session = LogiCircle(
client_id=client_id,
client_secret=client_secret,
api_key=api_key,
redirect_uri=redirect_uri,
)
self.hass.http.register_view(LogiCircleAuthCallbackView())
return logi_session.authorize_url
async def async_step_code(self, code=None):
"""Received code for authentication."""
self._async_abort_entries_match()
return await self._async_create_session(code)
async def _async_create_session(self, code):
"""Create Logi Circle session and entries."""
flow = self.hass.data[DATA_FLOW_IMPL][DOMAIN]
client_id = flow[CONF_CLIENT_ID]
client_secret = flow[CONF_CLIENT_SECRET]
api_key = flow[CONF_API_KEY]
redirect_uri = flow[CONF_REDIRECT_URI]
sensors = flow[CONF_SENSORS]
logi_session = LogiCircle(
client_id=client_id,
client_secret=client_secret,
api_key=api_key,
redirect_uri=redirect_uri,
cache_file=self.hass.config.path(DEFAULT_CACHEDB),
)
try:
async with asyncio.timeout(_TIMEOUT):
await logi_session.authorize(code)
except AuthorizationFailed:
(self.hass.data[DATA_FLOW_IMPL][DOMAIN][EXTERNAL_ERRORS]) = "invalid_auth"
return self.async_abort(reason="external_error")
except TimeoutError:
(
self.hass.data[DATA_FLOW_IMPL][DOMAIN][EXTERNAL_ERRORS]
) = "authorize_url_timeout"
return self.async_abort(reason="external_error")
account_id = (await logi_session.account)["accountId"]
await logi_session.close()
return self.async_create_entry(
title=f"Logi Circle ({account_id})",
data={
CONF_CLIENT_ID: client_id,
CONF_CLIENT_SECRET: client_secret,
CONF_API_KEY: api_key,
CONF_REDIRECT_URI: redirect_uri,
CONF_SENSORS: sensors,
},
)
class LogiCircleAuthCallbackView(HomeAssistantView):
"""Logi Circle Authorization Callback View."""
requires_auth = False
url = AUTH_CALLBACK_PATH
name = AUTH_CALLBACK_NAME
async def get(self, request):
"""Receive authorization code."""
hass = request.app[KEY_HASS]
if "code" in request.query:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": "code"}, data=request.query["code"]
)
)
return self.json_message("Authorisation code saved")
return self.json_message(
"Authorisation code missing from query string",
status_code=HTTPStatus.BAD_REQUEST,
)

22
homeassistant/components/logi_circle/const.py
View File

@ -1,22 +0,0 @@
"""Constants in Logi Circle component."""
from __future__ import annotations
DOMAIN = "logi_circle"
DATA_LOGI = DOMAIN
CONF_REDIRECT_URI = "redirect_uri"
DEFAULT_CACHEDB = ".logi_cache.pickle"
LED_MODE_KEY = "LED"
RECORDING_MODE_KEY = "RECORDING_MODE"
SIGNAL_LOGI_CIRCLE_RECONFIGURE = "logi_circle_reconfigure"
SIGNAL_LOGI_CIRCLE_SNAPSHOT = "logi_circle_snapshot"
SIGNAL_LOGI_CIRCLE_RECORD = "logi_circle_record"
# Attribution
ATTRIBUTION = "Data provided by circle.logi.com"
DEVICE_BRAND = "Logitech"

7
homeassistant/components/logi_circle/icons.json
View File

@ -1,7 +0,0 @@
{
"services": {
"set_config": "mdi:cog",
"livestream_snapshot": "mdi:camera",
"livestream_record": "mdi:record-rec"
}
}

11
homeassistant/components/logi_circle/manifest.json
View File

@ -1,11 +0,0 @@
{
"domain": "logi_circle",
"name": "Logi Circle",
"codeowners": ["@evanjd"],
"config_flow": true,
"dependencies": ["ffmpeg", "http"],
"documentation": "https://www.home-assistant.io/integrations/logi_circle",
"iot_class": "cloud_polling",
"loggers": ["logi_circle"],
"requirements": ["logi-circle==0.2.3"]
}

164
homeassistant/components/logi_circle/sensor.py
View File

@ -1,164 +0,0 @@
"""Support for Logi Circle sensors."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_BATTERY_CHARGING,
CONF_MONITORED_CONDITIONS,
CONF_SENSORS,
PERCENTAGE,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.dt import as_local
from .const import ATTRIBUTION, DEVICE_BRAND, DOMAIN as LOGI_CIRCLE_DOMAIN
_LOGGER = logging.getLogger(__name__)
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="battery_level",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
),
SensorEntityDescription(
key="last_activity_time",
translation_key="last_activity",
icon="mdi:history",
),
SensorEntityDescription(
key="recording",
translation_key="recording_mode",
icon="mdi:eye",
),
SensorEntityDescription(
key="signal_strength_category",
translation_key="wifi_signal_category",
icon="mdi:wifi",
),
SensorEntityDescription(
key="signal_strength_percentage",
translation_key="wifi_signal_strength",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:wifi",
),
SensorEntityDescription(
key="streaming",
translation_key="streaming_mode",
icon="mdi:camera",
),
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up a sensor for a Logi Circle device. Obsolete."""
_LOGGER.warning("Logi Circle no longer works with sensor platform configuration")
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Logi Circle sensor based on a config entry."""
devices = await hass.data[LOGI_CIRCLE_DOMAIN].cameras
time_zone = str(hass.config.time_zone)
monitored_conditions = entry.data[CONF_SENSORS].get(CONF_MONITORED_CONDITIONS)
entities = [
LogiSensor(device, time_zone, description)
for description in SENSOR_TYPES
if description.key in monitored_conditions
for device in devices
if device.supports_feature(description.key)
]
async_add_entities(entities, True)
class LogiSensor(SensorEntity):
"""A sensor implementation for a Logi Circle camera."""
_attr_attribution = ATTRIBUTION
_attr_has_entity_name = True
def __init__(self, camera, time_zone, description: SensorEntityDescription) -> None:
"""Initialize a sensor for Logi Circle camera."""
self.entity_description = description
self._camera = camera
self._attr_unique_id = f"{camera.mac_address}-{description.key}"
self._activity: dict[Any, Any] = {}
self._tz = time_zone
self._attr_device_info = DeviceInfo(
identifiers={(LOGI_CIRCLE_DOMAIN, camera.id)},
manufacturer=DEVICE_BRAND,
model=camera.model_name,
name=camera.name,
sw_version=camera.firmware,
)
@property
def extra_state_attributes(self):
"""Return the state attributes."""
state = {
"battery_saving_mode": (
STATE_ON if self._camera.battery_saving else STATE_OFF
),
"microphone_gain": self._camera.microphone_gain,
}
if self.entity_description.key == "battery_level":
state[ATTR_BATTERY_CHARGING] = self._camera.charging
return state
@property
def icon(self):
"""Icon to use in the frontend, if any."""
sensor_type = self.entity_description.key
if sensor_type == "recording_mode" and self._attr_native_value is not None:
return "mdi:eye" if self._attr_native_value == STATE_ON else "mdi:eye-off"
if sensor_type == "streaming_mode" and self._attr_native_value is not None:
return (
"mdi:camera"
if self._attr_native_value == STATE_ON
else "mdi:camera-off"
)
return self.entity_description.icon
async def async_update(self) -> None:
"""Get the latest data and updates the state."""
_LOGGER.debug("Pulling data from %s sensor", self.name)
await self._camera.update()
if self.entity_description.key == "last_activity_time":
last_activity = await self._camera.get_last_activity(force_refresh=True)
if last_activity is not None:
last_activity_time = as_local(last_activity.end_time_utc)
self._attr_native_value = (
f"{last_activity_time.hour:0>2}:{last_activity_time.minute:0>2}"
)
else:
state = getattr(self._camera, self.entity_description.key, None)
if isinstance(state, bool):
self._attr_native_value = STATE_ON if state is True else STATE_OFF
else:
self._attr_native_value = state

53
homeassistant/components/logi_circle/services.yaml
View File

@ -1,53 +0,0 @@
# Describes the format for available Logi Circle services
set_config:
fields:
entity_id:
selector:
entity:
integration: logi_circle
domain: camera
mode:
required: true
selector:
select:
options:
- "LED"
- "RECORDING_MODE"
value:
required: true
selector:
boolean:
livestream_snapshot:
fields:
entity_id:
selector:
entity:
integration: logi_circle
domain: camera
filename:
required: true
example: "/tmp/snapshot_{{ entity_id }}.jpg"
selector:
text:
livestream_record:
fields:
entity_id:
selector:
entity:
integration: logi_circle
domain: camera
filename:
required: true
example: "/tmp/snapshot_{{ entity_id }}.mp4"
selector:
text:
duration:
required: true
selector:
number:
min: 1
max: 3600
unit_of_measurement: seconds

105
homeassistant/components/logi_circle/strings.json
View File

@ -1,105 +0,0 @@
{
"config": {
"step": {
"user": {
"title": "Authentication Provider",
"description": "Pick via which authentication provider you want to authenticate with Logi Circle.",
"data": {
"flow_impl": "Provider"
}
},
"auth": {
"title": "Authenticate with Logi Circle",
"description": "Please follow the link below and **Accept** access to your Logi Circle account, then come back and press **Submit** below.\n\n[Link]({authorization_url})"
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"follow_link": "Please follow the link and authenticate before pressing Submit."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"external_error": "Exception occurred from another flow.",
"external_setup": "Logi Circle successfully configured from another flow.",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]"
}
},
"entity": {
"sensor": {
"last_activity": {
"name": "Last activity"
},
"recording_mode": {
"name": "Recording mode"
},
"wifi_signal_category": {
"name": "Wi-Fi signal category"
},
"wifi_signal_strength": {
"name": "Wi-Fi signal strength"
},
"streaming_mode": {
"name": "Streaming mode"
}
}
},
"issues": {
"integration_removed": {
"title": "The Logi Circle integration has been deprecated and will be removed",
"description": "Logitech stopped accepting applications for access to the Logi Circle API in May 2022, and the Logi Circle integration will be removed from Home Assistant.\n\nTo resolve this issue, please remove the integration entries from your Home Assistant setup. [Click here to see your existing Logi Circle integration entries]({entries})."
}
},
"services": {
"set_config": {
"name": "Set config",
"description": "Sets a configuration property.",
"fields": {
"entity_id": {
"name": "Entity",
"description": "Name(s) of entities to apply the operation mode to."
},
"mode": {
"name": "[%key:common::config_flow::data::mode%]",
"description": "Operation mode. Allowed values: LED, RECORDING_MODE."
},
"value": {
"name": "Value",
"description": "Operation value."
}
}
},
"livestream_snapshot": {
"name": "Livestream snapshot",
"description": "Takes a snapshot from the camera's livestream. Will wake the camera from sleep if required.",
"fields": {
"entity_id": {
"name": "Entity",
"description": "Name(s) of entities to create snapshots from."
},
"filename": {
"name": "File name",
"description": "Template of a Filename. Variable is entity_id."
}
}
},
"livestream_record": {
"name": "Livestream record",
"description": "Takes a video recording from the camera's livestream.",
"fields": {
"entity_id": {
"name": "Entity",
"description": "Name(s) of entities to create recordings from."
},
"filename": {
"name": "File name",
"description": "[%key:component::logi_circle::services::livestream_snapshot::fields::filename::description%]"
},
"duration": {
"name": "Duration",
"description": "Recording duration."
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -325,7 +325,6 @@ FLOWS = {
"local_ip", "local_ip",
"local_todo", "local_todo",
"locative", "locative",
"logi_circle",
"lookin", "lookin",
"loqed", "loqed",
"luftdaten", "luftdaten",

6
homeassistant/generated/integrations.json
View File

@ -3339,12 +3339,6 @@
"config_flow": false, "config_flow": false,
"iot_class": "cloud_push" "iot_class": "cloud_push"
}, },
"logi_circle": {
"name": "Logi Circle",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"logitech": { "logitech": {
"name": "Logitech", "name": "Logitech",
"integrations": { "integrations": {

3
requirements_all.txt
View File

@ -1278,9 +1278,6 @@ lmcloud==1.1.13
# homeassistant.components.google_maps # homeassistant.components.google_maps
locationsharinglib==5.0.1 locationsharinglib==5.0.1
# homeassistant.components.logi_circle
logi-circle==0.2.3
# homeassistant.components.london_underground # homeassistant.components.london_underground
london-tube-status==0.5 london-tube-status==0.5

3
requirements_test_all.txt
View File

@ -1056,9 +1056,6 @@ linear-garage-door==0.2.9
# homeassistant.components.lamarzocco # homeassistant.components.lamarzocco
lmcloud==1.1.13 lmcloud==1.1.13
# homeassistant.components.logi_circle
logi-circle==0.2.3
# homeassistant.components.london_underground # homeassistant.components.london_underground
london-tube-status==0.5 london-tube-status==0.5

1
tests/components/logi_circle/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the Logi Circle component."""

227
tests/components/logi_circle/test_config_flow.py
View File

@ -1,227 +0,0 @@
"""Tests for Logi Circle config flow."""
import asyncio
from collections.abc import Generator
from http import HTTPStatus
from typing import Any
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from homeassistant import config_entries
from homeassistant.components.http import KEY_HASS
from homeassistant.components.logi_circle import config_flow
from homeassistant.components.logi_circle.config_flow import (
DOMAIN,
AuthorizationFailed,
LogiCircleAuthCallbackView,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import AbortFlow, FlowResultType
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
class MockRequest:
"""Mock request passed to HomeAssistantView."""
def __init__(self, hass: HomeAssistant, query: dict[str, Any]) -> None:
"""Init request object."""
self.app = {KEY_HASS: hass}
self.query = query
def init_config_flow(hass: HomeAssistant) -> config_flow.LogiCircleFlowHandler:
"""Init a configuration flow."""
config_flow.register_flow_implementation(
hass,
DOMAIN,
client_id="id",
client_secret="secret",
api_key="123",
redirect_uri="http://example.com",
sensors=None,
)
flow = config_flow.LogiCircleFlowHandler()
flow._get_authorization_url = Mock(return_value="http://example.com")
flow.hass = hass
return flow
@pytest.fixture
def mock_logi_circle() -> Generator[MagicMock]:
"""Mock logi_circle."""
with patch(
"homeassistant.components.logi_circle.config_flow.LogiCircle"
) as logi_circle:
future = asyncio.Future()
future.set_result({"accountId": "testId"})
LogiCircle = logi_circle()
LogiCircle.authorize = AsyncMock(return_value=True)
LogiCircle.close = AsyncMock(return_value=True)
LogiCircle.account = future
LogiCircle.authorize_url = "http://authorize.url"
yield LogiCircle
@pytest.mark.usefixtures("mock_logi_circle")
async def test_step_import(hass: HomeAssistant) -> None:
"""Test that we trigger import when configuring with client."""
flow = init_config_flow(hass)
result = await flow.async_step_import()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
@pytest.mark.usefixtures("mock_logi_circle")
async def test_full_flow_implementation(hass: HomeAssistant) -> None:
"""Test registering an implementation and finishing flow works."""
config_flow.register_flow_implementation(
hass,
"test-other",
client_id=None,
client_secret=None,
api_key=None,
redirect_uri=None,
sensors=None,
)
flow = init_config_flow(hass)
result = await flow.async_step_user()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await flow.async_step_user({"flow_impl": "test-other"})
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["description_placeholders"] == {
"authorization_url": "http://example.com"
}
result = await flow.async_step_code("123ABC")
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Logi Circle ({})".format("testId")
async def test_we_reprompt_user_to_follow_link(hass: HomeAssistant) -> None:
"""Test we prompt user to follow link if previously prompted."""
flow = init_config_flow(hass)
result = await flow.async_step_auth("dummy")
assert result["errors"]["base"] == "follow_link"
async def test_abort_if_no_implementation_registered(hass: HomeAssistant) -> None:
"""Test we abort if no implementation is registered."""
flow = config_flow.LogiCircleFlowHandler()
flow.hass = hass
result = await flow.async_step_user()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "missing_configuration"
async def test_abort_if_already_setup(hass: HomeAssistant) -> None:
"""Test we abort if Logi Circle is already setup."""
flow = init_config_flow(hass)
MockConfigEntry(domain=config_flow.DOMAIN).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
with pytest.raises(AbortFlow):
result = await flow.async_step_code()
result = await flow.async_step_auth()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "external_setup"
@pytest.mark.parametrize(
("side_effect", "error"),
[
(TimeoutError, "authorize_url_timeout"),
(AuthorizationFailed, "invalid_auth"),
],
)
async def test_abort_if_authorize_fails(
hass: HomeAssistant,
mock_logi_circle: MagicMock,
side_effect: type[Exception],
error: str,
) -> None:
"""Test we abort if authorizing fails."""
flow = init_config_flow(hass)
mock_logi_circle.authorize.side_effect = side_effect
result = await flow.async_step_code("123ABC")
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "external_error"
result = await flow.async_step_auth()
assert result["errors"]["base"] == error
async def test_not_pick_implementation_if_only_one(hass: HomeAssistant) -> None:
"""Test we bypass picking implementation if we have one flow_imp."""
flow = init_config_flow(hass)
result = await flow.async_step_user()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
@pytest.mark.usefixtures("mock_logi_circle")
async def test_gen_auth_url(hass: HomeAssistant) -> None:
"""Test generating authorize URL from Logi Circle API."""
config_flow.register_flow_implementation(
hass,
"test-auth-url",
client_id="id",
client_secret="secret",
api_key="123",
redirect_uri="http://example.com",
sensors=None,
)
flow = config_flow.LogiCircleFlowHandler()
flow.hass = hass
flow.flow_impl = "test-auth-url"
await async_setup_component(hass, "http", {})
result = flow._get_authorization_url()
assert result == "http://authorize.url"
async def test_callback_view_rejects_missing_code(hass: HomeAssistant) -> None:
"""Test the auth callback view rejects requests with no code."""
view = LogiCircleAuthCallbackView()
resp = await view.get(MockRequest(hass, {}))
assert resp.status == HTTPStatus.BAD_REQUEST
async def test_callback_view_accepts_code(
hass: HomeAssistant, mock_logi_circle: MagicMock
) -> None:
"""Test the auth callback view handles requests with auth code."""
init_config_flow(hass)
view = LogiCircleAuthCallbackView()
resp = await view.get(MockRequest(hass, {"code": "456"}))
assert resp.status == HTTPStatus.OK
await hass.async_block_till_done()
mock_logi_circle.authorize.assert_called_with("456")

68
tests/components/logi_circle/test_init.py
View File

@ -1,68 +0,0 @@
"""Tests for the Logi Circle integration."""
import asyncio
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from homeassistant.components.logi_circle import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from tests.common import MockConfigEntry
@pytest.fixture(name="disable_platforms")
def disable_platforms_fixture() -> Generator[None]:
"""Disable logi_circle platforms."""
with patch("homeassistant.components.logi_circle.PLATFORMS", []):
yield
@pytest.fixture
def mock_logi_circle() -> Generator[MagicMock]:
"""Mock logi_circle."""
auth_provider_mock = Mock()
auth_provider_mock.close = AsyncMock()
auth_provider_mock.clear_authorization = AsyncMock()
with patch("homeassistant.components.logi_circle.LogiCircle") as logi_circle:
future = asyncio.Future()
future.set_result({"accountId": "testId"})
LogiCircle = logi_circle()
LogiCircle.auth_provider = auth_provider_mock
LogiCircle.synchronize_cameras = AsyncMock()
yield LogiCircle
@pytest.mark.usefixtures("disable_platforms", "mock_logi_circle")
async def test_repair_issue(
hass: HomeAssistant,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test the LogiCircle configuration entry loading/unloading handles the repair."""
config_entry = MockConfigEntry(
title="Example 1",
domain=DOMAIN,
data={
"api_key": "blah",
"client_id": "blah",
"client_secret": "blah",
"redirect_uri": "blah",
},
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert issue_registry.async_get_issue(DOMAIN, DOMAIN)
# Remove the entry
await hass.config_entries.async_remove(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.NOT_LOADED
assert issue_registry.async_get_issue(DOMAIN, DOMAIN) is None