FEATURE: Add client side caching and pre-loading support

pyproject.toml

@@ -24,11 +24,11 @@ classifiers = [
 ]
 
 
-# FIXME: add ansys-api-edb version
 dependencies = [
-    "ansys-api-edb==1.0.10",
+    "ansys-api-edb==1.0.11",
     "protobuf>=3.19.3,<5",
-    "grpcio>=1.44.0"
+    "grpcio>=1.44.0",
+    "Django>=4.2.16"
 ]
 
 [project.urls]

src/ansys/edb/core/inner/base.py

@@ -2,6 +2,8 @@
 
 from ansys.api.edb.v1.edb_messages_pb2 import EDBObjMessage
 
+from ansys.edb.core.utility.cache import get_cache
+
 
 class ObjBase:
     """Provides the base object that all gRPC-related models extend from."""
@@ -14,6 +16,9 @@ def __init__(self, msg):
         msg : EDBObjMessage
         """
         self._id = 0 if msg is None else msg.id
+        cache = get_cache()
+        if cache is not None:
+            cache.add_from_cache_msg(msg.cache)
 
     @property
     def is_null(self):

src/ansys/edb/core/inner/conn_obj.py

@@ -48,6 +48,17 @@ def get_client_prim_type_from_class():
             return client_obj
         return cls(None)
 
+    @property
+    def obj_type(self):
+        """:class:`LayoutObjType <ansys.edb.core.edb_defs.LayoutObjType>`: Layout object type.
+
+        This property is read-only.
+        """
+        if self.layout_obj_type != LayoutObjType.INVALID_LAYOUT_OBJ:
+            return super().obj_type
+        else:
+            return LayoutObjType(self.__stub.GetObjType(self.msg).type)
+
     @classmethod
     def find_by_id(cls, layout, uid):
         """Find a :term:`Connectable` object by database ID in a given layout.

src/ansys/edb/core/inner/factory.py

@@ -1,13 +1,65 @@
 """This module allows for the creating of objects while avoid circular imports."""
 
 from ansys.edb.core.edb_defs import LayoutObjType
-from ansys.edb.core.hierarchy import cell_instance
-from ansys.edb.core.hierarchy.group import Group
-from ansys.edb.core.hierarchy.pin_group import PinGroup
-from ansys.edb.core.layout import voltage_regulator
-from ansys.edb.core.primitive.primitive import PadstackInstance, Primitive
-from ansys.edb.core.session import StubAccessor, StubType
-from ansys.edb.core.terminal.terminals import Terminal, TerminalInstance
+from ansys.edb.core.inner.conn_obj import ConnObj
+
+_type_creator_params_dict = None
+
+
+class _CreatorParams:
+    def __init__(self, obj_type, do_cast=False):
+        self.obj_type = obj_type
+        self.do_cast = do_cast
+
+
+def _initialize_type_creator_params_dict():
+    global _type_creator_params_dict
+
+    from ansys.edb.core.hierarchy.cell_instance import CellInstance
+    from ansys.edb.core.hierarchy.group import Group
+    from ansys.edb.core.hierarchy.pin_group import PinGroup
+    from ansys.edb.core.layout.voltage_regulator import VoltageRegulator
+    from ansys.edb.core.net.differential_pair import DifferentialPair
+    from ansys.edb.core.net.extended_net import ExtendedNet
+    from ansys.edb.core.net.net import Net
+    from ansys.edb.core.net.net_class import NetClass
+    from ansys.edb.core.primitive.primitive import PadstackInstance, Primitive
+    from ansys.edb.core.terminal.terminals import Terminal, TerminalInstance
+
+    _type_creator_params_dict = {
+        LayoutObjType.PRIMITIVE: _CreatorParams(Primitive, True),
+        LayoutObjType.PADSTACK_INSTANCE: _CreatorParams(PadstackInstance),
+        LayoutObjType.TERMINAL: _CreatorParams(Terminal, True),
+        LayoutObjType.TERMINAL_INSTANCE: _CreatorParams(TerminalInstance),
+        LayoutObjType.CELL_INSTANCE: _CreatorParams(CellInstance),
+        LayoutObjType.GROUP: _CreatorParams(Group, True),
+        LayoutObjType.PIN_GROUP: _CreatorParams(PinGroup),
+        LayoutObjType.VOLTAGE_REGULATOR: _CreatorParams(VoltageRegulator),
+        LayoutObjType.NET_CLASS: _CreatorParams(NetClass),
+        LayoutObjType.EXTENDED_NET: _CreatorParams(ExtendedNet),
+        LayoutObjType.DIFFERENTIAL_PAIR: _CreatorParams(DifferentialPair),
+        LayoutObjType.NET: _CreatorParams(Net),
+    }
+
+
+def _get_type_creator_dict():
+    if _type_creator_params_dict is None:
+        _initialize_type_creator_params_dict()
+    return _type_creator_params_dict
+
+
+def create_obj(msg, obj_type, do_cast):
+    """Create an object from the provided message of the provided type."""
+    obj = obj_type(msg)
+    if do_cast:
+        obj = obj.cast()
+    return obj
+
+
+def create_lyt_obj(msg, lyt_obj_type):
+    """Create a layout object from the provided message of the type corresponding to the provided layout object type."""
+    params = _get_type_creator_dict()[lyt_obj_type]
+    return create_obj(msg, params.obj_type, params.do_cast)
 
 
 def create_conn_obj(msg):
