Source code for here.content.hmc2.parsers

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.

"""
Parser functions for parsing HMC Object Types and parameters within them.
"""

from datetime import datetime
from typing import List

from here.content.hmc2.constants import Constant, FieldName
from here.content.hmc2.models import (
DTO,
Address,
AffiliationAttribute,
AlcoholServiceAttribute,
AmenitiesAttribute,
Author,
BasicInfoAttribute,
BlackSpot,
CapacityAttribute,
Contact,
ContactAdditionalData,
ContactInformation,
ElectricChargeAttribute,
ExternalIdentifier,
Feedback,
FuelTypeAttribute,
HotelAttribute,
InternetConnectionAttribute,
Location,
LocationInformation,
MatchLevelAttribute,
MediaReference,
MediaReferenceAttribute,
Name,
NoteTypeAttribute,
OfficeTypeAttribute,
OperatingTime,
OperationTime,
OtherInformation,
ParkingAttribute,
Payment,
Place,
PlaceAccessAttribute,
PlaceCategoryConfidenceAttribute,
PopularityAttribute,
PortAttribute,
PostalCode,
PriceRange,
QrCodeAttribute,
Qualifier,
QualityScore,
Relationship,
RelationshipAttribute,
RestaurantAttribute,
SafetyCamera,
SocialSignal,
SpokenLanguage,
Supplier,
Time,
TimeRange,
TransitAttribute,
TruckAttribute,
VehicleServicesAttribute,
VendorUrlAttribute,
Verification,
VerificationAttribute,
)
from here.content.utils.proto import decode_enum, decode_message, parse_multi_polygon, parse_point
from here.platform.adapter import Identifier, Partition, Ref

[docs]
def parse_simple_object_types(data, param: str, dto: DTO, object_type_class: type) -> List:
"""
Parser function to parse simple object types.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:param dto: Data Transfer Object that carries dynamic values.
:param object_type_class: class type in which list item needs to be parsed.
:returns: List of items of type :param:object_type_class.
"""

return [
object_type_class(
partition_id=Partition(getattr(dto, Constant.partition_id)),
value=decode_message(item),
)
for item in getattr(data, param)
]

[docs]
def parse_names(data, param: str) -> List[Name]:
"""
Parser function to parse names.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Names.
"""

supplier_names = []
for name in getattr(data, param):
n = Name(text=decode_message(name.text), type=decode_enum(name, FieldName.name_type))
if hasattr(name, FieldName.representation):
n.representation = [decode_message(rep) for rep in name.representation]
if hasattr(name, FieldName.primary):
n.primary = decode_message(name.primary)
supplier_names.append(n)
return supplier_names

[docs]
def parse_categories(data, param) -> List[Ref]:
"""
Parser function to parse categories.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Category references.
"""

return [
Ref(partition=cat.partition_name, identifier=cat.identifier)
for cat in getattr(data, param)
]

[docs]
def parse_suppliers(data, param: str, dto: DTO) -> List[Supplier]:
"""
Parser function to parse suppliers.

return [
Supplier(
partition_id=Partition(getattr(dto, Constant.partition_id)),
id=Identifier(sup.identifier),
names=parse_names(sup, FieldName.name),
category=Ref(
partition=sup.category.partition_name,
identifier=sup.category.identifier,
),
)
for sup in getattr(data, param)
]

[docs]
def parse_authors(data, param: str, dto: DTO) -> List[Author]:
"""
Parser function to parse authors.

return parse_simple_object_types(data, param, dto, Author)

