From 7fa0f56495bd6f29ee07aaf900a45637a5c3a2c7 Mon Sep 17 00:00:00 2001 From: yyc12345 Date: Thu, 25 Jun 2026 12:31:09 +0800 Subject: [PATCH] feat: remove astar, use bfs instead --- legacy/src/lcr_connector/__init__.py | 10 +- legacy/src/lcr_connector/common.py | 105 +++++++ legacy/src/lcr_connector/query.py | 131 ++------- legacy/src/lcr_connector/resolver/__init__.py | 4 +- legacy/src/lcr_connector/resolver/astar.py | 17 -- legacy/src/lcr_connector/resolver/bfs.py | 259 ++++++++++++++++++ legacy/src/lcr_connector/resolver/lut.py | 46 ++-- 7 files changed, 406 insertions(+), 166 deletions(-) delete mode 100644 legacy/src/lcr_connector/resolver/astar.py create mode 100644 legacy/src/lcr_connector/resolver/bfs.py diff --git a/legacy/src/lcr_connector/__init__.py b/legacy/src/lcr_connector/__init__.py index d9cfe61..65067ae 100644 --- a/legacy/src/lcr_connector/__init__.py +++ b/legacy/src/lcr_connector/__init__.py @@ -10,7 +10,7 @@ from .dataset import ( from_human_readable_value, ) from .query import Request, ResponsePriority, Response -from .resolver import Resolver, LutResolver, AStarResolver +from .resolver import Resolver, LutResolver, BfsResolver _TStrEnum = TypeVar("_TStrEnum", bound=enum.StrEnum) @@ -22,8 +22,8 @@ class AppResolver(enum.StrEnum): LUT = "lut" """The look-up table resolver.""" - ASTAR = "astar" - """The A* resolver.""" + BFS = "bfs" + """The BFS resolver.""" @dataclass @@ -65,8 +65,8 @@ class App: match self.__config.resolver: case AppResolver.LUT: self.__resolver = LutResolver(self.__dataset) - case AppResolver.ASTAR: - self.__resolver = AStarResolver(self.__dataset) + case AppResolver.BFS: + self.__resolver = BfsResolver(self.__dataset) def run(self) -> None: """ diff --git a/legacy/src/lcr_connector/common.py b/legacy/src/lcr_connector/common.py index 45e5299..8ed338c 100644 --- a/legacy/src/lcr_connector/common.py +++ b/legacy/src/lcr_connector/common.py @@ -1,4 +1,5 @@ import enum +from typing import Optional class LcrConnException(Exception): @@ -267,3 +268,107 @@ class Circuit: return self.__third_device_subckt.device_value else: raise LcrConnException("No third device") + + +class CircuitValueTrait: + """The trait for handful circuit computation""" + + __device_kind: DeviceKind + """The kind of the device""" + __target_value: float + """The target value""" + + def __init__(self, device_kind: DeviceKind, target_value: float) -> None: + self.__device_kind = device_kind + self.__target_value = target_value + + def value(self, circuit: Circuit) -> float: + """ + The value of this circuit. + + :param circuit: The circuit for computation. + :return: The value. + """ + return circuit.compute(self.__device_kind) + + def difference(self, circuit: Circuit, value: Optional[float] = None) -> float: + """ + The signed difference between the target value and the value of this circuit. + + Positive value indicates that the value of this circuit is greater than the target value. + Negative value indicates that the value of this circuit is less than the target value. + + :param circuit: The circuit for computation. + :param value: The value of the circuit computed by the `value` method + for reducing computation steps, or None if you request this method to compute the value. + :return: The signed difference. + """ + if value is None: + value = self.value(circuit) + return value - self.__target_value + + def unsigned_difference( + self, + circuit: Circuit, + value: Optional[float] = None, + difference: Optional[float] = None, + ) -> float: + """ + The unsigned difference between the target value and the value of this circuit. + + :param circuit: The circuit for computation. + :param value: The value of the circuit computed by the `value` method + for reducing computation steps, or None if you request this method to compute the value. + :param difference: The difference of the circuit computed by the `difference` method + for reducing computation steps, or None if you request this method to compute the difference. + :return: The unsigned difference. + """ + if difference is None: + difference = self.difference(circuit, value) + return abs(difference) + + def relative_difference( + self, + circuit: Circuit, + value: Optional[float] = None, + difference: Optional[float] = None, + ) -> float: + """ + The signed relative difference between the target value and the value of this circuit. + + Positive value indicates that the value of this circuit is greater than the target value. + Negative value indicates that the value of this circuit is less than the target value. + + :param circuit: The circuit for computation. + :param value: The value of the circuit computed by the `value` method + for reducing computation steps, or None if you request this method to compute the value. + :param difference: The difference of the circuit computed by the `difference` method + for reducing computation steps, or None if you request this method to compute the difference. + :return: The signed relative difference. + """ + if difference is None: + difference = self.difference(circuit, value) + return difference / self.__target_value + + def unsigned_relative_difference( + self, + circuit: Circuit, + value: Optional[float] = None, + difference: Optional[float] = None, + relative_difference: Optional[float] = None, + ) -> float: + """ + The unsigned relative difference between the target value and the value of this circuit. + + :param circuit: The circuit for computation. + :param value: The value of the circuit computed by the `value` method + for reducing computation steps, or None if you request this method to compute the value. + :param difference: The difference of the circuit computed by the `difference` method + for reducing computation steps, or None if you request this method to compute the difference. + :param relative_difference: The relative difference of the circuit computed by the `relative_difference` method + for reducing computation steps, or None if you request this method to compute the relative difference. + :return: The unsigned relative difference. + """ + if relative_difference is None: + relative_difference = self.relative_difference(circuit, value, difference) + return abs(relative_difference) diff --git a/legacy/src/lcr_connector/query.py b/legacy/src/lcr_connector/query.py index f72e395..f9dd630 100644 --- a/legacy/src/lcr_connector/query.py +++ b/legacy/src/lcr_connector/query.py @@ -1,9 +1,8 @@ import enum -import struct from functools import cached_property from dataclasses import dataclass from typing import Iterable, Iterator -from .common import DeviceKind, Circuit, CircuitDeviceScale +from .common import DeviceKind, Circuit, CircuitValueTrait class ResponsePriority(enum.Enum): @@ -35,92 +34,6 @@ class Request: """The limited count of results.""" -class ResponseDeduperItem: - """ - The item for response deduplicator. - """ - - __circuit: Circuit - """The circuit of this deduplicator item.""" - - def __init__(self, circuit: Circuit) -> None: - self.__circuit = circuit - - __ONE_PACKER = struct.Struct("=id") - __TWO_PACKER = struct.Struct("=iidd") - __THREE_PACKER = struct.Struct("=iiiddd") - - @cached_property - def __uniform_circuit_presentation(self) -> bytes: - c = self.__circuit - - match c.device_scale: - case CircuitDeviceScale.ONE: - return self.__ONE_PACKER.pack(1, c.first_device_value) - - case CircuitDeviceScale.TWO: - v1, v2 = sorted([c.first_device_value, c.second_device_value]) - return self.__TWO_PACKER.pack(2, int(c.second_device_joint), v1, v2) - - case CircuitDeviceScale.THREE: - v1, v2, v3 = ( - c.first_device_value, - c.second_device_value, - c.third_device_value, - ) - j2, j3 = int(c.second_device_joint), int(c.third_device_joint) - - if j2 == j3: - v1, v2, v3 = sorted([v1, v2, v3]) - else: - v1, v2 = sorted([v1, v2]) - - return self.__THREE_PACKER.pack(3, j2, j3, v1, v2, v3) - - @cached_property - def __uniform_circuit_hash(self) -> int: - return hash(self.__uniform_circuit_presentation) - - def __eq__(self, other: object) -> bool: - if not isinstance(other, ResponseDeduperItem): - return False - return ( - self.__uniform_circuit_presentation == other.__uniform_circuit_presentation - ) - - def __hash__(self) -> int: - return self.__uniform_circuit_hash - - @property - def circuit(self) -> Circuit: - """ - The circuit of this response item. - - :return: The circuit. - """ - return self.__circuit - - -class ResponseDeduper: - """ - The deduplicator for response circuits to deduplicate equivalent circuits. - """ - - __circuits: set[ResponseDeduperItem] - - def __init__(self) -> None: - self.__circuits = set() - - def add(self, circuit: Circuit) -> None: - self.__circuits.add(ResponseDeduperItem(circuit)) - - def __len__(self) -> int: - return len(self.__circuits) - - def __iter__(self) -> Iterator[Circuit]: - return map(lambda x: x.circuit, self.__circuits) - - class ResponseItem: """ The possible solution given by the resolver. @@ -128,20 +41,12 @@ class ResponseItem: __circuit: Circuit """The circuit of the response item.""" - __value: float - """The value of the response circuit.""" - __difference: float - """The signed difference between the target value and the value of this circuit.""" - __relative_difference: float - """The signed relative difference between the target value and the value of this circuit.""" + __cv_trait: CircuitValueTrait + """The trait for computing circuit values.""" - def __init__( - self, circuit: Circuit, device_kind: DeviceKind, target_value: float - ) -> None: + def __init__(self, circuit: Circuit, cv_trait: CircuitValueTrait) -> None: self.__circuit = circuit - self.__value = self.__circuit.compute(device_kind) - self.__difference = self.__value - target_value - self.__relative_difference = self.__difference / target_value + self.__cv_trait = cv_trait @property def circuit(self) -> Circuit: @@ -161,16 +66,16 @@ class ResponseItem: """ return self.__circuit.device_scale.to_device_count() - @property + @cached_property def value(self) -> float: """ The value of this circuit. :return: The value. """ - return self.__value + return self.__cv_trait.value(self.__circuit) - @property + @cached_property def difference(self) -> float: """ The signed difference between the target value and the value of this circuit. @@ -180,18 +85,18 @@ class ResponseItem: :return: The signed difference. """ - return self.__difference + return self.__cv_trait.difference(self.__circuit, value=self.value) - @property + @cached_property def unsigned_difference(self) -> float: """ The unsigned difference between the target value and the value of this circuit. :return: The unsigned difference. """ - return abs(self.__difference) + return self.__cv_trait.unsigned_difference(self.__circuit, difference=self.difference) - @property + @cached_property def relative_difference(self) -> float: """ The signed relative difference between the target value and the value of this circuit. @@ -201,16 +106,18 @@ class ResponseItem: :return: The signed relative difference. """ - return self.__relative_difference + return self.__cv_trait.relative_difference(self.__circuit, difference=self.difference) - @property + @cached_property def unsigned_relative_difference(self) -> float: """ The unsigned relative difference between the target value and the value of this circuit. :return: The unsigned relative difference. """ - return abs(self.__relative_difference) + return self.__cv_trait.unsigned_relative_difference( + self.__circuit, relative_difference=self.relative_difference + ) class Response: @@ -226,9 +133,9 @@ class Response: """The sorted items by priority and difference.""" def __init__(self, request: Request, candidates: Iterable[Circuit]) -> None: + cv_trait = CircuitValueTrait(request.device_kind, request.target_value) self.__sorted_items = list( - ResponseItem(item, request.device_kind, request.target_value) - for item in candidates + ResponseItem(item, cv_trait) for item in candidates ) # Sort by different strategy diff --git a/legacy/src/lcr_connector/resolver/__init__.py b/legacy/src/lcr_connector/resolver/__init__.py index ba4f47e..3d29ef9 100644 --- a/legacy/src/lcr_connector/resolver/__init__.py +++ b/legacy/src/lcr_connector/resolver/__init__.py @@ -1,9 +1,9 @@ from .common import Resolver from .lut import LutResolver -from .astar import AStarResolver +from .bfs import BfsResolver __all__ = [ 'Resolver', 'LutResolver', - 'AStarResolver' + 'BfsResolver' ] diff --git a/legacy/src/lcr_connector/resolver/astar.py b/legacy/src/lcr_connector/resolver/astar.py deleted file mode 100644 index c4fe1e9..0000000 --- a/legacy/src/lcr_connector/resolver/astar.py +++ /dev/null @@ -1,17 +0,0 @@ -from typing import Iterator -from .common import Resolver -from ..dataset import DatasetCollection -from ..common import Circuit -from ..query import Request, Response - -class AStarResolver(Resolver): - """ - A resolver that uses A* algorithm to find the best matching circuit. - """ - - def __init__(self, dataset: DatasetCollection): - pass - - - def resolve(self, request: Request) -> Iterator[Circuit]: - pass diff --git a/legacy/src/lcr_connector/resolver/bfs.py b/legacy/src/lcr_connector/resolver/bfs.py new file mode 100644 index 0000000..644f230 --- /dev/null +++ b/legacy/src/lcr_connector/resolver/bfs.py @@ -0,0 +1,259 @@ +import heapq +from itertools import chain, combinations_with_replacement, product +from typing import Iterable, Iterator +from functools import cached_property +from .common import Resolver +from ..dataset import DatasetCollection, Dataset +from ..common import Circuit, DeviceKind, JointKind, CircuitValueTrait +from ..query import Request, Response + + +class BfsItem: + """ + The entry used in BFS iteration storing circuit and value. + """ + + __circuit: Circuit + """The circuit represented by this item.""" + __cv_trait: CircuitValueTrait + """The trait for computing circuit values.""" + + def __init__(self, circuit: Circuit, cv_trait: CircuitValueTrait): + self.__circuit = circuit + self.__cv_trait = cv_trait + + @property + def circuit(self) -> Circuit: + return self.__circuit + + @cached_property + def value(self) -> float: + """ + The computed value of the circuit. + + :return: The computed value. + """ + return self.__cv_trait.value(self.__circuit) + + @cached_property + def unsigned_difference(self) -> float: + """ + The unsigned difference between the target value and the value of this circuit. + + :return: The unsigned difference. + """ + return self.__cv_trait.unsigned_difference(self.__circuit, value=self.value) + + +class ResultBucket(Iterable[BfsItem]): + """ + A bounded bucket that keeps up to `N` LutItem entries with the smallest floats. + + When the bucket is full, inserting a new item only succeeds if its float + is less than the current maximum; the maximum is then evicted. + """ + + class ResultBucketItem: + """ + An item stored in a :class:`ResultBucket`. + """ + + __score: float + """The score associated with this item.""" + __item: BfsItem + """The underlying LutItem.""" + __seq: int + """ + Monotonic counter used as a tiebreaker when scores are equal, + ensuring that heapq never compares :class:`LutItem` directly. + """ + + def __init__(self, score: float, item: BfsItem, seq: int): + self.__score = score + self.__item = item + self.__seq = seq + + @property + def score(self) -> float: + """The score associated with this item.""" + return self.__score + + @property + def item(self) -> BfsItem: + """The underlying LutItem.""" + return self.__item + + def __lt__(self, other: "ResultBucket.ResultBucketItem") -> bool: + # heapq is a min-heap: it always pops the smallest element. + # We invert the comparison so that an item with a larger score + # is considered "smaller", effectively turning the min-heap + # into a max-heap (largest-score item at the top). + if self.__score != other.__score: + return self.__score > other.__score + # Counter tiebreaker: when scores are equal the later-inserted + # item (higher seq) is considered "smaller" and gets evicted first. + return self.__seq > other.__seq + + __n: int + """Maximum number of items the bucket can hold.""" + __heap: list[ResultBucketItem] + """ + Min-heap of :class:`ResultBucketItem`. The heap invariant is inverted + via :meth:`ResultBucketItem.__lt__` so the entry with the largest score + sits at index 0. + """ + __counter: int + """ + Monotonic counter fed to each :class:`ResultBucketItem` as a tiebreaker, + preventing heapq from comparing :class:`LutItem` on score collisions. + """ + + def __init__(self, n: int): + self.__n = n + self.__heap = [] + self.__counter = 0 + + def __len__(self) -> int: + return len(self.__heap) + + def __iter__(self) -> Iterator[BfsItem]: + for entry in self.__heap: + yield entry.item + + def insert(self, item: BfsItem, score: float) -> bool: + """ + Insert a :class:`LutItem` with the given score. + + If the bucket is not yet full the item is always inserted. + Otherwise the item is only inserted when *score* is smaller + than the largest score currently in the bucket; the entry + with the largest score is then evicted. + + :param item: The LutItem to insert. + :param score: The score associated with the item. + :return: ``True`` if the item was inserted, ``False`` otherwise. + """ + entry = ResultBucket.ResultBucketItem(score, item, self.__counter) + if len(self.__heap) < self.__n: + heapq.heappush(self.__heap, entry) + self.__counter += 1 + return True + if score >= self.__heap[0].score: + return False + heapq.heapreplace(self.__heap, entry) + self.__counter += 1 + return True + + +class BfsResolver(Resolver): + __datasets: DatasetCollection + + def __init__(self, datasets: DatasetCollection): + self.__datasets = datasets + + # YYC MARK: + # Some circuit are equivalent in topology. + # If we deduplicate these equaivalent circuit in building result, + # there are too complex works. + # So we should deduplicated these equivalent circuit at the beginning, + # i.e. when generating them. + # So following 3 function are taking this job. + + @staticmethod + def iter_one_device_circuit(dataset: Dataset) -> Iterator[Circuit]: + """ + Iterate all possible circuits with one device without repeating equivalent topology. + + :param dataset: The dataset to iterate. + :return: The iterator of circuits with one device. + """ + # Every single device is unique so we directly output them. + # This feature is insured by dataset itself. + return (Circuit.from_one_device(v1) for v1 in dataset.values) + + @staticmethod + def iter_two_devices_circuit(dataset: Dataset) -> Iterator[Circuit]: + """ + Iterate all possible circuits with two devices without repeating equivalent topology. + + :param dataset: The dataset to iterate. + :return: The iterator of circuits with two devices. + """ + # The two devices in this circuit is always swapable, + # so we iterate them without repeating. + return ( + Circuit.from_two_devices(v1, v2, j2) + for (v1, v2), j2 in product( + combinations_with_replacement(dataset.values, 2), + tuple(JointKind), + ) + ) + + @staticmethod + def iter_three_devices_circuit(dataset: Dataset) -> Iterator[Circuit]: + """ + Iterate all possible circuits with three devices without repeating equivalent topology. + + :param dataset: The dataset to iterate. + :return: The iterator of circuits with three devices. + """ + # For generating three devices circuit, + # it should be consisted by 2 parts. + return chain( + # First, the whole circuit has only one joint type. + # In this case, 3 devices are swapable and we should iterate them without repeating + ( + Circuit.from_three_devices(v1, v2, j, v3, j) + for (v1, v2, v3), j in product( + combinations_with_replacement(dataset.values, 3), + tuple(JointKind), + ) + ), + # Second, if the joint type is different, then the first 2 devices are swapable. + # So we need iterate them without repeating. + ( + Circuit.from_three_devices(v1, v2, j, v3, j.flip()) + for (v1, v2), v3, j in product( + combinations_with_replacement(dataset.values, 2), + dataset.values, + tuple(JointKind), + ) + ), + ) + + @staticmethod + def __bfs_iteration( + dataset: Dataset, cv_trait: CircuitValueTrait + ) -> Iterator[BfsItem]: + return ( + BfsItem(circuit, cv_trait) + for circuit in chain( + BfsResolver.iter_one_device_circuit(dataset), + BfsResolver.iter_two_devices_circuit(dataset), + BfsResolver.iter_three_devices_circuit(dataset), + ) + ) + + def resolve(self, request: Request) -> Response: + # Pick dataset from collection + dataset: Dataset + match request.device_kind: + case DeviceKind.RESISTOR: + dataset = self.__datasets.resistor_values + case DeviceKind.CAPACITOR: + dataset = self.__datasets.capacitor_values + case DeviceKind.INDUCTOR: + dataset = self.__datasets.inductor_values + + # Iterate circuit item one by one + bucket = ResultBucket(request.count_limit) + cv_trait = CircuitValueTrait(request.device_kind, request.target_value) + for item in BfsResolver.__bfs_iteration(dataset, cv_trait): + # If circuit absolute difference is out of tolerance, skip it directly. + if item.unsigned_difference > request.tolerance: + continue + # put it into bucket + bucket.insert(item, item.unsigned_difference) + + # Return result + return Response(request, map(lambda item: item.circuit, bucket)) diff --git a/legacy/src/lcr_connector/resolver/lut.py b/legacy/src/lcr_connector/resolver/lut.py index b575fbb..a942150 100644 --- a/legacy/src/lcr_connector/resolver/lut.py +++ b/legacy/src/lcr_connector/resolver/lut.py @@ -2,9 +2,10 @@ import bisect from itertools import chain, product from functools import cached_property from .common import Resolver +from .bfs import BfsResolver from ..dataset import DatasetCollection, Dataset -from ..common import Circuit, DeviceKind, JointKind -from ..query import Request, Response, ResponseDeduper +from ..common import Circuit, DeviceKind, JointKind, CircuitValueTrait +from ..query import Request, Response class LutItem: @@ -14,25 +15,20 @@ class LutItem: __circuit: Circuit """The circuit represented by this item.""" - __device_kind: DeviceKind - """The device kind applied for this circuit.""" + __value: float + """The value of this circuit.""" def __init__(self, circuit: Circuit, device_kind: DeviceKind): self.__circuit = circuit - self.__device_kind = device_kind + self.__value = self.__circuit.compute(device_kind) @property def circuit(self) -> Circuit: return self.__circuit - @cached_property + @property def value(self) -> float: - """ - The computed value of the circuit. - - :return: The computed value. - """ - return self.__circuit.compute(self.__device_kind) + return self.__value class LutResolver(Resolver): @@ -59,22 +55,12 @@ class LutResolver(Resolver): ) def __build_lut(self, dataset: Dataset, device_kind: DeviceKind) -> list[LutItem]: - values = dataset.values - joints = tuple(JointKind) lut = [ LutItem(circuit, device_kind) for circuit in chain( - (Circuit.from_one_device(v1) for v1 in values), - ( - Circuit.from_two_devices(v1, v2, j2) - for v1, v2, j2 in product(values, values, joints) - ), - ( - Circuit.from_three_devices(v1, v2, j2, v3, j3) - for v1, v2, j2, v3, j3 in product( - values, values, joints, values, joints - ) - ), + BfsResolver.iter_one_device_circuit(dataset), + BfsResolver.iter_two_devices_circuit(dataset), + BfsResolver.iter_three_devices_circuit(dataset), ) ] lut.sort(key=lambda item: item.value) @@ -91,8 +77,8 @@ class LutResolver(Resolver): lut = self.__inductor_lut target = request.target_value - count_limit = min(request.count_limit, 100) - deduper = ResponseDeduper() + count_limit = request.count_limit + bucket: list[Circuit] = [] # Locate the insertion point of target in the sorted LUT. # left/right start at the two nearest neighbours and expand outward. @@ -106,7 +92,7 @@ class LutResolver(Resolver): # difference order, so the first N items within tolerance are exactly # the N best matches. while left >= 0 or right < len(lut): - if len(deduper) >= count_limit: + if len(bucket) >= count_limit: break if left < 0: @@ -134,6 +120,6 @@ class LutResolver(Resolver): right = len(lut) continue - deduper.add(item.circuit) + bucket.append(item.circuit) - return Response(request, deduper) + return Response(request, bucket)