From e05c11c56db2521b6610e6a377eeebfeb021a5c8 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Tue, 18 Jun 2024 10:17:32 +1000 Subject: [PATCH 01/24] Draft idea for signal typing on new virtualmux --- script.py | 31 +++++++++++++++++ src/fixate/core/switching.py | 66 ++++++++++++++++++++++++++++++++---- 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/script.py b/script.py index 25a0ad83..e8040ad1 100644 --- a/script.py +++ b/script.py @@ -52,3 +52,34 @@ class JigMuxGroup(MuxGroup): jig.mux.mux_two("sig5") jig.mux.mux_three("On") jig.mux.mux_three(False) + +try: + from typing import Annotated +except ImportError: + # 3.8 + from typing_extensions import Annotated + +from typing import Literal, Union + +# maybe we can create aliases to make it easier to understand how to create a MuxDef +SignalName = Literal +Signal = Annotated +MuxDef = Union + +MuxOneSigDef = MuxDef[ + Signal[SignalName["sig_a1"], "a0", "a2" ], + Signal[SignalName["sig_a2"], "a1"] + ] + +class MuxA(VirtualMux[MuxOneSigDef]): + """A mux definition used by a few scripts""" + +muxa = MuxA(update_pins=print) + +muxa("sig_a2") +muxa("sig_a1") + +muxb = VirtualMux[MuxOneSigDef](update_pins=print) + +muxb.multiplex("sig_a2") +muxb.multiplex("sig_a1") diff --git a/src/fixate/core/switching.py b/src/fixate/core/switching.py index 2a58dae3..149d47a0 100644 --- a/src/fixate/core/switching.py +++ b/src/fixate/core/switching.py @@ -43,15 +43,20 @@ Dict, FrozenSet, Iterable, + Literal, + get_origin, + get_args ) from dataclasses import dataclass from functools import reduce from operator import or_ Signal = str +EmptySignal = Literal[""] Pin = str PinList = Sequence[Pin] PinSet = FrozenSet[Pin] +# do we bother to add EmptySignal here? SignalMap = Dict[Signal, PinSet] TreeDef = Sequence[Union[Signal, "TreeDef"]] @@ -87,14 +92,26 @@ def __or__(self, other: PinUpdate) -> PinUpdate: PinUpdateCallback = Callable[[PinUpdate, bool], None] +S = TypeVar("S", bound=str) -class VirtualMux: +class VirtualMux(Generic[S]): pin_list: PinList = () clearing_time: float = 0.0 + def __class_getitem__(cls, *args, **kwargs): + # without calling getitem the class doesn't work as a generic + getitm = super().__class_getitem__(*args, **kwargs) # normally returns a generic + # create a proxy to force the __init_subclass__ hook + class Hack(getitm): + ... + def __name__(self) -> str: + return f"{cls.__name__}" + + + return Hack # now the actual class can be initialised + ########################################################################### # These methods are the public API for the class - def __init__(self, update_pins: Optional[PinUpdateCallback] = None): self._last_update_time = time.monotonic() @@ -129,7 +146,7 @@ def __init__(self, update_pins: Optional[PinUpdateCallback] = None): if hasattr(self, "default_signal"): raise ValueError("'default_signal' should not be set on a VirtualMux") - def __call__(self, signal_output: Signal, trigger_update: bool = True) -> None: + def __call__(self, signal_output: Union[S, EmptySignal], trigger_update: bool = True) -> None: """ Convenience to avoid having to type jig.mux..multiplex. @@ -138,7 +155,7 @@ def __call__(self, signal_output: Signal, trigger_update: bool = True) -> None: """ self.multiplex(signal_output, trigger_update) - def multiplex(self, signal_output: Signal, trigger_update: bool = True) -> None: + def multiplex(self, signal_output: Union[S, EmptySignal], trigger_update: bool = True) -> None: """ Update the multiplexer state to signal_output. @@ -230,11 +247,48 @@ def _map_signals(self) -> SignalMap: return self._map_tree(self.map_tree, self.pin_list, fixed_pins=frozenset()) elif hasattr(self, "map_list"): return {sig: frozenset(pins) for sig, *pins in self.map_list} + elif (self.__orig_bases__ != VirtualMux.__orig_bases__): + # the user has provided map_list using annotations + # if the type annotations have not been supplied, then self.__orig_bases__ == VirtualMux.__orig_bases__ + # if they have been overridden, then self.__orig_bases__ != VirtualMux.__orig_bases__ + # this will only work if VirtualMux was subclassed + # if creating an instance using VirtualMux[type]() then __orig_bases__ does not exist + # which is why __class_getitem__ is used to create a proxy subclass + return self._map_signals_from_annotations() else: raise ValueError( - "VirtualMux subclass must define either map_tree or map_list" + "VirtualMux subclass must define either map_tree or map_list or provide a type to VirtualMux" ) - + def _map_signals_from_annotations(self) -> SignalMap: + + # structure is: + # Union[ + # Annotated[Literal["sig_a1"], "pin1", "pin2", ...] + # Annotated[Literal["sig_a2"], "pin3"] + # ] + + # we are expecting exactly 1 value, unpack it + cls, = self.__orig_bases__ + assert get_origin(cls) == VirtualMux, "VirtualMux subclass must provide a type to VirtualMux" + muxdef, = get_args(cls) + assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" + signals = get_args(muxdef) + sigmap = {} + for s in signals: + # get_args gives Literal + # get_args ignores metadata before 3.10 + sigdef, *pins = get_args(s) + # 3.8 only + if not pins: + if not hasattr(s, "__metadata__"): + raise ValueError("VirtualMux Subclass must define the pins for each signal") + # s.__metadata__ is our pin list + pins = s.__metadata__ + # get_args gives members of Literal + signame, = get_args(sigdef) + sigmap[signame] = frozenset(pins) + + return sigmap def _map_tree(self, tree: TreeDef, pins: PinList, fixed_pins: PinSet) -> SignalMap: """recursively add nested signal lists to the signal map. tree: is the current sub-branch to be added. At the first call From b438728c2e5f6b2be70db24751e142f3065641d3 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Tue, 18 Jun 2024 10:58:12 +1000 Subject: [PATCH 02/24] runtime checks to make it obvious when muxdef structure incorrect --- src/fixate/core/switching.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/fixate/core/switching.py b/src/fixate/core/switching.py index 149d47a0..84f5cadd 100644 --- a/src/fixate/core/switching.py +++ b/src/fixate/core/switching.py @@ -286,6 +286,8 @@ def _map_signals_from_annotations(self) -> SignalMap: pins = s.__metadata__ # get_args gives members of Literal signame, = get_args(sigdef) + assert isinstance(signame, Signal), "Signal name must be signal type" + assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" sigmap[signame] = frozenset(pins) return sigmap From 7703016810a0ca8838d2492798921e01886e67b9 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Thu, 20 Jun 2024 15:16:39 +1000 Subject: [PATCH 03/24] Fix some formatting --- src/fixate/_switching.py | 87 ++++++++++++++++++------------- src/fixate/examples/jig_driver.py | 7 +-- 2 files changed, 54 insertions(+), 40 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 100a5c58..299d5e28 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -45,7 +45,7 @@ Iterable, Literal, get_origin, - get_args + get_args, ) from dataclasses import dataclass from functools import reduce @@ -94,21 +94,24 @@ def __or__(self, other: PinUpdate) -> PinUpdate: S = TypeVar("S", bound=str) + class VirtualMux(Generic[S]): pin_list: PinList = () clearing_time: float = 0.0 def __class_getitem__(cls, *args, **kwargs): # without calling getitem the class doesn't work as a generic - getitm = super().__class_getitem__(*args, **kwargs) # normally returns a generic + getitm = super().__class_getitem__( + *args, **kwargs + ) # normally returns a generic # create a proxy to force the __init_subclass__ hook class Hack(getitm): ... + def __name__(self) -> str: return f"{cls.__name__}" - - return Hack # now the actual class can be initialised + return Hack # now the actual class can be initialised ########################################################################### # These methods are the public API for the class @@ -146,7 +149,9 @@ def __init__(self, update_pins: Optional[PinUpdateCallback] = None): if hasattr(self, "default_signal"): raise ValueError("'default_signal' should not be set on a VirtualMux") - def __call__(self, signal: Union[S, EmptySignal], trigger_update: bool = True) -> None: + def __call__( + self, signal: Union[S, EmptySignal], trigger_update: bool = True + ) -> None: """ Convenience to avoid having to type jig.mux..multiplex. @@ -155,7 +160,9 @@ def __call__(self, signal: Union[S, EmptySignal], trigger_update: bool = True) - """ self.multiplex(signal, trigger_update) - def multiplex(self, signal: Union[S, EmptySignal], trigger_update: bool = True) -> None: + def multiplex( + self, signal: Union[S, EmptySignal], trigger_update: bool = True + ) -> None: """ Update the multiplexer state to signal. @@ -251,7 +258,7 @@ def _map_signals(self) -> SignalMap: return self._map_tree(self.map_tree, self.pin_list, fixed_pins=frozenset()) elif hasattr(self, "map_list"): return {sig: frozenset(pins) for sig, *pins in self.map_list} - elif (self.__orig_bases__ != VirtualMux.__orig_bases__): + elif self.__orig_bases__ != VirtualMux.__orig_bases__: # the user has provided map_list using annotations # if the type annotations have not been supplied, then self.__orig_bases__ == VirtualMux.__orig_bases__ # if they have been overridden, then self.__orig_bases__ != VirtualMux.__orig_bases__ @@ -263,38 +270,44 @@ def _map_signals(self) -> SignalMap: raise ValueError( "VirtualMux subclass must define either map_tree or map_list or provide a type to VirtualMux" ) + def _map_signals_from_annotations(self) -> SignalMap: - # structure is: - # Union[ - # Annotated[Literal["sig_a1"], "pin1", "pin2", ...] - # Annotated[Literal["sig_a2"], "pin3"] - # ] - - # we are expecting exactly 1 value, unpack it - cls, = self.__orig_bases__ - assert get_origin(cls) == VirtualMux, "VirtualMux subclass must provide a type to VirtualMux" - muxdef, = get_args(cls) - assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" - signals = get_args(muxdef) - sigmap = {} - for s in signals: - # get_args gives Literal - # get_args ignores metadata before 3.10 - sigdef, *pins = get_args(s) - # 3.8 only - if not pins: - if not hasattr(s, "__metadata__"): - raise ValueError("VirtualMux Subclass must define the pins for each signal") - # s.__metadata__ is our pin list - pins = s.__metadata__ - # get_args gives members of Literal - signame, = get_args(sigdef) - assert isinstance(signame, Signal), "Signal name must be signal type" - assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" - sigmap[signame] = frozenset(pins) - - return sigmap + # structure is: + # Union[ + # Annotated[Literal["sig_a1"], "pin1", "pin2", ...] + # Annotated[Literal["sig_a2"], "pin3"] + # ] + + # we are expecting exactly 1 value, unpack it + (cls,) = self.__orig_bases__ + assert ( + get_origin(cls) == VirtualMux + ), "VirtualMux subclass must provide a type to VirtualMux" + (muxdef,) = get_args(cls) + assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" + signals = get_args(muxdef) + sigmap = {} + for s in signals: + # get_args gives Literal + # get_args ignores metadata before 3.10 + sigdef, *pins = get_args(s) + # 3.8 only + if not pins: + if not hasattr(s, "__metadata__"): + raise ValueError( + "VirtualMux Subclass must define the pins for each signal" + ) + # s.__metadata__ is our pin list + pins = s.__metadata__ + # get_args gives members of Literal + (signame,) = get_args(sigdef) + assert isinstance(signame, Signal), "Signal name must be signal type" + assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" + sigmap[signame] = frozenset(pins) + + return sigmap + def _map_tree(self, tree: TreeDef, pins: PinList, fixed_pins: PinSet) -> SignalMap: """recursively add nested signal lists to the signal map. tree: is the current sub-branch to be added. At the first call diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index bac48617..8fea4cbd 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -73,13 +73,14 @@ class JigMuxGroup(MuxGroup): MuxDef = Union MuxOneSigDef = MuxDef[ - Signal[SignalName["sig_a1"], "a0", "a2" ], - Signal[SignalName["sig_a2"], "a1"] - ] + Signal[SignalName["sig_a1"], "a0", "a2"], Signal[SignalName["sig_a2"], "a1"] +] + class MuxA(VirtualMux[MuxOneSigDef]): """A mux definition used by a few scripts""" + muxa = MuxA(update_pins=print) muxa("sig_a2") From 4fad25a37c7a11cb2fc75e624be02ecd48721c45 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Thu, 20 Jun 2024 18:59:03 +1000 Subject: [PATCH 04/24] Get inheritance and some better dynamic proxying --- src/fixate/_switching.py | 41 ++++++++++++++++++++----------- src/fixate/examples/jig_driver.py | 29 ++++++++++++++++++---- 2 files changed, 50 insertions(+), 20 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 299d5e28..f24d7281 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -50,6 +50,7 @@ from dataclasses import dataclass from functools import reduce from operator import or_ +from types import new_class Signal = str EmptySignal = Literal[""] @@ -99,19 +100,29 @@ class VirtualMux(Generic[S]): pin_list: PinList = () clearing_time: float = 0.0 - def __class_getitem__(cls, *args, **kwargs): - # without calling getitem the class doesn't work as a generic - getitm = super().__class_getitem__( - *args, **kwargs - ) # normally returns a generic - # create a proxy to force the __init_subclass__ hook - class Hack(getitm): - ... + def __class_getitem__(cls, arg): + # https://peps.python.org/pep-0560 - def __name__(self) -> str: - return f"{cls.__name__}" + # so the problem we have to solve is the way this works is convoluted and the __orig_bases__ property we rely on is not preserved + # __orig_bases__ is an attribute set by the metaclass that is used to store the type information provided by __class_getitem__ + # expected class creation: + # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> VirtualMux.__init__() (with type info visible on __orig_bases__) + # what actually happens + # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> _GenericAlias(VirtualMux).__call__() -> VirtualMux.__init__() + # the _GenericAlias middle class means we lose the type information provided through the __class_getitem__ call - return Hack # now the actual class can be initialised + # the attribute __orig_class__ can be used to store the original class that was created with __class_getitem__ + # HOWEVER this relies on the __init__ of VirtualMux succeeding - we end up in a circular dependency of needing + # the init to succeed to set the attribute, but needing the attribute to be set to succeed in the init (due to our implementation) + + # there are two options + # 1. create a proxy class and wrap it to look like a normal VirtualMux + # 2. spend more time figuring out how the typing system works + + getitm = super().__class_getitem__(arg) # normally returns a GenericAlias + proxy = new_class(f"{cls}[{arg}]", bases=(getitm,)) + + return proxy # now the actual class can be initialised ########################################################################### # These methods are the public API for the class @@ -281,9 +292,9 @@ def _map_signals_from_annotations(self) -> SignalMap: # we are expecting exactly 1 value, unpack it (cls,) = self.__orig_bases__ - assert ( - get_origin(cls) == VirtualMux - ), "VirtualMux subclass must provide a type to VirtualMux" + assert issubclass( + get_origin(cls), VirtualMux + ), "{cls} must be an instance of VirtualMux" # I don't know if this would ever be false unless you literally copy a method rather than inherit (muxdef,) = get_args(cls) assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" signals = get_args(muxdef) @@ -551,7 +562,7 @@ def __init__( super().__init__(update_pins) -class RelayMatrixMux(VirtualMux): +class RelayMatrixMux(VirtualMux[S], Generic[S]): clearing_time = 0.01 def _calculate_pins( diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index 8fea4cbd..d85a14a3 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -10,6 +10,7 @@ MuxGroup, PinValueAddressHandler, VirtualSwitch, + RelayMatrixMux, ) @@ -59,19 +60,19 @@ class JigMuxGroup(MuxGroup): jig.mux.mux_three("On") jig.mux.mux_three(False) -try: - from typing import Annotated -except ImportError: - # 3.8 - from typing_extensions import Annotated + +from typing_extensions import Annotated from typing import Literal, Union # maybe we can create aliases to make it easier to understand how to create a MuxDef SignalName = Literal +SignalName.__repr__ = lambda self: f"SignalName" Signal = Annotated +Signal.__repr__ = lambda self: f"Signal{self.__args__}" MuxDef = Union +# fmt off MuxOneSigDef = MuxDef[ Signal[SignalName["sig_a1"], "a0", "a2"], Signal[SignalName["sig_a2"], "a1"] ] @@ -90,3 +91,21 @@ class MuxA(VirtualMux[MuxOneSigDef]): muxb.multiplex("sig_a2") muxb.multiplex("sig_a1") +try: + muxb.multiplex("1") +except ValueError: + ... +else: + raise ValueError("muxb.multiplex('1') should have raised a ValueError") + +# an example of generic subclasses of VirtualMux +rmm = RelayMatrixMux[MuxOneSigDef]() + +rmm("sig_a1") + +try: + rmm("sig") +except ValueError: + ... +else: + raise ValueError("rmm('sig') should have raised a ValueError") From 050c41148595aa75c26a7f2c190b3d02c7806b1f Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Fri, 21 Jun 2024 12:11:21 +1000 Subject: [PATCH 05/24] Add some tests --- src/fixate/_switching.py | 20 +++++++---- src/fixate/examples/jig_driver.py | 25 ++++++++++---- test/test_switching.py | 56 +++++++++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 13 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index f24d7281..10fd5d56 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -100,7 +100,7 @@ class VirtualMux(Generic[S]): pin_list: PinList = () clearing_time: float = 0.0 - def __class_getitem__(cls, arg): + def __class_getitem__(cls, muxdef): # https://peps.python.org/pep-0560 # so the problem we have to solve is the way this works is convoluted and the __orig_bases__ property we rely on is not preserved @@ -119,8 +119,8 @@ def __class_getitem__(cls, arg): # 1. create a proxy class and wrap it to look like a normal VirtualMux # 2. spend more time figuring out how the typing system works - getitm = super().__class_getitem__(arg) # normally returns a GenericAlias - proxy = new_class(f"{cls}[{arg}]", bases=(getitm,)) + getitm = super().__class_getitem__(muxdef) # normally returns a GenericAlias + proxy = new_class(f"{cls}[{muxdef}]", bases=(getitm,)) return proxy # now the actual class can be initialised @@ -135,16 +135,22 @@ def __init__(self, update_pins: Optional[PinUpdateCallback] = None): else: self._update_pins = update_pins + self._signal_map: SignalMap = self._map_signals() + # We annotate the pin_list to be an ordered sequence, because if the # mux is defined with a map_tree, we need the ordering. But after # initialisation, we only need set operations on the pin list, so # we convert here and keep a reference to the set for future use. - self._pin_set = frozenset(self.pin_list) + if self.pin_list: + self._pin_set = frozenset(self.pin_list) + else: + # in the case of we didn't explicitly define pin_list, collapse the signal map to get all pins + self._pin_set = frozenset( + itertools.chain.from_iterable(self._signal_map.values()) + ) self._state = "" - self._signal_map: SignalMap = self._map_signals() - # Define the implicit signal "" which can be used to turn off all pins. # If the signal map already has this defined, raise an error. In the old # implementation, it allows the map to set this, but when switching the @@ -562,7 +568,7 @@ def __init__( super().__init__(update_pins) -class RelayMatrixMux(VirtualMux[S], Generic[S]): +class RelayMatrixMux(VirtualMux[S]): clearing_time = 0.01 def _calculate_pins( diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index d85a14a3..d5502298 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -72,22 +72,35 @@ class JigMuxGroup(MuxGroup): Signal.__repr__ = lambda self: f"Signal{self.__args__}" MuxDef = Union -# fmt off +# fmt: off MuxOneSigDef = MuxDef[ - Signal[SignalName["sig_a1"], "a0", "a2"], Signal[SignalName["sig_a2"], "a1"] + Signal[SignalName["sig_a1"], "a0", "a2"], + Signal[SignalName["sig_a2"], "a1"] ] +# fmt: on +rmm = RelayMatrixMux[MuxOneSigDef]() + +from typing import TypeVar +S = TypeVar("S", bound =str) class MuxA(VirtualMux[MuxOneSigDef]): """A mux definition used by a few scripts""" - -muxa = MuxA(update_pins=print) - +muxa = MuxA() muxa("sig_a2") muxa("sig_a1") +muxa("sig_a") + +class MuxC(VirtualMux[S]): + ... + +muxc = MuxC[MuxOneSigDef]() +muxc("sig_a2") +muxc("sig_a1") +muxc("sig_a") -muxb = VirtualMux[MuxOneSigDef](update_pins=print) +muxb = VirtualMux[MuxOneSigDef]() muxb.multiplex("sig_a2") muxb.multiplex("sig_a1") diff --git a/test/test_switching.py b/test/test_switching.py index 527f1d2c..0c9f0605 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -15,6 +15,9 @@ JigDriver, ) +from typing import Literal, Union, TypeVar, Generic +from typing_extensions import Annotated # only for 3.8 + import pytest ################################################################ @@ -592,3 +595,56 @@ def test_pin_update_or(): 2.0, ) assert expected == a | b + + +# fmt: off +MuxASigDef = Union[ + Annotated[Literal["sig_a1"], "a0", "a1"], + Annotated[Literal["sig_a2"], "a1"] +] +# fmt: on + + +def test_typed_mux(): + clear = PinSetState(off=frozenset({"a0", "a1"})) + a1 = PinSetState(on=frozenset({"a0", "a1"})) + a2 = PinSetState(on=frozenset({"a1"}), off=frozenset({"a0"})) + + updates = [] + updatesa = [] + + mux = VirtualMux[MuxASigDef](lambda x, y: updates.append((x, y))) + mux_a = MuxA(lambda x, y: updatesa.append((x, y))) + assert mux_a._signal_map == mux._signal_map + assert mux_a._pin_set == mux._pin_set + + mux("sig_a1") + mux_a("sig_a1") + assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a1), True) + + mux.multiplex("sig_a2", trigger_update=False) + mux_a.multiplex("sig_a2", trigger_update=False) + assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a2), False) + + mux("") + mux_a("") + assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), clear), True) + + +def test_typed_mux_subclass(): + class SubMux(VirtualMux[MuxASigDef]): + pass + + sm = SubMux() + assert sm._signal_map == MuxA()._signal_map + assert sm._pin_set == MuxA()._pin_set + + +def test_typed_mux_generic_subclass(): + T = TypeVar("T", bound=str) + + class GenericSubMux(VirtualMux[T], Generic[T]): + pass + + gsm = GenericSubMux[MuxASigDef]() + assert gsm._signal_map == MuxA()._signal_map From bec311bcefd8e8c6f0af5dce9f1a6a0c7c147daa Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Fri, 21 Jun 2024 13:20:08 +1000 Subject: [PATCH 06/24] Remove hasattr, get rid of super call --- src/fixate/_switching.py | 103 +++++++++++++----------------- src/fixate/examples/jig_driver.py | 14 ++-- test/test_switching.py | 2 +- 3 files changed, 53 insertions(+), 66 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 10fd5d56..3cc26853 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -59,7 +59,7 @@ PinSet = FrozenSet[Pin] # do we bother to add EmptySignal here? SignalMap = Dict[Signal, PinSet] -TreeDef = Sequence[Union[Signal, "TreeDef"]] +TreeDef = Sequence[Union[Optional[Signal], "TreeDef"]] @dataclass(frozen=True) @@ -97,10 +97,12 @@ def __or__(self, other: PinUpdate) -> PinUpdate: class VirtualMux(Generic[S]): + map_tree: Optional[TreeDef] = None + map_list: Optional[Sequence[Sequence[str]]] = None pin_list: PinList = () clearing_time: float = 0.0 - def __class_getitem__(cls, muxdef): + def __class_getitem__(cls, arg): # https://peps.python.org/pep-0560 # so the problem we have to solve is the way this works is convoluted and the __orig_bases__ property we rely on is not preserved @@ -117,12 +119,44 @@ def __class_getitem__(cls, muxdef): # there are two options # 1. create a proxy class and wrap it to look like a normal VirtualMux - # 2. spend more time figuring out how the typing system works + # 2. dynamically create the pin_list map_list here + + proxy = new_class(f"{cls}[{arg}]", bases=(cls,)) + # two cases: either we are a passing in the generic typevar, or we are passing in the mux definition + if type(arg) != TypeVar: + pin_list, map_list = cls._unpack_muxdef(arg) + proxy.pin_list = pin_list + proxy.map_list = map_list + return proxy # now the actual class can be initialised - getitm = super().__class_getitem__(muxdef) # normally returns a GenericAlias - proxy = new_class(f"{cls}[{muxdef}]", bases=(getitm,)) + @staticmethod + def _unpack_muxdef(muxdef): + # two cases + # muxdef is the signal definition + assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" + signals = get_args(muxdef) + map_list: Sequence[Sequence[str]] = [] + pin_list: PinList = [] + for s in signals: + # get_args gives Literal + # get_args ignores metadata before 3.10 + sigdef, *pins = get_args(s) + # 3.8 only + if not pins: + if not hasattr(s, "__metadata__"): + raise ValueError( + "VirtualMux Subclass must define the pins for each signal" + ) + # s.__metadata__ is our pin list + pins = s.__metadata__ + # get_args gives members of Literal + (signame,) = get_args(sigdef) + assert isinstance(signame, Signal), "Signal name must be signal type" + assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" + pin_list.extend(pins) + map_list.append((signame, *pins)) - return proxy # now the actual class can be initialised + return pin_list, map_list ########################################################################### # These methods are the public API for the class @@ -135,19 +169,13 @@ def __init__(self, update_pins: Optional[PinUpdateCallback] = None): else: self._update_pins = update_pins + self._pin_set = frozenset(self.pin_list) self._signal_map: SignalMap = self._map_signals() # We annotate the pin_list to be an ordered sequence, because if the # mux is defined with a map_tree, we need the ordering. But after # initialisation, we only need set operations on the pin list, so # we convert here and keep a reference to the set for future use. - if self.pin_list: - self._pin_set = frozenset(self.pin_list) - else: - # in the case of we didn't explicitly define pin_list, collapse the signal map to get all pins - self._pin_set = frozenset( - itertools.chain.from_iterable(self._signal_map.values()) - ) self._state = "" @@ -271,60 +299,15 @@ def _map_signals(self) -> SignalMap: Avoid subclassing. Consider creating helper functions to build map_tree or map_list. """ - if hasattr(self, "map_tree"): + if self.map_tree is not None: return self._map_tree(self.map_tree, self.pin_list, fixed_pins=frozenset()) - elif hasattr(self, "map_list"): + elif self.map_list is not None: return {sig: frozenset(pins) for sig, *pins in self.map_list} - elif self.__orig_bases__ != VirtualMux.__orig_bases__: - # the user has provided map_list using annotations - # if the type annotations have not been supplied, then self.__orig_bases__ == VirtualMux.__orig_bases__ - # if they have been overridden, then self.__orig_bases__ != VirtualMux.__orig_bases__ - # this will only work if VirtualMux was subclassed - # if creating an instance using VirtualMux[type]() then __orig_bases__ does not exist - # which is why __class_getitem__ is used to create a proxy subclass - return self._map_signals_from_annotations() else: raise ValueError( "VirtualMux subclass must define either map_tree or map_list or provide a type to VirtualMux" ) - def _map_signals_from_annotations(self) -> SignalMap: - - # structure is: - # Union[ - # Annotated[Literal["sig_a1"], "pin1", "pin2", ...] - # Annotated[Literal["sig_a2"], "pin3"] - # ] - - # we are expecting exactly 1 value, unpack it - (cls,) = self.__orig_bases__ - assert issubclass( - get_origin(cls), VirtualMux - ), "{cls} must be an instance of VirtualMux" # I don't know if this would ever be false unless you literally copy a method rather than inherit - (muxdef,) = get_args(cls) - assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" - signals = get_args(muxdef) - sigmap = {} - for s in signals: - # get_args gives Literal - # get_args ignores metadata before 3.10 - sigdef, *pins = get_args(s) - # 3.8 only - if not pins: - if not hasattr(s, "__metadata__"): - raise ValueError( - "VirtualMux Subclass must define the pins for each signal" - ) - # s.__metadata__ is our pin list - pins = s.__metadata__ - # get_args gives members of Literal - (signame,) = get_args(sigdef) - assert isinstance(signame, Signal), "Signal name must be signal type" - assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" - sigmap[signame] = frozenset(pins) - - return sigmap - def _map_tree(self, tree: TreeDef, pins: PinList, fixed_pins: PinSet) -> SignalMap: """recursively add nested signal lists to the signal map. tree: is the current sub-branch to be added. At the first call diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index d5502298..51b420d2 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -66,10 +66,9 @@ class JigMuxGroup(MuxGroup): from typing import Literal, Union # maybe we can create aliases to make it easier to understand how to create a MuxDef +# this doesn't work well because these are instances of type, not instances of object SignalName = Literal -SignalName.__repr__ = lambda self: f"SignalName" Signal = Annotated -Signal.__repr__ = lambda self: f"Signal{self.__args__}" MuxDef = Union # fmt: off @@ -82,23 +81,26 @@ class JigMuxGroup(MuxGroup): from typing import TypeVar -S = TypeVar("S", bound =str) +S = TypeVar("S", bound=str) + class MuxA(VirtualMux[MuxOneSigDef]): """A mux definition used by a few scripts""" + muxa = MuxA() muxa("sig_a2") muxa("sig_a1") -muxa("sig_a") + class MuxC(VirtualMux[S]): ... + muxc = MuxC[MuxOneSigDef]() muxc("sig_a2") muxc("sig_a1") -muxc("sig_a") +muxa("sig_a") muxb = VirtualMux[MuxOneSigDef]() @@ -122,3 +124,5 @@ class MuxC(VirtualMux[S]): ... else: raise ValueError("rmm('sig') should have raised a ValueError") + +rmm("fuck") \ No newline at end of file diff --git a/test/test_switching.py b/test/test_switching.py index 0c9f0605..de153f77 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -643,7 +643,7 @@ class SubMux(VirtualMux[MuxASigDef]): def test_typed_mux_generic_subclass(): T = TypeVar("T", bound=str) - class GenericSubMux(VirtualMux[T], Generic[T]): + class GenericSubMux(VirtualMux[T]): pass gsm = GenericSubMux[MuxASigDef]() From 2e5edfe453433391beab639f4030c16f69c42dfd Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Fri, 21 Jun 2024 14:50:40 +1000 Subject: [PATCH 07/24] removed a gamer word --- src/fixate/examples/jig_driver.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index 51b420d2..10c6ad67 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -124,5 +124,3 @@ class MuxC(VirtualMux[S]): ... else: raise ValueError("rmm('sig') should have raised a ValueError") - -rmm("fuck") \ No newline at end of file From bcec245925e1d8667685b183593023b9bcd420d6 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Fri, 21 Jun 2024 14:52:49 +1000 Subject: [PATCH 08/24] move stuff and make proxy conditional --- src/fixate/_switching.py | 114 ++++++++++++++++++++------------------- test/test_switching.py | 1 + 2 files changed, 59 insertions(+), 56 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 3cc26853..3994cc39 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -102,62 +102,6 @@ class VirtualMux(Generic[S]): pin_list: PinList = () clearing_time: float = 0.0 - def __class_getitem__(cls, arg): - # https://peps.python.org/pep-0560 - - # so the problem we have to solve is the way this works is convoluted and the __orig_bases__ property we rely on is not preserved - # __orig_bases__ is an attribute set by the metaclass that is used to store the type information provided by __class_getitem__ - # expected class creation: - # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> VirtualMux.__init__() (with type info visible on __orig_bases__) - # what actually happens - # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> _GenericAlias(VirtualMux).__call__() -> VirtualMux.__init__() - # the _GenericAlias middle class means we lose the type information provided through the __class_getitem__ call - - # the attribute __orig_class__ can be used to store the original class that was created with __class_getitem__ - # HOWEVER this relies on the __init__ of VirtualMux succeeding - we end up in a circular dependency of needing - # the init to succeed to set the attribute, but needing the attribute to be set to succeed in the init (due to our implementation) - - # there are two options - # 1. create a proxy class and wrap it to look like a normal VirtualMux - # 2. dynamically create the pin_list map_list here - - proxy = new_class(f"{cls}[{arg}]", bases=(cls,)) - # two cases: either we are a passing in the generic typevar, or we are passing in the mux definition - if type(arg) != TypeVar: - pin_list, map_list = cls._unpack_muxdef(arg) - proxy.pin_list = pin_list - proxy.map_list = map_list - return proxy # now the actual class can be initialised - - @staticmethod - def _unpack_muxdef(muxdef): - # two cases - # muxdef is the signal definition - assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" - signals = get_args(muxdef) - map_list: Sequence[Sequence[str]] = [] - pin_list: PinList = [] - for s in signals: - # get_args gives Literal - # get_args ignores metadata before 3.10 - sigdef, *pins = get_args(s) - # 3.8 only - if not pins: - if not hasattr(s, "__metadata__"): - raise ValueError( - "VirtualMux Subclass must define the pins for each signal" - ) - # s.__metadata__ is our pin list - pins = s.__metadata__ - # get_args gives members of Literal - (signame,) = get_args(sigdef) - assert isinstance(signame, Signal), "Signal name must be signal type" - assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" - pin_list.extend(pins) - map_list.append((signame, *pins)) - - return pin_list, map_list - ########################################################################### # These methods are the public API for the class def __init__(self, update_pins: Optional[PinUpdateCallback] = None): @@ -510,6 +454,64 @@ def _default_update_pins( """ print(pin_updates, trigger_update) + def __class_getitem__(cls, arg): + # https://peps.python.org/pep-0560 + + # so the problem we have to solve is the way this works is convoluted and the __orig_bases__ property we rely on is not preserved + # __orig_bases__ is an attribute set by the metaclass that is used to store the type information provided by __class_getitem__ + # expected class creation: + # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> VirtualMux.__init__() (with type info visible on __orig_bases__) + # what actually happens + # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> _GenericAlias(VirtualMux).__call__() -> VirtualMux.__init__() + # the _GenericAlias middle class means we lose the type information provided through the __class_getitem__ call + + # the attribute __orig_class__ can be used to store the original class that was created with __class_getitem__ + # HOWEVER this relies on the __init__ of VirtualMux succeeding - we end up in a circular dependency of needing + # the init to succeed to set the attribute, but needing the attribute to be set to succeed in the init (due to our implementation) + + # there are two options + # 1. create a proxy class and wrap it to look like a normal VirtualMux + # 2. dynamically create the pin_list map_list here + # two cases: either we are a passing in the generic typevar, or we are passing in the mux definition + if type(arg) == TypeVar: + # we are just creating classes normally so fall back to default behaviour + return super().__class_getitem__(arg) + else: + # we are creating a mux definition + proxy = new_class(f"{cls.__name__}", bases=(cls,)) + pin_list, map_list = cls._unpack_muxdef(arg) + proxy.pin_list = pin_list + proxy.map_list = map_list + return proxy # now the actual class can be initialised + + @staticmethod + def _unpack_muxdef(muxdef): + # muxdef is the signal definition + assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" + signals = get_args(muxdef) + map_list: Sequence[Sequence[str]] = [] + pin_list: PinList = [] + for s in signals: + # get_args gives Literal + # get_args ignores metadata before 3.10 + sigdef, *pins = get_args(s) + # 3.8 only + if not pins: + if not hasattr(s, "__metadata__"): + raise ValueError( + "VirtualMux definition must define the pins for each signal" + ) + # s.__metadata__ is our pin list + pins = s.__metadata__ + # get_args gives members of Literal + (signame,) = get_args(sigdef) + assert isinstance(signame, Signal), "Signal name must be signal type" + assert all(isinstance(p, Pin) for p in pins), "Pins must be pin type" + pin_list.extend(pins) + map_list.append((signame, *pins)) + + return pin_list, map_list + class VirtualSwitch(VirtualMux): """ diff --git a/test/test_switching.py b/test/test_switching.py index de153f77..08fa49e1 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -648,3 +648,4 @@ class GenericSubMux(VirtualMux[T]): gsm = GenericSubMux[MuxASigDef]() assert gsm._signal_map == MuxA()._signal_map + assert gsm._pin_set == MuxA()._pin_set From 4d57e03bcd01ace4ef7cf42ee27a0d62089e9668 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Fri, 21 Jun 2024 16:26:48 +1000 Subject: [PATCH 09/24] Add in some stuff to make mypy happy --- src/fixate/_switching.py | 55 +++++++++++++++++++++++++++------------- 1 file changed, 38 insertions(+), 17 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 3994cc39..d24a858f 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -46,6 +46,7 @@ Literal, get_origin, get_args, + Any, ) from dataclasses import dataclass from functools import reduce @@ -57,6 +58,7 @@ Pin = str PinList = Sequence[Pin] PinSet = FrozenSet[Pin] +MapList = Sequence[Sequence[Union[Signal, Pin]]] # do we bother to add EmptySignal here? SignalMap = Dict[Signal, PinSet] TreeDef = Sequence[Union[Optional[Signal], "TreeDef"]] @@ -454,43 +456,62 @@ def _default_update_pins( """ print(pin_updates, trigger_update) - def __class_getitem__(cls, arg): + # This looks bad, maybe we should not type hint this + def __class_getitem__(cls, arg: Union[TypeVar, Any]) -> type[VirtualMux[S]]: # https://peps.python.org/pep-0560 - # so the problem we have to solve is the way this works is convoluted and the __orig_bases__ property we rely on is not preserved - # __orig_bases__ is an attribute set by the metaclass that is used to store the type information provided by __class_getitem__ + # so the problem we have to solve is the way this works is convoluted and + # the __orig_bases__ property we rely on is not preserved + # __orig_bases__ is an attribute set by the metaclass + # that is used to store the type information provided by __class_getitem__ # expected class creation: - # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> VirtualMux.__init__() (with type info visible on __orig_bases__) + # VirtualMux[type] -> VirtualMux.__class_getitem__(type) + # -> VirtualMux.__init__() (with type info visible on __orig_bases__) # what actually happens - # VirtualMux[type] -> VirtualMux.__class_getitem__(type) -> _GenericAlias(VirtualMux).__call__() -> VirtualMux.__init__() - # the _GenericAlias middle class means we lose the type information provided through the __class_getitem__ call + # VirtualMux[type] -> VirtualMux.__class_getitem__(type) + # -> _GenericAlias(VirtualMux).__call__() -> VirtualMux.__init__() + # the _GenericAlias middle class means we lose the type information provided + # through the __class_getitem__ call - # the attribute __orig_class__ can be used to store the original class that was created with __class_getitem__ - # HOWEVER this relies on the __init__ of VirtualMux succeeding - we end up in a circular dependency of needing - # the init to succeed to set the attribute, but needing the attribute to be set to succeed in the init (due to our implementation) + # the attribute __orig_class__ can be used to store the original class that was + # created with __class_getitem__ + # HOWEVER this relies on the __init__ of VirtualMux succeeding + # we end up in a circular dependency of needing the init to succeed to set the attribute, + # but needing the attribute to be set to succeed in the init (due to our implementation) # there are two options # 1. create a proxy class and wrap it to look like a normal VirtualMux # 2. dynamically create the pin_list map_list here - # two cases: either we are a passing in the generic typevar, or we are passing in the mux definition + # two cases: either we are a passing in the generic typevar, + # or we are passing in the mux definition if type(arg) == TypeVar: # we are just creating classes normally so fall back to default behaviour - return super().__class_getitem__(arg) + return super().__class_getitem__(arg) # type: ignore # __class_getitem__ does magic stuff at runtime else: + assert issubclass( + cls, VirtualMux + ), "class we are acting on should be a valid VirtualMux type" # we are creating a mux definition - proxy = new_class(f"{cls.__name__}", bases=(cls,)) pin_list, map_list = cls._unpack_muxdef(arg) - proxy.pin_list = pin_list - proxy.map_list = map_list + + def add_signals(ns: dict[str, Any]) -> dict[str, Any]: + ns["pin_list"] = pin_list + ns["map_list"] = map_list + return ns + + proxy = new_class(f"{cls.__name__}", bases=(cls,), exec_body=add_signals) + assert isinstance( + proxy, type(VirtualMux) + ), "class we have created should be a valid VirtualMux type" return proxy # now the actual class can be initialised @staticmethod - def _unpack_muxdef(muxdef): + def _unpack_muxdef(muxdef: Any) -> tuple[PinList, MapList]: # muxdef is the signal definition assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" signals = get_args(muxdef) - map_list: Sequence[Sequence[str]] = [] - pin_list: PinList = [] + map_list: list[tuple[str]] = [] + pin_list: list[str] = [] for s in signals: # get_args gives Literal # get_args ignores metadata before 3.10 From 1c9b0acb3256d14708a1c23addeedac0a651fda4 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 11:48:17 +1000 Subject: [PATCH 10/24] Remove busted aliases --- src/fixate/examples/jig_driver.py | 24 +++--------------------- 1 file changed, 3 insertions(+), 21 deletions(-) diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index 10c6ad67..21ddd49f 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -65,16 +65,11 @@ class JigMuxGroup(MuxGroup): from typing import Literal, Union -# maybe we can create aliases to make it easier to understand how to create a MuxDef -# this doesn't work well because these are instances of type, not instances of object -SignalName = Literal -Signal = Annotated -MuxDef = Union # fmt: off -MuxOneSigDef = MuxDef[ - Signal[SignalName["sig_a1"], "a0", "a2"], - Signal[SignalName["sig_a2"], "a1"] +MuxOneSigDef = Union[ + Annotated[Literal["sig_a1"], "a0", "a2"], + Annotated[Literal["sig_a2"], "a1"], ] # fmt: on rmm = RelayMatrixMux[MuxOneSigDef]() @@ -100,27 +95,14 @@ class MuxC(VirtualMux[S]): muxc = MuxC[MuxOneSigDef]() muxc("sig_a2") muxc("sig_a1") -muxa("sig_a") muxb = VirtualMux[MuxOneSigDef]() muxb.multiplex("sig_a2") muxb.multiplex("sig_a1") -try: - muxb.multiplex("1") -except ValueError: - ... -else: - raise ValueError("muxb.multiplex('1') should have raised a ValueError") # an example of generic subclasses of VirtualMux rmm = RelayMatrixMux[MuxOneSigDef]() rmm("sig_a1") -try: - rmm("sig") -except ValueError: - ... -else: - raise ValueError("rmm('sig') should have raised a ValueError") From cb3de69b4613ba0065e09ab0f25ea3998f6afd50 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 12:55:39 +1000 Subject: [PATCH 11/24] Add more tests --- src/fixate/_switching.py | 34 ++++++++++++++++-------------- test/test_switching.py | 45 ++++++++++++++++++++++++++++++++++++++-- 2 files changed, 62 insertions(+), 17 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index d24a858f..4e71ed84 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -53,6 +53,13 @@ from operator import or_ from types import new_class +import sys +if sys.version_info >= (3, 9): + from typing import Annotated +else: + from typing_extensions import Annotated + + Signal = str EmptySignal = Literal[""] Pin = str @@ -457,13 +464,10 @@ def _default_update_pins( print(pin_updates, trigger_update) # This looks bad, maybe we should not type hint this - def __class_getitem__(cls, arg: Union[TypeVar, Any]) -> type[VirtualMux[S]]: + def __class_getitem__(cls, arg: Union[TypeVar, type]) -> type[VirtualMux[S]]: # https://peps.python.org/pep-0560 # so the problem we have to solve is the way this works is convoluted and - # the __orig_bases__ property we rely on is not preserved - # __orig_bases__ is an attribute set by the metaclass - # that is used to store the type information provided by __class_getitem__ # expected class creation: # VirtualMux[type] -> VirtualMux.__class_getitem__(type) # -> VirtualMux.__init__() (with type info visible on __orig_bases__) @@ -473,15 +477,7 @@ def __class_getitem__(cls, arg: Union[TypeVar, Any]) -> type[VirtualMux[S]]: # the _GenericAlias middle class means we lose the type information provided # through the __class_getitem__ call - # the attribute __orig_class__ can be used to store the original class that was - # created with __class_getitem__ - # HOWEVER this relies on the __init__ of VirtualMux succeeding - # we end up in a circular dependency of needing the init to succeed to set the attribute, - # but needing the attribute to be set to succeed in the init (due to our implementation) - # there are two options - # 1. create a proxy class and wrap it to look like a normal VirtualMux - # 2. dynamically create the pin_list map_list here # two cases: either we are a passing in the generic typevar, # or we are passing in the mux definition if type(arg) == TypeVar: @@ -508,15 +504,23 @@ def add_signals(ns: dict[str, Any]) -> dict[str, Any]: @staticmethod def _unpack_muxdef(muxdef: Any) -> tuple[PinList, MapList]: # muxdef is the signal definition - assert get_origin(muxdef) == Union, "MuxDef must be a Union of signals" - signals = get_args(muxdef) + if get_origin(muxdef) == Union: + signals = get_args(muxdef) + elif get_origin(muxdef) == Annotated: + signals = (muxdef,) + else: + raise TypeError("Signal definition must be Union or Annotated") + map_list: list[tuple[str]] = [] pin_list: list[str] = [] for s in signals: + # not working, get_origin(Annotated) returns the wrapped type in 3.8 + # assert get_origin(s) == Annotated, "Signal definition must be annotated" # get_args gives Literal # get_args ignores metadata before 3.10 sigdef, *pins = get_args(s) - # 3.8 only + assert get_origin(sigdef) == Literal, "Signal definition must be string literal" + # 3.8 only, Annotated forces a type and at least one annotation if not pins: if not hasattr(s, "__metadata__"): raise ValueError( diff --git a/test/test_switching.py b/test/test_switching.py index 08fa49e1..a8332512 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -15,8 +15,12 @@ JigDriver, ) -from typing import Literal, Union, TypeVar, Generic -from typing_extensions import Annotated # only for 3.8 +from typing import Literal, Union, TypeVar, Generic, get_args, get_origin +import sys +if sys.version_info >= (3, 9): + from typing import Annotated +else: + from typing_extensions import Annotated import pytest @@ -649,3 +653,40 @@ class GenericSubMux(VirtualMux[T]): gsm = GenericSubMux[MuxASigDef]() assert gsm._signal_map == MuxA()._signal_map assert gsm._pin_set == MuxA()._pin_set + + +def test_annotated_preserve_pin_defs(): + annotated = Annotated[Literal["sig_a1"], "a0", "a1"] + sigdef, *pins = get_args(annotated) + +def test_annotated_raises_on_missing_pin_def(): + with pytest.raises(TypeError): + annotated = Annotated[Literal["sig_a1"]] + +def test_annotation_bad_pindefs(): + BadMuxDef = Union[ + Annotated[Literal["sig_a1"], "a0", "a1"], + Annotated[Literal["sig_a1"], "a0", 1], + ] + + with pytest.raises(AssertionError): + mux = VirtualMux[BadMuxDef]() + +def test_annotation_bad_brackets(): + """ + We put the brackets in the wrong spot and accidentally defined + one of the signals as one of the pins of the previous signal + """ + BadMuxDef = Union[ + Annotated[Literal["sig_a1"], "a0", "a1", + Annotated[Literal["sig_a2"], "a1"]], + Annotated[Literal["sig_a1"], "a0", "a1"], + ] + + with pytest.raises(AssertionError): + mux = VirtualMux[BadMuxDef]() + +def test_annotated_get_origin(): + # Annotated behaviour is different between python versions + # fails 3.8, passes >=3.9 + assert get_origin(Annotated[Literal["sig_a1"], "a0", "a1"]) == Annotated From 9bac0cf2fa1ef1c960402e9e754ae12c34c5dbf4 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:30:24 +1000 Subject: [PATCH 12/24] renamed tests remove class_getitem hack --- src/fixate/_switching.py | 65 ++++++++------------- test/test_switching.py | 122 +++++++++++++++++++++++++++------------ 2 files changed, 110 insertions(+), 77 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 4e71ed84..cce3f4a3 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -54,6 +54,7 @@ from types import new_class import sys + if sys.version_info >= (3, 9): from typing import Annotated else: @@ -104,6 +105,8 @@ def __or__(self, other: PinUpdate) -> PinUpdate: S = TypeVar("S", bound=str) +from types import get_original_bases, resolve_bases + class VirtualMux(Generic[S]): map_tree: Optional[TreeDef] = None @@ -114,6 +117,9 @@ class VirtualMux(Generic[S]): ########################################################################### # These methods are the public API for the class def __init__(self, update_pins: Optional[PinUpdateCallback] = None): + # digest all the typing information if there is any to sete pin_list and map_list + self._digest_type_hints() + self._last_update_time = time.monotonic() self._update_pins: PinUpdateCallback @@ -463,54 +469,31 @@ def _default_update_pins( """ print(pin_updates, trigger_update) - # This looks bad, maybe we should not type hint this - def __class_getitem__(cls, arg: Union[TypeVar, type]) -> type[VirtualMux[S]]: - # https://peps.python.org/pep-0560 - - # so the problem we have to solve is the way this works is convoluted and - # expected class creation: - # VirtualMux[type] -> VirtualMux.__class_getitem__(type) - # -> VirtualMux.__init__() (with type info visible on __orig_bases__) - # what actually happens - # VirtualMux[type] -> VirtualMux.__class_getitem__(type) - # -> _GenericAlias(VirtualMux).__call__() -> VirtualMux.__init__() - # the _GenericAlias middle class means we lose the type information provided - # through the __class_getitem__ call - - - # two cases: either we are a passing in the generic typevar, - # or we are passing in the mux definition - if type(arg) == TypeVar: - # we are just creating classes normally so fall back to default behaviour - return super().__class_getitem__(arg) # type: ignore # __class_getitem__ does magic stuff at runtime - else: - assert issubclass( - cls, VirtualMux - ), "class we are acting on should be a valid VirtualMux type" - # we are creating a mux definition - pin_list, map_list = cls._unpack_muxdef(arg) - - def add_signals(ns: dict[str, Any]) -> dict[str, Any]: - ns["pin_list"] = pin_list - ns["map_list"] = map_list - return ns - - proxy = new_class(f"{cls.__name__}", bases=(cls,), exec_body=add_signals) - assert isinstance( - proxy, type(VirtualMux) - ), "class we have created should be a valid VirtualMux type" - return proxy # now the actual class can be initialised + def _digest_type_hints(self): + # digest all the typing information if there is any + bases = get_original_bases(self.__class__) + resolved_bases = resolve_bases(bases) + first_resolved_base = resolved_bases[0] + assert issubclass( + first_resolved_base, VirtualMux + ), f"{first_resolved_base} should be VirtualMux subclass" + if bases != resolved_bases: + args = get_args(bases[0]) + plist, mlist = self._unpack_muxdef(args[0]) + self.pin_list = plist + self.map_list = mlist @staticmethod - def _unpack_muxdef(muxdef: Any) -> tuple[PinList, MapList]: + def _unpack_muxdef(muxdef: type) -> tuple[PinList, MapList]: # muxdef is the signal definition if get_origin(muxdef) == Union: signals = get_args(muxdef) elif get_origin(muxdef) == Annotated: + # Union FORCES you to have two or more types, so this handles the case of only one pin signals = (muxdef,) else: raise TypeError("Signal definition must be Union or Annotated") - + map_list: list[tuple[str]] = [] pin_list: list[str] = [] for s in signals: @@ -519,7 +502,9 @@ def _unpack_muxdef(muxdef: Any) -> tuple[PinList, MapList]: # get_args gives Literal # get_args ignores metadata before 3.10 sigdef, *pins = get_args(s) - assert get_origin(sigdef) == Literal, "Signal definition must be string literal" + assert ( + get_origin(sigdef) == Literal + ), "Signal definition must be string literal" # 3.8 only, Annotated forces a type and at least one annotation if not pins: if not hasattr(s, "__metadata__"): diff --git a/test/test_switching.py b/test/test_switching.py index a8332512..d1e2ebf8 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -15,8 +15,9 @@ JigDriver, ) -from typing import Literal, Union, TypeVar, Generic, get_args, get_origin +from typing import Literal, Union, TypeVar, get_args, get_origin import sys + if sys.version_info >= (3, 9): from typing import Annotated else: @@ -609,68 +610,79 @@ def test_pin_update_or(): # fmt: on -def test_typed_mux(): - clear = PinSetState(off=frozenset({"a0", "a1"})) - a1 = PinSetState(on=frozenset({"a0", "a1"})) - a2 = PinSetState(on=frozenset({"a1"}), off=frozenset({"a0"})) - - updates = [] - updatesa = [] - - mux = VirtualMux[MuxASigDef](lambda x, y: updates.append((x, y))) - mux_a = MuxA(lambda x, y: updatesa.append((x, y))) - assert mux_a._signal_map == mux._signal_map - assert mux_a._pin_set == mux._pin_set - - mux("sig_a1") - mux_a("sig_a1") - assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a1), True) - - mux.multiplex("sig_a2", trigger_update=False) - mux_a.multiplex("sig_a2", trigger_update=False) - assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a2), False) - - mux("") - mux_a("") - assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), clear), True) - - -def test_typed_mux_subclass(): +def test_typed_mux_using_subclass(): class SubMux(VirtualMux[MuxASigDef]): pass - sm = SubMux() + sm = SubMux(update_pins=print) assert sm._signal_map == MuxA()._signal_map assert sm._pin_set == MuxA()._pin_set +def test_typed_relaymux_using_subclass(): + class SubRelayMux(RelayMatrixMux[MuxASigDef]): + pass + + srm = SubRelayMux() + assert srm._signal_map == MuxA()._signal_map + assert srm._pin_set == MuxA()._pin_set + + def test_typed_mux_generic_subclass(): T = TypeVar("T", bound=str) class GenericSubMux(VirtualMux[T]): pass - gsm = GenericSubMux[MuxASigDef]() + class ConcreteMux(GenericSubMux[MuxASigDef]): + pass + + gsm = ConcreteMux() assert gsm._signal_map == MuxA()._signal_map assert gsm._pin_set == MuxA()._pin_set +def test_typed_mux_one_pin(): + + muxbdef = Annotated[Literal["sig1"], "a0"] + + class MuxB(VirtualMux[muxbdef]): + ... + + clear = PinSetState(off=frozenset({"a0"})) + a1 = PinSetState(on=frozenset({"a0"})) + + updates = [] + muxb = MuxB(update_pins=lambda x, y: updates.append((x, y))) + muxb("sig1") + assert updates.pop() == (PinUpdate(PinSetState(), a1), True) + + muxb("") + assert updates.pop() == (PinUpdate(PinSetState(), clear), True) + + def test_annotated_preserve_pin_defs(): annotated = Annotated[Literal["sig_a1"], "a0", "a1"] sigdef, *pins = get_args(annotated) + def test_annotated_raises_on_missing_pin_def(): with pytest.raises(TypeError): annotated = Annotated[Literal["sig_a1"]] + def test_annotation_bad_pindefs(): BadMuxDef = Union[ - Annotated[Literal["sig_a1"], "a0", "a1"], - Annotated[Literal["sig_a1"], "a0", 1], + Annotated[Literal["sig_a1"], "a0", "a1"], + Annotated[Literal["sig_a1"], "a0", 1], ] + class BadMux(VirtualMux[BadMuxDef]): + pass + with pytest.raises(AssertionError): - mux = VirtualMux[BadMuxDef]() + mux = BadMux() + def test_annotation_bad_brackets(): """ @@ -678,15 +690,51 @@ def test_annotation_bad_brackets(): one of the signals as one of the pins of the previous signal """ BadMuxDef = Union[ - Annotated[Literal["sig_a1"], "a0", "a1", - Annotated[Literal["sig_a2"], "a1"]], - Annotated[Literal["sig_a1"], "a0", "a1"], + Annotated[Literal["sig_a1"], "a0", "a1", Annotated[Literal["sig_a2"], "a1"]], + Annotated[Literal["sig_a1"], "a0", "a1"], ] + class BadMux(VirtualMux[BadMuxDef]): + pass + with pytest.raises(AssertionError): - mux = VirtualMux[BadMuxDef]() + mux = BadMux() + +@pytest.mark.xfail( + sys.version_info < (3, 9), + reason="Annotated behaviour is different between python versions", +) def test_annotated_get_origin(): # Annotated behaviour is different between python versions # fails 3.8, passes >=3.9 assert get_origin(Annotated[Literal["sig_a1"], "a0", "a1"]) == Annotated + + +@pytest.mark.skip( + reason="Revisit this idea once we have a way to stop Generic breaking getattr" +) +def test_typed_mux_class_getitem(): + clear = PinSetState(off=frozenset({"a0", "a1"})) + a1 = PinSetState(on=frozenset({"a0", "a1"})) + a2 = PinSetState(on=frozenset({"a1"}), off=frozenset({"a0"})) + + updates = [] + updatesa = [] + + mux = VirtualMux[MuxASigDef](lambda x, y: updates.append((x, y))) + mux_a = MuxA(lambda x, y: updatesa.append((x, y))) + assert mux_a._signal_map == mux._signal_map + assert mux_a._pin_set == mux._pin_set + + mux("sig_a1") + mux_a("sig_a1") + assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a1), True) + + mux.multiplex("sig_a2", trigger_update=False) + mux_a.multiplex("sig_a2", trigger_update=False) + assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a2), False) + + mux("") + mux_a("") + assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), clear), True) From 1d74d2b9323d06c82f2fb16f402e56398b633b05 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:32:09 +1000 Subject: [PATCH 13/24] rename test --- src/fixate/examples/jig_driver.py | 63 ++++++++++++++++++++----------- 1 file changed, 42 insertions(+), 21 deletions(-) diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index 21ddd49f..db979334 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -61,8 +61,8 @@ class JigMuxGroup(MuxGroup): jig.mux.mux_three(False) +# VirtualMuxes can be created with type annotations to provide the signal map from typing_extensions import Annotated - from typing import Literal, Union @@ -71,38 +71,59 @@ class JigMuxGroup(MuxGroup): Annotated[Literal["sig_a1"], "a0", "a2"], Annotated[Literal["sig_a2"], "a1"], ] -# fmt: on -rmm = RelayMatrixMux[MuxOneSigDef]() -from typing import TypeVar +MuxTwoSigDef = Union[ + Annotated[Literal["sig_b1"], "b0", "b2"], + Annotated[Literal["sig_b2"], "b1"], +] -S = TypeVar("S", bound=str) +SingleSingleDef = Annotated[Literal["sig_c1"], "c0"], +# fmt: on +# VirtualMuxes can now be created with type annotations to provide the signal map +muxa = VirtualMux[MuxOneSigDef]() -class MuxA(VirtualMux[MuxOneSigDef]): - """A mux definition used by a few scripts""" +muxa("sig_a1") +muxa("sig_a2") +# using the wrong signal name will be caught at runtime and by the type checker +try: + muxa("unknown signal name") # type: ignore[arg-type] +except ValueError as e: + print(f"An Exception would have occurred: {e}") +else: + raise ValueError("Should have raised an exception") -muxa = MuxA() -muxa("sig_a2") -muxa("sig_a1") +# alternatively, the mux definition can be passed in while subclassing -class MuxC(VirtualMux[S]): - ... +class SubClassed(VirtualMux[MuxTwoSigDef]): + """A mux definition used by a few scripts""" + + def extra_method(self) -> None: + print("Extra method") -muxc = MuxC[MuxOneSigDef]() -muxc("sig_a2") -muxc("sig_a1") +muxb = SubClassed() +muxb.multiplex("sig_b1") +muxb.multiplex("sig_b2") -muxb = VirtualMux[MuxOneSigDef]() +# # further generic types can be created by subclassing from VirtualMux using a TypeVar +# # compared to the above way of subclassing, this way lets you reuse the class -muxb.multiplex("sig_a2") -muxb.multiplex("sig_a1") +# from typing import TypeVar -# an example of generic subclasses of VirtualMux -rmm = RelayMatrixMux[MuxOneSigDef]() +# S = TypeVar("S", bound=str) +# class MyGenericMux(VirtualMux[S]): +# ... +# def extra_method(self) -> None: +# print("Extra method") -rmm("sig_a1") +# generic_mux = MyGenericMux[MuxOneSigDef]() +# generic_mux("sig_a2") +# generic_mux("sig_a1") +# # RelayMatrixMux is an example of this +# rmm = RelayMatrixMux[MuxOneSigDef]() +# rmm("sig_a1") +# rmm("sig_a2") From 1c49afc86ea8647be767ef36df6c589b243c8ce5 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:39:19 +1000 Subject: [PATCH 14/24] Fix typing --- src/fixate/_switching.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index cce3f4a3..a34ada35 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -469,7 +469,7 @@ def _default_update_pins( """ print(pin_updates, trigger_update) - def _digest_type_hints(self): + def _digest_type_hints(self) ->None: # digest all the typing information if there is any bases = get_original_bases(self.__class__) resolved_bases = resolve_bases(bases) From 0ef023bd9ac459eff199dda2be649a9bf17d07b5 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:41:24 +1000 Subject: [PATCH 15/24] rename test --- test/test_switching.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/test_switching.py b/test/test_switching.py index d1e2ebf8..fa022495 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -642,7 +642,7 @@ class ConcreteMux(GenericSubMux[MuxASigDef]): assert gsm._pin_set == MuxA()._pin_set -def test_typed_mux_one_pin(): +def test_typed_mux_one_signal(): muxbdef = Annotated[Literal["sig1"], "a0"] From 992fff53298c646e6755fa3bbcd91b4da4b01ab7 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:53:18 +1000 Subject: [PATCH 16/24] rename and new test --- test/test_switching.py | 44 ++++++++++++++++++++++++++++++------------ 1 file changed, 32 insertions(+), 12 deletions(-) diff --git a/test/test_switching.py b/test/test_switching.py index fa022495..a69d28d3 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -711,6 +711,14 @@ def test_annotated_get_origin(): assert get_origin(Annotated[Literal["sig_a1"], "a0", "a1"]) == Annotated +def test_annotated_get_args(): + assert get_args(Annotated[Literal["sig_a1"], "a0", "a1"]) == ( + Literal["sig_a1"], + "a0", + "a1", + ) + + @pytest.mark.skip( reason="Revisit this idea once we have a way to stop Generic breaking getattr" ) @@ -719,22 +727,34 @@ def test_typed_mux_class_getitem(): a1 = PinSetState(on=frozenset({"a0", "a1"})) a2 = PinSetState(on=frozenset({"a1"}), off=frozenset({"a0"})) - updates = [] - updatesa = [] + updates_class_mux = [] + updates_mux_a = [] - mux = VirtualMux[MuxASigDef](lambda x, y: updates.append((x, y))) - mux_a = MuxA(lambda x, y: updatesa.append((x, y))) - assert mux_a._signal_map == mux._signal_map - assert mux_a._pin_set == mux._pin_set + class_mux = VirtualMux[MuxASigDef](lambda x, y: updates_class_mux.append((x, y))) + mux_a = MuxA(lambda x, y: updates_mux_a.append((x, y))) + assert mux_a._signal_map == class_mux._signal_map + assert mux_a._pin_set == class_mux._pin_set - mux("sig_a1") + class_mux("sig_a1") mux_a("sig_a1") - assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a1), True) + assert ( + updates_class_mux.pop() + == updates_mux_a.pop() + == (PinUpdate(PinSetState(), a1), True) + ) - mux.multiplex("sig_a2", trigger_update=False) + class_mux.multiplex("sig_a2", trigger_update=False) mux_a.multiplex("sig_a2", trigger_update=False) - assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), a2), False) + assert ( + updates_class_mux.pop() + == updates_mux_a.pop() + == (PinUpdate(PinSetState(), a2), False) + ) - mux("") + class_mux("") mux_a("") - assert updates.pop() == updatesa.pop() == (PinUpdate(PinSetState(), clear), True) + assert ( + updates_class_mux.pop() + == updates_mux_a.pop() + == (PinUpdate(PinSetState(), clear), True) + ) From 8775c1b922b7efceeadf554cba1d859e2e68c69b Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:53:59 +1000 Subject: [PATCH 17/24] fix formatting --- src/fixate/_switching.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index a34ada35..7683f47e 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -469,7 +469,7 @@ def _default_update_pins( """ print(pin_updates, trigger_update) - def _digest_type_hints(self) ->None: + def _digest_type_hints(self) -> None: # digest all the typing information if there is any bases = get_original_bases(self.__class__) resolved_bases = resolve_bases(bases) From ba4e9a42bd515b7d33c7c86978205c57c6a2c851 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:54:11 +1000 Subject: [PATCH 18/24] Update jig driver example --- src/fixate/examples/jig_driver.py | 61 +++++++++++++++++-------------- 1 file changed, 34 insertions(+), 27 deletions(-) diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index db979334..9d129cb2 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -62,10 +62,13 @@ class JigMuxGroup(MuxGroup): # VirtualMuxes can be created with type annotations to provide the signal map -from typing_extensions import Annotated -from typing import Literal, Union - +from typing import Literal, Annotated, Union +# a signal is a typing Annotation +# the first Literal is the signal name, the rest are the pin names +# the signal name MUST be a Literal +# multiple signals can be combined with a Union +# assigning annotations to variables is possible # fmt: off MuxOneSigDef = Union[ Annotated[Literal["sig_a1"], "a0", "a2"], @@ -73,15 +76,20 @@ class JigMuxGroup(MuxGroup): ] MuxTwoSigDef = Union[ - Annotated[Literal["sig_b1"], "b0", "b2"], + Annotated[Literal["sig_b1"], "b0", "b2"], Annotated[Literal["sig_b2"], "b1"], ] -SingleSingleDef = Annotated[Literal["sig_c1"], "c0"], +SingleSingleDef = Annotated[Literal["sig_c1"], "c0"], # fmt: on # VirtualMuxes can now be created with type annotations to provide the signal map -muxa = VirtualMux[MuxOneSigDef]() +# this only works when subclassing +class MyMux(VirtualMux[MuxOneSigDef]): + "A helpful description for my mux that is used in this jig driver" + + +muxa = MyMux() muxa("sig_a1") muxa("sig_a2") @@ -94,36 +102,35 @@ class JigMuxGroup(MuxGroup): else: raise ValueError("Should have raised an exception") -# alternatively, the mux definition can be passed in while subclassing + +# further generic types can be created by subclassing from VirtualMux using a TypeVar +# compared to the above way of subclassing, this way lets you reuse the class + +from typing import TypeVar + +S = TypeVar("S", bound=str) -class SubClassed(VirtualMux[MuxTwoSigDef]): - """A mux definition used by a few scripts""" +class MyGenericMux(VirtualMux[S]): + ... def extra_method(self) -> None: print("Extra method") -muxb = SubClassed() -muxb.multiplex("sig_b1") -muxb.multiplex("sig_b2") +class MyConcreteMux(MyGenericMux[MuxTwoSigDef]): + pass -# # further generic types can be created by subclassing from VirtualMux using a TypeVar -# # compared to the above way of subclassing, this way lets you reuse the class -# from typing import TypeVar +generic_mux = MyConcreteMux() +generic_mux("sig_b1") +generic_mux("sig_b2") -# S = TypeVar("S", bound=str) -# class MyGenericMux(VirtualMux[S]): -# ... -# def extra_method(self) -> None: -# print("Extra method") +# RelayMatrixMux is an example of this +class MyRelayMatrixMux(RelayMatrixMux[MuxOneSigDef]): + pass -# generic_mux = MyGenericMux[MuxOneSigDef]() -# generic_mux("sig_a2") -# generic_mux("sig_a1") -# # RelayMatrixMux is an example of this -# rmm = RelayMatrixMux[MuxOneSigDef]() -# rmm("sig_a1") -# rmm("sig_a2") +rmm = MyRelayMatrixMux() +rmm("sig_a1") +rmm("sig_a2") From 6fec4d185f6d8993f0cc9c6691c20c2caa5896ef Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:54:46 +1000 Subject: [PATCH 19/24] make unused ignores in jig example throw errors --- mypy.ini | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/mypy.ini b/mypy.ini index 765bff0e..05bf0a23 100644 --- a/mypy.ini +++ b/mypy.ini @@ -90,4 +90,10 @@ follow_imports = silent [mypy-fixate.drivers,fixate.drivers.dso.helper,fixate.drivers.funcgen.helper,fixate.drivers.funcgen.rigol_dg1022,fixate.drivers.pps,fixate.drivers.pps.helper,fixate.drivers.ftdi] follow_imports = silent [mypy-fixate.examples.function_generator,fixate.examples.programmable_power_supply,fixate.examples.test_script] -follow_imports = silent \ No newline at end of file +follow_imports = silent + +# this example demos the type checker warning you when using the wrong arguments +# we explicitly suppress the errors in the script, but make sure that mypy +# actually found an error +[mypy-fixate.examples.jig_driver] +enable_error_code = unused-ignore \ No newline at end of file From 24fe77d813ade8889f9d2bee9db42a2c24a53d84 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 17:57:09 +1000 Subject: [PATCH 20/24] remove conditional import and abandon old versions --- test/test_switching.py | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/test/test_switching.py b/test/test_switching.py index a69d28d3..0c1e2c77 100644 --- a/test/test_switching.py +++ b/test/test_switching.py @@ -15,13 +15,7 @@ JigDriver, ) -from typing import Literal, Union, TypeVar, get_args, get_origin -import sys - -if sys.version_info >= (3, 9): - from typing import Annotated -else: - from typing_extensions import Annotated +from typing import Literal, Union, TypeVar, get_args, get_origin, Annotated import pytest @@ -701,10 +695,6 @@ class BadMux(VirtualMux[BadMuxDef]): mux = BadMux() -@pytest.mark.xfail( - sys.version_info < (3, 9), - reason="Annotated behaviour is different between python versions", -) def test_annotated_get_origin(): # Annotated behaviour is different between python versions # fails 3.8, passes >=3.9 From 33aec0286a847f9acc3d987353795ea523ef8793 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 18:05:35 +1000 Subject: [PATCH 21/24] remove some code for 3.8 --- src/fixate/_switching.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 7683f47e..67647ba2 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -500,19 +500,10 @@ def _unpack_muxdef(muxdef: type) -> tuple[PinList, MapList]: # not working, get_origin(Annotated) returns the wrapped type in 3.8 # assert get_origin(s) == Annotated, "Signal definition must be annotated" # get_args gives Literal - # get_args ignores metadata before 3.10 sigdef, *pins = get_args(s) assert ( get_origin(sigdef) == Literal ), "Signal definition must be string literal" - # 3.8 only, Annotated forces a type and at least one annotation - if not pins: - if not hasattr(s, "__metadata__"): - raise ValueError( - "VirtualMux definition must define the pins for each signal" - ) - # s.__metadata__ is our pin list - pins = s.__metadata__ # get_args gives members of Literal (signame,) = get_args(sigdef) assert isinstance(signame, Signal), "Signal name must be signal type" From c6ccf852710073bcac528bef9bbfcde4e2af9238 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Mon, 24 Jun 2024 18:06:42 +1000 Subject: [PATCH 22/24] single signal example --- src/fixate/examples/jig_driver.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index 9d129cb2..5a9e93a7 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -80,13 +80,14 @@ class JigMuxGroup(MuxGroup): Annotated[Literal["sig_b2"], "b1"], ] -SingleSingleDef = Annotated[Literal["sig_c1"], "c0"], +# if defining only a single signal, the Union is omitted in the definition +SingleSingleDef = Annotated[Literal["sig_c1"], "c0", "c1"] # fmt: on # VirtualMuxes can now be created with type annotations to provide the signal map # this only works when subclassing class MyMux(VirtualMux[MuxOneSigDef]): - "A helpful description for my mux that is used in this jig driver" + """A helpful description for my mux that is used in this jig driver""" muxa = MyMux() @@ -103,6 +104,14 @@ class MyMux(VirtualMux[MuxOneSigDef]): raise ValueError("Should have raised an exception") +class MultiPinSwitch(VirtualMux[SingleSingleDef]): + """This acts like a switch, but has to coordinate two pins""" + + +ls = MultiPinSwitch() +ls("sig_c1") +ls("") + # further generic types can be created by subclassing from VirtualMux using a TypeVar # compared to the above way of subclassing, this way lets you reuse the class From 10487f53b0801d4c9ad035368940ca503cb6a95f Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Tue, 25 Jun 2024 14:24:16 +1000 Subject: [PATCH 23/24] Remove old python versions from CI matrix --- .github/workflows/test.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 47f8cc17..2f7b6163 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -17,7 +17,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] + python-version: ["3.12"] steps: - uses: actions/checkout@v3 From 6218915fbdef3c6e5e697600c3a9e819f59e2aa1 Mon Sep 17 00:00:00 2001 From: Daniel Montanari Date: Tue, 25 Jun 2024 14:38:09 +1000 Subject: [PATCH 24/24] Address review comments --- src/fixate/_switching.py | 22 ++++++---------------- src/fixate/examples/jig_driver.py | 2 +- 2 files changed, 7 insertions(+), 17 deletions(-) diff --git a/src/fixate/_switching.py b/src/fixate/_switching.py index 67647ba2..48edec3a 100644 --- a/src/fixate/_switching.py +++ b/src/fixate/_switching.py @@ -44,21 +44,13 @@ FrozenSet, Iterable, Literal, + Annotated, get_origin, get_args, - Any, ) from dataclasses import dataclass from functools import reduce from operator import or_ -from types import new_class - -import sys - -if sys.version_info >= (3, 9): - from typing import Annotated -else: - from typing_extensions import Annotated Signal = str @@ -67,7 +59,6 @@ PinList = Sequence[Pin] PinSet = FrozenSet[Pin] MapList = Sequence[Sequence[Union[Signal, Pin]]] -# do we bother to add EmptySignal here? SignalMap = Dict[Signal, PinSet] TreeDef = Sequence[Union[Optional[Signal], "TreeDef"]] @@ -117,7 +108,7 @@ class VirtualMux(Generic[S]): ########################################################################### # These methods are the public API for the class def __init__(self, update_pins: Optional[PinUpdateCallback] = None): - # digest all the typing information if there is any to sete pin_list and map_list + # digest all the typing information if there is any to set pin_list and map_list self._digest_type_hints() self._last_update_time = time.monotonic() @@ -128,16 +119,16 @@ def __init__(self, update_pins: Optional[PinUpdateCallback] = None): else: self._update_pins = update_pins - self._pin_set = frozenset(self.pin_list) - self._signal_map: SignalMap = self._map_signals() - # We annotate the pin_list to be an ordered sequence, because if the # mux is defined with a map_tree, we need the ordering. But after # initialisation, we only need set operations on the pin list, so # we convert here and keep a reference to the set for future use. + self._pin_set = frozenset(self.pin_list) self._state = "" + self._signal_map: SignalMap = self._map_signals() + # Define the implicit signal "" which can be used to turn off all pins. # If the signal map already has this defined, raise an error. In the old # implementation, it allows the map to set this, but when switching the @@ -497,8 +488,7 @@ def _unpack_muxdef(muxdef: type) -> tuple[PinList, MapList]: map_list: list[tuple[str]] = [] pin_list: list[str] = [] for s in signals: - # not working, get_origin(Annotated) returns the wrapped type in 3.8 - # assert get_origin(s) == Annotated, "Signal definition must be annotated" + assert get_origin(s) == Annotated, "Signal definition must be Annotated" # get_args gives Literal sigdef, *pins = get_args(s) assert ( diff --git a/src/fixate/examples/jig_driver.py b/src/fixate/examples/jig_driver.py index 5a9e93a7..774082f0 100644 --- a/src/fixate/examples/jig_driver.py +++ b/src/fixate/examples/jig_driver.py @@ -135,7 +135,7 @@ class MyConcreteMux(MyGenericMux[MuxTwoSigDef]): generic_mux("sig_b1") generic_mux("sig_b2") -# RelayMatrixMux is an example of this +# RelayMatrixMux is an example of a reusable generic class class MyRelayMatrixMux(RelayMatrixMux[MuxOneSigDef]): pass