overturetoosm.objects

  1"""Pydantic models needed throughout the project."""
  2
  3# ruff: noqa: D415
  4
  5from enum import Enum
  6
  7try:
  8    from typing import Annotated
  9except ImportError:
 10    from typing import Annotated
 11
 12from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator
 13
 14from .resources import places_tags
 15
 16
 17class OvertureBaseModel(BaseModel):
 18    """Base model for Overture features."""
 19
 20    model_config = ConfigDict(extra="forbid")
 21
 22    version: int = Field(ge=0)
 23    theme: str | None = None
 24    type: str | None = None
 25    id: str | None = Field(None, pattern=r"^(\S.*)?\S$")
 26
 27
 28class Wikidata(RootModel):
 29    """Model for transportation segment wikidata."""
 30
 31    root: str = Field(description="Wikidata ID.", pattern=r"^Q\d+")
 32
 33
 34class Sources(BaseModel):
 35    """Overture sources model."""
 36
 37    property: str
 38    dataset: str
 39    license: str | None = None
 40    record_id: str | None = None
 41    confidence: float | None = Field(ge=0.0, le=1.0)
 42    update_time: str | None = Field(
 43        pattern=r"^([1-9]\d{3})-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])T([01]\d|2[0-3]):([0-5]\d):([0-5]\d|60)(\.\d{1,3})?(Z|[-+]([01]\d|2[0-3]):[0-5]\d)$"
 44    )
 45
 46    @field_validator("confidence")
 47    @classmethod
 48    def set_default_if_none(cls, v: float) -> float:
 49        """@private"""
 50        return v if v is not None else 0.0
 51
 52    def get_osm_link(self) -> str | None:
 53        """Return the OSM link for the source."""
 54        if (
 55            self.record_id
 56            and self.record_id.startswith(("n", "w", "r"))
 57            and self.dataset == "OpenStreetMap"
 58        ):
 59            type_dict = {"n": "node", "w": "way", "r": "relation"}
 60            return f"https://www.openstreetmap.org/{type_dict[self.record_id[0]]}/{self.record_id[1:]}"
 61
 62
 63class RulesVariant(str, Enum):
 64    """Overture name rules variant model."""
 65
 66    alternate = "alternate"
 67    common = "common"
 68    official = "official"
 69    short = "short"
 70
 71
 72class Between(RootModel):
 73    """Model for transportation segment between."""
 74
 75    root: Annotated[list, Field(float, min_length=2, max_length=2)]
 76
 77
 78class Mode(str, Enum):
 79    """Model for political perspectives from which a named feature is viewed."""
 80
 81    accepted_by = "accepted_by"
 82    disputed_by = "disputed_by"
 83
 84
 85class Perspectives(BaseModel):
 86    """Model for political perspectives from which a named feature is viewed."""
 87
 88    mode: Mode
 89    countries: list[str] = Field(min_length=1)
 90
 91
 92class Rules(BaseModel):
 93    """Overture name rules model."""
 94
 95    variant: RulesVariant
 96    language: str | None = None
 97    value: str
 98    between: Between | None = None
 99    side: str | None = None