[docs]
def parse_places(data, param: str, dto: DTO) -> List[Place]:
"""
Parser function to parse places.

places = []
for place in getattr(data, param):
p = Place(
partition_id=Partition(getattr(dto, Constant.partition_id)),
id=Identifier(place.identifier),
names=parse_names(place, FieldName.name),
location=Ref(
partition=place.location_ref.partition_name,
identifier=place.location_ref.identifier,
),
)
if hasattr(data, FieldName.category):
place.categories = parse_categories(place, FieldName.category)
if hasattr(data, FieldName.alt_category):
place.alt_categories = parse_categories(place, FieldName.alt_category)
if hasattr(data, FieldName.political_view_name_replacement):
place.political_view_name_replacement = [
decode_message(pvnr)
for pvnr in getattr(data, FieldName.political_view_name_replacement)
]
if hasattr(data, FieldName.political_view_action):
place.political_view_action = [
decode_message(pva) for pva in getattr(data, FieldName.political_view_action)
]
if hasattr(data, FieldName.valid_unnamed):
place.valid_unnamed = decode_message(place.valid_unnamed)
places.append(p)
return places

[docs]
def parse_external_identifiers(data, param: str, dto: DTO) -> List[ExternalIdentifier]:
"""
Parser function to parse external identifiers.

place_extracter = getattr(dto, Constant.place_extracter)
supplier_extracter = getattr(dto, Constant.supplier_extracter)
return [
ExternalIdentifier(
partition_id=Partition(getattr(dto, Constant.partition_id)),
id=Identifier(ei.external_identifier),
places=[place_extracter.get(i) for i in ei.place_index],
supplier=supplier_extracter.get(ei.supplier_index),
inactive=ei.inactive,
category=Ref(partition=ei.category.partition_name, identifier=ei.category.identifier),
type=ei.type.value,
)
for ei in getattr(data, param)
]

[docs]
def parse_contacts(data, param: str):
"""
Parser function to parse contacts.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Contacts.
"""

contact_infos = []
for ci in getattr(data, param):
contact_info = Contact(
type=decode_enum(ci, FieldName.type), value=ci.value, preferred=ci.preferred
)
if hasattr(ci, FieldName.label):
contact_info.label = decode_message(ci.label)
if hasattr(ci, FieldName.country_code):
contact_info.country_code = ci.country_code.value
if hasattr(ci, FieldName.additional_data):
contact_info.additional_data = [
ContactAdditionalData(
key=ad.key,
value=ad.value,
language_code=ad.language_code,
language_type=ad.language_type,
)
for ad in ci.additional_data
]
contact_infos.append(contact_info)
return contact_infos

[docs]
def parse_contact_information(data, param: str, dto: DTO) -> List[ContactInformation]:
"""
Parser function to parse contact information.

place_extracter = getattr(dto, Constant.place_extracter)
return [
ContactInformation(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in ci.place_index],
contacts=parse_contacts(ci, FieldName.contact_info),
)
for ci in getattr(data, param)
]

[docs]
def parse_operation_times(data, param: str) -> List[OperationTime]:
"""
Parser function to parse operation times.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Operation Times.
"""

return [
OperationTime(
excluded=op_time.excluded,
approx_seasonal_range=op_time.approximate_seasonal_range,
days_of_week=decode_message(op_time.days_of_week)[FieldName.day_of_week],
time_range=TimeRange(
start_time_of_day=Time(
seconds=int(op_time.time_range.start_time_of_day.seconds),
nanos=int(op_time.time_range.start_time_of_day.nanos),
),
end_time_of_day=Time(
seconds=int(op_time.time_range.end_time_of_day.seconds),
nanos=int(op_time.time_range.end_time_of_day.nanos),
),
),
)
for op_time in getattr(data, param)
]

[docs]
def parse_operating_times(data, param, dto: DTO) -> List[OperatingTime]:
"""
Parser function to parse operating times.

place_extracter = getattr(dto, Constant.place_extracter)
operating_times = []
for ot in getattr(data, param):
operating_time = OperatingTime(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in ot.place_index],
open24x7=ot.open24x7,
operation_times=parse_operation_times(ot, FieldName.operating_time),
)
if hasattr(ot, FieldName.category):
operating_time.categories = parse_categories(ot, FieldName.category)
if hasattr(ot, FieldName.additional_data):
operating_time.additional_data = [decode_message(ad) for ad in ot.additional_data]
operating_times.append(operating_time)
return operating_times

