Skip to content

Commit

Permalink
Further linting/formatting fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
@Snuffy2
Snuffy2 committed Dec 28, 2024
1 parent 2976c02 commit 5efed35
Show file tree
Hide file tree
Showing 12 changed files with 184 additions and 510 deletions.
35 changes: 9 additions & 26 deletions custom_components/opnsense/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -237,18 +237,12 @@ async def async_remove_config_entry_device(
"""Remove OPNsense Devices that aren't Device Tracker Devices and without any linked entities."""

if device_entry.via_device_id:
_LOGGER.error(
"Remove OPNsense Device Tracker Devices via the Integration Configuration"
)
_LOGGER.error("Remove OPNsense Device Tracker Devices via the Integration Configuration")
return False
entity_registry = er.async_get(hass)
for ent in er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
):
for ent in er.async_entries_for_config_entry(entity_registry, config_entry.entry_id):
if ent.device_id == device_entry.id:
_LOGGER.error(
"Cannot remove OPNsense Devices with linked entities at this time"
)
_LOGGER.error("Cannot remove OPNsense Devices with linked entities at this time")
return False
return True

Expand Down Expand Up @@ -318,8 +312,7 @@ async def _migrate_2_to_3(hass: HomeAssistant, config_entry: ConfigEntry) -> boo
is_main_dev: bool = any(t[0] == "opnsense" for t in dev.identifiers)
if is_main_dev:
new_identifiers = {
(t[0], new_device_unique_id) if t[0] == "opnsense" else t
for t in dev.identifiers
(t[0], new_device_unique_id) if t[0] == "opnsense" else t for t in dev.identifiers
}
_LOGGER.debug(
"[migrate_2_to_3] dev.identifiers: %s, new_identifiers: %s",
Expand All @@ -339,9 +332,7 @@ async def _migrate_2_to_3(hass: HomeAssistant, config_entry: ConfigEntry) -> boo
e,
)

for ent in er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
):
for ent in er.async_entries_for_config_entry(entity_registry, config_entry.entry_id):
# _LOGGER.debug(f"[migrate_2_to_3] ent: {ent}")
platform = ent.entity_id.split(".")[0]
try:
Expand Down Expand Up @@ -418,9 +409,7 @@ async def _migrate_3_to_4(hass: HomeAssistant, config_entry: ConfigEntry) -> boo
)
telemetry = await client.get_telemetry()

for ent in er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
):
for ent in er.async_entries_for_config_entry(entity_registry, config_entry.entry_id):
platform = ent.entity_id.split(".")[0]
if platform == Platform.SENSOR:
# _LOGGER.debug(f"[migrate_3_to_4] ent: {ent}")
Expand All @@ -429,15 +418,11 @@ async def _migrate_3_to_4(hass: HomeAssistant, config_entry: ConfigEntry) -> boo
"_telemetry_interface_", "_interface_"
)
elif "_telemetry_gateway_" in ent.unique_id:
new_unique_id = ent.unique_id.replace(
"_telemetry_gateway_", "_gateway_"
)
new_unique_id = ent.unique_id.replace("_telemetry_gateway_", "_gateway_")
elif "_connected_client_count" in ent.unique_id:
try:
entity_registry.async_remove(ent.entity_id)
_LOGGER.debug(
"[migrate_3_to_4] removed_entity_id: %s", ent.entity_id
)
_LOGGER.debug("[migrate_3_to_4] removed_entity_id: %s", ent.entity_id)
except (KeyError, ValueError) as e:
_LOGGER.error(
"Error removing entity: %s. %s: %s",
Expand All @@ -447,9 +432,7 @@ async def _migrate_3_to_4(hass: HomeAssistant, config_entry: ConfigEntry) -> boo
)
continue
elif "_telemetry_openvpn_" in ent.unique_id:
new_unique_id = ent.unique_id.replace(
"_telemetry_openvpn_", "_openvpn_"
)
new_unique_id = ent.unique_id.replace("_telemetry_openvpn_", "_openvpn_")
elif "_telemetry_filesystems_" in ent.unique_id:
new_unique_id = None
for filesystem in telemetry.get("filesystems", []):
Expand Down
10 changes: 4 additions & 6 deletions custom_components/opnsense/binary_sensor.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,9 @@ async def async_setup_entry(
) -> None:
"""Set up the OPNsense binary sensors."""

