LCRConnector/legacy/src/lcr_connector/resolver/lut.py

import bisect
from itertools import chain, product
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, ResponseDeduper


class LutItem:
    """
    An item in the lookup table.
    """

    __circuit: Circuit
    """The circuit represented by this item."""
    __device_kind: DeviceKind
    """The device kind applied for this circuit."""

    def __init__(self, circuit: Circuit, device_kind: DeviceKind):
        self.__circuit = circuit
        self.__device_kind = device_kind

    @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.__circuit.compute(self.__device_kind)


class LutResolver(Resolver):
    """
    A resolver that uses a lookup table to find the best matching circuit.
    """

    __resistor_lut: list[LutItem]
    """The lookup table for resistors."""
    __capacitor_lut: list[LutItem]
    """The lookup table for capacitors."""
    __inductor_lut: list[LutItem]
    """The lookup table for inductors."""

    def __init__(self, datasets: DatasetCollection):
        self.__resistor_lut = self.__build_lut(
            datasets.resistor_values, DeviceKind.RESISTOR
        )
        self.__capacitor_lut = self.__build_lut(
            datasets.capacitor_values, DeviceKind.CAPACITOR
        )
        self.__inductor_lut = self.__build_lut(
            datasets.inductor_values, DeviceKind.INDUCTOR
        )

    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
                    )
                ),
            )
        ]
        lut.sort(key=lambda item: item.value)
        return lut

    def resolve(self, request: Request) -> Response:
        lut: list[LutItem]
        match request.device_kind:
            case DeviceKind.RESISTOR:
                lut = self.__resistor_lut
            case DeviceKind.CAPACITOR:
                lut = self.__capacitor_lut
            case DeviceKind.INDUCTOR:
                lut = self.__inductor_lut

        target = request.target_value
        count_limit = min(request.count_limit, 100)
        deduper = ResponseDeduper()

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