Skip to content
29 changes: 13 additions & 16 deletions edg/core/Blocks.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,16 @@
from .Link import Link


class BaseBlockMeta(type):
"""Adds a hook to set the post-init elaboration state"""
def __call__(cls, *args, **kwargs):
block_context = builder.get_enclosing_block()
obj = super().__call__(*args, **kwargs)
if isinstance(obj, BaseBlock): # ignore block prototypes
obj._block_context = block_context
return obj


class Connection():
"""An incremental connection builder, that validates additional ports as they are added so
the stack trace can provide the problematic statement."""
Expand Down Expand Up @@ -186,7 +196,6 @@ def make_connection(self) -> Optional[Union[ConnectedLink, Export]]:
class BlockElaborationState(Enum):
pre_init = 1 # not sure if this is needed, doesn't actually get used
init = 2
post_init = 3
contents = 4
post_contents = 5
generate = 6
Expand Down Expand Up @@ -230,12 +239,13 @@ def set_elt_proto(self, pb, ref_map=None):
AbstractBlockProperty = EltPropertiesBase()

@non_library
class BaseBlock(HasMetadata, Generic[BaseBlockEdgirType]):
class BaseBlock(HasMetadata, Generic[BaseBlockEdgirType], metaclass=BaseBlockMeta):
"""Base block that has ports (IOs), parameters, and constraints between them.
"""
# __init__ should initialize the object with structural information (parameters, fields)
# as well as optionally initialization (parameter defaults)
def __init__(self) -> None:
self._block_context: Optional["Refable"] # set by metaclass, as lexical scope available pre-binding
self._parent: Optional[Union[BaseBlock, Port]] # refined from Optional[Refable] in base LibraryElement

super().__init__()
Expand Down Expand Up @@ -276,10 +286,6 @@ def _all_connects_of(self, base: Connection) -> IdentitySet[Connection]:

return delegated_connects

def _post_init(self):
assert self._elaboration_state == BlockElaborationState.init
self._elaboration_state = BlockElaborationState.post_init

def name(self) -> StringExpr:
return self._name

