Source code for here.content.base

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.

"""
Support classes common to all the bindings.
"""
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass, field
from functools import lru_cache
from typing import Callable, Dict, Iterable, Iterator, List, Mapping, Optional, Tuple, Type, Union

from google.protobuf.message import Message
from here.content.content import Content as ParentContent
from here.content.content import ContentBinding as ParentContentBinding
from here.platform.adapter import Adapter, ContentAdapter, Identifier, Partition, Ref
from here.platform.layer import LayerType

_CACHE_SIZE = 25 # on average, roughly equivalent to an area of 5x5 tiles

[docs]
@dataclass
class ContentIndexer:
"""
Used to inspect and index content objects.

Each field describes how to extract index entries from an object.
Each content class must have one indexer of this type, as static parameter.

None: not indexable
str: get the value from a dataclass field with the corresponding name
Callable: get the value through a function, passing the object as parameter

refs is a mapping because more than one reference type can be defined per class.
"""

partition: Union[None, str, Callable[[object], Partition]] = None
identifier: Union[None, str, Callable[[object], Identifier]] = None
refs: Mapping[str, Union[str, Callable[[object], Union[Ref, List[Ref]]]]] = field(
default_factory=dict
)

[docs]
class ContentPartition:
"""Class to store, index and lookup objects from a parsed partition."""

def init(self, binding_cls: Type["ContentBinding"], **objects: Iterable[object]):
"""
Initialize the content of a parsed partition by inserting and indexing
different types of object. This is supposed to be called by parse_and_index
functions of content bindings, after parsing the content.

Object types must match the ones declared in object_types in the bindings.
Objects are automatically index according to the indexer specified
in object_types, for each object type.

:param binding_cls: the class of the content bindings
:param objects: for each object type, a sequence of objects
"""

Regarding binding_cls parameter:

https://stackoverflow.com/questions/70624956/whats-the-type-of-cls-introduced-by-classmethod

self._object_types = binding_cls.object_types()
self._ref_types = binding_cls.ref_types()

To handle MultiLayer Bindings object types where primary layer data (Message) needs to

parsed_and_indexed before and then further associate with secondary dependent layers.

if hasattr(binding_cls, "_internal_object_types"):
self._object_types = {**self._object_types, **binding_cls._internal_object_types()} # type: ignore # noqa: E501

The objects, untouched

self._objects = objects

Automatic object indexing

def create_indexes_for_type(
indexer: ContentIndexer, obj_list: Iterable[object]
) -> Tuple[Mapping[str, object], Mapping[str, Mapping[Ref, Iterable[object]]]]:
by_id =
by_refs: Mapping[str, Mapping[Ref, List[object]]] = defaultdict(
lambda: defaultdict(list)
)
for x in obj_list:
if indexer.identifier is not None:
obj_id = ContentAdapter._extract_identifier(indexer.identifier, x)
by_id[obj_id] = x
if indexer.refs is not None:
for ref_type, index_ref in indexer.refs.items():
for ref in ContentAdapter._extract_refs(index_ref, x):
by_refs[ref_type][ref].append(x)
return by_id, by_refs

