From 934a6088ca9f324032b11225a031bf04f3656c1f Mon Sep 17 00:00:00 2001 From: Ashley Gittins Date: Tue, 2 Apr 2024 20:16:13 +0000 Subject: [PATCH 1/2] linting: initial round of type checking tightening. --- custom_components/bermuda/__init__.py | 79 +++++++++++++++------------ custom_components/bermuda/const.py | 1 + 2 files changed, 45 insertions(+), 35 deletions(-) diff --git a/custom_components/bermuda/__init__.py b/custom_components/bermuda/__init__.py index b633d3b..88f28e4 100644 --- a/custom_components/bermuda/__init__.py +++ b/custom_components/bermuda/__init__.py @@ -62,6 +62,7 @@ from .const import DEFAULT_SMOOTHING_SAMPLES from .const import DEFAULT_UPDATE_INTERVAL from .const import DEVICE_TRACKER +from .const import DISTANCE_INFINITE from .const import DISTANCE_TIMEOUT from .const import DOMAIN from .const import DOMAIN_PRIVATE_BLE_DEVICE @@ -165,10 +166,12 @@ def __init__( self.area_id: str = area_id self.parent_device = device_address - self.stamp: float = 0 - self.new_stamp = None # Set when a new advert is loaded from update + self.stamp: float | None = 0 + self.new_stamp: float | None = ( + None # Set when a new advert is loaded from update + ) self.hist_stamp = [] - self.rssi: float = None + self.rssi: float | None = None self.hist_rssi = [] self.hist_distance = [] self.hist_distance_by_interval = [] # updated per-interval @@ -177,9 +180,10 @@ def __init__( self.stale_update_count = ( 0 # How many times we did an update but no new stamps were found. ) - self.tx_power: float = None - self.rssi_distance: float = None - self.rssi_distance_raw: float = None + self.tx_power: float | None = None + self.rssi_distance: float | None = None + self.rssi_distance_raw: float | None = None + self.adverts: dict[str, bytes] = {} # Just pass the rest on to update... self.update_advertisement(device_address, scandata, area_id, options) @@ -202,7 +206,7 @@ def update_advertisement( # In case the scanner has changed it's details since startup: self.name: str = scandata.scanner.name self.area_id: str = area_id - new_stamp = None + new_stamp: float | None = None # Only remote scanners log timestamps here (local usb adaptors do not), if hasattr(scandata.scanner, "_discovered_device_timestamps"): @@ -252,9 +256,9 @@ def update_advertisement( if len(self.hist_stamp) == 0 or new_stamp is not None: # this is the first entry or a new one... - self.rssi: float = scandata.advertisement.rssi + self.rssi = scandata.advertisement.rssi self.hist_rssi.insert(0, self.rssi) - self.rssi_distance_raw: float = rssi_to_metres( + self.rssi_distance_raw = rssi_to_metres( self.rssi, options.get(CONF_REF_POWER), options.get(CONF_ATTENUATION), @@ -270,7 +274,11 @@ def update_advertisement( # of the true inter-packet interval. For stamps from local bluetooth # adaptors (usb dongles) it reflects "Which update cycle last saw a # different rssi", which will be a multiple of our update interval. - self.hist_interval.insert(0, new_stamp - self.stamp) + if new_stamp is not None and self.stamp is not None: + _interval = new_stamp - self.stamp + else: + _interval = None + self.hist_interval.insert(0, _interval) self.stamp = new_stamp self.hist_stamp.insert(0, self.stamp) @@ -292,8 +300,9 @@ def update_advertisement( device_address, scandata.advertisement.tx_power, ) - self.tx_power: float = scandata.advertisement.tx_power - self.adverts: dict[str, bytes] = scandata.advertisement.service_data.items() + self.tx_power = scandata.advertisement.tx_power + for ad_str, ad_bytes in scandata.advertisement.service_data.items(): + self.adverts[ad_str] = ad_bytes self.options = options self.new_stamp = new_stamp @@ -435,9 +444,9 @@ def calculate_data(self): # slope angle (other than increasing bucket count) might be # helpful, but probably dependent on use-case. # - dist_total = 0 - dist_count = 0 - local_min = self.rssi_distance_raw + dist_total: float = 0 + dist_count: int = 0 + local_min: float = self.rssi_distance_raw or DISTANCE_INFINITE for i, distance in enumerate(self.hist_distance_by_interval): if distance <= local_min: dist_total += distance @@ -492,35 +501,35 @@ class BermudaDevice(dict): def __init__(self, address, options): """Initial (empty) data""" - self.name: str = None - self.local_name: str = None - self.prefname: str = None # "preferred" name - ideally local_name + self.name: str | None = None + self.local_name: str | None = None + self.prefname: str | None = None # "preferred" name - ideally local_name self.address: str = address self.options = options - self.unique_id: str = None # mac address formatted. + self.unique_id: str | None = None # mac address formatted. self.mac_is_random: bool = False - self.area_id: str = None - self.area_name: str = None - self.area_distance: float = None # how far this dev is from that area - self.area_rssi: float = None # rssi from closest scanner - self.area_scanner: str = None # name of closest scanner + self.area_id: str | None = None + self.area_name: str | None = None + self.area_distance: float | None = None # how far this dev is from that area + self.area_rssi: float | None = None # rssi from closest scanner + self.area_scanner: str | None = None # name of closest scanner self.zone: str = STATE_UNAVAILABLE # STATE_HOME or STATE_NOT_HOME - self.manufacturer: str = None + self.manufacturer: str | None = None self.connectable: bool = False self.is_scanner: bool = False - self.beacon_type: bool = BEACON_NOT_A_BEACON + self.beacon_type = BEACON_NOT_A_BEACON self.beacon_sources = ( [] ) # list of MAC addresses that have advertised this beacon - self.beacon_unique_id: str = ( + self.beacon_unique_id: str | None = ( None # combined uuid_major_minor for *really* unique id ) - self.beacon_uuid: str = None - self.beacon_major: str = None - self.beacon_minor: str = None - self.beacon_power: float = None + self.beacon_uuid: str | None = None + self.beacon_major: str | None = None + self.beacon_minor: str | None = None + self.beacon_power: float | None = None - self.entry_id: str = None # used for scanner devices + self.entry_id: str | None = None # used for scanner devices self.create_sensor: bool = False # Create/update a sensor for this device self.create_sensor_done: bool = False # Sensor should now exist self.create_tracker_done: bool = False # device_tracker should now exist @@ -653,7 +662,7 @@ def __init__( # Track the list of Private BLE devices, noting their entity id # and current "last address". - self.pb_state_sources: dict[str, str] = {} + self.pb_state_sources: dict[str, str | None] = {} @callback def handle_state_changes(ev: Event): @@ -806,7 +815,7 @@ def device_tracker_created(self, address): else: _LOGGER.warning("Very odd, we got sensor_created for non-tracked device") - def _get_device(self, address: str) -> BermudaDevice: + def _get_device(self, address: str) -> BermudaDevice | None: """Search for a device entry based on mac address""" mac = format_mac(address).lower() # format_mac tries to return a lower-cased, colon-separated mac address. @@ -1240,7 +1249,7 @@ def _refresh_areas_by_min_distance(self): def _refresh_area_by_min_distance(self, device: BermudaDevice): """Very basic Area setting by finding closest beacon to a given device""" assert device.is_scanner is not True - closest_scanner: BermudaDeviceScanner = None + closest_scanner: BermudaDeviceScanner | None = None for scanner in device.scanners.values(): # Check each scanner and keep note of the closest one based on rssi_distance. diff --git a/custom_components/bermuda/const.py b/custom_components/bermuda/const.py index 6ed2400..046c809 100644 --- a/custom_components/bermuda/const.py +++ b/custom_components/bermuda/const.py @@ -40,6 +40,7 @@ DISTANCE_TIMEOUT = 30 # seconds to wait before marking a sensor distance measurement # as unknown/none/stale/away. Separate from device_tracker. +DISTANCE_INFINITE = 999 # arbitrary distance for infinite/unknown rssi range UPDATE_INTERVAL = 1.05 # Seconds between bluetooth data processing cycles # Note: this is separate from the CONF_UPDATE_INTERVAL which allows the From 80407875216fe286433c3afb0d7d4221f1a5def7 Mon Sep 17 00:00:00 2001 From: Ashley Gittins Date: Tue, 2 Apr 2024 20:34:29 +0000 Subject: [PATCH 2/2] chore: Remove deprecated async_add_job --- custom_components/bermuda/__init__.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/custom_components/bermuda/__init__.py b/custom_components/bermuda/__init__.py index 88f28e4..57329e1 100644 --- a/custom_components/bermuda/__init__.py +++ b/custom_components/bermuda/__init__.py @@ -107,9 +107,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry): for platform in PLATFORMS: coordinator.platforms.append(platform) - hass.async_add_job( - hass.config_entries.async_forward_entry_setup(entry, platform) - ) + await hass.config_entries.async_forward_entry_setup(entry, platform) entry.add_update_listener(async_reload_entry) return True