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