[docs]
def parse_feedbacks(data, param: str, dto: DTO) -> List[Feedback]:
"""
Parser function to parse feedbacks.

return parse_simple_object_types(data, param, dto, Feedback)

[docs]
def parse_payments(data, param: str, dto: DTO) -> List[Payment]:
"""
Parser function to parse payments.

place_extracter = getattr(dto, Constant.place_extracter)
return [
Payment(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in pmt.place_index],
type_qualifier=pmt.type_qualifier,
type=decode_enum(pmt, FieldName.payment_type),
accepted=pmt.accepted,
)
for pmt in getattr(data, param)
]

[docs]
def parse_relationships(data, param: str) -> List[Relationship]:
"""
Parser function to parse relationships.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Relationships.
"""

return [
Relationship(
references=[decode_message(refs) for refs in rel.reference],
type=rel.type,
primary=rel.primary,
system=rel.system,
name=rel.name,
id=rel.id,
)
for rel in getattr(data, param)
]

[docs]
def parse_relationship_attributes(data, param: str, dto: DTO) -> List[RelationshipAttribute]:
"""
Parser function to parse relation attributes.

place_extracter = getattr(dto, Constant.place_extracter)
return [
RelationshipAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in ra.place_index],
relationships=parse_relationships(ra, FieldName.relationship),
)
for ra in getattr(data, param)
]

[docs]
def parse_capacity_attribute(data, param: str, dto: DTO) -> List[CapacityAttribute]:
"""
Parser function to parse capacity attributes.

return parse_simple_object_types(data, param, dto, CapacityAttribute)

[docs]
def parse_internet_connection_atttributes(
data, param: str, dto: DTO
) -> List[InternetConnectionAttribute]:
"""
Parser function to parse internet connection attributes.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:param dto: Data Transfer Object that carries dynamic values.
:returns: List of Internet Connection Attributes.
"""

return parse_simple_object_types(data, param, dto, InternetConnectionAttribute)

[docs]
def parse_qualifiers(data, param: str) -> List[Qualifier]:
"""
Parser function to parse qualifiers.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Qualifiers.
"""

return [Qualifier(value=q.value, language=q.language) for q in getattr(data, param)]

[docs]
def parse_price_ranges(data, param: str, dto: DTO) -> List[PriceRange]:
"""
Parser function to parse price ranges.

place_extracter = getattr(dto, Constant.place_extracter)
return [
PriceRange(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in pr.place_index],
qualifiers=parse_qualifiers(pr, FieldName.qualifier),
notes=[decode_message(n) for n in pr.note],
min_range=pr.min_range,
max_range=pr.max_range,
currency_type=decode_enum(pr, FieldName.currency_type),
)
for pr in getattr(data, param)
]

[docs]
def parse_spoken_languages(data, param: str, dto: DTO) -> List[SpokenLanguage]:
"""
Parser function to parse spoken languages.

return parse_simple_object_types(data, param, dto, SpokenLanguage)

[docs]
def parse_qr_code_attributes(data, param: str, dto: DTO) -> List[QrCodeAttribute]:
"""
Parser function to parse QR code attributes.

return parse_simple_object_types(data, param, dto, QrCodeAttribute)

[docs]
def parse_datetime(date_str: str) -> str:
"""
Parser function to parse datetimes.

:param date_str: Datetime in string format.
:returns: Data parsed in datetime object.
"""

dt: int = int(date_str)
if dt > 100000000000000: # hack to cope with invalid timestamps
dt //= 100

remove the milli-seconds

dt //= 1000

return str(datetime.fromtimestamp(dt))

[docs]
def parse_verification_attributes(data, param: str, dto: DTO) -> List[VerificationAttribute]:
"""
Parser function to parse verification attributes.

place_extracter = getattr(dto, Constant.place_extracter)
return [
VerificationAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in va.place_index],
verifications=[
Verification(
attributes=v.attribute,
method=v.method,
status=decode_enum(v, FieldName.status),
type=decode_enum(v, FieldName.type),
system=v.system,
verified=decode_enum(v, FieldName.verified),
start_date=parse_datetime(v.start_date),
end_date=parse_datetime(v.end_date),
expiration_date=parse_datetime(v.expiration_date),
)
for v in va.verification
],
)
for va in getattr(data, param)
]

