From 9d1fdb5f185f6a8a6228f42cde9c1774223b3bc9 Mon Sep 17 00:00:00 2001 From: dmiller Date: Thu, 24 Jul 2025 15:26:03 -0600 Subject: [PATCH] F605: Added support for multiple spatial queries in lyt inst queries --- doc/source/api/glossary.rst | 8 ++ src/ansys/edb/core/inner/utils.py | 5 + .../core/layout_instance/layout_instance.py | 123 ++++++++++++------ 3 files changed, 95 insertions(+), 41 deletions(-) diff --git a/doc/source/api/glossary.rst b/doc/source/api/glossary.rst index 5c9870f9b0..a1ccdda6c0 100644 --- a/doc/source/api/glossary.rst +++ b/doc/source/api/glossary.rst @@ -120,3 +120,11 @@ Glossary - ``Square45`` -> [``inner size``, ``channel width``, ``isolation gap``] - ``Square90`` -> [``inner size``, ``channel width``, ``isolation gap``] - ``N-Sided Polygon`` -> [``side length``, ``number of sides``] + + LayoutInstanceQueryResult + + If a polygonal spatial filter is specified, a tuple of lists of hits is returned in this + format: ``[, ]``. + Otherwise, a list containing all hits is returned. + + :obj:`list` of :class:`.LayoutObjInstance` or :obj:`tuple` of (:obj:`list` of :class:`.LayoutObjInstance`, :obj:`list` of :class:`.LayoutObjInstance`) \ No newline at end of file diff --git a/src/ansys/edb/core/inner/utils.py b/src/ansys/edb/core/inner/utils.py index 47c19142d7..d23086cdcf 100644 --- a/src/ansys/edb/core/inner/utils.py +++ b/src/ansys/edb/core/inner/utils.py @@ -17,6 +17,11 @@ def map_list(iterable_to_operate_on, operator=None): ) +def ensure_is_list(obj): + """If the object is a list, return it. Otherwise, return a list where the sole entry is the provided obj.""" + return obj if isinstance(obj, list) else [obj] + + def query_lyt_object_collection( owner, obj_type, unary_rpc, unary_streaming_rpc, request_requires_type=True ): diff --git a/src/ansys/edb/core/layout_instance/layout_instance.py b/src/ansys/edb/core/layout_instance/layout_instance.py index 55edbbbaef..dc120c7974 100644 --- a/src/ansys/edb/core/layout_instance/layout_instance.py +++ b/src/ansys/edb/core/layout_instance/layout_instance.py @@ -1,4 +1,11 @@ """Layout instance.""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from ansys.edb.core.typing import LayerLike, NetLike + import ansys.api.edb.v1.layout_instance_pb2 as layout_instance_pb2 from ansys.edb.core.geometry.point_data import PointData @@ -11,9 +18,9 @@ polygon_data_message, strings_message, ) +from ansys.edb.core.inner.utils import client_stream_iterator from ansys.edb.core.layout_instance import layout_obj_instance from ansys.edb.core.session import LayoutInstanceServiceStub, StubAccessor, StubType -from ansys.edb.core.utility.io_manager import get_cache class LayoutInstance(ObjBase): @@ -25,72 +32,106 @@ def refresh(self): """Refresh the layout instance so it contains an up-to-date geometry.""" self.__stub.Refresh(self.msg) - def query_layout_obj_instances(self, layer_filter=None, net_filter=None, spatial_filter=None): + @staticmethod + def _query_request_iterator(requests): + chunk_entry_creator = lambda request: request + chunk_entries_getter = lambda chunk: chunk.queries + return client_stream_iterator( + requests, + layout_instance_pb2.LayoutObjInstancesQueriesMessage, + chunk_entry_creator, + chunk_entries_getter, + ) + + def query_layout_obj_instances( + self, + layer_filter: LayerLike | list[LayerLike] = None, + net_filter: NetLike | list[NetLike] = None, + spatial_filter: PolygonData | PointData | None | list[PolygonData | PointData] = None, + ): """Query layout object instances using the provided filters. Parameters ---------- - layer_filter : list[:class:`.Layer` or str or None], default: None + layer_filter : :term:`LayerLike` or list of :term:`LayerLike`, default: None Layers to query. The default is ``None``, in which case all layers are queried. - net_filter : list[:class:`.Net` or str or None], default: None + net_filter : :term:`NetLike` or list of :term:`NetLike`, default: None Nets to query. The default is ``None``, in which case all nets are queried. - spatial_filter : :class:`.PolygonData` or - :class:`.PointData` or ``None``, default: None - Area of the design to query. The default is ``None``, in which case the entire + spatial_filter : .PolygonData or .PointData or :obj:`None` or \ + list of .PolygonData or .PointData, default: None + Area of the design to query. The default is :obj:`None`, in which case the entire spatial domain of the design is queried. Returns ------- - list[:class:`.LayoutObjInstance`] or tuple[list[:class:`.LayoutObjInstance`], list[:class:`.LayoutObjInstance`]] - If a polygonal spatial filter is specified, a tuple of lists of hits is returned in this - format: ``[, ]``. - Otherwise, a list containing all hits is returned. + :term:`LayoutInstanceQueryResult` or list of :term:`LayoutInstanceQueryResult` + If a single query is provided, one :term:`LayoutInstanceQueryResult` is returned. If a \ + list of spatial queries is provided, then a list of :term:`LayoutInstanceQueryResult` is \ + returned where each entry maps to the spatial query at the same index. """ def to_msg_filter_list(client_filter, ref_msg_type): - return ( - utils.map_list(client_filter, ref_msg_type) if client_filter is not None else None + if client_filter is None: + return None + return utils.map_list(utils.ensure_is_list(client_filter), ref_msg_type) + + def spatial_filter_to_msg(_spatial_filter): + is_point_filter = isinstance(_spatial_filter, PointData) + spatial_filter_field = "point_filter" if is_point_filter else "region_filter" + spatial_filter_msg = ( + point_message(_spatial_filter) + if is_point_filter + else polygon_data_message(_spatial_filter) + ) + return layout_instance_pb2.LayoutObjInstancesQueryMessage( + **{spatial_filter_field: spatial_filter_msg} ) - msg_params = { + # Create queries + lyt_inst_net_filter_lyr_filter_params = { "layout_inst": self.msg, "layer_filter": to_msg_filter_list(layer_filter, layer_ref_message), "net_filter": to_msg_filter_list(net_filter, net_ref_message), } - if spatial_filter is not None: - is_point_filter = isinstance(spatial_filter, PointData) - spatial_filter_field = "point_filter" if is_point_filter else "region_filter" - spatial_filter_msg = ( - point_message(spatial_filter) - if is_point_filter - else polygon_data_message(spatial_filter) + requests = [ + layout_instance_pb2.LayoutObjInstancesQueryMessage( + **lyt_inst_net_filter_lyr_filter_params ) - msg_params[spatial_filter_field] = spatial_filter_msg + ] + has_spatial_filter = spatial_filter is not None + if has_spatial_filter: + for sf in utils.ensure_is_list(spatial_filter): + requests.append(spatial_filter_to_msg(sf)) - request = layout_instance_pb2.LayoutObjInstancesQueryMessage(**msg_params) all_hits = [] - if (cache := get_cache()) is not None: - for hits_chunk in self.__stub.StreamLayoutObjInstancesQuery(request): - all_hits.append(hits_chunk) - else: - all_hits.append(self.__stub.QueryLayoutObjInstances(request)) + for hits_chunk in self.__stub.StreamLayoutObjInstancesQuery( + self._query_request_iterator(requests) + ): + for hit in hits_chunk.query_results: + all_hits.append(hit) - if isinstance(spatial_filter, PolygonData): + def process_hits(_spatial_filter, hits_iter): full_hits = [] partial_hits = [] - for hits in all_hits: - full_partial_hits = hits.full_partial_hits - for full_hit in full_partial_hits.full.items: - full_hits.append(layout_obj_instance.LayoutObjInstance(full_hit)) - for partial_hit in full_partial_hits.partial.items: - partial_hits.append(layout_obj_instance.LayoutObjInstance(partial_hit)) - return full_hits, partial_hits + for hit in hits_iter: + if hit.is_end_of_query_results_group: + break + lyt_obj_inst = layout_obj_instance.LayoutObjInstance(hit.edb_obj) + if hit.is_partial: + partial_hits.append(lyt_obj_inst) + else: + full_hits.append(lyt_obj_inst) + return ( + (full_hits, partial_hits) if isinstance(_spatial_filter, PolygonData) else full_hits + ) + + all_hits_iter = iter(all_hits) + if not has_spatial_filter: + return process_hits(None, all_hits_iter) + elif len(spatial_filter) == 1: + return process_hits(spatial_filter[1], all_hits_iter) else: - return [ - layout_obj_instance.LayoutObjInstance(hit) - for hits in all_hits - for hit in hits.hits.items - ] + return [process_hits(sf, all_hits_iter) for sf in spatial_filter] def get_layout_obj_instance_in_context(self, layout_obj, context): """Get the layout object instance of the given :term:`connectable ` in the provided context.