Copyright (C) 2022-2023 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
Bindings for Here Places.
"""

from typing import List, Mapping

from google.protobuf.message import Message
from here.content.base import ContentPartition, SingleLayerBinding
from here.content.content import hmc_schema_version
from here.content.hmc2.constants import ObjectType
from here.content.hmc2.extracters import CachedExtracter
from here.content.hmc2.models import (
DTO,
Address,
AffiliationAttribute,
AlcoholServiceAttribute,
AmenitiesAttribute,
Author,
BasicInfoAttribute,
BlackSpot,
CapacityAttribute,
ContactInformation,
ElectricChargeAttribute,
ExternalIdentifier,
Feedback,
FuelTypeAttribute,
HotelAttribute,
InternetConnectionAttribute,
Location,
LocationInformation,
MatchLevelAttribute,
MediaReferenceAttribute,
NoteTypeAttribute,
OfficeTypeAttribute,
OperatingTime,
OtherInformation,
ParkingAttribute,
Payment,
Place,
PlaceAccessAttribute,
PlaceCategoryConfidenceAttribute,
PopularityAttribute,
PortAttribute,
PriceRange,
QrCodeAttribute,
QualityScore,
RelationshipAttribute,
RestaurantAttribute,
SafetyCamera,
SocialSignal,
SpokenLanguage,
Supplier,
TransitAttribute,
TruckAttribute,
VehicleServicesAttribute,
VendorUrlAttribute,
VerificationAttribute,
)
from here.content.hmc2.parsers import (
parse_addresses,
parse_affiliation_attributes,
parse_alcohol_service_attributes,
parse_amenities_attributes,
parse_authors,
parse_basic_info_attributes,
parse_black_spots,
parse_capacity_attribute,
parse_contact_information,
parse_electric_charge_attributes,
parse_external_identifiers,
parse_feedbacks,
parse_fuel_type_attributes,
parse_hotel_attributes,
parse_internet_connection_atttributes,
parse_location_information,
parse_locations,
parse_match_level_attribute,
parse_media_reference_attributes,
parse_note_type_attributes,
parse_office_type_attributes,
parse_operating_times,
parse_other_information,
parse_parking_attributes,
parse_payments,
parse_place_access_attributes,
parse_place_catagory_confidence_attributes,
parse_places,
parse_populatity_attributes,
parse_port_attributes,
parse_price_ranges,
parse_qr_code_attributes,
parse_quality_scores,
parse_relationship_attributes,
parse_restaurant_attributes,
parse_safety_cameras,
parse_social_signals,
parse_spoken_languages,
parse_suppliers,
parse_transit_attributes,
parse_truck_attributes,
parse_vehicle_services_attributes,
parse_vendor_url_attributes,
parse_verification_attributes,
)

[docs]
class HerePlaces(SingleLayerBinding):
"""
Bindings for the HMC Here Places.
"""

[docs]
@classmethod
def schema_hrn(cls) -> List[str]:
"""
:return: the HRN of schema this binding can parse and index
"""
return [f"hrn:here:schema:::com.here.schema.rib:places_v2:{hmc_schema_version}"]

[docs]
@classmethod
def object_types(cls) -> Mapping[str, type]:
"""
:return: the types of attributes supported by this binding
"""
return {ObjectType.supplier: Supplier, ObjectType.author: Author, ObjectType.place: Place, ObjectType.external_identifier: ExternalIdentifier, ObjectType.contact_information: ContactInformation, ObjectType.operating_time: OperatingTime, ObjectType.feedback: Feedback, ObjectType.payment: Payment, ObjectType.relationship_attribute: RelationshipAttribute, ObjectType.capacity_attribute: CapacityAttribute, ObjectType.internet_connection_attribute: InternetConnectionAttribute, ObjectType.price_range: PriceRange, ObjectType.spoken_language: SpokenLanguage, ObjectType.qr_code_attribute: QrCodeAttribute, ObjectType.verification_attribute: VerificationAttribute, ObjectType.media_reference_attribute: MediaReferenceAttribute, ObjectType.quality_score: QualityScore, ObjectType.place_access_attribute: PlaceAccessAttribute, ObjectType.amenities_attribute: AmenitiesAttribute, ObjectType.alcohol_service_attribute: AlcoholServiceAttribute, ObjectType.basic_info_attribute: BasicInfoAttribute, ObjectType.other_information: OtherInformation, ObjectType.electric_charge_attribute: ElectricChargeAttribute, ObjectType.hotel_attribute: HotelAttribute, ObjectType.location_information: LocationInformation, ObjectType.parking_attribute: ParkingAttribute, ObjectType.restaurant_attribute: RestaurantAttribute, ObjectType.transit_attribute: TransitAttribute, ObjectType.vehicle_services_attribute: VehicleServicesAttribute, ObjectType.address: Address, ObjectType.location: Location, ObjectType.truck_attribute: TruckAttribute, ObjectType.fuel_type_attribute: FuelTypeAttribute, ObjectType.social_signal: SocialSignal, ObjectType.port_attribute: PortAttribute, ObjectType.safety_camera: SafetyCamera, ObjectType.black_spot: BlackSpot, ObjectType.match_level_attribute: MatchLevelAttribute, ObjectType.affiliation_attribute: AffiliationAttribute, ObjectType.office_type_attribute: OfficeTypeAttribute, ObjectType.note_type_attribute: NoteTypeAttribute, ObjectType.place_category_confidence_attribute: PlaceCategoryConfidenceAttribute, ObjectType.popularity_attribute: PopularityAttribute, ObjectType.vendor_url_attribute: VendorUrlAttribute,}

[docs]
@classmethod
def parse_and_index(cls, partition_id: str, msg: Message) -> ContentPartition:
"""
Parse and index decoded message.