[docs]
def parse_media_references(data, param: str) -> List[MediaReference]:
"""
Parser function to parse media references.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Media References.
"""

media_refs = []
for mr in getattr(data, param):
media_ref = MediaReference(
labels=mr.label, url=mr.url, type=decode_enum(mr, FieldName.type)
)
if hasattr(mr, FieldName.id):
media_ref.id = mr.id
media_refs.append(media_ref)
return media_refs

[docs]
def parse_media_reference_attributes(data, param: str, dto: DTO) -> List[MediaReferenceAttribute]:
"""
Parser function to parse media reference attributes.

place_extracter = getattr(dto, Constant.place_extracter)
return [
MediaReferenceAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in mra.place_index],
media_refs=parse_media_references(mra, FieldName.media_reference),
)
for mra in getattr(data, param)
]

[docs]
def parse_quality_scores(data, param: str, dto: DTO) -> List[QualityScore]:
"""
Parser function to parse quality scores.

place_extracter = getattr(dto, Constant.place_extracter)
quality_scores = []
for qs in getattr(data, param):
quality_score = QualityScore(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in qs.place_index],
overall_confidence=qs.overall_confidence,
place_confidence=qs.place_confidence,
open_confidence=qs.open_confidence,
address_confidence=qs.address_confidence,
name_confidence=qs.name_confidence,
phone_confidence=qs.phone_confidence,
model_version=qs.model_version,
place_quality_level=qs.place_quality_level,
open_quality_level=qs.open_quality_level,
address_quality_level=qs.address_quality_level,
name_quality_level=qs.name_quality_level,
phone_quality_level=qs.phone_quality_level,
reality_confidence=qs.reality_confidence,
reality_quality_level=qs.reality_quality_level,
)
if hasattr(qs, FieldName.category):
quality_score.category = Ref(
partition=qs.category.partition_name, identifier=qs.category.identifier
)
quality_scores.append(quality_score)
return quality_scores

[docs]
def parse_place_access_attributes(data, param: str, dto: DTO) -> List[PlaceAccessAttribute]:
"""
Parser function to parse place access attributes.

place_extracter = getattr(dto, Constant.place_extracter)
place_access_attributes = []
for paa in getattr(data, param):
place_access_attribute = PlaceAccessAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in paa.place_index],
group_restrictions=[decode_message(gr) for gr in paa.group_restrictions],
dogs_allowed=paa.dogs_allowed,
handicap_accessible=decode_enum(paa, FieldName.handicap_accessible),
private_access=paa.private_access,
)
if hasattr(paa, FieldName.category):
place_access_attribute.category = Ref(
partition=paa.category.partition_name,
identifier=paa.category.identifier,
)
place_access_attributes.append(place_access_attribute)
return place_access_attributes

[docs]
def parse_alcohol_service_attributes(data, param: str, dto: DTO) -> List[AlcoholServiceAttribute]:
"""
Parser function to parse alcohol service attributes.

place_extracter = getattr(dto, Constant.place_extracter)
alcohol_service_attrs = []
for asa in getattr(data, param):
alcohol_service_attr = AlcoholServiceAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in asa.place_index],
alcohol_notes=[decode_message(an) for an in asa.alcohol_note],
beer=asa.beer,
wine=asa.wine,
liquor=asa.liquor,
)
if hasattr(asa, FieldName.category):
alcohol_service_attr.category = Ref(
partition=asa.category.partition_name,
identifier=asa.category.identifier,
)
alcohol_service_attrs.append(alcohol_service_attr)
return alcohol_service_attrs

