import math from abc import ABC, abstractmethod from notifications_utils.formatters import formatted_list from notifications_utils.polygons import Polygons from notifications_utils.serialised_model import SerialisedModelCollection from rtreelib import Rect from werkzeug.utils import cached_property from .populations import CITY_OF_LONDON from .repo import BroadcastAreasRepository, rtree_index class SortableMixin: def __repr__(self): return f'{self.__class__.__name__}(<{self.id}>)' def __lt__(self, other): # Implementing __lt__ means any classes inheriting from this # method are sortable return self.name < other.name def __eq__(self, other): return self.id == other.id def __hash__(self): return hash(self.id) class GetItemByIdMixin: def get(self, id): for item in self: if item.id == id: return item raise KeyError(id) class BaseBroadcastArea(ABC): @property @abstractmethod def simple_polygons(self): pass @property @abstractmethod def polygons(self): pass @property @abstractmethod def count_of_phones(self): pass @cached_property def simple_polygons_with_bleed(self): return self.simple_polygons.bleed_by(self.estimated_bleed_in_degrees) @cached_property def phone_density(self): if not self.polygons.estimated_area: return 0 return self.count_of_phones / self.polygons.estimated_area @property def estimated_bleed_in_m(self): ''' Estimates the amount of bleed based on the population of an area. Higher density areas tend to have short range masts, so the bleed is low (down to 500m). Lower density areas have longer range masts, so the typical bleed will be high (up to 5,000m). ''' if self.phone_density < 1: return Polygons.approx_bleed_in_degrees * Polygons.approx_metres_to_degree estimated_bleed = 5_900 - (math.log(self.phone_density, 10) * 1_250) return max(500, min(estimated_bleed, 5000)) @property def estimated_bleed_in_degrees(self): return self.estimated_bleed_in_m / Polygons.approx_metres_to_degree class BroadcastArea(BaseBroadcastArea, SortableMixin): def __init__(self, row): self.id, self.name, self._count_of_phones, self.library_id = row @cached_property def is_lower_tier_local_authority(self): return self.id.startswith('lad20-') and self.parent @cached_property def is_electoral_ward(self): return self.id.startswith('wd20-') @classmethod def from_row_with_simple_polygons(cls, row): instance = cls(row[:4]) instance.simple_polygons = Polygons(row[4]) return instance @cached_property def polygons(self): return Polygons( BroadcastAreasRepository().get_polygons_for_area(self.id) ) @cached_property def simple_polygons(self): return Polygons( BroadcastAreasRepository().get_simple_polygons_for_area(self.id) ) @cached_property def sub_areas(self): return [ BroadcastArea(row) for row in BroadcastAreasRepository().get_all_areas_for_group(self.id) ] @property def count_of_phones(self): if self.id.endswith(CITY_OF_LONDON.WARDS): return CITY_OF_LONDON.DAYTIME_POPULATION * ( self.polygons.estimated_area / CITY_OF_LONDON.AREA_SQUARE_MILES ) if self.sub_areas: return sum(area.count_of_phones for area in self.sub_areas) # TODO: remove the `or 0` once missing data is fixed, see # https://www.pivotaltracker.com/story/show/174837293 return self._count_of_phones or 0 @cached_property def ancestors(self): return list(self._ancestors_iterator) @cached_property def parent(self): return next(iter(self.ancestors), None) @property def _ancestors_iterator(self): id = self.id while True: parent = BroadcastAreasRepository().get_parent_for_area(id) if not parent: return parent_broadcast_area = BroadcastArea(parent) yield parent_broadcast_area id = parent_broadcast_area.id class CustomBroadcastArea(BaseBroadcastArea): def __init__(self, *, name, polygons=None): self.name = name self._polygons = polygons or [] @classmethod def from_polygon_objects(cls, polygon_objects): return cls(name=None, polygons=polygon_objects.as_coordinate_pairs_lat_long) @property def polygons(self): return Polygons( # Polygons in the DB are stored with the coordinate pair # order flipped – this flips them back again Polygons(self._polygons).as_coordinate_pairs_lat_long ) simple_polygons = polygons @cached_property def overlapping_electoral_wards(self): return [ area for area in self.nearby_electoral_wards if area.simple_polygons.intersects(self.polygons) ] @cached_property def nearby_electoral_wards(self): if not self.polygons: return [] return broadcast_area_libraries.get_areas_with_simple_polygons([ # We only index electoral wards in the RTree overlap.data for overlap in rtree_index.query( Rect(*self.polygons.bounds) ) ]) @cached_property def count_of_phones(self): return sum( area.simple_polygons.ratio_of_intersection_with(self.polygons) * area.count_of_phones for area in self.nearby_electoral_wards ) class CustomBroadcastAreas(SerialisedModelCollection): model = CustomBroadcastArea def __init__(self, *, area_ids, polygons): self.items = area_ids self._polygons = polygons def __getitem__(self, index): return self.model( name=self.items[index], polygons=self._polygons if index == 0 else None, ) class BroadcastAreaLibrary(SerialisedModelCollection, SortableMixin, GetItemByIdMixin): model = BroadcastArea def __init__(self, row): id, name, name_singular, is_group = row self.id = id self.name = name self.name_singular = name_singular self.is_group = bool(is_group) self.items = BroadcastAreasRepository().get_all_areas_for_library(self.id) def get_examples(self): # we show up to four things. three areas, then either a fourth area if there are exactly four, or "and X more". areas_to_show = sorted(area.name for area in self)[:4] count_of_areas_not_named = len(self.items) - 3 # if there's exactly one area not named, there are exactly four - we should just show all four. if count_of_areas_not_named > 1: areas_to_show = areas_to_show[:3] + [f'{count_of_areas_not_named} more…'] return formatted_list(areas_to_show, before_each='', after_each='') class BroadcastAreaLibraries(SerialisedModelCollection, GetItemByIdMixin): model = BroadcastAreaLibrary def __init__(self): self.items = BroadcastAreasRepository().get_libraries() def get_areas(self, area_ids): areas = BroadcastAreasRepository().get_areas(area_ids) return [BroadcastArea(area) for area in areas] def get_areas_with_simple_polygons(self, area_ids): areas = BroadcastAreasRepository().get_areas_with_simple_polygons(area_ids) return [BroadcastArea.from_row_with_simple_polygons(area) for area in areas] broadcast_area_libraries = BroadcastAreaLibraries()