Expand All @@ -295,7 +301,7 @@ def _elaborated_def_to_proto(self) -> BaseBlockEdgirType:
prev_element = builder.push_element(self)
assert prev_element is None
try:
assert self._elaboration_state == BlockElaborationState.post_init
assert self._elaboration_state == BlockElaborationState.init
self._elaboration_state = BlockElaborationState.contents
self.contents()
self._elaboration_state = BlockElaborationState.post_contents
Expand Down Expand Up @@ -409,15 +415,6 @@ def _get_ref_map(self, prefix: edgir.LocalPath) -> IdentityDict[Refable, edgir.L
def _bind_in_place(self, parent: Union[BaseBlock, Port]):
self._parent = parent

SelfType = TypeVar('SelfType', bound='BaseBlock')
def _bind(self: SelfType, parent: Union[BaseBlock, Port]) -> SelfType:
"""Returns a clone of this object with the specified binding. This object must be unbound."""
assert self._parent is None, "can't clone bound block"
assert builder.get_enclosing_block() is self._block_context, "can't clone to different context"
clone = type(self)(*self._initializer_args[0], **self._initializer_args[1]) # type: ignore
clone._bind_in_place(parent)
return clone

def _check_constraint(self, constraint: ConstraintExpr) -> None:
def check_subexpr(expr: Union[ConstraintExpr, BasePort]) -> None: # TODO rewrite this whole method
if isinstance(expr, ConstraintExpr) and isinstance(expr.binding, ParamBinding):
Expand Down
22 changes: 1 addition & 21 deletions edg/core/Core.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,20 +142,6 @@ def values(self) -> ValuesView[ElementType]:
return self.container.values()


class ElementMeta(type):
"""Hook on construction to store some metadata about its creation.
This hooks the top-level __init__ only."""
def __call__(cls, *args, **kwargs):
block_context = builder.get_enclosing_block()

obj = type.__call__(cls, *args, **kwargs)
obj._initializer_args = (args, kwargs) # stores args so it is clone-able
obj._block_context = block_context
obj._post_init()

return obj


class Refable():
"""Object that could be referenced into a edgir.LocalPath"""
def __repr__(self) -> str:
Expand Down Expand Up @@ -185,25 +171,19 @@ def non_library(decorated: NonLibraryType) -> NonLibraryType:


@non_library
class LibraryElement(Refable, metaclass=ElementMeta):
class LibraryElement(Refable):
"""Defines a library element, which optionally contains other library elements."""
_elt_properties: Dict[Tuple[Type[LibraryElement], EltPropertiesBase], Any] = {}

def __repr__(self) -> str:
return "%s@%02x" % (self._get_def_name(), (id(self) // 4) & 0xff)

def __init__(self) -> None:
self._block_context: Optional["Refable"] # set by metaclass, as lexical scope available pre-binding
self._parent: Optional[LibraryElement] = None # set by binding, None means not bound
self._initializer_args: Tuple[Tuple[Any, ...], Dict[str, Any]] # set by metaclass

self.manager = SubElementManager()
self.manager_ignored: Set[str] = set(['_parent'])

"""Optionally overloaded to run anything post-__init__"""
def _post_init(self):
pass

def __setattr__(self, name: str, value: Any) -> None:
if hasattr(self, 'manager_ignored') and name not in self.manager_ignored:
self.manager.add_element(name, value)
Expand Down
5 changes: 3 additions & 2 deletions edg/core/DesignTop.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ def make_packing_refinement(multipack_part: Union[Block, PackedBlockAllocate], p
if isinstance(multipack_part, Block):
return path, type(multipack_part)
elif isinstance(multipack_part, PackedBlockAllocate):
return path, type(multipack_part.parent._tpe)
assert multipack_part.parent._elt_sample is not None
return path, type(multipack_part.parent._elt_sample)
else:
raise TypeError

Expand All @@ -55,7 +56,7 @@ def _elaborated_def_to_proto(self) -> edgir.HierarchyBlock:
prev_element = builder.push_element(self)
assert prev_element is None
try:
assert self._elaboration_state == BlockElaborationState.post_init
assert self._elaboration_state == BlockElaborationState.init
self._elaboration_state = BlockElaborationState.contents
self.contents()
self.multipack()
Expand Down
2 changes: 1 addition & 1 deletion edg/core/Generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ def _generated_def_to_proto(self, generate_values: Iterable[Tuple[edgir.LocalPat
assert prev_element is None

try:
assert self._elaboration_state == BlockElaborationState.post_init # TODO dedup w/ elaborated_def_to_proto
assert self._elaboration_state == BlockElaborationState.init
self._elaboration_state = BlockElaborationState.contents
self.contents()
self._elaboration_state = BlockElaborationState.generate
Expand Down
89 changes: 76 additions & 13 deletions edg/core/HierarchyBlock.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,10 @@
ArrayFloatLike, ArrayRangeLike, ArrayStringLike
from .Array import BaseVector, Vector
from .Binding import InitParamBinding, AssignBinding
from .Blocks import BaseBlock, Connection, BlockElaborationState, AbstractBlockProperty
from .Blocks import BaseBlock, Connection, BlockElaborationState, AbstractBlockProperty, BaseBlockMeta
from .ConstraintExpr import BoolLike, FloatLike, IntLike, RangeLike, StringLike
from .ConstraintExpr import ConstraintExpr, BoolExpr, FloatExpr, IntExpr, RangeExpr, StringExpr
from .Core import Refable, non_library, ElementMeta
from .Core import Refable, non_library
from .HdlUserExceptions import *
from .IdentityDict import IdentityDict
from .IdentitySet import IdentitySet
Expand Down Expand Up @@ -100,7 +100,41 @@ def __iter__(self):
return iter((tuple(self.blocks), self))


class BlockMeta(ElementMeta):
BlockPrototypeType = TypeVar('BlockPrototypeType', bound='Block')
class BlockPrototype(Generic[BlockPrototypeType]):
"""A block prototype, that contains a type and arguments, but without constructing the entire block
and running its (potentially quite expensive) __init__.

This class is automatically created on Block instantiations by the BlockMeta metaclass __init__ hook."""
def __init__(self, tpe: Type[BlockPrototypeType], args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> None:
self._tpe = tpe
self._args = args
self._kwargs = kwargs

def __repr__(self) -> str:
return f"{self.__class__.__name__}({self._tpe}, args={self._args}, kwargs={self._kwargs})"

def _bind(self, parent: Union[BaseBlock, Port]) -> BlockPrototypeType:
"""Binds the prototype into an actual Block instance."""
Block._next_bind = self._tpe
block = self._tpe(*self._args, **self._kwargs) # type: ignore
block._bind_in_place(parent)
return block

def __getattribute__(self, item: str) -> Any:
if item.startswith("_"):
return super().__getattribute__(item)
else:
raise AttributeError(f"{self.__class__.__name__} has no attributes, must bind to get a concrete instance, tried to get {item}")

def __setattr__(self, key: str, value: Any) -> None:
if key.startswith("_"):
super().__setattr__(key, value)
else:
raise AttributeError(f"{self.__class__.__name__} has no attributes, must bind to get a concrete instance, tried to set {key}")


class BlockMeta(BaseBlockMeta):
"""This provides a hook on __init__ that replaces argument values with empty ConstraintExpr
based on the type annotation and stores the supplied argument to the __init__ (if any) in the binding.

Expand Down Expand Up @@ -233,6 +267,23 @@ class Block(BaseBlock[edgir.HierarchyBlock], metaclass=BlockMeta):
"""Part with a statically-defined subcircuit.
Relations between contained parameters may only be expressed in the given constraint language.
"""
_next_bind: Optional[Type[Block]] = None # set when binding, to avoid creating a prototype

def __new__(cls, *args: Any, **kwargs: Any) -> Block:
if Block._next_bind is not None:
assert Block._next_bind is cls
Block._next_bind = None
return super().__new__(cls)
elif builder.get_enclosing_block() is None: # always construct if top-level
return super().__new__(cls)
else:
return BlockPrototype(cls, args, kwargs) # type: ignore

SelfType = TypeVar('SelfType', bound='BaseBlock')
def _bind(self: SelfType, parent: Union[BaseBlock, Port]) -> SelfType:
# for type checking only
raise TypeError("_bind should be called from BlockPrototype")

def __init__(self) -> None:
super().__init__()

Expand Down Expand Up @@ -410,14 +461,19 @@ def _def_to_proto(self) -> edgir.HierarchyBlock:
def with_mixin(self, tpe: MixinType) -> MixinType:
"""Adds an interface mixin for this Block. Mainly useful for abstract blocks, e.g. IoController with HasI2s."""
from .BlockInterfaceMixin import BlockInterfaceMixin
if not (isinstance(tpe, BlockInterfaceMixin) and tpe._is_mixin()):
if isinstance(tpe, BlockPrototype):
tpe_cls = tpe._tpe
else:
tpe_cls = tpe.__class__

if not (issubclass(tpe_cls, BlockInterfaceMixin) and tpe_cls._is_mixin()):
raise TypeError("param to with_mixin must be a BlockInterfaceMixin")
if isinstance(self, BlockInterfaceMixin) and self._is_mixin():
raise BlockDefinitionError(self, "mixins can not have with_mixin")
if (self.__class__, AbstractBlockProperty) not in self._elt_properties:
raise BlockDefinitionError(self, "mixins can only be added to abstract classes")
if not isinstance(self, tpe._get_mixin_base()):
raise TypeError(f"block {self.__class__.__name__} not an instance of mixin base {tpe._get_mixin_base().__name__}")
if not isinstance(self, tpe_cls._get_mixin_base()):
raise TypeError(f"block {self.__class__.__name__} not an instance of mixin base {tpe_cls._get_mixin_base().__name__}")
assert self._parent is not None

elt = tpe._bind(self._parent)
Expand Down Expand Up @@ -569,16 +625,23 @@ def Export(self, port: ExportType, tags: Iterable[PortTag]=[], *, optional: bool
def Block(self, tpe: BlockType) -> BlockType:
from .BlockInterfaceMixin import BlockInterfaceMixin
from .DesignTop import DesignTop
if not isinstance(tpe, Block):
raise TypeError(f"param to Block(...) must be Block, got {tpe} of type {type(tpe)}")
if isinstance(tpe, BlockInterfaceMixin) and tpe._is_mixin():
raise TypeError("param to Block(...) must not be BlockInterfaceMixin")
if isinstance(tpe, DesignTop):
raise TypeError(f"param to Block(...) may not be DesignTop")

if self._elaboration_state not in \
[BlockElaborationState.init, BlockElaborationState.contents, BlockElaborationState.generate]:
[BlockElaborationState.init, BlockElaborationState.contents, BlockElaborationState.generate]:
raise BlockDefinitionError(self, "can only define blocks in init, contents, or generate")

if isinstance(tpe, BlockPrototype):
tpe_cls = tpe._tpe
else:
tpe_cls = tpe.__class__

if not issubclass(tpe_cls, Block):
raise TypeError(f"param to Block(...) must be Block, got {tpe_cls}")
if issubclass(tpe_cls, BlockInterfaceMixin) and tpe_cls._is_mixin():
raise TypeError("param to Block(...) must not be BlockInterfaceMixin")
if issubclass(tpe_cls, DesignTop):
raise TypeError(f"param to Block(...) may not be DesignTop")

elt = tpe._bind(self)
self._blocks.register(elt)

Expand Down
6 changes: 3 additions & 3 deletions edg/core/Link.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,15 @@

from .. import edgir
from .Array import BaseVector, DerivedVector
from .Blocks import BaseBlock, Connection
from .Blocks import BaseBlock, Connection, BaseBlockMeta
from .Builder import builder
from .Core import Refable, non_library, ElementMeta
from .Core import Refable, non_library
from .HdlUserExceptions import UnconnectableError
from .IdentityDict import IdentityDict
from .Ports import Port


class LinkMeta(ElementMeta):
class LinkMeta(BaseBlockMeta):
def __new__(cls, *args: Any, **kwargs: Any) -> Any:
new_cls = super().__new__(cls, *args, **kwargs)

Expand Down
9 changes: 7 additions & 2 deletions edg/core/MultipackBlock.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from .Core import non_library, SubElementDict
from .ConstraintExpr import ConstraintExpr, BoolExpr, IntExpr, FloatExpr, RangeExpr, StringExpr
from .Ports import BasePort, Port
from .HierarchyBlock import Block
from .HierarchyBlock import Block, BlockPrototype


class PackedBlockAllocate(NamedTuple):
Expand Down Expand Up @@ -134,7 +134,12 @@ def __init__(self):
def PackedPart(self, tpe: PackedPartType) -> PackedPartType:
"""Adds a block type that can be packed into this block.
The block is a "virtual block" that will not appear in the design tree."""
if not isinstance(tpe, (Block, PackedBlockArray)):
if isinstance(tpe, BlockPrototype):
tpe_cls = tpe._tpe
else:
tpe_cls = tpe.__class__

if not issubclass(tpe_cls, (Block, PackedBlockArray)):
raise TypeError(f"param to PackedPart(...) must be Block, got {tpe} of type {type(tpe)}")
if self._elaboration_state != BlockElaborationState.init:
raise BlockDefinitionError(self, "can only define multipack in init")
Expand Down
16 changes: 15 additions & 1 deletion edg/core/Ports.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,26 @@
from .PortBlocks import PortBridge, PortAdapter


class InitializerContextMeta(type):
def __call__(cls, *args, **kwargs):
"""Hook on construction to store some metadata about its creation.
This hooks the top-level __init__ only."""
obj = type.__call__(cls, *args, **kwargs)
obj._initializer_args = (args, kwargs) # stores args so it is clone-able
return obj


PortParentTypes = Union['BaseContainerPort', 'BaseBlock']
@non_library
class BasePort(HasMetadata):
class BasePort(HasMetadata, metaclass=InitializerContextMeta):
SelfType = TypeVar('SelfType', bound='BasePort')

def __init__(self) -> None:
"""Abstract Base Class for ports"""
self._parent: Optional[PortParentTypes] # refined from Optional[Refable] in base LibraryElement
self._block_context: Optional["Refable"] # set by metaclass, as lexical scope available pre-binding
self._initializer_args: Tuple[Tuple[Any, ...], Dict[str, Any]] # set by metaclass
self._block_context = builder.get_enclosing_block()

super().__init__()

Expand Down Expand Up @@ -157,12 +168,15 @@ def init_from(self: SelfType, other: SelfType):

def _bridge(self) -> Optional[PortBridge]:
"""Creates a (unbound) bridge and returns it."""
from .HierarchyBlock import Block

if self.bridge_type is None:
return None
if self._bridge_instance is not None:
return self._bridge_instance
assert self._is_bound(), "not bound, can't create bridge"

Block._next_bind = self.bridge_type
self._bridge_instance = self.bridge_type()
return self._bridge_instance

Expand Down
2 changes: 1 addition & 1 deletion edg/core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from .MultiBiDict import MultiBiDict

# Features for library builders
from .Core import LibraryElement, SubElementDict, ElementDict, ElementMeta, non_library
from .Core import LibraryElement, SubElementDict, ElementDict, non_library
from .Blocks import BasePort, BaseBlock
from .Categories import InternalBlock
from .Builder import builder
Expand Down
Loading
Loading