[docs]
def parse_amenities_attributes(data, param: str, dto: DTO) -> List[AmenitiesAttribute]:
"""
Parser function to parse amenities attributes.

place_extracter = getattr(dto, Constant.place_extracter)
amenities_attributes = []
for aa in getattr(data, param):
amenities_attribute = AmenitiesAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in aa.place_index],
)

for f in [
FieldName.air_conditioning,
FieldName.elevators,
FieldName.escalators,
FieldName.family_friendly,
FieldName.onsite_heart_defribillator,
FieldName.restrooms_available,
FieldName.security_guard,
FieldName.smoking,
FieldName.television,
FieldName.seating_capacity,
]:
if hasattr(aa, f):
setattr(amenities_attribute, f, getattr(aa, f))

if hasattr(aa, FieldName.seating_options):
amenities_attribute.seating_options = decode_enum(aa, FieldName.seating_options)
if hasattr(aa, FieldName.category):
amenities_attribute.category = Ref(
partition=aa.category.partition_name, identifier=aa.category.identifier
)
amenities_attributes.append(amenities_attribute)
return amenities_attributes

[docs]
def parse_basic_info_attributes(data, param: str, dto: DTO) -> List[BasicInfoAttribute]:
"""
Parser function to parse basic information attributes.

place_extracter = getattr(dto, Constant.place_extracter)
basic_info_attr = []

for bi in getattr(data, param):
basic_info = BasicInfoAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in bi.place_index],
access_type=bi.access_type,
building_type=bi.building_type,
capital_indicator=bi.capital_indicator,
capital_level=bi.capital_level,
population=bi.population,
category=bi.category,
)
basic_info_attr.append(basic_info)
return basic_info_attr

[docs]
def parse_other_information(data, param: str, dto: DTO) -> List[OtherInformation]:
"""
Parser function to parse other information.

place_extracter = getattr(dto, Constant.place_extracter)
other_infos = []
for oi in getattr(data, param):
other_info = OtherInformation(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in oi.place_index],
additional_notes=[decode_message(an) for an in oi.additional_note],
national_importance=oi.national_importance,
)
if hasattr(oi, FieldName.category):
other_info.category = Ref(
partition=oi.category.partition_name, identifier=oi.category.identifier
)
other_infos.append(other_info)
return other_infos

[docs]
def parse_electric_charge_attributes(data, param: str, dto: DTO) -> List[ElectricChargeAttribute]:
"""
Parser function to parse electric charge attributes.

return parse_simple_object_types(data, param, dto, ElectricChargeAttribute)

[docs]
def parse_hotel_attributes(data, param: str, dto: DTO) -> List[HotelAttribute]:
"""
Parser function to parse hotel attributes.

place_extracter = getattr(dto, Constant.place_extracter)
hotel_attrs = []
for ha in getattr(data, param):
hotel_attr = HotelAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in ha.place_index],
)
for f in [
FieldName.conference_hall,
FieldName.handicap_rooms,
FieldName.hot_tub,
FieldName.sauna,
FieldName.spa,
FieldName.sports_infrastructure,
FieldName.swimming,
FieldName.swimming_pool,
FieldName.swimming_pool_indoor,
FieldName.swimming_pool_outdoor,
]:
setattr(hotel_attr, f, getattr(ha, f))

if hasattr(ha, FieldName.category):
hotel_attr.category = Ref(
partition=ha.category.partition_name, identifier=ha.category.identifier
)
hotel_attrs.append(hotel_attr)
return hotel_attrs

[docs]
def parse_location_information(data, param: str, dto: DTO) -> List[LocationInformation]:
"""
Parser function to parse location information.

return parse_simple_object_types(data, param, dto, LocationInformation)

[docs]
def parse_parking_attributes(data, param: str, dto: DTO) -> List[ParkingAttribute]:
"""
Parser function to parse parking attributes.

return parse_simple_object_types(data, param, dto, ParkingAttribute)

[docs]
def parse_restaurant_attributes(data, param: str, dto: DTO) -> List[RestaurantAttribute]:
"""
Parser function to parse restaurant attributes.