@@ -21,21 +73,4 @@ def create_conn_obj(msg):
     -------
     ansys.edb.core.inner.ConnObj
     """
-    type = LayoutObjType(StubAccessor(StubType.connectable).__get__().GetObjType(msg).type)
-    if type == LayoutObjType.PRIMITIVE:
-        return Primitive(msg).cast()
-    elif type == LayoutObjType.PADSTACK_INSTANCE:
-        return PadstackInstance(msg)
-    elif type == LayoutObjType.TERMINAL:
-        return Terminal(msg).cast()
-    elif type == LayoutObjType.TERMINAL_INSTANCE:
-        return TerminalInstance(msg)
-    elif type == LayoutObjType.CELL_INSTANCE:
-        return cell_instance.CellInstance(msg)
-    elif type == LayoutObjType.GROUP:
-        return Group(msg).cast()
-    elif type == LayoutObjType.PIN_GROUP:
-        return PinGroup(msg)
-    elif type == LayoutObjType.VOLTAGE_REGULATOR:
-        return voltage_regulator.VoltageRegulator(msg)
-    raise TypeError("Encountered an unknown layout object type.")
+    return create_lyt_obj(msg, ConnObj(msg).obj_type)

src/ansys/edb/core/inner/interceptors.py

@@ -1,14 +1,21 @@
 """Client-side gRPC interceptors."""
 
 import abc
+from collections import namedtuple
 import logging
 
-from grpc import StatusCode, UnaryUnaryClientInterceptor
+from grpc import (
+    ClientCallDetails,
+    StatusCode,
+    UnaryStreamClientInterceptor,
+    UnaryUnaryClientInterceptor,
+)
 
 from ansys.edb.core.inner.exceptions import EDBSessionException, ErrorCode, InvalidArgumentException
+from ansys.edb.core.utility.cache import get_cache
 
 
-class Interceptor(UnaryUnaryClientInterceptor, metaclass=abc.ABCMeta):
+class Interceptor(UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor, metaclass=abc.ABCMeta):
     """Provides the base interceptor class."""
 
     def __init__(self, logger):
@@ -20,14 +27,21 @@ def __init__(self, logger):
     def _post_process(self, response):
         pass
 
+    def _continue_unary_unary(self, continuation, client_call_details, request):
+        return continuation(client_call_details, request)
+
     def intercept_unary_unary(self, continuation, client_call_details, request):
         """Intercept a gRPC call."""
-        response = continuation(client_call_details, request)
+        response = self._continue_unary_unary(continuation, client_call_details, request)
 
         self._post_process(response)
 
         return response
 
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        """Intercept a gRPC streaming call."""
+        return continuation(client_call_details, request)
+
 
 class LoggingInterceptor(Interceptor):
     """Logs EDB errors on each request."""
@@ -76,3 +90,78 @@ def _post_process(self, response):
 
         if exception is not None:
             raise exception
+
+
+class CachingInterceptor(Interceptor):
+    """Returns cached values if a given request has already been made and caching is enabled."""
+
+    def __init__(self, logger, rpc_counter):
+        """Initialize a caching interceptor with a logger and rpc counter."""
+        super().__init__(logger)
+        self._rpc_counter = rpc_counter
+        self._reset_cache_entry_data()
+
+    def _reset_cache_entry_data(self):
+        self._current_rpc_method = ""
+        self._current_cache_key_details = None
+
+    def _should_log_traffic(self):
+        return self._rpc_counter is not None
+
+    class _ClientCallDetails(
+        namedtuple("_ClientCallDetails", ("method", "timeout", "metadata", "credentials")),
+        ClientCallDetails,
+    ):
+        pass
+
+    @classmethod
+    def _get_client_call_details_with_caching_options(cls, client_call_details):
+        if get_cache() is None:
+            return client_call_details
+        metadata = []
+        if client_call_details.metadata is not None:
+            metadata = list(client_call_details.metadata)
+        metadata.append(("enable-caching", "1"))
+        return cls._ClientCallDetails(
+            client_call_details.method,
+            client_call_details.timeout,
+            metadata,
+            client_call_details.credentials,
+        )
+
+    def _continue_unary_unary(self, continuation, client_call_details, request):
+        if self._should_log_traffic():
+            self._current_rpc_method = client_call_details.method
+        cache = get_cache()
+        if cache is not None:
+            method_tokens = client_call_details.method.strip("/").split("/")
+            cache_key_details = method_tokens[0], method_tokens[1], request
+            cached_response = cache.get(*cache_key_details)
+            if cached_response is not None:
+                return cached_response
+            else:
+                self._current_cache_key_details = cache_key_details
+        return super()._continue_unary_unary(
+            continuation,
+            self._get_client_call_details_with_caching_options(client_call_details),
+            request,
+        )
+
+    def _cache_missed(self):
+        return self._current_cache_key_details is not None
+
+    def _post_process(self, response):
+        cache = get_cache()
+        if cache is not None and self._cache_missed():
+            cache.add(*self._current_cache_key_details, response.result())
+        if self._should_log_traffic() and (cache is None or self._cache_missed()):
+            self._rpc_counter[self._current_rpc_method] += 1
+        self._reset_cache_entry_data()
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        """Intercept a gRPC streaming call."""
+        return super().intercept_unary_stream(
+            continuation,
+            self._get_client_call_details_with_caching_options(client_call_details),
+            request,
+        )

src/ansys/edb/core/inner/messages.py

@@ -83,14 +83,9 @@
 )
 from ansys.api.edb.v1.hierarchy_obj_pb2 import ObjectNameInLayoutMessage
 from ansys.api.edb.v1.inst_array_pb2 import InstArrayCreationMessage
-from ansys.api.edb.v1.layout_pb2 import (
-    LayoutConvertP2VMessage,
-    LayoutExpandedExtentMessage,
-    LayoutGetItemsMessage,
-)
+from ansys.api.edb.v1.layout_pb2 import LayoutConvertP2VMessage, LayoutExpandedExtentMessage
 from ansys.api.edb.v1.material_def_pb2 import MaterialDefPropertiesMessage
 from ansys.api.edb.v1.mcad_model_pb2 import *  # noqa
-from ansys.api.edb.v1.net_pb2 import NetGetLayoutObjMessage
 from ansys.api.edb.v1.package_def_pb2 import HeatSinkMessage, SetHeatSinkMessage
 from ansys.api.edb.v1.padstack_inst_term_pb2 import (
     PadstackInstTermCreationsMessage,
@@ -507,11 +502,6 @@ def point_3d_property_message(target, val):
     return Point3DPropertyMessage(target=edb_obj_message(target), value=point3d_message(val))
 
 
-def layout_get_items_message(layout, item_type):
-    """Convert to a ``LayoutGetItemsMessage`` object."""
-    return LayoutGetItemsMessage(layout=layout.msg, obj_type=item_type.value)
-
-
 def layout_expanded_extent_message(
     layout, nets, extent, exp, exp_unitless, use_round_corner, num_increments
 ):
@@ -1080,11 +1070,6 @@ def double_property_message(edb_obj, double):
     return DoublePropertyMessage(target=edb_obj.msg, value=double)
 
 
-def net_get_layout_obj_message(obj, layout_obj_type):
-    """Convert to a ``NetGetLayoutObjMessage`` object."""
-    return NetGetLayoutObjMessage(net=edb_obj_message(obj), obj_type=layout_obj_type.value)
-
-
 def differential_pair_creation_message(layout, name, pos_net, neg_net):
     """Convert to a ``DifferentialPairCreationMessage`` object."""
     return DifferentialPairCreationMessage(

src/ansys/edb/core/inner/utils.py

@@ -1,14 +1,38 @@
 """This module contains utility functions for API development work."""
+from ansys.api.edb.v1.layout_obj_pb2 import LayoutObjTargetMessage
+
+from ansys.edb.core.inner.factory import create_lyt_obj
+from ansys.edb.core.utility.cache import get_cache
 
 
 def map_list(iterable_to_operate_on, operator=None):
     """Apply the given operator to each member of an iterable and return the modified list.
 
     Parameters
     ---------
-    iterable_to_operate on
+    iterable_to_operate_on
     operator
     """
     return list(
         iterable_to_operate_on if operator is None else map(operator, iterable_to_operate_on)
     )
+
+
+def query_lyt_object_collection(owner, obj_type, unary_rpc, unary_streaming_rpc):
+    """For the provided request, retrieve a collection of objects using the unary_rpc or unary_streaming_rpc methods \
+    depending on whether caching is enabled."""
+    items = []
+    cache = get_cache()
+    request = LayoutObjTargetMessage(target=owner.msg, type=obj_type.value)
+
+    def add_msgs_to_items(edb_obj_collection_msg):
+        nonlocal items
+        for item in edb_obj_collection_msg.items:
+            items.append(create_lyt_obj(item, obj_type))
+
+    if cache is None:
+        add_msgs_to_items(unary_rpc(request))
+    else:
+        for streamed_items in unary_streaming_rpc(request):
+            add_msgs_to_items(streamed_items)
+    return items