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