100    perspectives: Perspectives | None = None
101
102
103class Names(BaseModel):
104    """Overture names model."""
105
106    primary: str
107    common: list[tuple[str, str]] | None
108    rules: list[Rules] | None
109
110    def to_osm(self) -> dict[str, str]:
111        """Convert names to OSM tags."""
112        names = {}
113        if self.primary:
114            names["name"] = self.primary
115
116        return names
117
118
119class PlaceAddress(BaseModel):
120    """Overture addresses model."""
121
122    freeform: str | None
123    locality: str | None
124    postcode: str | None
125    region: str | None
126    country: str | None = Field(pattern=r"^[A-Z]{2}$")
127
128    def to_osm(self, region_tag: str) -> dict[str, str]:
129        """Convert address to OSM tags."""
130        address_info = {}
131        if self.freeform:
132            address_info["addr:full"] = self.freeform
133        if self.country:
134            address_info["addr:country"] = self.country
135        if self.postcode:
136            address_info["addr:postcode"] = self.postcode
137        if self.locality:
138            address_info["addr:city"] = self.locality
139        if self.region:
140            address_info[region_tag] = self.region
141
142        return address_info
143
144
145class Categories(BaseModel):
146    """Overture categories model."""
147
148    primary: str
149    alternate: list[str] | None
150
151    def to_osm(self, unmatched: str) -> dict[str, str]:
152        """Convert categories to OSM tags."""
153        prim = places_tags.get(self.primary)
154        if prim:
155            return prim
156        elif unmatched == "force":
157            return {"type": self.primary}
158        elif unmatched == "error":
159            raise UnmatchedError(self.primary)
160        return {}
161
162
163class Brand(BaseModel):
164    """Overture brand model."""
165
166    wikidata: Wikidata | None = None
167    names: Names
168
169    def to_osm(self) -> dict[str, str]:
170        """Convert brand properties to OSM tags."""
171        osm = {"brand": self.names.primary}
172        if self.wikidata:
173            osm.update({"brand:wikidata": str(self.wikidata.root)})
174        return osm
175
176
177class Socials(RootModel):
178    """Overture socials model."""
179
180    root: list[str]
181
182    def to_osm(self) -> dict[str, str]:
183        """Convert socials properties to OSM tags."""
184        new_props = {}
185        for social in self.root:
186            if "facebook" in social:
187                new_props["contact:facebook"] = social
188            elif "twitter" in str(social):
189                new_props["contact:twitter"] = social
190        return new_props
191
192
193class OperatingStatus(str, Enum):
194    """Enum for place operating status."""
195
196    open = "open"
197    permanently_closed = "permanently_closed"
198    temporarily_closed = "temporarily_closed"
199
200
201class PlaceProps(OvertureBaseModel):
202    """Overture properties model.
203
204    Use this model directly if you want to manipulate the `place` properties yourself.
205    """
206
207    model_config = ConfigDict(extra="ignore")
208
209    sources: list[Sources]
210    names: Names
211    brand: Brand | None = None
212    categories: Categories | None = None
213    basic_category: str | None = Field(pattern=r"^[a-z0-9]+(_[a-z0-9]+)*$")
214    confidence: float = Field(ge=0.0, le=1.0)
215    websites: list[str | None] | None = None
216    socials: Socials | None = None
217    emails: list[str | None] | None = None
218    phones: list[str | None] | None = None
219    addresses: list[PlaceAddress]
220    operating_status: OperatingStatus | None = None
221
222    def _validate_license(self, required_license: str | None) -> None:
223        """Validate that sources meet license requirements.
224
225        Args:
226            required_license: The required license string (e.g., "CDLA").
227                             If None, no validation is performed.
228
229        Raises:
230            LicenseError: If no sources have the required license and at least
231                         one source has a non-null license.
232        """
233        if required_license is None:
234            return
235
236        found_licenses = [s.license for s in self.sources if s.license is not None]
237
238        # If all licenses are null, pass the validation
239        if not found_licenses:
240            return
241
242        # Check if any source has the required license
243        for source in self.sources:
244            if source.license and required_license in source.license:
245                return
246
247        # If we get here, no source has the required license
248        raise LicenseError(required_license, found_licenses)
249
250    def to_osm(
251        self,
252        confidence: float,
253        region_tag: str,
254        unmatched: str,
255        required_license: str | None = None,
256    ) -> dict[str, str]:
257        """Convert Overture's place properties to OSM tags.
258
259        Used internally by the `overturetoosm.process_place` function.
260        """
261        if self.confidence < confidence:
262            raise ConfidenceError(confidence, self.confidence)
263
264        self._validate_license(required_license)
265
266        new_props = {}
267
268        # Categories
269        if self.categories:
270            new_props.update(self.categories.to_osm(unmatched))
271
272        # Names
273        if self.names:
274            new_props.update(self.names.to_osm())
275
276        # Contact information
277        new_props.update(self._process_contact_info())
278
279        # Addresses
280        if self.addresses:
281            new_props.update(self.addresses[0].to_osm(region_tag))
282
283        # Sources
284        new_props["source"] = source_statement(self.sources)
285
286        # Socials and Brand
287        if self.socials:
288            new_props.update(self.socials.to_osm())
289        if self.brand:
290            new_props.update(self.brand.to_osm())
291
292        return new_props
293
294    def _process_contact_info(self) -> dict[str, str]:
295        """Process contact information."""
296        contact_info = {}
297        if not is_none_or_list_of_nones(self.phones):
298            contact_info["phone"] = self.phones[0]
299        if not is_none_or_list_of_nones(self.websites):
300            contact_info["website"] = str(self.websites[0])
301        if not is_none_or_list_of_nones(self.emails):
302            contact_info["email"] = self.emails[0]
303        return contact_info
304
305
306class ConfidenceError(Exception):
307    """Confidence error exception.
308
309    This exception is raised when the confidence level of an item is below the
310    user-defined level. It contains the original confidence level and the confidence
311    level of the item.
312
313    Attributes:
314        confidence_level (float): The set confidence level.
315        confidence_item (float): The confidence of the item.
316        message (str): The error message.
317    """
318
319    def __init__(
320        self,
321        confidence_level: float,
322        confidence_item: float,
323        message: str = "Confidence in this item is too low.",
324    ) -> None:
325        """@private"""
326        self.confidence_level = confidence_level
327        self.confidence_item = confidence_item
328        self.message = message
329        super().__init__(message)
330
331    def __str__(self) -> str:
332        """@private"""
333        lev = f"confidence_level={self.confidence_level}"
334        item = f"confidence_item={self.confidence_item}"
335        return f"""{self.message} {lev}, {item}"""
336
337
338class UnmatchedError(Exception):
339    """Unmatched category error.
340
341    This exception is raised when an item's Overture category does not have a
342    corresponding OSM definition. Edit
343    [the OSM Wiki page](https://wiki.openstreetmap.org/wiki/Overture_categories)
344    to add a definition to this category.
345
346    Attributes:
347        category (str): The Overture category that is unmatched.
348        message (str): The error message.
349    """
350
351    def __init__(
352        self, category: str, message: str = "Overture category is unmatched."
353    ) -> None:
354        """@private"""
355        self.category = category
356        self.message = message
357        super().__init__(message)
358
359    def __str__(self) -> str:
360        """@private"""
361        return f"{self.message} {{category={self.category}}}"
362
363
364class LicenseError(Exception):
365    """License compatibility error.
366
367    This exception is raised when a feature's source licenses do not meet
368    the required license criteria.
369
370    Attributes:
371        required_license (str): The required license string.
372        found_licenses (list[str]): The licenses found in the sources.
373        message (str): The error message.
374    """
375
376    def __init__(
377        self,
378        required_license: str,
379        found_licenses: list[str],
380        message: str = "Feature does not meet license requirements.",
381    ) -> None:
382        """@private"""
383        self.required_license = required_license
384        self.found_licenses = found_licenses
385        self.message = message
386        super().__init__(message)
387
388    def __str__(self) -> str:
389        """@private"""
390        return f"{self.message} {{required_license={self.required_license}, found_licenses={self.found_licenses}}}"
391
392
393class BuildingProps(OvertureBaseModel):
394    """Overture building properties.
395
396    Use this model if you want to manipulate the `building` properties yourself.
397    """
398
399    has_parts: bool
400    sources: list[Sources]
401    class_: str | None = Field(alias="class", default=None)
402    subtype: str | None = None
403    names: Names | None = None
404    level: int | None = None
405    height: float | None = None
406    is_underground: bool | None = None
407    num_floors: int | None = Field(serialization_alias="building:levels", default=None)
408    num_floors_underground: int | None = Field(
409        serialization_alias="building:levels:underground", default=None
410    )
411    min_height: float | None = None
412    min_floor: int | None = Field(
413        serialization_alias="building:min_level", default=None
414    )
415    facade_color: str | None = Field(
416        serialization_alias="building:colour", default=None
417    )
418    facade_material: str | None = Field(
419        serialization_alias="building:material", default=None
420    )
421    roof_material: str | None = Field(serialization_alias="roof:material", default=None)
422    roof_shape: str | None = Field(serialization_alias="roof:shape", default=None)
423    roof_direction: str | None = Field(
424        serialization_alias="roof:direction", default=None
425    )
426    roof_orientation: str | None = Field(
427        serialization_alias="roof:orientation", default=None
428    )
429    roof_color: str | None = Field(serialization_alias="roof:colour", default=None)
430    roof_height: float | None = Field(serialization_alias="roof:height", default=None)
431
432    def _validate_license(self, required_license: str | None) -> None:
433        """Validate that sources meet license requirements.
434
435        Args:
436            required_license: The required license string (e.g., "CDLA").
437                             If None, no validation is performed.
438
439        Raises:
440            LicenseError: If no sources have the required license and at least
441                         one source has a non-null license.
442        """
443        if required_license is None:
444            return
445
446        found_licenses = [s.license for s in self.sources if s.license is not None]
447
448        # If all licenses are null, pass the validation
449        if not found_licenses:
450            return
451
452        # Check if any source has the required license
453        for source in self.sources:
454            if source.license and required_license in source.license:
455                return
456
457        # If we get here, no source has the required license
458        raise LicenseError(required_license, found_licenses)
459
460    def to_osm(
461        self, confidence: float, required_license: str | None = None
462    ) -> dict[str, str]:
463        """Convert properties to OSM tags.
464
465        Used internally by`overturetoosm.process_building` function.
466        """
467        new_props = {}
468        confidences = {source.confidence for source in self.sources}
469        if any(conf and conf < confidence for conf in confidences):
470            raise ConfidenceError(confidence, max({i for i in confidences if i}))
471
472        self._validate_license(required_license)
473
474        new_props["building"] = self.class_ if self.class_ else "yes"
475
476        new_props["source"] = source_statement(self.sources)
477
478        prop_obj = self.model_dump(exclude_none=True, by_alias=True).items()
479        new_props.update(
480            {k: v for k, v in prop_obj if k.startswith(("roof", "building"))}
481        )
482        new_props.update({k: round(v, 2) for k, v in prop_obj if k.endswith("height")})
483
484        if self.is_underground:
485            new_props["location"] = "underground"
486        if self.names:
487            new_props["name"] = self.names.primary
488        return new_props
489
490
491class AddressLevel(BaseModel):
492    """Overture address level model."""
493
494    value: str
495
496
497class AddressProps(OvertureBaseModel):
498    """Overture address properties.
499
500    Use this model directly if you want to manipulate the `address` properties yourself.
501    """
502
503    number: str | None = Field(serialization_alias="addr:housenumber", default=None)
504    street: str | None = Field(serialization_alias="addr:street", default=None)
505    unit: str | None = Field(serialization_alias="addr:unit", default=None)
506    postcode: str | None = Field(serialization_alias="addr:postcode", default=None)
507    postal_city: str | None = Field(serialization_alias="addr:city", default=None)
508    country: str | None = Field(serialization_alias="addr:country", default=None)
509    address_levels: (
510        None | (Annotated[list[AddressLevel], Field(min_length=1, max_length=5)])
511    ) = Field(default_factory=list)
512    sources: list[Sources]
513
514    def _validate_license(self, required_license: str | None) -> None:
515        """Validate that sources meet license requirements.
516
517        Args:
518            required_license: The required license string (e.g., "CDLA").
519                             If None, no validation is performed.
520
521        Raises:
522            LicenseError: If no sources have the required license and at least
523                         one source has a non-null license.
524        """
525        if required_license is None:
526            return
527
528        found_licenses = [s.license for s in self.sources if s.license is not None]
529
530        # If all licenses are null, pass the validation
531        if not found_licenses:
532            return
533
534        # Check if any source has the required license
535        for source in self.sources:
536            if source.license and required_license in source.license:
537                return
538
539        # If we get here, no source has the required license
540        raise LicenseError(required_license, found_licenses)
541
542    def to_osm(self, style: str, required_license: str | None = None) -> dict[str, str]:
543        """Convert properties to OSM tags.
544
545        Used internally by `overturetoosm.process_address`.
546        """
547        self._validate_license(required_license)
548
549        obj_dict = {
550            k: v
551            for k, v in self.model_dump(exclude_none=True, by_alias=True).items()
552            if k.startswith("addr:")
553        }
554        obj_dict["source"] = source_statement(self.sources)
555
556        if self.address_levels and len(self.address_levels) > 0 and style == "US":
557            obj_dict["addr:state"] = str(self.address_levels[0].value)
558
559        return obj_dict
560
561
562def source_statement(source: list[Sources]) -> str:
563    """Return a source statement from a list of sources."""
564    return (
565        ", ".join(sorted({i.dataset.strip(", ") for i in source}))
566        + " via overturetoosm"
567    )
568
569
570def is_none_or_list_of_nones(value) -> bool:
571    """Check whether a given value is either None or a list containing only None values.
572
573    Args:
574        value: The value to check. Can be of any type.
575
576    Returns:
577        bool: True if the value is None or a list containing only None values,
578              False otherwise.
579    """
580    if value is None:
581        return True
582
583    if isinstance(value, list):
584        return len(value) > 0 and all(item is None for item in value)
585
586    return False