place_extracter = getattr(dto, Constant.place_extracter)
rest_attrs = []
for ra in getattr(data, param):
rest_attr = RestaurantAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in ra.place_index],
)
if hasattr(ra, FieldName.beer_selection):
rest_attr.beer_selections = [decode_message(bs) for bs in ra.beer_selection]
if hasattr(ra, FieldName.cooking_types):
rest_attr.cooking_types = [decode_message(ct) for ct in ra.cooking_types]
if hasattr(ra, FieldName.specialties):
rest_attr.specialities = [decode_message(s) for s in ra.specialties]
if hasattr(ra, FieldName.wine_selection):
rest_attr.wine_selections = [decode_message(ws) for ws in ra.wine_selection]
if hasattr(ra, FieldName.catering):
rest_attr.catering = ra.catering
if hasattr(ra, FieldName.childrens_menu):
rest_attr.childrens_menu = ra.childrens_menu
if hasattr(ra, FieldName.delivery):
rest_attr.delivery = ra.delivery
if hasattr(ra, FieldName.dress_code):
rest_attr.dress_code = decode_enum(ra, FieldName.dress_code)
if hasattr(ra, FieldName.reservation_required):
rest_attr.reservation_required = ra.reservation_required
if hasattr(ra, FieldName.takeout):
rest_attr.takeout = ra.takeout
if hasattr(ra, FieldName.category):
rest_attr.category = Ref(
partition=ra.category.partition_name, identifier=ra.category.identifier
)
rest_attrs.append(rest_attr)
return rest_attrs

[docs]
def parse_transit_attributes(data, param: str, dto: DTO) -> List[TransitAttribute]:
"""
Parser function to parse transit attributes.

return parse_simple_object_types(data, param, dto, TransitAttribute)

[docs]
def parse_vehicle_services_attributes(
data, param: str, dto: DTO
) -> List[VehicleServicesAttribute]:
"""
Parser function to parse vehicle service attributes.

return parse_simple_object_types(data, param, dto, VehicleServicesAttribute)

[docs]
def parse_postal_codes(data, param: str) -> List[PostalCode]:
"""
Parser function to parse postal codes.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:returns: List of Postal Codes.
"""

postal_codes = []
for postal_code in getattr(data, param):
pc = PostalCode(
text=decode_message(postal_code.text),
type=decode_enum(postal_code, FieldName.name_type),
)
if hasattr(postal_code, FieldName.representation):
pc.representation = [decode_message(rep) for rep in postal_code.representation]
if hasattr(postal_code, FieldName.primary):
pc.primary = decode_message(postal_code.primary)
postal_codes.append(pc)
return postal_codes

[docs]
def parse_addresses(data, param: str, dto: DTO) -> List[Address]:
"""
Parser function to parse addresses.

addresses = []
for a in getattr(data, param):
addr = Address(
partition_id=Partition(getattr(dto, Constant.partition_id)),
postal_codes=parse_postal_codes(a, FieldName.postal_code),
location_address_type=decode_enum(a, FieldName.location_address_type),
source_type=decode_enum(a, FieldName.source_type),
)
if hasattr(a, FieldName.city):
addr.cities = [decode_message(c) for c in a.city]
if hasattr(a, FieldName.state):
addr.states = [decode_message(s) for s in a.state]
if hasattr(a, FieldName.county):
addr.counties = [decode_message(c) for c in a.county]
if hasattr(a, FieldName.region):
addr.regions = [decode_message(r) for r in a.region]
if hasattr(a, FieldName.district):
addr.districts = [decode_message(d) for d in a.district]
if hasattr(a, FieldName.area):
addr.areas = [decode_message(a) for a in a.area]
if hasattr(a, FieldName.street):
addr.streets = [decode_message(s) for s in a.street]
if hasattr(a, FieldName.house):
addr.houses = [decode_message(h) for h in a.house]
if hasattr(a, FieldName.building):
addr.buildings = [decode_message(b) for b in a.building]
if hasattr(a, FieldName.building_unit):
addr.building_units = [decode_message(bu) for bu in a.building_unit]
if hasattr(a, FieldName.level_name):
addr.level_names = [decode_message(ln) for ln in a.level_name]
if hasattr(a, FieldName.unit_name):
addr.unit_names = [decode_message(un) for un in a.unit_name]
if hasattr(a, FieldName.postal_mapping):
addr.postal_mappings = [decode_message(pm) for pm in a.postal_mapping]
if hasattr(a, FieldName.state_code):
addr.state_codes = [decode_message(sc) for sc in a.state_code]
if hasattr(a, FieldName.country_code):
addr.country_code = a.country_code.official_country_code
if hasattr(a, FieldName.category):
addr.category = Ref(
partition=a.category.partition_name, identifier=a.category.identifier
)
addresses.append(addr)
return addresses

[docs]
def parse_locations(data, param: str, dto: DTO) -> List[Location]:
"""
Parser function to parse locations.

