LOG = logging.getLogger()
+
def _to_float(numstr: str, max_value: float) -> float:
""" Convert the number in string into a float. The number is expected
to be in the range of [-max_value, max_value]. Otherwise rises a
return num
+
class _PostcodeCollector:
""" Collector for postcodes of a single country.
"""
self.collected: Dict[str, PointsCentroid] = defaultdict(PointsCentroid)
self.normalization_cache: Optional[Tuple[str, Optional[str]]] = None
-
def add(self, postcode: str, x: float, y: float) -> None:
""" Add the given postcode to the collection cache. If the postcode
already existed, it is overwritten with the new centroid.
if normalized:
self.collected[normalized] += (x, y)
-
def commit(self, conn: Connection, analyzer: AbstractAnalyzer, project_dir: Path) -> None:
""" Update postcodes for the country from the postcodes selected so far
as well as any externally supplied postcodes.
""").format(pysql.Literal(self.country)),
to_update)
-
- def _compute_changes(self, conn: Connection) \
- -> Tuple[List[Tuple[str, float, float]], List[str], List[Tuple[float, float, str]]]:
+ def _compute_changes(
+ self, conn: Connection
+ ) -> Tuple[List[Tuple[str, float, float]], List[str], List[Tuple[float, float, str]]]:
""" Compute which postcodes from the collected postcodes have to be
added or modified and which from the location_postcode table
have to be deleted.
return to_add, to_delete, to_update
-
def _update_from_external(self, analyzer: AbstractAnalyzer, project_dir: Path) -> None:
""" Look for an external postcode file for the active country in
the project directory and add missing postcodes when found.
finally:
csvfile.close()
-
def _open_external(self, project_dir: Path) -> Optional[TextIO]:
fname = project_dir / f'{self.country}_postcodes.csv'
analyzer.update_postcodes_from_db()
+
def can_compute(dsn: str) -> bool:
"""
Check that the place table exists so that