:param partition_id: identifier of the partition
:param msg: decoded content from the layer, in the form of a Protobuf message
:return: the content of a partition parsed and indexed in its own data structure
"""

content =

for object_type, parser in [
(ObjectType.supplier, parse_suppliers),
(ObjectType.author, parse_authors),
(ObjectType.place, parse_places),
(ObjectType.feedback, parse_feedbacks),
(ObjectType.capacity_attribute, parse_capacity_attribute),
(
ObjectType.internet_connection_attribute,
parse_internet_connection_atttributes,
),
(ObjectType.spoken_language, parse_spoken_languages),
(ObjectType.qr_code_attribute, parse_qr_code_attributes),
(ObjectType.electric_charge_attribute, parse_electric_charge_attributes),
(ObjectType.location_information, parse_location_information),
(ObjectType.parking_attribute, parse_parking_attributes),
(ObjectType.transit_attribute, parse_transit_attributes),
(ObjectType.vehicle_services_attribute, parse_vehicle_services_attributes),
(ObjectType.address, parse_addresses),
(ObjectType.social_signal, parse_social_signals),
(ObjectType.port_attribute, parse_port_attributes),
(ObjectType.safety_camera, parse_safety_cameras),
(ObjectType.black_spot, parse_black_spots),
(ObjectType.affiliation_attribute, parse_affiliation_attributes),
(ObjectType.office_type_attribute, parse_office_type_attributes),
(ObjectType.note_type_attribute, parse_note_type_attributes),
(
ObjectType.place_category_confidence_attribute,
parse_place_catagory_confidence_attributes,
),
(ObjectType.popularity_attribute, parse_populatity_attributes),
(ObjectType.vendor_url_attribute, parse_vendor_url_attributes),
]:
if hasattr(msg, object_type):
content[object_type] = parser(msg, object_type, DTO(partition_id=partition_id))

places_cached_extracter = CachedExtracter(content[ObjectType.place]) # type: ignore
supplier_cached_extracter = CachedExtracter(content[ObjectType.supplier]) # type: ignore
address_cached_extracter = CachedExtracter(content[ObjectType.address]) # type: ignore

if hasattr(msg, ObjectType.external_identifier):
content[ObjectType.external_identifier] = parse_external_identifiers(
msg,
ObjectType.external_identifier,
DTO(
partition_id=partition_id,
place_extracter=places_cached_extracter,
supplier_extracter=supplier_cached_extracter,
),
)

for object_type, parser in [
(ObjectType.contact_information, parse_contact_information),
(ObjectType.operating_time, parse_operating_times),
(ObjectType.payment, parse_payments),
(ObjectType.relationship_attribute, parse_relationship_attributes),
(ObjectType.price_range, parse_price_ranges),
(ObjectType.verification_attribute, parse_verification_attributes),
(ObjectType.media_reference_attribute, parse_media_reference_attributes),
(ObjectType.quality_score, parse_quality_scores),
(ObjectType.place_access_attribute, parse_place_access_attributes),
(ObjectType.alcohol_service_attribute, parse_alcohol_service_attributes),
(ObjectType.amenities_attribute, parse_amenities_attributes),
(ObjectType.other_information, parse_other_information),
(ObjectType.hotel_attribute, parse_hotel_attributes),
(ObjectType.restaurant_attribute, parse_restaurant_attributes),
(ObjectType.truck_attribute, parse_truck_attributes),
(ObjectType.fuel_type_attribute, parse_fuel_type_attributes),
(ObjectType.match_level_attribute, parse_match_level_attribute),
(ObjectType.basic_info_attribute, parse_basic_info_attributes),
]:
content[object_type] = parser(
msg,
object_type,
DTO(partition_id=partition_id, place_extracter=places_cached_extracter),
)

if hasattr(msg, ObjectType.location):
content[ObjectType.location] = parse_locations(
msg,
ObjectType.location,
DTO(
partition_id=partition_id,
address_extracter=address_cached_extracter,
),
)

return ContentPartition(cls, **content) # type: ignore