address_extracter = getattr(dto, Constant.address_extracter)
locations = []
for loc in getattr(data, param):
location = Location(
partition_id=Partition(getattr(dto, Constant.partition_id)),
location_id=Identifier(loc.identifier),
location_type=decode_enum(loc, FieldName.location_type),
display_position=parse_point(loc.display_position),
address=address_extracter.get(loc.address_index.value),
address_relation=decode_enum(loc, FieldName.address_relation),
area_type=decode_enum(loc, FieldName.area_type),
segment_anchor=decode_message(loc.segment_anchor),
segment_anchor_side=decode_enum(loc, FieldName.segment_anchor_side),
)
if hasattr(loc, FieldName.geometry):
location.geometry = parse_multi_polygon(loc.geometry)
if hasattr(loc, FieldName.level_information):
location.level_info = decode_message(loc.level_information)
if hasattr(loc, FieldName.accessors):
location.accessors = [decode_message(a) for a in loc.accessors]
if hasattr(loc, FieldName.political_geometry):
location.political_geometry = [decode_message(a) for a in loc.political_geometry]
if hasattr(loc, FieldName.alternate_geometry):
location.alternate_geometry = [decode_message(a) for a in loc.alternate_geometry]
if hasattr(loc, FieldName.within_location_ref):
location.within_location_ref = decode_message(loc.within_location_ref)
if hasattr(loc, FieldName.bounding_box):
location.bounding_box = decode_message(loc.bounding_box)
locations.append(location)
return locations

[docs]
def parse_truck_attributes(data, param: str, dto: DTO) -> List[TruckAttribute]:
"""
Parser function to parse truck attributes.

place_extracter = getattr(dto, Constant.place_extracter)
truck_attrs = []
for ta in getattr(data, param):
truck_attr = TruckAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in ta.place_index],
)
for f in [
FieldName.high_canopy,
FieldName.idle_reduction_system,
FieldName.showers,
FieldName.truck_parking,
FieldName.truck_scales,
FieldName.truck_service,
FieldName.truck_wash,
FieldName.truck_secure_parking,
FieldName.truck_stop,
FieldName.truck_night_parking_only,
]:
if hasattr(ta, f):
setattr(truck_attr, f, decode_message(getattr(ta, f)))
if hasattr(ta, FieldName.number_of_showers):
truck_attr.number_of_showers = ta.number_of_showers
if hasattr(ta, FieldName.category):
truck_attr.category = Ref(
partition=ta.category.partition_name, identifier=ta.category.identifier
)
truck_attrs.append(truck_attr)
return truck_attrs

[docs]
def parse_fuel_type_attributes(data, param: str, dto: DTO) -> List[FuelTypeAttribute]:
"""
Parser function to parse fuel type attributes.

place_extracter = getattr(dto, Constant.place_extracter)
fuel_types_attrs = []
for fta in getattr(data, param):
fuel_types_attr = FuelTypeAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in fta.place_index],
)

