Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 14 additions & 17 deletions edg/core/Blocks.py
Original file line number Diff line number Diff line change
Expand Up @@ -407,35 +407,32 @@ def _bind_in_place(self, parent: Union[BaseBlock, Port]):
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_curr_context() is self._lexical_parent, "can't clone to different context"
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):
if isinstance(expr.parent, BaseBlock):
block_parent = expr.parent
elif isinstance(expr.parent, BasePort):
block_parent = cast(BaseBlock, expr.parent._block_parent()) # TODO make less ugly
assert block_parent is not None
else:
raise ValueError(f"unknown parent {expr.parent} of {expr}")

if isinstance(block_parent._parent, BasePort):
block_parent_parent: Any = block_parent._parent._block_parent()
else:
block_parent_parent = block_parent._parent

if not (block_parent is self or block_parent_parent is self):
raise UnreachableParameterError(f"In {type(self)}, constraint references unreachable parameter {expr}. "
expr_parent = expr.binding.parent
if isinstance(expr_parent, BasePort):
expr_block_parent = expr_parent._block_parent()
assert expr_block_parent is not None
expr_parent = expr_block_parent

expr_parent_parent = expr_parent._parent # may be None for top-level
if isinstance(expr_parent_parent, BasePort): # resolve Link parent port to parent block
expr_parent_parent = expr_parent_parent._block_parent()

if not (expr_parent is self or expr_parent_parent is self):
raise UnreachableParameterError(f"In {self}, constraint references unreachable parameter {expr}. "
"Only own parameters, or immediate contained blocks' parameters can be accessed.")
elif isinstance(expr, BasePort):
block_parent = cast(BaseBlock, expr._block_parent())
assert block_parent is not None
if not block_parent is self or block_parent._parent is self:
raise UnreachableParameterError(f"In {type(self)}, constraint references unreachable port {expr}. "
raise UnreachableParameterError(f"In {self}, constraint references unreachable port {expr}. "
"Only own ports, or immediate contained blocks' ports can be accessed.")

for subexpr in constraint._get_exprs():
Expand Down
34 changes: 18 additions & 16 deletions edg/core/Builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,42 +2,44 @@

from typing import *

from deprecated import deprecated

from .. import edgir

if TYPE_CHECKING:
from .Core import Refable
from .Blocks import BaseBlock


class Builder:
def __init__(self) -> None:
self.stack: List[Refable] = []
self.stack: List[BaseBlock] = []

def push_element(self, elt: Refable) -> None:
self.stack.append(elt)
def push_element(self, elt: BaseBlock) -> Optional[BaseBlock]:
"""Pushes a new element onto the context stack, returning the previous top element.
Ignores if the element is already on top of the stack."""
prev_elt = self.get_enclosing_block()
if not self.stack or self.stack[-1] is not elt: # prevent double-pushing
self.stack.append(elt)
return prev_elt

def pop_to(self, elt: Optional[Refable]) -> None:
self.stack.pop()
assert self.get_curr_context() is elt
def pop_to(self, elt: Optional[BaseBlock]) -> None:
while (elt is None and self.stack) or (elt is not None and self.stack[-1] is not elt):
self.stack.pop()

def get_enclosing_block(self) -> Optional[BaseBlock]:
from .Blocks import BaseBlock
# traverse down the stack and get the first BaseBlock, ignoring things like ports
for elt in reversed(self.stack):
if isinstance(elt, BaseBlock):
return elt
return None

def get_curr_context(self) -> Optional[Refable]:
if not self.stack:
return None
else:
return self.stack[-1]

@deprecated("use get_enclosing_block() instead, context frames can only be blocks now")
def get_curr_context(self) -> Optional[BaseBlock]:
return self.get_enclosing_block()

def elaborate_toplevel(self, block: BaseBlock, *,
is_generator: bool = False,
generate_values: Iterable[Tuple[edgir.LocalPath, edgir.ValueLit]] = []) -> edgir.HierarchyBlock:
assert self.get_curr_context() is None
assert self.get_enclosing_block() is None
self.push_element(block)
try:
if is_generator: # TODO this is kind of nasty =(
Expand Down
5 changes: 3 additions & 2 deletions edg/core/ConstraintExpr.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

if TYPE_CHECKING:
from .Ports import BasePort
from .Blocks import BaseBlock


SelfType = TypeVar('SelfType', bound='ConstraintExpr')
Expand Down Expand Up @@ -64,7 +65,7 @@ def __init__(self: SelfType, initializer: Optional[Union[SelfType, WrappedType]]
self.initializer = None
else:
self.initializer = self._to_expr_type(initializer)
self.parent = builder.get_curr_context()
self._context: Optional[BaseBlock] = builder.get_enclosing_block()

def _get_exprs(self) -> Iterable[Union[ConstraintExpr, BasePort]]:
assert self.binding is not None
Expand All @@ -80,7 +81,7 @@ def _new_bind(cls: Type[SelfType], binding: Binding) -> SelfType:
def _bind(self: SelfType, binding: Binding) -> SelfType:
"""Returns a clone of this object with the specified binding. This object must be unbound."""
assert not self._is_bound()
assert builder.get_curr_context() is self.parent, f"can't clone in original context {self.parent} to different new context {builder.get_curr_context()}"
assert builder.get_enclosing_block() is self._context, f"can't clone in original context {self._context} to different new context {builder.get_enclosing_block()}"
if not isinstance(binding, ParamBinding):
assert self.initializer is None, "Only Parameters may have initializers"
clone: SelfType = type(self)(self.initializer)
Expand Down
19 changes: 6 additions & 13 deletions edg/core/Core.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,17 +146,12 @@ 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):
parent = builder.get_curr_context()
block_context = builder.get_enclosing_block()
try:
obj = type.__call__(cls, *args, **kwargs)
obj._initializer_args = (args, kwargs) # stores args so it is clone-able
obj._lexical_parent = parent # stores context for error checking
obj._block_context = block_context
obj._post_init()
finally:
if builder.get_curr_context() is not parent: # in case the constructor skipped internal element init
builder.pop_to(parent)

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

Expand Down Expand Up @@ -198,12 +193,10 @@ def __repr__(self) -> str:
return "%s@%02x" % (self._get_def_name(), (id(self) // 4) & 0xff)

def __init__(self) -> None:
self._lexical_parent: Optional[LibraryElement] # set by metaclass
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

builder.push_element(self)

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

Expand Down
33 changes: 15 additions & 18 deletions edg/core/HierarchyBlock.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,12 +147,11 @@ def __new__(cls, *args: Any, **kwargs: Any) -> Any:
arg_data: List[Tuple[str, inspect.Parameter, Type[ConstraintExpr]]] = []
# discard param 0 (self)
for arg_name, arg_param in list(inspect.signature(orig_init).parameters.items())[1:]:
if arg_param.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
param_expr_type = BlockMeta._ANNOTATION_EXPR_MAP.get(arg_param.annotation, None)
if param_expr_type is None:
raise BlockDefinitionError(new_cls, f"in {new_cls}.__init__, unknown annotation type for {arg_name}: {arg_param.annotation}")
else:
param_expr_type = ConstraintExpr # dummy placeholder
if arg_param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
continue # pass through of *args, **kwargs handled later
param_expr_type = BlockMeta._ANNOTATION_EXPR_MAP.get(arg_param.annotation, None)
if param_expr_type is None:
raise BlockDefinitionError(new_cls, f"in {new_cls}.__init__, unknown annotation type for {arg_name}: {arg_param.annotation}")

arg_data.append((arg_name, arg_param, param_expr_type))

Expand Down Expand Up @@ -182,21 +181,13 @@ def remap_arg(arg_name: str, arg_type: Type[ConstraintExpr], arg_value: Any) ->

return arg_type()._bind(InitParamBinding(self, typed_arg_value))

# create wrapper ConstraintExpr in new object scope
builder_prev = builder.get_curr_context()
builder.push_element(self)
builder_prev = builder.push_element(self)
try:
# rebuild args and kwargs by traversing the args list
new_args: List[Any] = []
new_kwargs: Dict[str, Any] = {}
for arg_pos, (arg_name, arg_param, param_expr_type) in enumerate(arg_data):
if arg_param.kind == inspect.Parameter.VAR_POSITIONAL: # pass-through *args, handled at lower level
new_args.extend(args[arg_pos:])
elif arg_param.kind == inspect.Parameter.VAR_KEYWORD: # pass-through *kwargs, handled at lower level
for arg_name in kwargs:
if arg_name not in new_kwargs:
new_kwargs[arg_name] = kwargs[arg_name]
elif arg_pos < len(args) and arg_param.kind in (inspect.Parameter.POSITIONAL_ONLY,
if arg_pos < len(args) and arg_param.kind in (inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD): # present positional arg
new_arg = remap_arg(arg_name, param_expr_type, args[arg_pos])
new_args.append(new_arg)
Expand All @@ -221,11 +212,17 @@ def remap_arg(arg_name: str, arg_type: Type[ConstraintExpr], arg_value: Any) ->
new_arg = remap_arg(arg_name, param_expr_type, None)
new_kwargs[arg_name] = new_arg
self._init_params[arg_name] = new_arg

# unconditionally pass through all args and kwargs
new_args.extend(args[len(new_args):])
for arg_name in kwargs:
if arg_name not in new_kwargs:
new_kwargs[arg_name] = kwargs[arg_name]

orig_init(self, *new_args, **new_kwargs)
finally:
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Corrected spelling of 'unconditioally' to 'unconditionally'.

Copilot uses AI. Check for mistakes.
builder.pop_to(builder_prev)

orig_init(self, *new_args, **new_kwargs)

new_cls.__init__ = functools.update_wrapper(wrapped_init, orig_init)

return new_cls
Expand Down
24 changes: 22 additions & 2 deletions edg/core/Link.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,38 @@
from __future__ import annotations

import functools
from typing import *

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


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

if '__init__' in new_cls.__dict__:
orig_init = new_cls.__dict__['__init__']
def wrapped_init(self, *args, **kwargs) -> None:
builder_prev = builder.push_element(self)
try:
orig_init(self, *args, **kwargs)
finally:
builder.pop_to(builder_prev)

new_cls.__init__ = functools.update_wrapper(wrapped_init, orig_init)

return new_cls


@non_library
class Link(BaseBlock[edgir.Link]):
class Link(BaseBlock[edgir.Link], metaclass=LinkMeta):
def __init__(self) -> None:
super().__init__()
self.parent: Optional[Port] = None
Expand Down
4 changes: 2 additions & 2 deletions edg/core/MultipackBlock.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ def packed_assign(self, self_param: ConstraintExpr, packed_param: PackedParamTyp
raise BlockDefinitionError(self, "can only define multipack in init")
if isinstance(packed_param, ConstraintExpr):
assert type(self_param) == type(packed_param), "packed_assign parameters must be of the same type"
block_parent = packed_param.parent
block_parent = packed_param._context
assert isinstance(block_parent, Block)
self._packed_assigns_by_packed_block[block_parent][self_param] = packed_param
elif isinstance(packed_param, PackedBlockParamArray):
Expand Down Expand Up @@ -234,7 +234,7 @@ def unpacked_assign(self, packed_param: UnpackedParamTypes, self_param: Constrai
raise BlockDefinitionError(self, "can only define multipack in init")
if isinstance(packed_param, ConstraintExpr):
assert type(packed_param) == type(self_param), "unpacked_assign parameters must be of the same type"
block_parent = packed_param.parent
block_parent = packed_param._context
assert isinstance(block_parent, Block)
self._unpacked_assigns_by_packed_block[block_parent][self_param] = packed_param
elif isinstance(packed_param, PackedBlockParam):
Expand Down
1 change: 0 additions & 1 deletion edg/core/Ports.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ class BasePort(HasMetadata):
def __init__(self) -> None:
"""Abstract Base Class for ports"""
self._parent: Optional[PortParentTypes] # refined from Optional[Refable] in base LibraryElement
self._block_context: Optional[BaseBlock] # set by metaclass, as lexical scope available pre-binding
self._initializer_args: Tuple[Tuple[Any, ...], Dict[str, Any]] # set by metaclass

super().__init__()
Expand Down
2 changes: 1 addition & 1 deletion edg/jlcparts/JlcPartsDiode.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,4 +70,4 @@ def _entry_to_table_row(cls, row_dict: Dict[PartsTableColumn, Any], filename: st
return None


lambda: JlcPartsDiode(), JlcPartsZenerDiode() # ensure class is instantiable (non-abstract)
lambda: (JlcPartsDiode(), JlcPartsZenerDiode()) # ensure class is instantiable (non-abstract)
2 changes: 1 addition & 1 deletion edg/jlcparts/JlcPartsFet.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,4 +68,4 @@ class JlcPartsSwitchFet(PartsTableSelectorFootprint, JlcPartsBaseFet, TableSwitc
pass


lambda: JlcPartsFet(), JlcPartsSwitchFet() # ensure class is instantiable (non-abstract)
lambda: (JlcPartsFet(), JlcPartsSwitchFet()) # ensure class is instantiable (non-abstract)
2 changes: 1 addition & 1 deletion edg/parts/JlcFet.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,4 +196,4 @@ class JlcSwitchFet(PartsTableSelectorFootprint, JlcBaseFet, FetFallbackGateCharg
pass


lambda: JlcFet, JlcSwitchFet() # ensure class is instantiable (non-abstract)
lambda: (JlcFet(), JlcSwitchFet()) # ensure class is instantiable (non-abstract)