yyc12345/LCRConnector
1
0
Fork 0
Code Activity

feat: remove astar, use bfs instead

Browse Source
This commit is contained in:
yyc12345
2026-06-25 12:31:09 +08:00
parent 625628cc1a
commit 7fa0f56495
7 changed files with 406 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

259
legacy/src/lcr_connector/resolver/bfs.py Normal file
View File

@@ -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))