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, CircuitCalculator 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.""" __ccalc: CircuitCalculator """The trait for computing circuit values.""" def __init__(self, circuit: Circuit, ccalc: CircuitCalculator): self.__circuit = circuit self.__ccalc = ccalc @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.__ccalc.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.__ccalc.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, ccalc: CircuitCalculator ) -> Iterator[BfsItem]: return ( BfsItem(circuit, ccalc) 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) ccalc = CircuitCalculator(request.device_kind, request.target_value) for item in BfsResolver.__bfs_iteration(dataset, ccalc): # 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))