feat: update legacy
- use sorted lut and bisect to optimize lut resolver - add circuit decuper - add signed diff for response item
This commit is contained in:
@@ -1,11 +1,10 @@
|
||||
import heapq
|
||||
import bisect
|
||||
from itertools import chain, 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
|
||||
from ..query import Request, Response
|
||||
from ..query import Request, Response, ResponseDeduper
|
||||
|
||||
|
||||
class LutItem:
|
||||
@@ -36,106 +35,6 @@ class LutItem:
|
||||
return self.__circuit.compute(self.__device_kind)
|
||||
|
||||
|
||||
class ResultBucket(Iterable[LutItem]):
|
||||
"""
|
||||
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: LutItem
|
||||
"""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: LutItem, 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) -> LutItem:
|
||||
"""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[LutItem]:
|
||||
for entry in self.__heap:
|
||||
yield entry.item
|
||||
|
||||
def insert(self, item: LutItem, 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 LutResolver(Resolver):
|
||||
"""
|
||||
A resolver that uses a lookup table to find the best matching circuit.
|
||||
@@ -149,21 +48,20 @@ class LutResolver(Resolver):
|
||||
"""The lookup table for inductors."""
|
||||
|
||||
def __init__(self, datasets: DatasetCollection):
|
||||
self.__resistor_lut = LutResolver.__build_lut(
|
||||
self.__resistor_lut = self.__build_lut(
|
||||
datasets.resistor_values, DeviceKind.RESISTOR
|
||||
)
|
||||
self.__capacitor_lut = LutResolver.__build_lut(
|
||||
self.__capacitor_lut = self.__build_lut(
|
||||
datasets.capacitor_values, DeviceKind.CAPACITOR
|
||||
)
|
||||
self.__inductor_lut = LutResolver.__build_lut(
|
||||
self.__inductor_lut = self.__build_lut(
|
||||
datasets.inductor_values, DeviceKind.INDUCTOR
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __build_lut(dataset: Dataset, device_kind: DeviceKind) -> list[LutItem]:
|
||||
def __build_lut(self, dataset: Dataset, device_kind: DeviceKind) -> list[LutItem]:
|
||||
values = dataset.values
|
||||
joints = tuple(JointKind)
|
||||
return [
|
||||
lut = [
|
||||
LutItem(circuit, device_kind)
|
||||
for circuit in chain(
|
||||
(Circuit.from_one_device(v1) for v1 in values),
|
||||
@@ -179,9 +77,10 @@ class LutResolver(Resolver):
|
||||
),
|
||||
)
|
||||
]
|
||||
lut.sort(key=lambda item: item.value)
|
||||
return lut
|
||||
|
||||
def resolve(self, request: Request) -> Response:
|
||||
# Fetch LUT by device kind
|
||||
lut: list[LutItem]
|
||||
match request.device_kind:
|
||||
case DeviceKind.RESISTOR:
|
||||
@@ -191,16 +90,50 @@ class LutResolver(Resolver):
|
||||
case DeviceKind.INDUCTOR:
|
||||
lut = self.__inductor_lut
|
||||
|
||||
# Check LUT item one by one
|
||||
bucket = ResultBucket(min(request.count_limit, 100))
|
||||
for item in lut:
|
||||
# compute absolute difference
|
||||
difference = abs(request.target_value - item.value)
|
||||
# If it is out of tolerance, skip it directly.
|
||||
if difference > request.tolerance:
|
||||
continue
|
||||
# put it into bucket
|
||||
bucket.insert(item, difference)
|
||||
target = request.target_value
|
||||
count_limit = min(request.count_limit, 100)
|
||||
deduper = ResponseDeduper()
|
||||
|
||||
# Return result
|
||||
return Response(request, map(lambda item: item.circuit, bucket))
|
||||
# Locate the insertion point of target in the sorted LUT.
|
||||
# left/right start at the two nearest neighbours and expand outward.
|
||||
idx = bisect.bisect_left(lut, target, key=lambda item: item.value)
|
||||
left = idx - 1
|
||||
right = idx
|
||||
|
||||
# Expand outward non-symmetrically: at each step compare the two
|
||||
# candidates on each side and advance the one that is closer to the
|
||||
# target. This guarantees items are visited in strictly increasing
|
||||
# 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:
|
||||
break
|
||||
|
||||
if left < 0:
|
||||
go_left = False
|
||||
elif right >= len(lut):
|
||||
go_left = True
|
||||
else:
|
||||
go_left = (target - lut[left].value) <= (lut[right].value - target)
|
||||
|
||||
if go_left:
|
||||
item = lut[left]
|
||||
left -= 1
|
||||
else:
|
||||
item = lut[right]
|
||||
right += 1
|
||||
|
||||
diff = abs(target - item.value)
|
||||
# Since the LUT is sorted, values on each side only move further
|
||||
# from target as we advance. Once one side exceeds tolerance,
|
||||
# the rest of that side is guaranteed out of range — disable it.
|
||||
if diff > request.tolerance:
|
||||
if go_left:
|
||||
left = -1
|
||||
else:
|
||||
right = len(lut)
|
||||
continue
|
||||
|
||||
deduper.add(item.circuit)
|
||||
|
||||
return Response(request, deduper)
|
||||
|
||||
Reference in New Issue
Block a user