-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
2 changed files
with
342 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,236 @@ | ||
| """Helper for Duplicati integration.""" | ||
|
|
||
| import logging | ||
|
|
||
| import aiohttp | ||
| from homeassistant import config_entries | ||
| from homeassistant.const import ( | ||
| CONF_PASSWORD, | ||
| CONF_SCAN_INTERVAL, | ||
| CONF_URL, | ||
| CONF_VERIFY_SSL, | ||
| Platform, | ||
| ) | ||
| from homeassistant.core import HomeAssistant | ||
| from homeassistant.exceptions import HomeAssistantError | ||
| from homeassistant.helpers import device_registry as dr | ||
| from homeassistant.helpers.device_registry import DeviceEntry | ||
| from homeassistant.helpers.entity_platform import EntityPlatform | ||
| from homeassistant.helpers.selector import ( | ||
| SelectOptionDict, | ||
| ) | ||
|
|
||
| from .api import ApiResponseError, CannotConnect, DuplicatiBackendAPI | ||
| from .button import create_backup_buttons | ||
| from .const import DEFAULT_SCAN_INTERVAL, DOMAIN | ||
| from .sensor import create_backup_sensors | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class DuplicatiHelper: | ||
| """Helper class for the Duplicati integration.""" | ||
|
|
||
| def __init__( | ||
| self, hass: HomeAssistant, config_entry: config_entries.ConfigEntry | ||
| ) -> None: | ||
| """Initialize options flow.""" | ||
| self.config_entry = config_entry | ||
| self.api = self.__create_api() | ||
| self.hass = hass | ||
| self.sensor_platform = self.get_platform(Platform.SENSOR) | ||
| self.button_platform = self.get_platform(Platform.BUTTON) | ||
|
|
||
| async def __async_get_backups(self) -> dict: | ||
| """Get available backups.""" | ||
| try: | ||
| response = await self.api.list_backups() | ||
| if "Error" in response: | ||
| raise ApiResponseError(response["Error"]) | ||
| # Check if backups are available | ||
| if len(response) == 0: | ||
| raise ValueError( | ||
| f"No backups found for server '{self.api.get_api_host()}'" | ||
| ) | ||
| except aiohttp.ClientConnectionError as e: | ||
| raise CannotConnect(str(e)) from e | ||
| return response | ||
|
|
||
| def __create_api(self) -> DuplicatiBackendAPI: | ||
| """Create an instance of DuplicatiBackendAPI.""" | ||
| base_url = self.config_entry.data[CONF_URL] | ||
| password = self.config_entry.data.get(CONF_PASSWORD) | ||
| verify_ssl = self.config_entry.data[CONF_VERIFY_SSL] | ||
| # Create an instance of DuplicatiBackendAPI | ||
| return DuplicatiBackendAPI(base_url, verify_ssl, password) | ||
|
|
||
| def get_api(self) -> DuplicatiBackendAPI: | ||
| """Return the DuplicatiBackendAPI instance.""" | ||
| return self.api | ||
|
|
||
| def get_backup_id_from_serial_number(self, serial_number: str | None) -> str | None: | ||
| """Get backup ID from serial number.""" | ||
| if not isinstance(serial_number, str): | ||
| return None | ||
| if "/" in serial_number: | ||
| return serial_number.split("/", 1)[1] | ||
| return None | ||
|
|
||
| def get_platform(self, type: str) -> EntityPlatform: | ||
| """Get the EntityPlatform for the given type.""" | ||
| platforms = self.hass.data["entity_platform"][DOMAIN] | ||
| for platform in platforms: | ||
| if ( | ||
| platform.config_entry.entry_id == self.config_entry.entry_id | ||
| and platform.domain == type | ||
| ): | ||
| return platform | ||
| _LOGGER.error( | ||
| "No platform found for config entry %s", | ||
| self.config_entry.entry_id, | ||
| ) | ||
| raise HomeAssistantError( | ||
| "No platform found for config entry %s", | ||
| self.config_entry.entry_id, | ||
| ) | ||
|
|
||
| async def async_get_available_backups(self) -> dict[str, str]: | ||
| """Return a dictionary of available backup names.""" | ||
| backups = {} | ||
| for backup in await self.__async_get_backups(): | ||
| backup_id = backup["Backup"]["ID"] | ||
| backup_name = backup["Backup"]["Name"] | ||
| backups[backup_id] = backup_name | ||
| return backups | ||
|
|
||
| def get_backup_select_options_list( | ||
| self, backups: dict[str, str] | ||
| ) -> list[SelectOptionDict]: | ||
| """Return a dictionary of available backup names.""" | ||
| return [ | ||
| SelectOptionDict( | ||
| label=value, | ||
| value=key, | ||
| ) | ||
| for key, value in backups.items() | ||
| ] | ||
|
|
||
| def get_integration_device_entries(self) -> list[DeviceEntry]: | ||
| """Get device entries for the config entry.""" | ||
| device_entries = [] | ||
| device_registry = self.hass.data[dr.DATA_REGISTRY] | ||
| for device_entry in device_registry.devices.data.values(): | ||
| for config_entry in device_entry.config_entries: | ||
| if config_entry == self.config_entry.entry_id: | ||
| device_entries.append(device_entry) | ||
| break | ||
| if len(device_entries) == 0: | ||
| _LOGGER.error( | ||
| "No devices found for config entry %s", | ||
| self.config_entry.entry_id, | ||
| ) | ||
| return device_entries | ||
|
|
||
| async def async_remove_backup_from_hass( | ||
| self, device: DeviceEntry, backup_id: str | ||
| ) -> bool: | ||
| """Remove a backup from Home Assistant.""" | ||
| device_registry = self.hass.data[dr.DATA_REGISTRY] | ||
| # for device in device_registry.devices.data.values(): | ||
| for config_entry in device.config_entries: | ||
| if ( | ||
| config_entry == self.config_entry.entry_id | ||
| and self.get_backup_id_from_serial_number(device.serial_number) | ||
| == backup_id | ||
| ): | ||
| if ( | ||
| backup_id | ||
| in self.hass.data[DOMAIN][self.config_entry.entry_id][ | ||
| "coordinators" | ||
| ] | ||
| ): | ||
| # Unregister coordinator | ||
| coordinator = self.hass.data[DOMAIN][self.config_entry.entry_id][ | ||
| "coordinators" | ||
| ][backup_id] | ||
| host = self.hass.data[DOMAIN][self.config_entry.entry_id]["host"] | ||
| service = self.hass.data[DOMAIN][host]["service"] | ||
| service.unregister_coordinator(coordinator) | ||
| # Remove coordinator | ||
| self.hass.data[DOMAIN][self.config_entry.entry_id][ | ||
| "coordinators" | ||
| ].pop(backup_id) | ||
| else: | ||
| _LOGGER.debug("Coordinator for resource %s not found", backup_id) | ||
| # Remove device including its entities | ||
| device_registry.async_remove_device(device.id) | ||
| _LOGGER.debug("Removed device registry entry: %s.%s", DOMAIN, backup_id) | ||
| return True | ||
| return False | ||
|
|
||
| async def async_add_backup_to_hass(self, backup_id: str, backup_name: str) -> bool: | ||
| """Add a backup to Home Assistant.""" | ||
| try: | ||
| # Get device registry | ||
| device_registry = self.hass.data[dr.DATA_REGISTRY] | ||
| # Create coordinator | ||
| from .coordinator import DuplicatiDataUpdateCoordinator | ||
|
|
||
| coordinator = DuplicatiDataUpdateCoordinator( | ||
| self.hass, | ||
| api=self.api, | ||
| backup_id=backup_id, | ||
| update_interval=int( | ||
| self.config_entry.data.get( | ||
| CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL | ||
| ) | ||
| ), | ||
| ) | ||
| # Create sensors | ||
| sensors = create_backup_sensors( | ||
| self.hass, | ||
| self.config_entry, | ||
| {"id": backup_id, "name": backup_name}, | ||
| coordinator, | ||
| ) | ||
| # Create buttons | ||
| buttons = create_backup_buttons( | ||
| self.hass, self.config_entry, {"id": backup_id, "name": backup_name} | ||
| ) | ||
| # Register device | ||
| device_entry = device_registry.async_get_or_create( | ||
| config_entry_id=self.config_entry.entry_id, | ||
| name=sensors[0].device_info["name"], | ||
| model=sensors[0].device_info["model"], | ||
| manufacturer=sensors[0].device_info["manufacturer"], | ||
| sw_version=sensors[0].device_info["sw_version"], | ||
| identifiers=sensors[0].device_info["identifiers"], | ||
| entry_type=sensors[0].device_info["entry_type"], | ||
| ) | ||
| # Link sensors to device | ||
| for sensor in sensors: | ||
| sensor.device_entry = device_entry | ||
| # Link buttons to device | ||
| for button in buttons: | ||
| button.device_entry = device_entry | ||
| # Add sensors to hass | ||
| await self.sensor_platform.async_add_entities(sensors) | ||
| # Add buttons to hass | ||
| await self.button_platform.async_add_entities(buttons) | ||
| # Add coordinator to config entry | ||
| self.hass.data[DOMAIN][self.config_entry.entry_id]["coordinators"][ | ||
| backup_id | ||
| ] = coordinator | ||
| # Register coordinator | ||
| host = self.hass.data[DOMAIN][self.config_entry.entry_id]["host"] | ||
| service = self.hass.data[DOMAIN][host]["service"] | ||
| service.register_coordinator(coordinator) | ||
| # Add backup to config entry | ||
| self.hass.data[DOMAIN][self.config_entry.entry_id]["backups"][backup_id] = ( | ||
| backup_name | ||
| ) | ||
| except Exception as e: # noqa: BLE001 | ||
| _LOGGER.error("Error adding backup to Home Assistant: %s", e) | ||
| return False | ||
| else: | ||
| return True |