merge

Author: ducky64

edg/abstract_parts/AbstractInductor.py

@@ -1,7 +1,7 @@
 from typing import Dict, Optional, cast
 
 from ..electronics_model import *
-from .PartsTable import PartsTableColumn, PartsTableRow
+from .PartsTable import PartsTableColumn, PartsTableRow, ExperimentalUserFnPartsTable
 from .PartsTablePart import PartsTableSelector
 from .Categories import *
 from .StandardFootprint import StandardFootprint, HasStandardFootprint
@@ -94,7 +94,9 @@ def symbol_pinning(self, symbol_name: str) -> Dict[str, BasePort]:
   def __init__(self, inductance: RangeLike,
                current: RangeLike = RangeExpr.ZERO,
                frequency: RangeLike = RangeExpr.ZERO,
-               resistance_dc: RangeLike = (0, 1)*Ohm  # generic sane choice?
+               resistance_dc: RangeLike = (0, 1)*Ohm,  # generic sane choice?
+               *,
+               experimental_filter_fn: StringLike = ""
                ) -> None:
     super().__init__()
 
@@ -105,6 +107,7 @@ def __init__(self, inductance: RangeLike,
     self.current = self.ArgParameter(current)  # defined as operating current range, non-directioned
     self.frequency = self.ArgParameter(frequency)  # defined as operating frequency range
     self.resistance_dc = self.ArgParameter(resistance_dc)
+    self.experimental_filter_fn = self.ArgParameter(experimental_filter_fn)
 
     self.actual_inductance = self.Parameter(RangeExpr())
     self.actual_current_rating = self.Parameter(RangeExpr())
@@ -136,15 +139,23 @@ class TableInductor(PartsTableSelector, Inductor):
   @init_in_parent
   def __init__(self, *args, **kwargs) -> None:
     super().__init__(*args, **kwargs)
-    self.generator_param(self.inductance, self.current, self.frequency, self.resistance_dc)
+    self.generator_param(self.inductance, self.current, self.frequency, self.resistance_dc,
+                         self.experimental_filter_fn)
 
   def _row_filter(self, row: PartsTableRow) -> bool:
     # TODO eliminate arbitrary DCR limit in favor of exposing max DCR to upper levels
+    filter_fn_str = self.get(self.experimental_filter_fn)
+    if filter_fn_str:
+      filter_fn = ExperimentalUserFnPartsTable.deserialize_fn(filter_fn_str)
+    else:
+      filter_fn = None
+
     return super()._row_filter(row) and \
       row[self.INDUCTANCE].fuzzy_in(self.get(self.inductance)) and \
       self.get(self.current).fuzzy_in(row[self.CURRENT_RATING]) and \
       row[self.DC_RESISTANCE].fuzzy_in(self.get(self.resistance_dc)) and \
-      self.get(self.frequency).fuzzy_in(row[self.FREQUENCY_RATING])
+      self.get(self.frequency).fuzzy_in(row[self.FREQUENCY_RATING]) and\
+      (filter_fn is None or filter_fn(row))
 
   def _row_generate(self, row: PartsTableRow) -> None:
     super()._row_generate(row)

edg/abstract_parts/AbstractPowerConverters.py

@@ -106,6 +106,25 @@ def __init__(self, forced_current_draw: RangeLike) -> None:
     ), [Output])
 
 
+class ForcedVoltageCurrentLimit(DummyDevice, NetBlock):
+  """Forces some output current limit, which should be tighter than the input's actual current draw."""
+  @init_in_parent
+  def __init__(self, forced_current_limit: RangeLike) -> None:
+    super().__init__()
+
+    self.pwr_in = self.Port(VoltageSink(
+      current_draw=RangeExpr(),
+      voltage_limits=RangeExpr.ALL
+    ), [Input])
+
+    self.pwr_out = self.Port(VoltageSource(
+      voltage_out=self.pwr_in.link().voltage,
+      current_limits=forced_current_limit
+    ), [Output])
+
+    self.assign(self.pwr_in.current_draw, self.pwr_out.link().current_drawn)
+
+
 class ForcedVoltage(DummyDevice, NetBlock):
   """Forces some voltage on the output regardless of the input's actual voltage.
   Current draw is passed through unchanged."""

edg/abstract_parts/DummyDevices.py

@@ -2,14 +2,12 @@
 
 import csv
 import itertools
-import sys
 from typing import TypeVar, Generic, Type, overload, Union, Callable, List, Dict, Any, KeysView, Optional, OrderedDict, \
-  cast
+  cast, Tuple, Sequence, Protocol
 