for f in [
FieldName.biodiesel,
FieldName.cng,
FieldName.diesel,
FieldName.diesel_with_additives,
FieldName.e85,
FieldName.e10,
FieldName.ethanol,
FieldName.ethanol_with_additives,
FieldName.gasoline,
FieldName.high_volume_pumps,
FieldName.hydrogen,
FieldName.lpg,
FieldName.midgrade,
FieldName.pay_at_pump,
FieldName.premium,
FieldName.premium_with_additives,
FieldName.regular,
FieldName.regular_with_additives,
FieldName.lng,
]:
setattr(fuel_types_attr, f, getattr(getattr(fta, f), FieldName.value))

for f in [
FieldName.biodiesel_grade,
FieldName.diesel_additive,
FieldName.diesel_additive_type,
FieldName.hydrogen_pressure,
]:
setattr(fuel_types_attr, f, getattr(fta, f))

if hasattr(fta, FieldName.category):
fuel_types_attr.category = Ref(
partition=fta.category.partition_name,
identifier=fta.category.identifier,
)

fuel_types_attrs.append(fuel_types_attr)
return fuel_types_attrs

[docs]
def parse_social_signals(data, param: str, dto: DTO) -> List[SocialSignal]:
"""
Parser function to parse social signals.

return parse_simple_object_types(data, param, dto, SocialSignal)

[docs]
def parse_port_attributes(data, param: str, dto: DTO) -> List[PortAttribute]:
"""
Parser function to parse port attributes.

return parse_simple_object_types(data, param, dto, PortAttribute)

[docs]
def parse_safety_cameras(data, param: str, dto: DTO) -> List[SafetyCamera]:
"""
Parser function to parse port cameras.

return parse_simple_object_types(data, param, dto, SafetyCamera)

[docs]
def parse_black_spots(data, param: str, dto: DTO) -> List[BlackSpot]:
"""
Parser function to parse black spots.

return parse_simple_object_types(data, param, dto, BlackSpot)

[docs]
def parse_match_level_attribute(data, param: str, dto: DTO) -> List[MatchLevelAttribute]:
"""
Parser function to parse match level attributes.

place_extracter = getattr(dto, Constant.place_extracter)
return [
MatchLevelAttribute(
partition_id=Partition(getattr(dto, Constant.partition_id)),
places=[place_extracter.get(i) for i in mla.place_index],
match_level=decode_enum(mla, FieldName.match_level),
)
for mla in getattr(data, param)
]

[docs]
def parse_affiliation_attributes(data, param: str, dto: DTO) -> List[AffiliationAttribute]:
"""
Parser function to parse affiliation attributes.

return parse_simple_object_types(data, param, dto, AffiliationAttribute)

[docs]
def parse_office_type_attributes(data, param: str, dto: DTO) -> List[OfficeTypeAttribute]:
"""
Parser function to parse office type attributes.

return parse_simple_object_types(data, param, dto, OfficeTypeAttribute)

[docs]
def parse_note_type_attributes(data, param: str, dto: DTO) -> List[NoteTypeAttribute]:
"""
Parser function to parse note type attributes.

return parse_simple_object_types(data, param, dto, NoteTypeAttribute)

[docs]
def parse_place_catagory_confidence_attributes(
data, param: str, dto: DTO
) -> List[PlaceCategoryConfidenceAttribute]:
"""
Parser function to parse place category confidence attributes.

:param data: Object that contains :param:param.
:param param: Parameter that needs to be parsed from :param:data.
:param dto: Data Transfer Object that carries dynamic values.
:returns: List of Place Category Confidence Attributes.
"""

return parse_simple_object_types(data, param, dto, PlaceCategoryConfidenceAttribute)

[docs]
def parse_populatity_attributes(data, param: str, dto: DTO) -> List[PopularityAttribute]:
"""
Parser function to parse popularity attributes.

return parse_simple_object_types(data, param, dto, PopularityAttribute)

[docs]
def parse_vendor_url_attributes(data, param: str, dto: DTO) -> List[VendorUrlAttribute]:
"""
Parser function to parse vendor URL attributes.

return parse_simple_object_types(data, param, dto, VendorUrlAttribute)