Copyright (C) 2021-2022 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.

"""
Provides access to HMC external references
"""
from typing import Dict, List, Optional, Union, cast

from here.platform import Platform
from here.platform.adapter import Ref, make_ref
from here.platform.catalog import Catalog
from here.platform.layer import VersionedLayer
from pandas import DataFrame

hmc_external_references_catalog_hrn = "hrn:here:data::olp-here:rib-external-references-2"
hmc_external_references_layer_id = "external-reference-attributes"

[docs]
class HMCExternalReferences:
"""
Provides access to HMC external references
"""

platform: Platform = None catalog: Catalog = None
layer: VersionedLayer = None pvid_segment: Dict[Union[str, int], Ref] =
_segment_pvid: Dict[Ref, Union[str, int]] =
_partitions: Dict[Union[str, int], DataFrame] =
_partition_segment_refs: Dict[Union[str, int], List[Ref]] =

@classmethod
def _get_external_references_partition(
cls, partition_id: Union[str, int]
) -> Optional[DataFrame]:
"""
Reads partition for the HERE External-References catalog layer.

:param partition_id: id of the partition to be read.
:returns: dataframe containing the resulting partition data
"""
if partition_id in cls._partitions:
return cls._partitions.get(partition_id)
else:
if cls._platform is None:
cls._platform = Platform()
if cls._catalog is None:
cls._catalog = cls._platform.get_catalog(hmc_external_references_catalog_hrn)
if cls._layer is None:
cls._layer = cast(
VersionedLayer, cls._catalog.get_layer(hmc_external_references_layer_id)
)
partitions: DataFrame = cls._layer.read_partitions(partition_ids=[partition_id])
if not partitions.empty:
cls._partitions[partition_id] = partitions
return partitions
return None

@classmethod
def _parse_segment_refs(cls, partition_id: Union[str, int], partition: DataFrame) -> List[Ref]:
"""
Parses segment references from the partition dataframe.

:param partition_id: id of the partition being parsed.
:param partition: dataframe containing partition data.
:returns: list of segment references
"""
if partition_id not in cls._partition_segment_refs:
cls._partition_segment_refs[partition_id] = [
make_ref(
segment_ref["segment_ref"].get("partition_name"),
segment_ref["segment_ref"].get("identifier"),
)
for segment_anchor_col in partition.head(1)["segment_anchor"]
for segment_anchor in segment_anchor_col
for segment_ref in segment_anchor["oriented_segment_ref"]
]
return cls._partition_segment_refs[partition_id]

@classmethod
def _parse_external_references_in_partition(cls, partition_id: Union[str, int]):
"""
Parses external references within a partition and caches result in a dict.

:param partition_id: id of the partition being parsed.
"""
partitions = cls._get_external_references_partition(partition_id)
if partitions is not None:
segment_refs = cls._parse_segment_refs(partition_id, partitions)
for core_map_link_objs in partitions.head(1)["core_map_link_id"]:
for core_map_link_obj in core_map_link_objs:
core_map_link_id = core_map_link_obj["core_map_link_id"]
segment_ref_index = (
core_map_link_obj["segment_anchor_index"][0]
if core_map_link_obj["segment_anchor_index"]
else None
)
if segment_ref_index is not None:
cls._pvid_segment[core_map_link_id] = segment_refs[segment_ref_index]
cls._segment_pvid[segment_refs[segment_ref_index]] = core_map_link_id

[docs]
@classmethod
def pvid_to_segment(cls, partition_id: Union[str, int], pvid: Union[str, int]) -> Ref:
"""
Provides segment reference from HMC for the provided link PVID.

:param partition_id: id of the partition containing the link PVID.
:param pvid: PVID of the link to be mapped to segment reference.
:returns: segment reference for the provided link pvid.
"""
if pvid not in cls._pvid_segment:
cls._parse_external_references_in_partition(partition_id)
return cls._pvid_segment.get(pvid)

[docs]
@classmethod
def segment_to_pvid(
cls, partition_id: Union[str, int], segment_ref: Ref
) -> Union[str, int, None]:
"""
Provides link PVID for the segment reference from HMC
.

:param partition_id: id of the partition containing the link PVID.
:param segment_ref: segment reference to be mapped to link PVID.
:returns: link PVID for the provided segment ref.
"""
if segment_ref not in cls._segment_pvid:
cls._parse_external_references_in_partition(partition_id)
return cls._segment_pvid.get(segment_ref)

[docs]
@classmethod
def check_catalog_access(cls) -> bool:
"""
Verifies if the user has access to the HERE External-Reference catalog
"""
try:
Platform().get_catalog(hmc_external_references_catalog_hrn)
return True
except ValueError:
return False