coordinator: OPNsenseDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
][COORDINATOR]
coordinator: OPNsenseDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id][
COORDINATOR
]

entities: list = []
entities.extend(
Expand Down Expand Up @@ -75,9 +75,7 @@ def __init__(
) -> None:
"""Initialize OPNsense Binary Sensor entity."""
name_suffix: str | None = (
entity_description.name
if isinstance(entity_description.name, str)
else None
entity_description.name if isinstance(entity_description.name, str) else None
)
unique_id_suffix: str | None = (
entity_description.key if isinstance(entity_description.key, str) else None
Expand Down
92 changes: 25 additions & 67 deletions custom_components/opnsense/config_flow.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,7 @@
import awesomeversion
import voluptuous as vol

from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
from homeassistant.const import (
CONF_NAME,
CONF_PASSWORD,
Expand Down Expand Up @@ -113,9 +108,7 @@ async def validate_input(
message=f"Missing Device Unique ID Error. {err.__class__.__qualname__}: {err}",
)
except PluginMissing:
_log_and_set_error(
errors=errors, key="plugin_missing", message="OPNsense Plugin Missing"
)
_log_and_set_error(errors=errors, key="plugin_missing", message="OPNsense Plugin Missing")
except (aiohttp.InvalidURL, InvalidURL) as err:
_log_and_set_error(
errors=errors,
Expand Down Expand Up @@ -158,10 +151,7 @@ async def validate_input(
)
except xmlrpc.client.ProtocolError as err:
error_message = str(err)
if (
"307 Temporary Redirect" in error_message
or "301 Moved Permanently" in error_message
):
if "307 Temporary Redirect" in error_message or "301 Moved Permanently" in error_message:
errors["base"] = "url_redirect"
else:
errors["base"] = "cannot_connect"
Expand Down Expand Up @@ -218,9 +208,7 @@ async def _clean_and_parse_url(user_input: MutableMapping[str, Any]) -> None:
_LOGGER.debug("[config_flow] Cleaned URL: %s", user_input[CONF_URL])


async def _get_client(
user_input: MutableMapping[str, Any], hass: HomeAssistant
) -> OPNsenseClient:
async def _get_client(user_input: MutableMapping[str, Any], hass: HomeAssistant) -> OPNsenseClient:
"""Create and return the OPNsense client."""
return OPNsenseClient(
url=user_input[CONF_URL],
Expand All @@ -244,18 +232,14 @@ def _validate_firmware_version(firmware_version: str) -> None:
raise BelowMinFirmware


async def _handle_user_input(
user_input: MutableMapping[str, Any], hass: HomeAssistant
) -> None:
async def _handle_user_input(user_input: MutableMapping[str, Any], hass: HomeAssistant) -> None:
"""Handle and validate the user input."""
await _clean_and_parse_url(user_input)

client: OPNsenseClient = await _get_client(user_input, hass)

user_input[CONF_FIRMWARE_VERSION] = await client.get_host_firmware_version()
_LOGGER.debug(
"[config_flow] Firmware Version: %s", user_input[CONF_FIRMWARE_VERSION]
)
_LOGGER.debug("[config_flow] Firmware Version: %s", user_input[CONF_FIRMWARE_VERSION])

try:
_validate_firmware_version(user_input[CONF_FIRMWARE_VERSION])
Expand All @@ -272,17 +256,13 @@ async def _handle_user_input(
user_input[CONF_NAME] = system_info.get("name") or "OPNsense"

user_input[CONF_DEVICE_UNIQUE_ID] = await client.get_device_unique_id()
_LOGGER.debug(
"[config_flow] Device Unique ID: %s", user_input[CONF_DEVICE_UNIQUE_ID]
)
_LOGGER.debug("[config_flow] Device Unique ID: %s", user_input[CONF_DEVICE_UNIQUE_ID])

if not user_input.get(CONF_DEVICE_UNIQUE_ID):
raise MissingDeviceUniqueID


def _log_and_set_error(
errors: MutableMapping[str, Any], key: str, message: str
) -> None:
def _log_and_set_error(errors: MutableMapping[str, Any], key: str, message: str) -> None:
"""Log the error and set it in the errors dictionary."""
_LOGGER.error(message)
errors["base"] = key
Expand All @@ -303,9 +283,7 @@ async def async_step_user(
errors: MutableMapping[str, Any] = {}
firmware: str = "Unknown"
if user_input is not None:
errors = await validate_input(
hass=self.hass, user_input=user_input, errors=errors
)
errors = await validate_input(hass=self.hass, user_input=user_input, errors=errors)
firmware = user_input.get(CONF_FIRMWARE_VERSION, "Unknown")
if not errors:
# https://developers.home-assistant.io/docs/config_entries_config_flow_handler#unique-ids
Expand All @@ -327,9 +305,7 @@ async def async_step_user(
user_input = {}
schema = vol.Schema(
{
vol.Required(
CONF_URL, default=user_input.get(CONF_URL, "https://")
): str,
vol.Required(CONF_URL, default=user_input.get(CONF_URL, "https://")): str,
vol.Optional(
CONF_VERIFY_SSL,
default=user_input.get(CONF_VERIFY_SSL, DEFAULT_VERIFY_SSL),
Expand All @@ -338,9 +314,7 @@ async def async_step_user(
CONF_USERNAME,
default=user_input.get(CONF_USERNAME, ""),
): str,
vol.Required(
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
): str,
vol.Required(CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")): str,
vol.Optional(CONF_NAME, default=user_input.get(CONF_NAME, "")): str,
}
)
Expand All @@ -365,9 +339,7 @@ async def async_step_reconfigure(
firmware: str = "Unknown"
if user_input is not None:
user_input[CONF_NAME] = prev_data.get(CONF_NAME, "")
errors = await validate_input(
hass=self.hass, user_input=user_input, errors=errors
)
errors = await validate_input(hass=self.hass, user_input=user_input, errors=errors)
firmware = user_input.get(CONF_FIRMWARE_VERSION, "Unknown")
if not errors:
# https://developers.home-assistant.io/docs/config_entries_config_flow_handler#unique-ids
Expand All @@ -391,9 +363,7 @@ async def async_step_reconfigure(
{
vol.Required(
CONF_URL,
default=user_input.get(
CONF_URL, prev_data.get(CONF_URL, "https://")
),
default=user_input.get(CONF_URL, prev_data.get(CONF_URL, "https://")),
): str,
vol.Optional(
CONF_VERIFY_SSL,
Expand All @@ -404,15 +374,11 @@ async def async_step_reconfigure(
): bool,
vol.Required(
CONF_USERNAME,
default=user_input.get(
CONF_USERNAME, prev_data.get(CONF_USERNAME, "")
),
default=user_input.get(CONF_USERNAME, prev_data.get(CONF_USERNAME, "")),
): str,
vol.Required(
CONF_PASSWORD,
default=user_input.get(
CONF_PASSWORD, prev_data.get(CONF_PASSWORD, "")
),
default=user_input.get(CONF_PASSWORD, prev_data.get(CONF_PASSWORD, "")),
): str,
}
)
Expand Down Expand Up @@ -455,9 +421,7 @@ async def async_step_init(self, user_input=None) -> ConfigFlowResult:
return await self.async_step_device_tracker()
return self.async_create_entry(title="", data=user_input)

scan_interval = self.config_entry.options.get(
CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
)
scan_interval = self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
device_tracker_enabled = self.config_entry.options.get(
CONF_DEVICE_TRACKER_ENABLED, DEFAULT_DEVICE_TRACKER_ENABLED
)
Expand All @@ -472,9 +436,7 @@ async def async_step_init(self, user_input=None) -> ConfigFlowResult:
vol.Optional(CONF_SCAN_INTERVAL, default=scan_interval): vol.All(
vol.Coerce(int), vol.Clamp(min=10, max=300)
),
vol.Optional(
CONF_DEVICE_TRACKER_ENABLED, default=device_tracker_enabled
): bool,
vol.Optional(CONF_DEVICE_TRACKER_ENABLED, default=device_tracker_enabled): bool,
vol.Optional(
CONF_DEVICE_TRACKER_SCAN_INTERVAL, default=device_tracker_scan_interval
): vol.All(vol.Coerce(int), vol.Clamp(min=30, max=300)),
Expand All @@ -490,9 +452,7 @@ async def async_step_device_tracker(self, user_input=None) -> ConfigFlowResult:
url = self.config_entry.data[CONF_URL].strip()
username: str = self.config_entry.data[CONF_USERNAME]
password: str = self.config_entry.data[CONF_PASSWORD]
verify_ssl: bool = self.config_entry.data.get(
CONF_VERIFY_SSL, DEFAULT_VERIFY_SSL
)
verify_ssl: bool = self.config_entry.data.get(CONF_VERIFY_SSL, DEFAULT_VERIFY_SSL)
client = OPNsenseClient(
url=url,
username=username,
Expand Down Expand Up @@ -540,9 +500,9 @@ async def async_step_device_tracker(self, user_input=None) -> ConfigFlowResult:
step_id="device_tracker",
data_schema=vol.Schema(
{
vol.Optional(
CONF_DEVICES, default=selected_devices
): cv.multi_select(dict(sorted_entries)),
vol.Optional(CONF_DEVICES, default=selected_devices): cv.multi_select(
dict(sorted_entries)
),
vol.Optional(CONF_MANUAL_DEVICES): selector.TextSelector(
selector.TextSelectorConfig()
),
Expand All @@ -552,19 +512,17 @@ async def async_step_device_tracker(self, user_input=None) -> ConfigFlowResult:
if user_input:
_LOGGER.debug("[options_flow device_tracker] user_input: %s", user_input)
macs: list = []
if isinstance(
user_input.get(CONF_MANUAL_DEVICES, None), str
) and user_input.get(CONF_MANUAL_DEVICES, None):
if isinstance(user_input.get(CONF_MANUAL_DEVICES, None), str) and user_input.get(
CONF_MANUAL_DEVICES, None
):
for item in user_input.get(CONF_MANUAL_DEVICES).split(","):
if not isinstance(item, str) or not item:
continue
item = item.strip()
if is_valid_mac_address(item):
macs.append(item)
_LOGGER.debug("[async_step_device_tracker] Manual Devices: %s", macs)
_LOGGER.debug(
"[async_step_device_tracker] Devices: %s", user_input.get(CONF_DEVICES)
)
_LOGGER.debug("[async_step_device_tracker] Devices: %s", user_input.get(CONF_DEVICES))
self.new_options[CONF_DEVICES] = user_input.get(CONF_DEVICES) + macs
return self.async_create_entry(title="", data=self.new_options)

Expand Down
4 changes: 1 addition & 3 deletions custom_components/opnsense/coordinator.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -155,9 +155,7 @@ async def _async_update_data(self):

self._state.update(await self._get_states(categories))
if self._state.get("device_unique_id") is None:
_LOGGER.warning(
"Coordinator failed to confirm OPNsense Router Unique ID. Will retry"
)
_LOGGER.warning("Coordinator failed to confirm OPNsense Router Unique ID. Will retry")
return {}
if self._state.get("device_unique_id") != self._device_unique_id:
_LOGGER.debug(
Expand Down
Loading

0 comments on commit 5efed35

Please sign in to comment.