feat: finish legacy project and move it as packaged python project
This commit is contained in:
364
legacy/src/lcr_connector/__init__.py
Normal file
364
legacy/src/lcr_connector/__init__.py
Normal file
@@ -0,0 +1,364 @@
|
||||
import enum
|
||||
import argparse
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import TypeVar, Callable
|
||||
from .common import Circuit, CircuitDeviceScale, JointKind, DeviceKind
|
||||
from .dataset import (
|
||||
DatasetCollection,
|
||||
to_human_readable_value,
|
||||
from_human_readable_value,
|
||||
)
|
||||
from .query import Request, ResponsePriority, Response
|
||||
from .resolver import Resolver, LutResolver, AStarResolver
|
||||
|
||||
_TStrEnum = TypeVar("_TStrEnum", bound=enum.StrEnum)
|
||||
|
||||
|
||||
class AppResolver(enum.StrEnum):
|
||||
"""
|
||||
The resolver for the app.
|
||||
"""
|
||||
|
||||
LUT = "lut"
|
||||
"""The look-up table resolver."""
|
||||
ASTAR = "astar"
|
||||
"""The A* resolver."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class AppConfig:
|
||||
"""
|
||||
The configuration for the app.
|
||||
"""
|
||||
|
||||
resolver: AppResolver
|
||||
"""The resolver for the app."""
|
||||
|
||||
resistor_dataset: Path
|
||||
"""The path to the resistor dataset file."""
|
||||
capacitor_dataset: Path
|
||||
"""The path to the capacitor dataset file."""
|
||||
inductor_dataset: Path
|
||||
"""The path to the inductor dataset file."""
|
||||
|
||||
|
||||
class App:
|
||||
"""
|
||||
The app.
|
||||
"""
|
||||
|
||||
__config: AppConfig
|
||||
"""The configuration for the app."""
|
||||
__dataset: DatasetCollection
|
||||
"""The dataset for the app."""
|
||||
__resolver: Resolver
|
||||
"""The resolver for the app."""
|
||||
|
||||
def __init__(self, config: AppConfig) -> None:
|
||||
self.__config = config
|
||||
self.__dataset = DatasetCollection.from_file(
|
||||
self.__config.resistor_dataset,
|
||||
self.__config.capacitor_dataset,
|
||||
self.__config.inductor_dataset,
|
||||
)
|
||||
match self.__config.resolver:
|
||||
case AppResolver.LUT:
|
||||
self.__resolver = LutResolver(self.__dataset)
|
||||
case AppResolver.ASTAR:
|
||||
self.__resolver = AStarResolver(self.__dataset)
|
||||
|
||||
def run(self) -> None:
|
||||
"""
|
||||
Run the app.
|
||||
"""
|
||||
print("LCR Connector")
|
||||
print('Type "help" for more info. Type "exit" to quit.')
|
||||
self.__op_main()
|
||||
|
||||
class MainCmd(enum.StrEnum):
|
||||
QUERY = "query"
|
||||
HELP = "help"
|
||||
EXIT = "exit"
|
||||
|
||||
class QueryDeviceChoice(enum.StrEnum):
|
||||
RESISTOR = "r"
|
||||
CAPACITOR = "c"
|
||||
INDUCTOR = "l"
|
||||
|
||||
def to_device_kind(self) -> DeviceKind:
|
||||
match self:
|
||||
case App.QueryDeviceChoice.RESISTOR:
|
||||
return DeviceKind.RESISTOR
|
||||
case App.QueryDeviceChoice.CAPACITOR:
|
||||
return DeviceKind.CAPACITOR
|
||||
case App.QueryDeviceChoice.INDUCTOR:
|
||||
return DeviceKind.INDUCTOR
|
||||
|
||||
class QuerySortPriority(enum.StrEnum):
|
||||
LESS_DEVICES = "l"
|
||||
MORE_ACCURACY = "a"
|
||||
|
||||
def to_response_priority(self) -> ResponsePriority:
|
||||
match self:
|
||||
case App.QuerySortPriority.LESS_DEVICES:
|
||||
return ResponsePriority.LESS_DEVICES
|
||||
case App.QuerySortPriority.MORE_ACCURACY:
|
||||
return ResponsePriority.MORE_ACCURACY
|
||||
|
||||
class PageViewerCmd(enum.StrEnum):
|
||||
PREVIOUS_PAGE = "f"
|
||||
NEXT_PAGE = "b"
|
||||
QUIT = "q"
|
||||
|
||||
def __op_main(self) -> None:
|
||||
while True:
|
||||
match self.__accept_command(App.MainCmd):
|
||||
case "query":
|
||||
self.__op_query()
|
||||
case "help":
|
||||
print("LCR Connector Help:")
|
||||
print("")
|
||||
print("query: do a query.")
|
||||
print("help: show all command.")
|
||||
print("exit: exit this app.")
|
||||
case "exit":
|
||||
break
|
||||
|
||||
def __op_query(self) -> None:
|
||||
# collecting request infos
|
||||
print("What are you connecting?")
|
||||
print("r: resistor")
|
||||
print("l: inductor")
|
||||
print("c: capacitor")
|
||||
device_kind = self.__accept_command(App.QueryDeviceChoice).to_device_kind()
|
||||
|
||||
print("Your target value?")
|
||||
print('Example: "2.1k", "0.75m", "3.2M" and etc.')
|
||||
target_value = self.__accept_device_value()
|
||||
|
||||
print("Your tolerance?")
|
||||
print('It can be absolute value like "2.1k".')
|
||||
print('Or relative value to your target value like "19.5%".')
|
||||
tolerance = self.__accept_device_value_tolerance(target_value)
|
||||
|
||||
print("How to sort result?")
|
||||
print("l: less component")
|
||||
print("a: more accuracy")
|
||||
response_priority = self.__accept_command(
|
||||
App.QuerySortPriority
|
||||
).to_response_priority()
|
||||
|
||||
# build request and ask resolver
|
||||
request = Request(device_kind, target_value, tolerance, response_priority, 100)
|
||||
response = self.__resolver.resolve(request)
|
||||
|
||||
# use page viewer to show result
|
||||
self.__op_page_viewer(response)
|
||||
|
||||
def __op_page_viewer(self, response: Response) -> None:
|
||||
cnt = len(response)
|
||||
if cnt == 0:
|
||||
print("Sorry, no result!")
|
||||
print("Please consider adjusting your requirements and try again.")
|
||||
return
|
||||
|
||||
ITEMS_PER_PAGE: int = 10
|
||||
all_page = cnt // ITEMS_PER_PAGE
|
||||
current_page = 0
|
||||
|
||||
while True:
|
||||
# print list
|
||||
for i in range(ITEMS_PER_PAGE - 1):
|
||||
# build index and check it
|
||||
index = current_page * (ITEMS_PER_PAGE - 1) + i
|
||||
if index >= cnt:
|
||||
continue
|
||||
# fetch item and print it
|
||||
item = response[index]
|
||||
print(
|
||||
"Plan {0}\t{1}\t{2:.2%}".format(
|
||||
index + 1,
|
||||
to_human_readable_value(item.value),
|
||||
item.relative_difference,
|
||||
)
|
||||
)
|
||||
# print page footer
|
||||
print("")
|
||||
print("Page {} of {}.".format(current_page + 1, all_page + 1))
|
||||
print("f: previous page. b: next page. q: quit this viewer.")
|
||||
# check command
|
||||
match self.__accept_command(App.PageViewerCmd):
|
||||
case App.PageViewerCmd.PREVIOUS_PAGE:
|
||||
current_page = max(0, current_page - 1)
|
||||
case App.PageViewerCmd.NEXT_PAGE:
|
||||
current_page = min(all_page, current_page + 1)
|
||||
case App.PageViewerCmd.QUIT:
|
||||
break
|
||||
|
||||
def __accept_command(self, cmd_enum: type[_TStrEnum]) -> _TStrEnum:
|
||||
"""
|
||||
Accept a command from the user.
|
||||
|
||||
:param cmd_enum: The type of the command. It must be a subclass of `enum.StrEnum`.
|
||||
:return: The command. It is an instance of `cmd_enum`.
|
||||
"""
|
||||
while True:
|
||||
print("> ", end=None)
|
||||
words = input()
|
||||
words = words.strip()
|
||||
if words in cmd_enum:
|
||||
return cmd_enum(words)
|
||||
print("Unknown command, please try again.")
|
||||
|
||||
def __accept_device_value(self) -> float:
|
||||
while True:
|
||||
words = input()
|
||||
value = self.__parse_human_readable_value(words)
|
||||
if value is None:
|
||||
print("Wrong value, please try again.")
|
||||
else:
|
||||
return value
|
||||
|
||||
def __accept_device_value_tolerance(self, target_value: float) -> float:
|
||||
while True:
|
||||
words = input()
|
||||
|
||||
if words.endswith("%"):
|
||||
value = self.__parse_plain_float(
|
||||
words[:-1], lambda x: x >= 0 and x <= 100
|
||||
)
|
||||
if value is not None:
|
||||
value = value / 100 * target_value
|
||||
else:
|
||||
value = self.__parse_human_readable_value(words)
|
||||
|
||||
if value is None:
|
||||
print("Wrong value, please try again.")
|
||||
else:
|
||||
return value
|
||||
|
||||
def __parse_plain_float(
|
||||
self, user_value: str, checker: Callable[[float], bool]
|
||||
) -> float | None:
|
||||
"""
|
||||
Parse a plain float value.
|
||||
|
||||
:param user_value: The value to parse.
|
||||
:param checker: A function that checks if the input is valid.
|
||||
It takes a float as input and returns a bool. True means the input is valid, otherwise False.
|
||||
:return: The parsed value if it is valid, otherwise None.
|
||||
"""
|
||||
# try parsing it first
|
||||
try:
|
||||
value = float(user_value)
|
||||
except ValueError:
|
||||
return None
|
||||
# then check it by checker
|
||||
if checker(value):
|
||||
return value
|
||||
else:
|
||||
return None
|
||||
|
||||
def __parse_human_readable_value(self, user_value: str) -> float | None:
|
||||
"""
|
||||
Parse a human-readable device value.
|
||||
|
||||
:param user_value: The value to parse.
|
||||
:return: The parsed value if it is valid, otherwise None.
|
||||
"""
|
||||
# parse it
|
||||
try:
|
||||
value = from_human_readable_value(user_value)
|
||||
except ValueError:
|
||||
return None
|
||||
# then check its range
|
||||
if value > 0:
|
||||
return value
|
||||
else:
|
||||
return None
|
||||
|
||||
def __get_joint_kind_symbol(self, joint_kind: JointKind) -> str:
|
||||
match joint_kind:
|
||||
case JointKind.SERIES:
|
||||
return "S"
|
||||
case JointKind.PARALLEL:
|
||||
return "P"
|
||||
|
||||
def __illustrate_circuit(self, circuit: Circuit) -> None:
|
||||
match circuit.device_scale:
|
||||
case CircuitDeviceScale.ONE:
|
||||
dev1 = to_human_readable_value(circuit.first_device_value)
|
||||
print(f"{dev1}")
|
||||
case CircuitDeviceScale.TWO:
|
||||
dev1 = to_human_readable_value(circuit.first_device_value)
|
||||
j2 = self.__get_joint_kind_symbol(circuit.second_device_joint)
|
||||
dev2 = to_human_readable_value(circuit.second_device_value)
|
||||
print(f"[{j2}] ┬ {dev1}")
|
||||
print(f" └ {dev2}")
|
||||
case CircuitDeviceScale.THREE:
|
||||
dev1 = to_human_readable_value(circuit.first_device_value)
|
||||
j2 = self.__get_joint_kind_symbol(circuit.second_device_joint)
|
||||
dev2 = to_human_readable_value(circuit.second_device_value)
|
||||
j3 = self.__get_joint_kind_symbol(circuit.third_device_joint)
|
||||
dev3 = to_human_readable_value(circuit.third_device_value)
|
||||
print(f"[{j3}] ┬ [{j2}] ┬ {dev1}")
|
||||
print(f" │ └ {dev2}")
|
||||
print(f" └ {dev3}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="LCR Connector",
|
||||
description="Get the resistor, capacitor, or inductor circuit which has the closest value for your given value within at most 3 devices.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-s",
|
||||
"--resolver",
|
||||
dest="resolver",
|
||||
action="store",
|
||||
type=AppResolver,
|
||||
choices=[resolver.value for resolver in AppResolver],
|
||||
required=True,
|
||||
help="The resolver you want to use",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-r",
|
||||
"--resistor",
|
||||
dest="resistor_dataset",
|
||||
action="store",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="The path to the resistor dataset file",
|
||||
metavar="RESISTOR.TXT",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-l",
|
||||
"--inductor",
|
||||
dest="inductor_dataset",
|
||||
action="store",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="The path to the inductor dataset file",
|
||||
metavar="INDUCTOR.TXT",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-c",
|
||||
"--capacitor",
|
||||
dest="capacitor_dataset",
|
||||
action="store",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="The path to the capacitor dataset file",
|
||||
metavar="CAPACITOR.TXT",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
app_config = AppConfig(
|
||||
resolver=args.resolver,
|
||||
resistor_dataset=args.resistor_dataset,
|
||||
capacitor_dataset=args.capacitor_dataset,
|
||||
inductor_dataset=args.inductor_dataset,
|
||||
)
|
||||
app = App(app_config)
|
||||
app.run()
|
||||
269
legacy/src/lcr_connector/common.py
Normal file
269
legacy/src/lcr_connector/common.py
Normal file
@@ -0,0 +1,269 @@
|
||||
import enum
|
||||
|
||||
|
||||
class LcrConnException(Exception):
|
||||
"""The exception thrown by LCR Connector"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class DeviceKind(enum.IntEnum):
|
||||
"""The kind of device"""
|
||||
|
||||
RESISTOR = enum.auto()
|
||||
"""Resistor device"""
|
||||
CAPACITOR = enum.auto()
|
||||
"""Capacitor device"""
|
||||
INDUCTOR = enum.auto()
|
||||
"""Inductor device"""
|
||||
|
||||
|
||||
class JointKind(enum.IntEnum):
|
||||
"""The joint type between 2 devices"""
|
||||
|
||||
SERIES = enum.auto()
|
||||
"""Series connection"""
|
||||
PARALLEL = enum.auto()
|
||||
"""Parallel connection"""
|
||||
|
||||
def flip(self) -> "JointKind":
|
||||
"""
|
||||
Flip the joint kind
|
||||
|
||||
Flip the joint kind from series to parallel or vice versa
|
||||
|
||||
:return: The flipped joint kind
|
||||
"""
|
||||
match self:
|
||||
case JointKind.SERIES:
|
||||
return JointKind.PARALLEL
|
||||
case JointKind.PARALLEL:
|
||||
return JointKind.SERIES
|
||||
|
||||
|
||||
class SubCircuit:
|
||||
"""The part of circuit composed of two devices and the joint kind"""
|
||||
|
||||
__device_value: float
|
||||
"""The value of the device"""
|
||||
__joint_kind: JointKind
|
||||
"""The joint kind between this device and the next device"""
|
||||
|
||||
def __init__(self, device_value: float, joint_kind: JointKind):
|
||||
self.__device_value = device_value
|
||||
self.__joint_kind = joint_kind
|
||||
|
||||
def compute(self, value: float, device_kind: DeviceKind) -> float:
|
||||
"""
|
||||
Compute the joint value
|
||||
|
||||
:param value: The value computed from previous devices
|
||||
:param device_kind: The kind of the device
|
||||
:return: The joint value computed
|
||||
"""
|
||||
if self.__device_value <= 0 or value <= 0:
|
||||
raise LcrConnException("Device value must be greater than 0")
|
||||
|
||||
# We perform series connect for: series resistor, series inductor and parallel capacitor.
|
||||
# We perform parallel connect for: parallel resistor, parallel inductor and series capacitor.
|
||||
joint_kind = self.__joint_kind
|
||||
if device_kind == DeviceKind.CAPACITOR:
|
||||
joint_kind = joint_kind.flip()
|
||||
|
||||
match joint_kind:
|
||||
case JointKind.SERIES:
|
||||
return self.__device_value + value
|
||||
case JointKind.PARALLEL:
|
||||
return (self.__device_value * value) / (self.__device_value + value)
|
||||
|
||||
@property
|
||||
def device_value(self) -> float:
|
||||
"""
|
||||
Get the device value
|
||||
|
||||
:return: The device value
|
||||
"""
|
||||
return self.__device_value
|
||||
|
||||
@property
|
||||
def joint_kind(self) -> JointKind:
|
||||
"""
|
||||
Get the joint kind
|
||||
|
||||
:return: The joint kind
|
||||
"""
|
||||
return self.__joint_kind
|
||||
|
||||
|
||||
class CircuitDeviceScale(enum.IntEnum):
|
||||
"""The scale of devices in the circuit"""
|
||||
|
||||
ONE = enum.auto()
|
||||
"""One device"""
|
||||
TWO = enum.auto()
|
||||
"""Two devices"""
|
||||
THREE = enum.auto()
|
||||
"""Three devices"""
|
||||
|
||||
def to_device_count(self) -> int:
|
||||
"""
|
||||
Convert circuit device scale to device count
|
||||
|
||||
:return: The device count
|
||||
"""
|
||||
match self:
|
||||
case CircuitDeviceScale.ONE:
|
||||
return 1
|
||||
case CircuitDeviceScale.TWO:
|
||||
return 2
|
||||
case CircuitDeviceScale.THREE:
|
||||
return 3
|
||||
|
||||
|
||||
class Circuit:
|
||||
"""The circuit composed of multiple joints"""
|
||||
|
||||
__first_device_value: float
|
||||
"""The value of the first device"""
|
||||
__second_device_subckt: SubCircuit | None
|
||||
"""The second device and its joint property"""
|
||||
__third_device_subckt: SubCircuit | None
|
||||
"""The third device and its joint property"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
first_device_value: float,
|
||||
second_device_subckt: SubCircuit | None,
|
||||
third_device_subckt: SubCircuit | None,
|
||||
):
|
||||
"""
|
||||
Initialize the circuit
|
||||
|
||||
:param first_device_value: The value of the first device
|
||||
:param second_device_subckt: The second device and its joint property
|
||||
:param third_device_subckt: The third device and its joint property
|
||||
"""
|
||||
if second_device_subckt is None and third_device_subckt is not None:
|
||||
raise LcrConnException("Third device cannot exist without second device")
|
||||
|
||||
self.__first_device_value = first_device_value
|
||||
self.__second_device_subckt = second_device_subckt
|
||||
self.__third_device_subckt = third_device_subckt
|
||||
|
||||
@staticmethod
|
||||
def from_one_device(device1_value: float) -> "Circuit":
|
||||
return Circuit(device1_value, None, None)
|
||||
|
||||
@staticmethod
|
||||
def from_two_devices(
|
||||
device1_value: float, device2_value: float, device2_joint: JointKind
|
||||
) -> "Circuit":
|
||||
return Circuit(device1_value, SubCircuit(device2_value, device2_joint), None)
|
||||
|
||||
@staticmethod
|
||||
def from_three_devices(
|
||||
device1_value: float,
|
||||
device2_value: float,
|
||||
device2_joint: JointKind,
|
||||
device3_value: float,
|
||||
device3_joint: JointKind,
|
||||
) -> "Circuit":
|
||||
return Circuit(
|
||||
device1_value,
|
||||
SubCircuit(device2_value, device2_joint),
|
||||
SubCircuit(device3_value, device3_joint),
|
||||
)
|
||||
|
||||
def compute(self, device_kind: DeviceKind) -> float:
|
||||
"""
|
||||
Compute the circuit value
|
||||
|
||||
:param device_kind: The kind of the device
|
||||
:return: The circuit value
|
||||
"""
|
||||
if self.__first_device_value <= 0:
|
||||
raise LcrConnException("Device value must be greater than 0")
|
||||
|
||||
value = self.__first_device_value
|
||||
if self.__second_device_subckt is None:
|
||||
return value
|
||||
value = self.__second_device_subckt.compute(value, device_kind)
|
||||
if self.__third_device_subckt is None:
|
||||
return value
|
||||
value = self.__third_device_subckt.compute(value, device_kind)
|
||||
return value
|
||||
|
||||
@property
|
||||
def device_scale(self) -> CircuitDeviceScale:
|
||||
"""
|
||||
Get the device scale
|
||||
|
||||
:return: The device scale
|
||||
"""
|
||||
if self.__third_device_subckt is not None:
|
||||
return CircuitDeviceScale.THREE
|
||||
elif self.__second_device_subckt is not None:
|
||||
return CircuitDeviceScale.TWO
|
||||
else:
|
||||
return CircuitDeviceScale.ONE
|
||||
|
||||
@property
|
||||
def first_device_value(self) -> float:
|
||||
"""
|
||||
Get the value of the first device
|
||||
|
||||
:return: The value of the first device
|
||||
"""
|
||||
return self.__first_device_value
|
||||
|
||||
@property
|
||||
def second_device_joint(self) -> JointKind:
|
||||
"""
|
||||
Get the joint kind of the second device
|
||||
|
||||
:return: The joint kind of the second device
|
||||
:raises LcrConnException: If there is no second device
|
||||
"""
|
||||
if self.__second_device_subckt is not None:
|
||||
return self.__second_device_subckt.joint_kind
|
||||
else:
|
||||
raise LcrConnException("No second device")
|
||||
|
||||
@property
|
||||
def second_device_value(self) -> float:
|
||||
"""
|
||||
Get the value of the second device
|
||||
|
||||
:return: The value of the second device
|
||||
:raises LcrConnException: If there is no second device
|
||||
"""
|
||||
if self.__second_device_subckt is not None:
|
||||
return self.__second_device_subckt.device_value
|
||||
else:
|
||||
raise LcrConnException("No second device")
|
||||
|
||||
@property
|
||||
def third_device_joint(self) -> JointKind:
|
||||
"""
|
||||
Get the joint kind of the third device
|
||||
|
||||
:return: The joint kind of the third device
|
||||
:raises LcrConnException: If there is no third device
|
||||
"""
|
||||
if self.__third_device_subckt is not None:
|
||||
return self.__third_device_subckt.joint_kind
|
||||
else:
|
||||
raise LcrConnException("No third device")
|
||||
|
||||
@property
|
||||
def third_device_value(self) -> float:
|
||||
"""
|
||||
Get the value of the third device
|
||||
|
||||
:return: The value of the third device
|
||||
:raises LcrConnException: If there is no third device
|
||||
"""
|
||||
if self.__third_device_subckt is not None:
|
||||
return self.__third_device_subckt.device_value
|
||||
else:
|
||||
raise LcrConnException("No third device")
|
||||
184
legacy/src/lcr_connector/dataset.py
Normal file
184
legacy/src/lcr_connector/dataset.py
Normal file
@@ -0,0 +1,184 @@
|
||||
from typing import Iterable
|
||||
from pathlib import Path
|
||||
from .common import LcrConnException
|
||||
|
||||
|
||||
class Dataset:
|
||||
"""
|
||||
A list holding available standard values for resistor, capacitor or inductor.
|
||||
|
||||
Standard values is a collection of all possible values of specific device manufactured by electronic factory.
|
||||
In reality, it also can be replaced by all possible values of specific device provided by your laboratory.
|
||||
For example, your laboratory only provide resistor with 100 Ohm and 4.7k Ohm.
|
||||
This list will only contain 100 and 4.7k.
|
||||
"""
|
||||
|
||||
__values: tuple[float, ...]
|
||||
"""A list of available device gauge values"""
|
||||
|
||||
def __init__(self, values: tuple[float, ...]):
|
||||
# Check redundant parts
|
||||
valueset = set(values)
|
||||
if len(valueset) != len(values):
|
||||
raise LcrConnException(f"Duplicate item in standard value list")
|
||||
if len(valueset) == 0:
|
||||
raise LcrConnException(f"Empty standard value list is not allowed")
|
||||
# Ok, assign it
|
||||
self.__values = values
|
||||
|
||||
@staticmethod
|
||||
def from_iterable(stringfied_values: Iterable[str]) -> "Dataset":
|
||||
return Dataset(
|
||||
tuple(
|
||||
from_human_readable_value(stringfied_value)
|
||||
for stringfied_value in stringfied_values
|
||||
)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def from_text(text: str) -> "Dataset":
|
||||
lines = text.split("\n")
|
||||
legal_lines = filter(lambda line: line != "", (line.strip() for line in lines))
|
||||
return Dataset.from_iterable(legal_lines)
|
||||
|
||||
@staticmethod
|
||||
def from_file(filename: Path) -> "Dataset":
|
||||
with open(filename, "r", encoding="utf-8") as f:
|
||||
legal_lines = filter(lambda line: line != "", (line.strip() for line in f))
|
||||
return Dataset.from_iterable(legal_lines)
|
||||
|
||||
@property
|
||||
def values(self) -> tuple[float, ...]:
|
||||
"""
|
||||
Get the available standard values
|
||||
|
||||
:return: A tuple of available standard values
|
||||
"""
|
||||
return self.__values
|
||||
|
||||
|
||||
class DatasetCollection:
|
||||
"""
|
||||
The collection holding all standard values for resistor, capacitor and inductor respectively.
|
||||
"""
|
||||
|
||||
__resistor: Dataset
|
||||
"""A list of available device gauge values for resistor"""
|
||||
__capacitor: Dataset
|
||||
"""A list of available device gauge values for capacitor"""
|
||||
__inductor: Dataset
|
||||
"""A list of available device gauge values for inductor"""
|
||||
|
||||
def __init__(self, resistor: Dataset, capacitor: Dataset, inductor: Dataset):
|
||||
self.__resistor = resistor
|
||||
self.__capacitor = capacitor
|
||||
self.__inductor = inductor
|
||||
|
||||
@staticmethod
|
||||
def from_iterable(
|
||||
resistor: Iterable[str], capacitor: Iterable[str], inductor: Iterable[str]
|
||||
) -> "DatasetCollection":
|
||||
return DatasetCollection(
|
||||
Dataset.from_iterable(resistor),
|
||||
Dataset.from_iterable(capacitor),
|
||||
Dataset.from_iterable(inductor),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def from_text(resistor: str, capacitor: str, inductor: str) -> "DatasetCollection":
|
||||
return DatasetCollection(
|
||||
Dataset.from_text(resistor),
|
||||
Dataset.from_text(capacitor),
|
||||
Dataset.from_text(inductor),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def from_file(
|
||||
resistor: Path, capacitor: Path, inductor: Path
|
||||
) -> "DatasetCollection":
|
||||
return DatasetCollection(
|
||||
Dataset.from_file(resistor),
|
||||
Dataset.from_file(capacitor),
|
||||
Dataset.from_file(inductor),
|
||||
)
|
||||
|
||||
@property
|
||||
def resistor_values(self) -> Dataset:
|
||||
"""
|
||||
Get the available standard values for resistor
|
||||
|
||||
:return: A tuple of available standard values for resistor
|
||||
"""
|
||||
return self.__resistor
|
||||
|
||||
@property
|
||||
def capacitor_values(self) -> Dataset:
|
||||
"""
|
||||
Get the available standard values for capacitor
|
||||
|
||||
:return: A tuple of available standard values for capacitor
|
||||
"""
|
||||
return self.__capacitor
|
||||
|
||||
@property
|
||||
def inductor_values(self) -> Dataset:
|
||||
"""
|
||||
Get the available standard values for inductor
|
||||
|
||||
:return: A tuple of available standard values for inductor
|
||||
"""
|
||||
return self.__inductor
|
||||
|
||||
|
||||
def from_human_readable_value(strl: str) -> float:
|
||||
"""
|
||||
Convert human readable value to float
|
||||
|
||||
:param strl: The human readable value
|
||||
:return: The parsed float value
|
||||
:raises ValueError: If the input string is not a valid number
|
||||
"""
|
||||
strl = strl.strip()
|
||||
|
||||
if strl.endswith("n"):
|
||||
return float(strl[0:-1]) * 1e-12
|
||||
if strl.endswith("p"):
|
||||
return float(strl[0:-1]) * 1e-9
|
||||
if strl.endswith("u"):
|
||||
return float(strl[0:-1]) * 1e-6
|
||||
if strl.endswith("m"):
|
||||
return float(strl[0:-1]) * 1e-3
|
||||
if strl.endswith("k"):
|
||||
return float(strl[0:-1]) * 1e3
|
||||
if strl.endswith("M"):
|
||||
return float(strl[0:-1]) * 1e6
|
||||
if strl.endswith("G"):
|
||||
return float(strl[0:-1]) * 1e9
|
||||
return float(strl)
|
||||
|
||||
|
||||
def to_human_readable_value(v: float) -> str:
|
||||
"""
|
||||
Convert float value to human readable value
|
||||
|
||||
:param value: The float value
|
||||
:return: The human readable value
|
||||
"""
|
||||
if v / 1e-12 < 1e3:
|
||||
return "{:e} n".format(v / 1e-12)
|
||||
if v / 1e-9 < 1e3:
|
||||
return "{:.4f} p".format(v / 1e-9)
|
||||
if v / 1e-6 < 1e3:
|
||||
return "{:.4f} u".format(v / 1e-6)
|
||||
if v / 1e-3 < 1e3:
|
||||
return "{:.4f} m".format(v / 1e-3)
|
||||
if v < 1e3:
|
||||
return "{:.4f}".format(v)
|
||||
if v / 1e3 < 1e3:
|
||||
return "{:.4f} k".format(v / 1e3)
|
||||
if v / 1e6 < 1e3:
|
||||
return "{:.4f} M".format(v / 1e6)
|
||||
if v / 1e9 < 1e3:
|
||||
return "{:.4f} G".format(v / 1e9)
|
||||
|
||||
return "{:e}".format(v)
|
||||
345
legacy/src/lcr_connector/main.py
Normal file
345
legacy/src/lcr_connector/main.py
Normal file
@@ -0,0 +1,345 @@
|
||||
import math
|
||||
import struct
|
||||
import os
|
||||
import sys
|
||||
|
||||
OneComponentList = []
|
||||
TwoComponentList = []
|
||||
ThreeComponentList = []
|
||||
|
||||
ResultList = []
|
||||
|
||||
class OneComponent(object):
|
||||
Value1 = 0.0
|
||||
|
||||
Value = 0.0
|
||||
def PrintCircuit(self):
|
||||
print(OutputAsHuman(self.Value1))
|
||||
|
||||
class TwoComponent(object):
|
||||
Value1 = 0.0
|
||||
Value2 = 0.0
|
||||
IsSeries = True
|
||||
|
||||
Value = 0.0
|
||||
def PrintCircuit(self):
|
||||
print("[{}] ┬ {}".format('S' if self.IsSeries else 'P', OutputAsHuman(self.Value1)))
|
||||
print(" └ {}".format(OutputAsHuman(self.Value2)))
|
||||
|
||||
class ThreeComponent(object):
|
||||
Value1 = 0.0
|
||||
Value2 = 0.0
|
||||
Value3 = 0.0
|
||||
IsSeries1 = True
|
||||
IsSeries2 = True
|
||||
|
||||
Value = 0.0
|
||||
def PrintCircuit(self):
|
||||
print("[{}] ┬ [{}] ┬ {}".format('S' if self.IsSeries2 else 'P', 'S' if self.IsSeries1 else 'P', OutputAsHuman(self.Value1)))
|
||||
print(" │ └ {}".format(OutputAsHuman(self.Value2)))
|
||||
print(" └ {}".format(OutputAsHuman(self.Value3)))
|
||||
|
||||
class ResultStruct(object):
|
||||
Subtraction = 0.0
|
||||
ComponentCount = 0
|
||||
CorrespondingClass = None
|
||||
|
||||
def LoadFromFile(name):
|
||||
f = open(name, 'r', encoding = 'utf-8')
|
||||
|
||||
# read file to get one elements state
|
||||
while True:
|
||||
cache = f.readline()
|
||||
if cache == '':
|
||||
break
|
||||
|
||||
(status, value) = InputAsHuman(cache.strip())
|
||||
if status:
|
||||
newobj = None
|
||||
newobj = OneComponent()
|
||||
newobj.Value = value
|
||||
newobj.Value1 = value
|
||||
OneComponentList.append(newobj)
|
||||
|
||||
# construct two component list
|
||||
length = len(OneComponentList)
|
||||
for i_index in range(length):
|
||||
for j_index in range(i_index, length):
|
||||
i = OneComponentList[i_index]
|
||||
j = OneComponentList[j_index]
|
||||
|
||||
# series
|
||||
newobj = TwoComponent()
|
||||
newobj.Value1 = i.Value
|
||||
newobj.Value2 = j.Value
|
||||
newobj.IsSeries = True
|
||||
newobj.Value = i.Value + j.Value
|
||||
TwoComponentList.append(newobj)
|
||||
|
||||
# parallel
|
||||
newobj = TwoComponent()
|
||||
newobj.Value1 = i.Value
|
||||
newobj.Value2 = j.Value
|
||||
newobj.IsSeries = False
|
||||
newobj.Value = (i.Value * j.Value) / (i.Value + j.Value)
|
||||
TwoComponentList.append(newobj)
|
||||
|
||||
# construct three component list
|
||||
for i in OneComponentList:
|
||||
for j in TwoComponentList:
|
||||
# series
|
||||
newobj = ThreeComponent()
|
||||
newobj.Value1 = j.Value1
|
||||
newobj.Value2 = j.Value2
|
||||
newobj.Value3 = i.Value
|
||||
newobj.IsSeries1 = j.IsSeries
|
||||
newobj.IsSeries2 = True
|
||||
newobj.Value = j.Value + i.Value
|
||||
ThreeComponentList.append(newobj)
|
||||
|
||||
# parallel
|
||||
newobj = ThreeComponent()
|
||||
newobj.Value1 = j.Value1
|
||||
newobj.Value2 = j.Value2
|
||||
newobj.Value3 = i.Value
|
||||
newobj.IsSeries1 = j.IsSeries
|
||||
newobj.IsSeries2 = False
|
||||
newobj.Value = (j.Value * i.Value) / (j.Value + i.Value)
|
||||
ThreeComponentList.append(newobj)
|
||||
|
||||
f.close()
|
||||
SaveAsCache(name)
|
||||
|
||||
def LoadFromCache(name):
|
||||
f = open(name + '.cache', 'rb')
|
||||
counter = 0
|
||||
(counter, ) = struct.unpack("I", f.read(4))
|
||||
for i in range(counter):
|
||||
newobj = OneComponent()
|
||||
newobj.Value1 = ReadDouble(f)
|
||||
newobj.Value = ReadDouble(f)
|
||||
OneComponentList.append(newobj)
|
||||
(counter, ) = struct.unpack("I", f.read(4))
|
||||
for i in range(counter):
|
||||
newobj = TwoComponent()
|
||||
newobj.Value1 = ReadDouble(f)
|
||||
newobj.Value2 = ReadDouble(f)
|
||||
newobj.IsSeries = ReadBoolean(f)
|
||||
newobj.Value = ReadDouble(f)
|
||||
TwoComponentList.append(newobj)
|
||||
(counter, ) = struct.unpack("I", f.read(4))
|
||||
for i in range(counter):
|
||||
newobj = ThreeComponent()
|
||||
newobj.Value1 = ReadDouble(f)
|
||||
newobj.Value2 = ReadDouble(f)
|
||||
newobj.Value3 = ReadDouble(f)
|
||||
newobj.IsSeries1 = ReadBoolean(f)
|
||||
newobj.IsSeries2 = ReadBoolean(f)
|
||||
newobj.Value = ReadDouble(f)
|
||||
ThreeComponentList.append(newobj)
|
||||
|
||||
f.close()
|
||||
|
||||
def SaveAsCache(name):
|
||||
# in cache, is series should follow resistor mode
|
||||
f = open(name + '.cache', 'wb')
|
||||
WriteInt(f, len(OneComponentList))
|
||||
for i in OneComponentList:
|
||||
WriteDouble(f, i.Value1)
|
||||
WriteDouble(f, i.Value)
|
||||
WriteInt(f, len(TwoComponentList))
|
||||
for i in TwoComponentList:
|
||||
WriteDouble(f, i.Value1)
|
||||
WriteDouble(f, i.Value2)
|
||||
WriteBoolean(f, i.IsSeries)
|
||||
WriteDouble(f, i.Value)
|
||||
WriteInt(f, len(ThreeComponentList))
|
||||
for i in ThreeComponentList:
|
||||
WriteDouble(f, i.Value1)
|
||||
WriteDouble(f, i.Value2)
|
||||
WriteDouble(f, i.Value3)
|
||||
WriteBoolean(f, i.IsSeries1)
|
||||
WriteBoolean(f, i.IsSeries2)
|
||||
WriteDouble(f, i.Value)
|
||||
|
||||
f.close()
|
||||
|
||||
def ReadDouble(fs):
|
||||
return struct.unpack("d", fs.read(8))[0]
|
||||
|
||||
def ReadInt(fs):
|
||||
return struct.unpack("I", fs.read(4))[0]
|
||||
|
||||
def ReadBoolean(fs):
|
||||
return struct.unpack("?", fs.read(1))[0]
|
||||
|
||||
def WriteDouble(fs, num):
|
||||
fs.write(struct.pack("d", num))
|
||||
|
||||
def WriteInt(fs, num):
|
||||
fs.write(struct.pack("I", num))
|
||||
|
||||
def WriteBoolean(fs, num):
|
||||
fs.write(struct.pack("?", num))
|
||||
|
||||
def OutputAsHuman(v):
|
||||
if v / 1e-12 < 1e3:
|
||||
return "{:e} n".format(v / 1e-12)
|
||||
if v / 1e-9 < 1e3:
|
||||
return "{:.4f} p".format(v / 1e-9)
|
||||
if v / 1e-6 < 1e3:
|
||||
return "{:.4f} u".format(v / 1e-6)
|
||||
if v / 1e-3 < 1e3:
|
||||
return "{:.4f} m".format(v / 1e-3)
|
||||
if v < 1e3:
|
||||
return "{:.4f}".format(v)
|
||||
if v / 1e3 < 1e3:
|
||||
return "{:.4f} k".format(v / 1e3)
|
||||
if v / 1e6 < 1e3:
|
||||
return "{:.4f} M".format(v / 1e6)
|
||||
|
||||
return "{:e}".format(v)
|
||||
|
||||
def InputAsHuman(strl):
|
||||
try:
|
||||
if strl.endswith('n'):
|
||||
return (True, float(strl[0:-1]) * 1e-12)
|
||||
if strl.endswith('p'):
|
||||
return (True, float(strl[0:-1]) * 1e-9)
|
||||
if strl.endswith('u'):
|
||||
return (True, float(strl[0:-1]) * 1e-6)
|
||||
if strl.endswith('m'):
|
||||
return (True, float(strl[0:-1]) * 1e-3)
|
||||
if strl.endswith('k'):
|
||||
return (True, float(strl[0:-1]) * 1e3)
|
||||
if strl.endswith('M'):
|
||||
return (True, float(strl[0:-1]) * 1e6)
|
||||
return (True, float(strl))
|
||||
except:
|
||||
return (False, 0.0)
|
||||
|
||||
def ValidCommandInput(valid_list):
|
||||
while True:
|
||||
cache = input()
|
||||
if cache not in valid_list:
|
||||
print('Wrong command, please input again.')
|
||||
else:
|
||||
return cache
|
||||
|
||||
def ValidNumberInput():
|
||||
while True:
|
||||
cache = input()
|
||||
(status, num) = InputAsHuman(cache)
|
||||
if not status:
|
||||
print('Wrong number, please input again.')
|
||||
else:
|
||||
return num
|
||||
|
||||
def DoQuery():
|
||||
# get config
|
||||
|
||||
print('What are you connecting?')
|
||||
print('r: resistor')
|
||||
print('l: inductor')
|
||||
print('c: capacitor')
|
||||
mode = ValidCommandInput(['c', 'l', 'r'])
|
||||
is_resistor_mode = mode != 'c'
|
||||
|
||||
print('Your target value?')
|
||||
target = ValidNumberInput()
|
||||
|
||||
print('Your tolerance?')
|
||||
target_tolerance = ValidNumberInput()
|
||||
|
||||
print('How to sort result?')
|
||||
print('l: less component')
|
||||
print('a: more accuracy')
|
||||
sort_mode = ValidCommandInput(['l', 'a'])
|
||||
|
||||
# start computing
|
||||
print('Collecting and sorting data...')
|
||||
ResultList.clear()
|
||||
for i in OneComponentList:
|
||||
cache = i.Value - target
|
||||
if abs(cache) < target_tolerance:
|
||||
newobj = ResultStruct()
|
||||
newobj.Subtraction = cache
|
||||
newobj.ComponentCount = 1
|
||||
newobj.CorrespondingClass = i
|
||||
ResultList.append(newobj)
|
||||
for i in TwoComponentList:
|
||||
cache = i.Value - target
|
||||
if abs(cache) < target_tolerance:
|
||||
newobj = ResultStruct()
|
||||
newobj.Subtraction = cache
|
||||
newobj.ComponentCount = 2
|
||||
newobj.CorrespondingClass = i
|
||||
ResultList.append(newobj)
|
||||
for i in ThreeComponentList:
|
||||
cache = i.Value - target
|
||||
if abs(cache) < target_tolerance:
|
||||
newobj = ResultStruct()
|
||||
newobj.Subtraction = cache
|
||||
newobj.ComponentCount = 3
|
||||
newobj.CorrespondingClass = i
|
||||
ResultList.append(newobj)
|
||||
|
||||
if sort_mode == 'l':
|
||||
ResultList.sort(key = lambda x: (x.ComponentCount, abs(x.Subtraction)))
|
||||
else:
|
||||
ResultList.sort(key = lambda x: abs(x.Subtraction))
|
||||
|
||||
# display result
|
||||
if len(ResultList) == 0:
|
||||
print('Sorry, no result!')
|
||||
return
|
||||
|
||||
count = len(ResultList)
|
||||
all_page = int(count / 10)
|
||||
current_page = 0
|
||||
while current_page <= all_page:
|
||||
for i in range(9):
|
||||
picked_index = current_page * 9 + i
|
||||
if (picked_index < count):
|
||||
picked_item = ResultList[picked_index]
|
||||
print("Plan {}\t{}\t{:.2%}".format(picked_index + 1, OutputAsHuman(picked_item.CorrespondingClass.Value), picked_item.Subtraction / target))
|
||||
picked_item.CorrespondingClass.PrintCircuit()
|
||||
|
||||
print('')
|
||||
print("Page {} of {}. p: next page. q: exit".format(current_page + 1, all_page + 1))
|
||||
command = ValidCommandInput(['p', 'q'])
|
||||
if command == 'p':
|
||||
current_page += 1
|
||||
elif command == 'q':
|
||||
break
|
||||
|
||||
def DoHelp():
|
||||
print('LCR Connect Help:')
|
||||
print('')
|
||||
print('query: do a query immediately. following the guider and find the result.')
|
||||
print('help: print this')
|
||||
print('exit: exit this app')
|
||||
|
||||
# ===================================================== program start
|
||||
|
||||
print('LCR Connect')
|
||||
|
||||
# loading file
|
||||
print('Input the component value list file name:')
|
||||
filename = input()
|
||||
if not os.path.isfile(filename + '.cache'):
|
||||
LoadFromFile(filename)
|
||||
else:
|
||||
LoadFromCache(filename)
|
||||
|
||||
# ready for command
|
||||
while True:
|
||||
sys.stdout.write('> ')
|
||||
sys.stdout.flush()
|
||||
cmd = ValidCommandInput(['exit', 'query', 'help'])
|
||||
if cmd == 'exit':
|
||||
break
|
||||
elif cmd == 'query':
|
||||
DoQuery()
|
||||
elif cmd == 'help':
|
||||
DoHelp()
|
||||
137
legacy/src/lcr_connector/query.py
Normal file
137
legacy/src/lcr_connector/query.py
Normal file
@@ -0,0 +1,137 @@
|
||||
import enum
|
||||
from functools import cached_property
|
||||
from dataclasses import dataclass
|
||||
from typing import Iterator
|
||||
from .common import DeviceKind, Circuit
|
||||
|
||||
|
||||
class ResponsePriority(enum.Enum):
|
||||
"""
|
||||
The priority of the result.
|
||||
"""
|
||||
|
||||
LESS_DEVICES = enum.auto()
|
||||
"""Less devices is the first priority."""
|
||||
MORE_ACCURACY = enum.auto()
|
||||
"""More accuracy is the first priority."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class Request:
|
||||
"""
|
||||
All request infomation for the resolver.
|
||||
"""
|
||||
|
||||
device_kind: DeviceKind
|
||||
"""The kind of device to resolve."""
|
||||
target_value: float
|
||||
"""The target value of the device."""
|
||||
tolerance: float
|
||||
"""The tolerance of the device in absolute value."""
|
||||
response_priority: ResponsePriority
|
||||
"""The priority principle when sorting response items."""
|
||||
count_limit: int
|
||||
"""The limited count of results."""
|
||||
|
||||
|
||||
class ResponseItem:
|
||||
"""
|
||||
The possible solution given by the resolver.
|
||||
"""
|
||||
|
||||
__circuit: Circuit
|
||||
"""The circuit of the response item."""
|
||||
__device_kind: DeviceKind
|
||||
"""The kind of device of this circuit."""
|
||||
__target_value: float
|
||||
"""The target value of this circuit."""
|
||||
|
||||
def __init__(
|
||||
self, circuit: Circuit, device_kind: DeviceKind, target_value: float
|
||||
) -> None:
|
||||
self.__circuit = circuit
|
||||
self.__device_kind = device_kind
|
||||
self.__target_value = target_value
|
||||
|
||||
@property
|
||||
def circuit(self) -> Circuit:
|
||||
"""
|
||||
The circuit of this response item.
|
||||
|
||||
:return: The circuit.
|
||||
"""
|
||||
return self.circuit
|
||||
|
||||
@cached_property
|
||||
def device_count(self) -> int:
|
||||
"""
|
||||
The device count of this circuit.
|
||||
|
||||
:return: The device count.
|
||||
"""
|
||||
return self.circuit.device_scale.to_device_count()
|
||||
|
||||
@cached_property
|
||||
def value(self) -> float:
|
||||
"""
|
||||
The value of this circuit.
|
||||
|
||||
:return: The value.
|
||||
"""
|
||||
return self.__circuit.compute(self.__device_kind)
|
||||
|
||||
@cached_property
|
||||
def difference(self) -> float:
|
||||
"""
|
||||
The absolute difference between the target value and the value of this circuit.
|
||||
|
||||
:return: The absolute difference.
|
||||
"""
|
||||
return abs(self.__target_value - self.value)
|
||||
|
||||
@cached_property
|
||||
def relative_difference(self) -> float:
|
||||
"""
|
||||
The relative difference between the target value and the value of this circuit.
|
||||
|
||||
:return: The relative difference.
|
||||
"""
|
||||
return self.difference / self.__target_value
|
||||
|
||||
|
||||
class Response:
|
||||
"""
|
||||
The collection of possible solutions given by the resolver.
|
||||
|
||||
For getting the response items, please use `response[index]`.
|
||||
For iterating the response items, please use the iterator protocol.
|
||||
For getting the count of response items, please use the ``len`` function.
|
||||
"""
|
||||
|
||||
__sorted_items: list[ResponseItem]
|
||||
"""The sorted items by priority and difference."""
|
||||
|
||||
def __init__(self, request: Request, candidates: Iterator[Circuit]) -> None:
|
||||
self.__sorted_items = list(
|
||||
ResponseItem(item, request.device_kind, request.target_value)
|
||||
for item in candidates
|
||||
)
|
||||
|
||||
# Sort by different strategy
|
||||
match request.response_priority:
|
||||
case ResponsePriority.LESS_DEVICES:
|
||||
self.__sorted_items.sort(key=lambda x: (x.device_count, x.difference))
|
||||
case ResponsePriority.MORE_ACCURACY:
|
||||
self.__sorted_items.sort(key=lambda x: x.difference)
|
||||
|
||||
# Cut item by limit
|
||||
self.__sorted_items = self.__sorted_items[:request.count_limit]
|
||||
|
||||
def __getitem__(self, index: int) -> ResponseItem:
|
||||
return self.__sorted_items[index]
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.__sorted_items)
|
||||
|
||||
def __iter__(self) -> Iterator[ResponseItem]:
|
||||
return iter(self.__sorted_items)
|
||||
9
legacy/src/lcr_connector/resolver/__init__.py
Normal file
9
legacy/src/lcr_connector/resolver/__init__.py
Normal file
@@ -0,0 +1,9 @@
|
||||
from .common import Resolver
|
||||
from .lut import LutResolver
|
||||
from .astar import AStarResolver
|
||||
|
||||
__all__ = [
|
||||
'Resolver',
|
||||
'LutResolver',
|
||||
'AStarResolver'
|
||||
]
|
||||
17
legacy/src/lcr_connector/resolver/astar.py
Normal file
17
legacy/src/lcr_connector/resolver/astar.py
Normal file
@@ -0,0 +1,17 @@
|
||||
from typing import Iterator
|
||||
from .common import Resolver
|
||||
from ..dataset import DatasetCollection
|
||||
from ..common import Circuit
|
||||
from ..query import Request, Response
|
||||
|
||||
class AStarResolver(Resolver):
|
||||
"""
|
||||
A resolver that uses A* algorithm to find the best matching circuit.
|
||||
"""
|
||||
|
||||
def __init__(self, dataset: DatasetCollection):
|
||||
pass
|
||||
|
||||
|
||||
def resolve(self, request: Request) -> Iterator[Circuit]:
|
||||
pass
|
||||
11
legacy/src/lcr_connector/resolver/common.py
Normal file
11
legacy/src/lcr_connector/resolver/common.py
Normal file
@@ -0,0 +1,11 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from ..query import Request, Response
|
||||
|
||||
class Resolver(ABC):
|
||||
"""
|
||||
Abstract base class for all resolvers.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def resolve(self, request: Request) -> Response:
|
||||
pass
|
||||
206
legacy/src/lcr_connector/resolver/lut.py
Normal file
206
legacy/src/lcr_connector/resolver/lut.py
Normal file
@@ -0,0 +1,206 @@
|
||||
import heapq
|
||||
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
|
||||
|
||||
|
||||
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 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.
|
||||
"""
|
||||
|
||||
__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 = LutResolver.__build_lut(
|
||||
datasets.resistor_values, DeviceKind.RESISTOR
|
||||
)
|
||||
self.__capacitor_lut = LutResolver.__build_lut(
|
||||
datasets.capacitor_values, DeviceKind.CAPACITOR
|
||||
)
|
||||
self.__inductor_lut = LutResolver.__build_lut(
|
||||
datasets.inductor_values, DeviceKind.INDUCTOR
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __build_lut(dataset: Dataset, device_kind: DeviceKind) -> list[LutItem]:
|
||||
values = dataset.values
|
||||
joints = tuple(JointKind)
|
||||
return [
|
||||
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
|
||||
)
|
||||
),
|
||||
)
|
||||
]
|
||||
|
||||
def resolve(self, request: Request) -> Response:
|
||||
# Fetch LUT by device kind
|
||||
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
|
||||
|
||||
# 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)
|
||||
|
||||
# Return result
|
||||
return Response(request, map(lambda item: item.circuit, bucket))
|
||||
Reference in New Issue
Block a user