-if sys.version_info[1] < 8:
-  from typing_extensions import Protocol
-else:
-  from typing import Protocol
+from typing_extensions import ParamSpec
+
+from ..core import Range
 
 
 # from https://stackoverflow.com/questions/47965083/comparable-types-with-mypy
@@ -152,3 +150,90 @@ def first(self, err="no elements in list") -> PartsTableRow:
     if not self.rows:
       raise IndexError(err)
     return self.rows[0]
+
+
+UserFnMetaParams = ParamSpec('UserFnMetaParams')
+UserFnType = TypeVar('UserFnType', bound=Callable, covariant=True)
+class UserFnSerialiable(Protocol[UserFnMetaParams, UserFnType]):
+  """A protocol that marks functions as usable in deserialize, that they have been registered."""
+  _is_serializable: None  # guard attribute
+
+  def __call__(self, *args: UserFnMetaParams.args, **kwargs: UserFnMetaParams.kwargs) -> UserFnType: ...
+  __name__: str
+
+
+class ExperimentalUserFnPartsTable(PartsTable):
+  """A PartsTable that can take in a user-defined function for filtering and (possibly) other operations.
+  These functions are serialized to a string by an internal name (cannot execute arbitrary code,
+  bounded to defined functions in the codebase), and some arguments can be serialized with the name
+  (think partial(...)).
+  Functions must be pre-registered using the @ExperimentalUserFnPartsTable.user_fn(...) decorator,
+  non-pre-registered functions will not be available.
+
+  This is intended to support searches on parts tables that are cross-coupled across multiple parameters,
+  but still restricted to within on table (e.g., no cross-optimizing RC filters).
+
+  EXPERIMENTAL - subject to change without notice."""
+
+  _FN_SERIALIZATION_SEPARATOR = ";"
+
+  _user_fns: Dict[str, Tuple[Callable, Sequence[Type]]] = {}  # name -> fn, [arg types]
+  _fn_name_dict: Dict[Callable, str] = {}
+
+  @staticmethod
+  def user_fn(param_types: Sequence[Type] = []) -> Callable[[Callable[UserFnMetaParams, UserFnType]],
+      UserFnSerialiable[UserFnMetaParams, UserFnType]]:
+    def decorator(fn: Callable[UserFnMetaParams, UserFnType]) -> UserFnSerialiable[UserFnMetaParams, UserFnType]:
+      """Decorator to register a user function that can be used in ExperimentalUserFnPartsTable."""
+      if fn.__name__ in ExperimentalUserFnPartsTable._user_fns or fn in ExperimentalUserFnPartsTable._fn_name_dict:
+          raise ValueError(f"Function {fn.__name__} already registered.")
+      ExperimentalUserFnPartsTable._user_fns[fn.__name__] = (fn, param_types)
+      ExperimentalUserFnPartsTable._fn_name_dict[fn] = fn.__name__
+      return fn  # type: ignore
+    return decorator
+
+  @classmethod
+  def serialize_fn(cls, fn: UserFnSerialiable[UserFnMetaParams, UserFnType],
+                   *args: UserFnMetaParams.args, **kwargs: UserFnMetaParams.kwargs) -> str:
+    """Serializes a user function to a string."""
+    assert not kwargs, "kwargs not supported in serialization"
+    if fn not in cls._fn_name_dict:
+      raise ValueError(f"Function {fn} not registered.")
+    fn_ctor, fn_argtypes = cls._user_fns[fn.__name__]
+    def serialize_arg(tpe: Type, val: Any) -> str:
+      assert isinstance(val, tpe), f"in serialize {val}, expected {tpe}, got {type(val)}"
+      if tpe is bool:
+        return str(val)
+      elif tpe is int:
+        return str(val)
+      elif tpe is float:
+        return str(val)
+      elif tpe is Range:
+        return f"({val.lower},{val.upper})"
+      else:
+        raise TypeError(f"cannot serialize type {tpe} in user function serialization")
+    serialized_args = [serialize_arg(tpe, arg) for tpe, arg in zip(fn_argtypes, args)]
+    return cls._FN_SERIALIZATION_SEPARATOR.join([fn.__name__] + serialized_args)
+
+  @classmethod
+  def deserialize_fn(cls, serialized: str) -> Callable:
+    """Deserializes a user function from a string."""
+    split = serialized.split(cls._FN_SERIALIZATION_SEPARATOR)
+    if split[0] not in cls._user_fns:
+      raise ValueError(f"Function {serialized} not registered.")
+    fn_ctor, fn_argtypes = cls._user_fns[split[0]]
+    assert len(split) == len(fn_argtypes) + 1
+    def deserialize_arg(tpe: Type, val: str) -> Any:
+      if tpe is bool:
+        return val == 'True'
+      elif tpe is int:
+        return int(val)
+      elif tpe is float:
+        return float(val)
+      elif tpe is Range:
+        parts = val[1:-1].split(",")
+        return Range(float(parts[0]), float(parts[1]))  # type: ignore
+      else:
+        raise TypeError(f"cannot deserialize type {tpe} in user function serialization")
+    deserialized_args = [deserialize_arg(tpe, arg) for tpe, arg in zip(fn_argtypes, split[1:])]
+    return fn_ctor(*deserialized_args)

