LCRConnector/legacy/resolver/lut.py

import struct
from typing import Iterator, BinaryIO
from pathlib import Path
from .common import Resolver, ResolverRequest, ResolverResult, ResultPriority
from ..dataset import DatasetCollection
from ..common import Circuit, SubCircuit, JointKind, LcrConnException

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

    lut: tuple[Circuit]

    def __init__(self, lut: tuple[Circuit]):
        self.lut = lut

    @staticmethod
    def from_dataset(dataset: DatasetCollection) -> 'LutResolver':
        pass

    @staticmethod
    def from_cache(filename: Path) -> 'LutResolver':
        with open(filename, "rb") as f:
            cnt = _read_int(f)
            return LutResolver(tuple(LutItem.from_cache(f) for _ in range(cnt)))
    
    def save_as_cache(self, filename: Path) -> None:
        with open(filename, "wb") as f:
            _write_int(f, len(self.lut))
            for item in self.lut:
                item.save_as_cache(f)

    def resolve(self, request: ResolverRequest) -> Iterator[ResolverResult]:
        pass


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

    circuit: Circuit
    """The circuit represented by this item."""
    __value_cache: float | None
    """The cached computed value of the circuit, or None if it has not been cached yet."""

    def __init__(self, circuit: Circuit):
        self.circuit = circuit

    @staticmethod
    def from_cache(f: BinaryIO) -> 'LutItem':
        cnt = _read_int(f)

        if cnt < 1:
            raise LcrConnException("Invalid circuit count in LUT item")
        device_value = _read_double(f)
        circuit = Circuit(device_value)
        cnt -= 1

        for _ in range(cnt):
            j = JointKind.SERIES if _read_bool(f) else JointKind.PARALLEL
            dev = _read_double(f)
            joint = SubCircuit(j, dev)
            circuit.add_joint(joint)
        
        return LutItem(circuit)

    def save_as_cache(self, f: BinaryIO) -> None:
        _write_int(f, self.circuit.len_devices())
        _write_double(f, self.circuit.__first_device_value)
        for joint in self.circuit.joints():
            _write_bool(f, joint.kind == JointKind.SERIES)
            _write_double(f, joint.value)

    def compute(self) -> float:
        """The computed value of the circuit."""
        if self.__value_cache is None:
            self.__value_cache = self.circuit.value()
        return self.__value_cache



DOUBLE_PACKER = struct.Struct("d")
INT_PACKER = struct.Struct("I")
BOOL_PACKER = struct.Struct("?")

def _read_double(fs) -> float:
    return DOUBLE_PACKER.unpack(fs.read(DOUBLE_PACKER.size))[0]

def _read_int(fs) -> int:
    return INT_PACKER.unpack(fs.read(INT_PACKER.size))[0]

def _read_bool(fs) -> bool:
    return BOOL_PACKER.unpack(fs.read(BOOL_PACKER.size))[0]

def _write_double(fs, num: float):
    fs.write(DOUBLE_PACKER.pack(num))

def _write_int(fs, num: int):
    fs.write(INT_PACKER.pack(num))

def _write_bool(fs, num: bool):
    fs.write(BOOL_PACKER.pack(num))