import logging
from collections import defaultdict
from qgis.core import QgsField, QgsFields
from qgis.PyQt.QtCore import QVariant
from catatom2osm import config, hgwnames, translate
from catatom2osm.geo.geometry import Geometry
from catatom2osm.geo.layer.base import BaseLayer
from catatom2osm.geo.tools import get_attributes
from catatom2osm.geo.types import WKBPoint
from catatom2osm.report import instance as report
log = logging.getLogger(config.app_name)
[documentos]class AddressLayer(BaseLayer):
"""Class for addresses."""
def __init__(
self, path="Point", baseName="address", providerLib="memory", source_date=None
):
super(AddressLayer, self).__init__(path, baseName, providerLib)
if self.fields().isEmpty():
self.writer.addAttributes(
[
QgsField("localId", QVariant.String, len=254),
QgsField("spec", QVariant.String, len=254),
QgsField("designator", QVariant.String, len=254),
QgsField("TN_text", QVariant.String, len=254),
QgsField("postCode", QVariant.Int),
QgsField("image", QVariant.String, len=254),
QgsField("task", QVariant.String, len=254),
QgsField("PD_id", QVariant.String, len=254),
QgsField("TN_id", QVariant.String, len=254),
QgsField("AU_id", QVariant.String, len=254),
]
)
self.updateFields()
self.rename = {"spec": "specification"}
self.resolve = {
"PD_id": ("component_href", r"[\w\.]+PD[\.0-9]+"),
"TN_id": ("component_href", r"[\w\.]+TN[\.0-9]+"),
"AU_id": ("component_href", r"[\w\.]+AU[\.0-9]+"),
}
self.source_date = source_date
[documentos] @staticmethod
def create_shp(name, crs, fields=QgsFields(), geom_type=WKBPoint):
BaseLayer.create_shp(name, crs, fields, geom_type)
[documentos] @staticmethod
def is_address(feature):
"""Address features have '.' but not '_' in its localId field."""
return "." in feature["localId"] and "_" not in feature["localId"]
[documentos] @staticmethod
def get_id(feat):
"""Trim to parcel id."""
return feat["localId"].split("_")[0].split(".")[-1]
[documentos] def to_osm(self, data=None, tags={}, upload="never"):
"""Export to OSM."""
return super(AddressLayer, self).to_osm(
translate.address_tags, data, tags=tags, upload=upload
)
[documentos] def conflate(self, current_address):
"""
Delete address existing in current_address.
Args:
current_address (OSM): dataset
"""
to_clean = [
feat.id()
for feat in self.getFeatures()
if feat["TN_text"] + feat["designator"] in current_address
]
if to_clean:
self.writer.deleteFeatures(to_clean)
log.debug(
_("Refused %d addresses because they exist in OSM") % len(to_clean)
)
report.refused_addresses = len(to_clean)
to_clean = [
feat.id() for feat in self.search("designator = '%s'" % config.no_number)
]
if to_clean:
self.writer.deleteFeatures(to_clean)
log.debug(_("Deleted %d addresses without house number") % len(to_clean))
report.addresses_without_number = len(to_clean)
[documentos] def get_names(self, highway=None, place=None):
"""
Return a dictionary with the translation for each street name.
Args:
highway (HighwayLayer): Current OSM highway data for conflation.
If highway is None, only parse names.
Returns:
(dict) highway names translations
"""
if highway is None or highway.featureCount() == 0:
highway_names = {
f["TN_text"]: (hgwnames.parse(f["TN_text"]), "CAT")
for f in self.getFeatures()
}
return highway_names
highway_names = defaultdict(list)
hgw_ndx = highway.get_index()
hgw_feats = {feat.id(): feat for feat in highway.getFeatures()}
plc_ndx = place.get_index()
plc_feats = {feat.id(): feat for feat in place.getFeatures()}
for f in self.getFeatures():
if f["TN_text"]:
highway_names[f["TN_text"]].append(f.geometry().asPoint())
for name, points in highway_names.items():
hgw_type = hgwnames.parse(name).split(" ")[0].lower()
bbox = Geometry.fromMultiPointXY(points).boundingBox()
bbox.grow(config.bbox_buffer * 100000)
if hgw_type in config.place_types:
choices = [plc_feats[fid]["name"] for fid in plc_ndx.intersects(bbox)]
highway_names[name] = hgwnames.match(name, choices)
elif name.split(" ")[0] == "PZ":
choices = [plc_feats[fid]["name"] for fid in plc_ndx.intersects(bbox)]
highway_names[name] = hgwnames.match(name, choices)
if highway_names[name][1] == "CAT":
choices = [
hgw_feats[fid]["name"] for fid in hgw_ndx.intersects(bbox)
]
highway_names[name] = hgwnames.match(name, choices)
else:
highway_names[name] = (
"square" + highway_names[name][0],
highway_names[name][1],
)
else:
choices = [hgw_feats[fid]["name"] for fid in hgw_ndx.intersects(bbox)]
highway_names[name] = hgwnames.match(name, choices)
return highway_names
[documentos] def get_image_links(self):
to_change = {}
for feat in self.getFeatures():
url = config.cadastre_doc_url.format(feat["localId"][-14:])
feat["image"] = url
to_change[feat.id()] = get_attributes(feat)
self.writer.changeAttributeValues(to_change)
[documentos] def remove_address_wo_building(self, buildings):
"""Remove address without associated building."""
bu_refs = [
f["localId"] for f in buildings.getFeatures() if buildings.is_building(f)
]
to_clean = [f.id() for f in self.getFeatures() if self.get_id(f) not in bu_refs]
if to_clean:
self.writer.deleteFeatures(to_clean)
msg = _("Removed %d addresses without building")
log.debug(msg, len(to_clean))
report.orphaned_addresses = len(to_clean)