edg/abstract_parts/PartsTable.py

@@ -46,6 +46,8 @@ def __init__(self, frequency: RangeLike, fet_rds: RangeLike = (0, 1)*Ohm,
         self.fet_rds = self.ArgParameter(fet_rds)
         self.gate_res = self.ArgParameter(gate_res)
 
+        self.actual_current_limits = self.Parameter(RangeExpr())
+
     def contents(self):
         super().contents()
         self.driver = self.Block(HalfBridgeDriver(has_boot_diode=True))
@@ -94,6 +96,9 @@ def contents(self):
             self.out)
         self.connect(self.out.as_ground((0, 0)*Amp), self.driver.high_gnd)  # TODO model driver current
 
+        self.assign(self.actual_current_limits, self.low_fet.actual_drain_current_rating.intersect(
+            self.high_fet.actual_drain_current_rating))
+
 
 class FetHalfBridgeIndependent(FetHalfBridge, HalfBridgeIndependent):
     def contents(self):

edg/abstract_parts/PowerCircuits.py

@@ -108,8 +108,8 @@
 
 from .DummyDevices import DummyPassive, DummyGround, DummyVoltageSource, DummyVoltageSink, DummyDigitalSink, \
     DummyAnalogSource, DummyAnalogSink
-from .DummyDevices import ForcedVoltageCurrentDraw, ForcedVoltage, ForcedVoltageCurrent, ForcedAnalogVoltage,\
-    ForcedAnalogSignal, ForcedDigitalSinkCurrentDraw
+from .DummyDevices import ForcedVoltageCurrentDraw, ForcedVoltageCurrentLimit, ForcedVoltage, ForcedVoltageCurrent, \
+    ForcedAnalogVoltage, ForcedAnalogSignal, ForcedDigitalSinkCurrentDraw
 from .MergedBlocks import MergedVoltageSource, MergedDigitalSource, MergedAnalogSource, MergedSpiController
 
 from .Nonstrict3v3Compatible import Nonstrict3v3Compatible

edg/abstract_parts/__init__.py

@@ -75,3 +75,49 @@ def test_map(self) -> None:
 
   def test_first(self) -> None:
     self.assertEqual(self.table.first().value, {'header1': '1', 'header2': 'foo', 'header3': '9'})
+
+
+class UserFnPartsTableTest(unittest.TestCase):
+  @staticmethod
+  @ExperimentalUserFnPartsTable.user_fn()
+  def user_fn_false() -> Callable[[], bool]:
+    def inner() -> bool:
+      return False
+    return inner
+
+  @staticmethod
+  @ExperimentalUserFnPartsTable.user_fn([bool])
+  def user_fn_bool_pass(meta_arg: bool) -> Callable[[], bool]:
+    def inner() -> bool:
+      return meta_arg
+    return inner
+
+  @staticmethod
+  @ExperimentalUserFnPartsTable.user_fn([float])
+  def user_fn_float_pass(meta_arg: float) -> Callable[[], float]:
+    def inner() -> float:
+      return meta_arg
+    return inner
+
+  @staticmethod
+  def user_fn_unserialized() -> Callable[[], None]:
+    def inner() -> None:
+      return None
+    return inner
+
+  def test_serialize_deserialize(self) -> None:
+    self.assertEqual(ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_false), 'user_fn_false')
+    self.assertEqual(ExperimentalUserFnPartsTable.deserialize_fn(
+      ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_false))(), False)
+
+    self.assertEqual(ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_bool_pass, False),
+                     'user_fn_bool_pass;False')
+    self.assertEqual(ExperimentalUserFnPartsTable.deserialize_fn(
+      ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_bool_pass, False))(), False)
+    self.assertEqual(ExperimentalUserFnPartsTable.deserialize_fn(
+      ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_bool_pass, True))(), True)
+
+    self.assertEqual(ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_float_pass, 0.42),
+                     'user_fn_float_pass;0.42')
+    self.assertEqual(ExperimentalUserFnPartsTable.deserialize_fn(
+      ExperimentalUserFnPartsTable.serialize_fn(self.user_fn_float_pass, 0.42))(), 0.42)