indexers: Dict[str, ContentIndexer] = {s: t.indexer for s, t in self._object_types.items() # type: ignore}
indexes_by_obj_type: Dict[
str, Tuple[Mapping[str, object], Mapping[str, Mapping[Ref, Iterable[object]]]]
] = {obj_type: create_indexes_for_type(indexers[obj_type], obj_list) for obj_type, obj_list in objects.items() if obj_type in indexers}
self._indexes: Mapping[str, Mapping[str, object]] = {obj_type: index_by_id for obj_type, (index_by_id, _) in indexes_by_obj_type.items() if index_by_id}
self._ref_indexes: Mapping[str, Mapping[str, Mapping[Ref, Iterable[object]]]] = {obj_type: index_by_refs for obj_type, (_, index_by_refs) in indexes_by_obj_type.items() if index_by_refs}

[docs]
def objects(self, obj_type: str) -> Iterable[object]:
"""
List all the objects contained in the content partition.
Objects are returned in no special order and without duplicates.

:param obj_type: type of the object to list
:return: a collection of objects
:raises ValueError: in case of unsupported object type
"""
if obj_type not in self._object_types:
raise ValueError(f"Unsupported object type '{obj_type}'")
return self._objects.get(obj_type, [])

[docs]
def index(self, obj_type: str) -> Mapping[str, object]:
"""
Index all the objects contained in the content partition by their identifier.
This con be considered a primary index of the content.

:param obj_type: type of the object to index
:return: a collection of objects, indexed by their identifier
:raises ValueError: in case of unsupported object type
"""
if obj_type not in self._object_types:
raise ValueError(f"Unsupported object type '{obj_type}'")
return self._indexes.get(obj_type, )

[docs]
def ref_index(self, obj_type: str, ref_type: str) -> Mapping[Ref, Iterable[object]]:
"""
Index all the objects contained in the content partition by their references
to other objects, in the same partition or in another partition.
This con be considered a secondary index of the content.

:param obj_type: type of the object to index
:param ref_type: type of the reference to index
:return: a collection of objects, indexed by their external references
:raises ValueError: in case of unsupported object or reference type
"""
if obj_type not in self._object_types:
raise ValueError(f"Unsupported object type '{obj_type}'")
if ref_type not in self._ref_types[obj_type]:
raise ValueError(
f"Unsupported reference type '{ref_type}' for object type '{obj_type}'"
)
return self._ref_indexes.get(obj_type, ).get(ref_type, )

The following and other classes in base.py can become the base classes among many more

content bindings, in addition to the new HMC bindings, like HDLM, traffic, weather, ...

In this case, ParentContent and ParentContentBinding from the parent package

will then collapse to this ContentBinding, removing the duplication

[docs]
class ContentBinding(ParentContentBinding, ABC):
"""
The entry point for developers.

This is the base class for all the bindings: this class represents the main
developer-facing API to obtain data form bindings and is common across all the bindings.

Bindings exposes objects.

Features are objects indexed by their partition and identifier.
Identifiers cannot be duplicate in the same partition.

Attributes are objects indexed by the feature they apply to.
"""

def init(self, content: ParentContent, adapter: Optional[Adapter] = None):
"""
Instantiate an instance of content binding.

:param content: Content object that references a catalog at a fixed version
:param adapter: the Adapter to transform data between different representations,
None to use the adapter defined in content.
"""
super().init(content, adapter)
self._catalog = self.content._catalog
self._schema_hrn: List[str] = self.schema_hrn()
self._object_types: Mapping[str, type] = self.object_types()

[docs]
@classmethod
@abstractmethod
def schema_hrn(cls) -> List[str]:
"""
:return: the HRN of schemas this binding can parse and index
"""

[docs]
@classmethod
@abstractmethod
def object_types(cls) -> Mapping[str, type]:
"""
:return: the name and types of objects supported by this binding
"""

[docs]
@classmethod
def ref_types(cls) -> Mapping[str, List[str]]:
"""
:return: the reference types supported by each object type of this binding
"""
return {type_name: list(getattr(object_type, "indexer").refs.keys()) for type_name, object_type in cls.object_types().items()}

[docs]
@abstractmethod
def get(
self,
partition: Union[str, int, Iterable[Union[str, int]]],
object_type: str,
identifier: Optional[str] = None,
):
"""
Get a selected or all the objects of a specific type from the content.

:param partition: the partition(s) where the objects are stored
:param object_type: the type of the objects
:param identifier: the identifier of the object, in case of looking up one specific object
:return: one or all the objects of the requested type contained in the requested
partition(s), format is adapter-dependent
:raises ValueError: if the requested object type is not supported
:raises KeyError: if a specific object is requested but it cannot be found
"""

[docs]
def get_ref(self, object_type: str, ref: Ref):
"""
Get a selected object of a specific type from the content via a direct reference to it.

:param object_type: the type of the objects
:param ref: reference to a partition and object
:return: the object of the requested type referenced by ref, that in turn
specifies partition and object identifier, format is adapter-dependent
:raises ValueError: if the requested object type is not supported # noqa: DAR402
:raises KeyError: if the requested object cannot be found # noqa: DAR402
"""
return self.get(ref.partition, object_type, ref.identifier)

[docs]
@abstractmethod
def get_referencing(
self,
partition: Union[str, int, Iterable[Union[str, int]]],
object_type: str,
ref_type: str,
referenced_obj: Optional[Ref] = None,
):
"""
Get objects of a specific type that reference other objects of the content.

:param partition: the partition(s) where the objects are stored
:param object_type: the type of the objects
:param ref_type: the type of the reference, among the ones supported by the object type
:param referenced_obj: return only object referencing this specific partition and object
:return: objects containing the requested reference, format is adapter-dependent
:raises ValueError: if the requested object or reference type is not supported
"""

class _ContentBindingImpl(ContentBinding, ABC):
"""Common functions needed by each binding, common implementation, and technical interface."""

def _verify_object_type(self, object_type: str) -> None:
if object_type not in self._object_types:
raise ValueError(f"Unsupported object type '{object_type}'")

def _partitions(self, partition: Union[str, int, Iterable[Union[str, int]]]) -> Iterable[str]:
return [str(partition)] if isinstance(partition, (str, int)) else map(str, partition)

def _indexer(self, object_type: str) -> ContentIndexer:

We assume every type has an indexer as property

indexer = getattr(self._object_types[object_type], "indexer")
assert isinstance(indexer, ContentIndexer)
return indexer

@abstractmethod
def get_partition(self, partition_id: Union[str, int]) -> Optional[ContentPartition]:
"""
Main method for the implementation of the binding. Get, decode, parse and index content
of a partition. Return None if no content is available for the partition.

The result is cached by where the method is called.

:param partition_id: identifier of the partition
:return: content of the partition parsed and indexed
"""

def get(
self,
partition: Union[str, int, Iterable[Union[str, int]]],
object_type: str,
identifier: Optional[str] = None,
):
"""
Get a selected or all the objects of a specific type from the content.

if identifier:
if not isinstance(partition, (str, int)):
raise ValueError(
"To get a specific object, the partition parameter "
"must be one single partition identifier"
)
assert isinstance(partition, (str, int))
parsed_partition = self.get_partition(partition)
if parsed_partition:
feature = parsed_partition.index(object_type)[identifier]
return self._content_adapter.from_objects(
self._object_types[object_type],
[feature],
single_element=True,
index_partition=indexer.partition,
index_id=indexer.identifier,
)
else:
raise KeyError(f"Partition '{partition}' contains no content or does not exists")

else:

def all_objects() -> Iterator[object]:
parsed_partitions: List[ContentPartition] = []
absent_partitions: List[str] = []
for partition_id in self._partitions(partition):
parsed_partition = self.get_partition(partition_id)
if parsed_partition:
parsed_partitions.append(parsed_partition)
else:
absent_partitions.append(partition_id)

if absent_partitions:
raise KeyError(
f"Partitions '{absent_partitions}' contains no content or does not exist"
)
else:
for p_id in parsed_partitions:
yield from p_id.objects(object_type)

return self._content_adapter.from_objects(
self._object_types[object_type],
all_objects(),
index_partition=indexer.partition,
index_id=indexer.identifier,
)

def get_referencing(
self,
partition: Union[str, int, Iterable[Union[str, int]]],
object_type: str,
ref_type: str,
referenced_obj: Optional[Ref] = None,
):
"""
Get objects of a specific type that reference other objects of the content.

if ref_type not in indexer.refs:
raise ValueError(f"Unsupported reference type '{ref_type}' for '{object_type}'")

def referencing_objects() -> Iterator[object]:
for partition_id in self._partitions(partition):
parsed_partition = self.get_partition(partition_id)
if parsed_partition:
ref_index = parsed_partition.ref_index(object_type, ref_type)
if referenced_obj:
yield from ref_index.get(referenced_obj, [])
else:

Same object could be present more than once in this case,

for example when it has two references of the same type.

We clean up the answer by removing duplicates.

seen = set()
for objs in ref_index.values():
for x in objs:
if id(x) not in seen:
seen.add(id(x))
yield x

When looking for all the objects that reference a given one, we return all the objects

with that reference, otherwise we build the index of references

index_ref = None if referenced_obj else indexer.refs[ref_type]

return self._content_adapter.from_objects(
self._object_types[object_type],
referencing_objects(),
index_ref=index_ref,
)

[docs]
class SingleLayerBinding(_ContentBindingImpl, ABC):
"""
The base class for all the bindings based on one single layer.
"""

def init(
self,
content: ParentContent,
layer_id: str,
adapter: Optional[Adapter] = None,
layer_type: Optional[LayerType] = None,
):
"""
Instantiate a content single-layer binding.

:param content: Content object that references a catalog at a fixed version
:param layer_id: ID of the only layer that contains the content
:param adapter: the Adapter to transform data between different representations,
None to use the adapter defined in content.
:param layer_type: the layer type to be supported by the binding.
:raises ValueError: if the specified layer is not of a supported type
"""
super().init(content, adapter)
self._layer = self._catalog.get_layer(layer_id)
self._schema = self._layer.get_schema()

Only versioned layers are supported so far,

mainly due to caching of volatile data that is problematic

if self._layer.configuration.type not in {LayerType.VERSIONED, layer_type}:
raise ValueError(f"The layer type of the layer '{layer_id}' is not supported")

[docs]
@classmethod
@abstractmethod
def parse_and_index(cls, partition_id: str, msg: Message) -> ContentPartition:
"""
Parse and index decoded message from the single layer into whatever
form is comfortable for the bindings. The resulting, parsed content is cached.

This is used by get_partition to parse and index content
retrieved and decoded from the layer, before caching it and returning it
to the implementation of the binding.

It is responsibility of the binding to implement the parser.

: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
"""

def _retrieve_and_decode_partition(self, partition_id: str) -> Optional[Message]:
"""
Retrieve and decode a partition.

Result is not cached, because caching happens at higher level, in get_partition.

:param partition_id: partition ID
:return: the partition data, decoded into a Protobuf :class:Message when this
is available, otherwise return None.
"""

get the blob for the pid

blob = next(
self._layer.read_partitions(
[partition_id], decode=False, version=self.content.version
),
None,
)

if blob is None:
return None

_, data = blob
assert isinstance(data, bytes)

Possible improvement to remove code duplication: the layer can decode the data directly

decoded = self._schema.decode_blob(data)
assert isinstance(decoded, Message)
return decoded

[docs]
@lru_cache(maxsize=_CACHE_SIZE)
def get_partition(self, partition_id: Union[str, int]) -> Optional[ContentPartition]:
"""
Main method for the implementation of the binding. Get, decode, parse and index content
of a partition. Return None if no content is available for the partition.

The result is cached.

:param partition_id: identifier of the partition
:return: content of the partition parsed and indexed
"""

Partition ids are in general int or string. It's good that user-facing function

support both cases. In HMC, partition ids are always string, so we convert

int to string internally as soon as possible to avoid having a mix of int and

strings in data models and partition identifiers. Partition ids are always

strings when exposed back to the user, both when they're part of the content

and when they're part of indexing structures like dictionaries.

partition_id = str(partition_id)
message = self._retrieve_and_decode_partition(partition_id)
return self.parse_and_index(partition_id, message) if message else None

TODO: this is a preliminary design

[docs]
class MultiLayerBinding(_ContentBindingImpl, ABC):
"""
The base class for all the bindings based on multiple layers.
"""

TODO: this is a preliminary design

[docs]
class CompositeBinding(_ContentBindingImpl, ABC):
"""
The base class for all the bindings based on other, multiple bindings.
"""