diff --git a/edg/abstract_parts/Connectors.py b/edg/abstract_parts/Connectors.py index 42261a839..8d07f565a 100644 --- a/edg/abstract_parts/Connectors.py +++ b/edg/abstract_parts/Connectors.py @@ -32,9 +32,9 @@ def __init__(self) -> None: class RfConnectorTestPoint(BlockInterfaceMixin[RfConnector]): """Test point mixin that allows the footprint to take a name""" - def __init__(self, name: StringLike): + def __init__(self, tp_name: StringLike): super().__init__() - self.tp_name = self.ArgParameter(name) + self.tp_name = self.ArgParameter(tp_name) class RfConnectorAntenna(Antenna): diff --git a/edg/abstract_parts/IoController.py b/edg/abstract_parts/IoController.py index 4826c02da..394269463 100644 --- a/edg/abstract_parts/IoController.py +++ b/edg/abstract_parts/IoController.py @@ -172,7 +172,7 @@ def connect_port_transformed(self_io: BasePort, inner_io: Callable[[], BasePort] if transform_fn is None: self.connect(self_io, inner_io) else: - connect_port_transformed(self_io, lambda: inner_io, self_io._name_from(self)) + connect_port_transformed(self_io, lambda: inner_io, self._path_name_of(self_io)) else: raise NotImplementedError(f"unknown port type {self_io}") @@ -293,9 +293,7 @@ def _make_pinning(self) -> Mapping[Union[Iterable[str], str], Union[Passive, Has allocation_list.append((io_port.elt_type(), self.get(io_port.requested()))) elif isinstance(io_port, Port): # derive Port connections from is_connected if self.get(io_port.is_connected()): - requested = [ - self._name_of_child(io_port, self) - ] # generate requested name from port name if connected + requested = [self._name_of(io_port)] # generate requested name from port name if connected else: requested = [] allocation_list.append((type(io_port), requested)) diff --git a/edg/abstract_parts/IoControllerWrapper.py b/edg/abstract_parts/IoControllerWrapper.py index 6a8a453d6..332aa08f7 100644 --- a/edg/abstract_parts/IoControllerWrapper.py +++ b/edg/abstract_parts/IoControllerWrapper.py @@ -80,8 +80,8 @@ def recurse_port(port: Port, prefix: str) -> None: assert prefix not in pin_dict, f"duplicate pin name {prefix}" pin_dict[prefix] = port - for subport_name, subport in port._ports.items(): - recurse_port(subport, f"{prefix}.{subport_name}") + for subport in port._ports: + recurse_port(subport, f"{prefix}.{port._name_of(subport)}") for io_port in self._io_ports: if isinstance(io_port, Vector): @@ -91,7 +91,7 @@ def recurse_port(port: Port, prefix: str) -> None: recurse_port(subport, subport_name) elif isinstance(io_port, Port): if self.get(io_port.is_connected()): - port_name = io_port._name_from(self) + port_name = self._path_name_of(io_port) recurse_port(io_port, port_name) else: raise NotImplementedError(f"unknown port type {io_port}") @@ -203,8 +203,8 @@ def recurse_port(port: Port, prefix: str) -> None: assert prefix not in pin_type_dict, f"duplicate pin name {prefix}" pin_type_dict[prefix] = type(port) - for subport_name, subport in port._ports.items(): - recurse_port(subport, f"{prefix}.{subport_name}") + for subport in port._ports: + recurse_port(subport, f"{prefix}.{port._name_of(subport)}") for io_port in self._io_ports: if isinstance(io_port, Vector): @@ -212,7 +212,7 @@ def recurse_port(port: Port, prefix: str) -> None: recurse_port(io_port._tpe.empty(), subport_name) elif isinstance(io_port, Port): if self.get(io_port.is_connected()): - port_name = io_port._name_from(self) + port_name = self._name_of(io_port) recurse_port(io_port, port_name) else: raise NotImplementedError(f"unknown port type {io_port}") diff --git a/edg/abstract_parts/PinMappable.py b/edg/abstract_parts/PinMappable.py index 2f6f84969..63d1b135c 100644 --- a/edg/abstract_parts/PinMappable.py +++ b/edg/abstract_parts/PinMappable.py @@ -435,7 +435,8 @@ def try_allocate_resource( resource_model = resource.port_model # typer gets confused if this is put directly where it is used inner_models = {} resource_name = resource.name # typer gets confused if this is put directly where it is used - for inner_name, inner_model in resource.port_model._ports.items(): + for inner_model in resource.port_model._ports: + inner_name = resource.port_model._name_of(inner_model) inner_type = type(inner_model) if inner_type in self.transforms: # apply transform to search for the resource type, if needed inner_type = self.transforms[inner_type][0] diff --git a/edg/core/Array.py b/edg/core/Array.py index bd5824db2..11d5e67d1 100644 --- a/edg/core/Array.py +++ b/edg/core/Array.py @@ -155,36 +155,20 @@ def items(self) -> ItemsView[str, VectorType]: # unlike most other LibraryElement types, the names are stored in _elts and _allocates @override - def _name_of_child(self, subelt: Any, context: Any, allow_unknown: bool = False) -> str: - from .HierarchyBlock import Block - - block_parent = self._block_parent() - assert isinstance(block_parent, Block) - - if context is block_parent: - # in block defining this port (direct elt definition), or in test top - assert self._elts is not None, "can't get name on undefined vector" + def _name_of(self, subelt: Any, *, debug: bool = False) -> str: + if self._elts is not None: for name, elt in self._elts.items(): if subelt is elt: return name - if allow_unknown: - return f"(unknown {subelt.__class__.__name__})" - else: - raise ValueError(f"no name for {subelt}") - elif context is block_parent._parent: - # in block enclosing the block defining this port (allocate required) - for i, (suggested_name, allocate_elt) in enumerate(self._requests): - if subelt is allocate_elt: - if suggested_name is not None: - return suggested_name - else: - return f"_allocate_{i}" - if allow_unknown: - return f"(unknown {subelt.__class__.__name__})" - else: - raise ValueError(f"allocated elt not found {subelt}") - else: - raise ValueError(f"unknown context of array") + + for i, (suggested_name, allocate_elt) in enumerate(self._requests): + if subelt is allocate_elt: + if suggested_name is not None: + return suggested_name + else: + return f"_allocate_{i}" + + raise ValueError(f"{subelt} not in Vector") @override def _get_elt_sample(self) -> BasePort: @@ -338,14 +322,19 @@ def validate_selector(expected: Type[SelectorType], result: ConstraintExpr) -> S # See the note in ArrayExpr for why this is expanded. @overload def map_extract(self, selector: Callable[[VectorType], BoolExpr]) -> ArrayBoolExpr: ... + @overload def map_extract(self, selector: Callable[[VectorType], IntExpr]) -> ArrayIntExpr: ... + @overload def map_extract(self, selector: Callable[[VectorType], FloatExpr]) -> ArrayFloatExpr: ... + @overload def map_extract(self, selector: Callable[[VectorType], RangeExpr]) -> ArrayRangeExpr: ... + @overload def map_extract(self, selector: Callable[[VectorType], StringExpr]) -> ArrayStringExpr: ... + @overload def map_extract(self, selector: Callable[[VectorType], ExtractPortType]) -> DerivedVector[ExtractPortType]: ... @@ -377,8 +366,10 @@ def count(self, selector: Callable[[VectorType], BoolExpr]) -> IntExpr: @overload def sum(self, selector: Callable[[VectorType], RangeExpr]) -> RangeExpr: ... + @overload def sum(self, selector: Callable[[VectorType], FloatExpr]) -> FloatExpr: ... + def sum(self, selector: Callable[[VectorType], Union[RangeExpr, FloatExpr]]) -> Union[RangeExpr, FloatExpr]: param = selector(self._elt_sample) if isinstance(param, FloatExpr): diff --git a/edg/core/BlockInterfaceMixin.py b/edg/core/BlockInterfaceMixin.py index 699cfee86..fe6daed77 100644 --- a/edg/core/BlockInterfaceMixin.py +++ b/edg/core/BlockInterfaceMixin.py @@ -1,7 +1,7 @@ from typing import Generic, Type, List, Optional, get_args, get_origin, Tuple, Callable, Any from typing_extensions import TypeVar, override -from .Core import non_library, HasMetadata +from .Core import non_library, HasMetadata, LibraryElement from .Blocks import AbstractBlockProperty from .HdlUserExceptions import BlockDefinitionError from .HierarchyBlock import Block @@ -30,7 +30,7 @@ class BlockInterfaceMixin(Block, Generic[MixinBaseType]): TODO: what about cases where it's a bit mixed, e.g. HasI2s also needs to register the self.i2s port? """ - BaseType = TypeVar("BaseType", bound=HasMetadata) + BaseType = TypeVar("BaseType", bound=LibraryElement) @classmethod @override diff --git a/edg/core/Blocks.py b/edg/core/Blocks.py index d1a7420ca..4d8512ab9 100644 --- a/edg/core/Blocks.py +++ b/edg/core/Blocks.py @@ -12,7 +12,15 @@ from .Array import BaseVector, Vector from .Binding import AssignBinding, NameBinding from .ConstraintExpr import ConstraintExpr, BoolExpr, ParamBinding, AssignExpr, StringExpr, BoolLike -from .Core import Refable, HasMetadata, builder, SubElementDict, non_library, EltPropertiesBase +from .Core import ( + Refable, + HasMetadata, + builder, + non_library, + EltPropertiesBase, + SourceLocator, + NamingContainer, +) from .HdlUserExceptions import * from .IdentityDict import IdentityDict from .IdentitySet import IdentitySet @@ -110,7 +118,7 @@ def add_ports(self, ports: Iterable[BasePort]) -> None: initializers = ext_port._get_initializers([]) if initializers: raise UnconnectableError( - f"Connected boundary port {ext_port._name_from(self.parent, allow_unknown=True)} may not have initializers, " + f"Connected boundary port {self.parent._path_name_of(ext_port, debug=True)} may not have initializers, " f"got {', '.join(['.'.join(path) + '=' + str(value) for _, path, value in initializers])}" ) return # is an export, not a connection @@ -125,7 +133,7 @@ def add_ports(self, ports: Iterable[BasePort]) -> None: assert not self.link_connected_ports assert not self.bridged_ports assert not self.available_link_ports_by_type - for name, port in link._ports.items(): + for port in link._ports: self.available_link_ports_by_type.setdefault(type(self._baseport_leaf_type(port)), []).append(port) else: link = self.link_instance @@ -136,23 +144,23 @@ def add_ports(self, ports: Iterable[BasePort]) -> None: if port._block_parent() is self.parent: if port._get_initializers([]): raise UnconnectableError( - f"Connected boundary port {port._name_from(self.parent, allow_unknown=True)} may not have initializers" + f"Connected boundary port {self.parent._path_name_of(port, debug=True)} may not have initializers" ) if not isinstance(port, Port): raise UnconnectableError( - f"Can't generate bridge for non-Port {port._name_from(self.parent, allow_unknown=True)}" + f"Can't generate bridge for non-Port {self.parent._path_name_of(port, debug=True)}" ) bridge = port._bridge() if bridge is None: - raise UnconnectableError(f"No bridge for {port._name_from(self.parent, allow_unknown=True)}") + raise UnconnectableError(f"No bridge for {self.parent._path_name_of(port, debug=True)}") link_facing_port = self.bridged_ports[port] = bridge.inner_link else: link_facing_port = port elif isinstance(self.parent, Link): # links don't bridge, all ports are treated as internal if port._block_parent() is not self.parent: raise UnconnectableError( - f"Port {port._name_from(self.parent, allow_unknown=True)} not in containing link" + f"Port {self.parent._path_name_of(port, debug=True)} not in containing link" ) link_facing_port = port else: @@ -165,18 +173,16 @@ def add_ports(self, ports: Iterable[BasePort]) -> None: # allocate the connection if self._baseport_leaf_type(link_facing_port).link_type is not type(link): raise UnconnectableError( - f"Can't connect {port._name_from(self.parent, allow_unknown=True)} to link of type {type(link)}" + f"Can't connect {self.parent._path_name_of(port, debug=True)} to link of type {type(link)}" ) port_type = type(self._baseport_leaf_type(link_facing_port)) allocatable_link_ports = self.available_link_ports_by_type.get(port_type, None) if allocatable_link_ports is None: raise UnconnectableError( - f"No link port for {port._name_from(self.parent, allow_unknown=True)} of type {port_type}" + f"No link port for {self.parent._path_name_of(port, debug=True)} of type {port_type}" ) if not allocatable_link_ports: - raise UnconnectableError( - f"No remaining link ports to {port._name_from(self.parent, allow_unknown=True)}" - ) + raise UnconnectableError(f"No remaining link ports to {self.parent._path_name_of(port, debug=True)}") allocated_link_port = allocatable_link_ports[0] if isinstance(allocated_link_port, BaseVector): # array on link side, can connected multiple ports @@ -268,7 +274,7 @@ def set_elt_proto(self, pb: edgir.BlockLikeTypes, ref_map: Refable.RefMapType) - @non_library -class BaseBlock(HasMetadata, metaclass=BaseBlockMeta): +class BaseBlock(HasMetadata, NamingContainer, metaclass=BaseBlockMeta): """Base block that has ports (IOs), parameters, and constraints between them.""" # __init__ should initialize the object with structural information (parameters, fields) @@ -285,21 +291,19 @@ def __init__(self) -> None: None # additional string field to be displayed as part of the tooltip for blocks ) - self._parameters: SubElementDict[ConstraintExpr] = self.manager.new_dict(ConstraintExpr) + self._parameters: List[ConstraintExpr] = [] self._param_docs = IdentityDict[ConstraintExpr, str]() - self._ports: SubElementDict[BasePort] = self.manager.new_dict(BasePort) + self._ports: List[BasePort] = [] self._required_ports = IdentityDict[BasePort, Optional[BoolExpr]]() # port -> precondition for required self._port_docs = IdentityDict[BasePort, str]() - self._connects = self.manager.new_dict(Connection, anon_prefix="anon_link") + self._connects: List[Connection] = [] self._connects_by_port = IdentityDict[BasePort, Connection]() # port -> connection self._connect_delegateds = IdentityDict[ Connection, List[Connection] ]() # for net joins, joined connect -> prior connects - self._constraints: SubElementDict[ConstraintExpr] = self.manager.new_dict( - ConstraintExpr, anon_prefix="anon_constr" - ) + self._constraints: List[ConstraintExpr] = [] self._name = StringExpr()._bind(NameBinding(self)) @@ -350,12 +354,12 @@ def _add_doc_metadata(self) -> None: metadata_dict: Dict[str, str] = {} if self.__doc__: metadata_dict[""] = inspect.cleandoc(self.__doc__) - for name, param in self._parameters.items(): + for param in self._parameters: if param in self._param_docs: - metadata_dict[name] = self._param_docs[param] - for name, port in self._ports.items(): + metadata_dict[self._name_of(param)] = self._param_docs[param] + for port in self._ports: if port in self._port_docs: - metadata_dict[name] = self._port_docs[port] + metadata_dict[self._name_of(port)] = self._port_docs[port] self._docs = self.Metadata(metadata_dict) def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: @@ -365,9 +369,6 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: or self._elaboration_state == BlockElaborationState.post_generate ) - self._parameters.finalize() - self._ports.finalize() - if (self.__class__, AbstractBlockProperty) in self._elt_properties: assert isinstance(pb, edgir.HierarchyBlock) pb.is_abstract = True @@ -387,15 +388,15 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: for cls in indirect_bases: pb.super_superclasses.add().target.name = cls._static_def_name() - for name, param in self._parameters.items(): + for param in self._parameters: assert isinstance(param.binding, ParamBinding) - param._populate_decl_proto(edgir.add_pair(pb.params, name)) + param._populate_decl_proto(edgir.add_pair(pb.params, self._name_of(param))) - for name, port in self._ports.items(): - port._populate_portlike_proto(edgir.add_pair(pb.ports, name)) + for port in self._ports: + port._populate_portlike_proto(edgir.add_pair(pb.ports, self._name_of(port))) ref_map = self._create_ref_map() - for name, port in self._ports.items(): + for port in self._ports: if port in self._required_ports: if isinstance(port, Port): connected_condition = port.is_connected() @@ -410,16 +411,18 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: connected_condition = precondition.implies(connected_condition) # add required constraints first - connected_condition._populate_expr_proto(edgir.add_pair(pb.constraints, f"(reqd){name}"), ref_map) - - self._constraints.finalize() # needed for source locator generation + # TODO move to port declaration time + connected_condition._populate_expr_proto( + edgir.add_pair(pb.constraints, f"(reqd){self._name_of(port)}"), ref_map + ) self._add_doc_metadata() - self._populate_metadata(pb.meta, self._metadata, ref_map) + named_metadata = {self._name_of(metadata): metadata for metadata in self._metadata} + self._populate_metadata(pb.meta, named_metadata, ref_map) def _populate_def_proto_port_init(self, pb: edgir.BlockLikeTypes, ref_map: Refable.RefMapType) -> None: - for name, port in self._ports.items(): - for param, path, initializer in port._get_initializers([name]): + for port in self._ports: + for param, path, initializer in port._get_initializers([self._name_of(port)]): AssignBinding.populate_assign_proto( edgir.add_pair(pb.constraints, f"(init){'.'.join(path)}"), param, @@ -427,13 +430,6 @@ def _populate_def_proto_port_init(self, pb: edgir.BlockLikeTypes, ref_map: Refab ref_map, ) - def _populate_def_proto_param_init(self, pb: edgir.BlockLikeTypes, ref_map: Refable.RefMapType) -> None: - for name, param in self._parameters.items(): - if param.initializer is not None: - AssignBinding.populate_assign_proto( - edgir.add_pair(pb.constraints, f"(init){name}"), param, param.initializer, ref_map - ) - def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: Refable.RefMapType) -> None: """Populates the contents of a block proto: constraints""" assert ( @@ -441,7 +437,8 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: or self._elaboration_state == BlockElaborationState.post_generate ) - for name, constraint in self._constraints.items(): + for i, constraint in enumerate(self._constraints): + name = self._name_of(constraint) constraint._populate_expr_proto(edgir.add_pair(pb.constraints, name), ref_map) def _populate_def_proto_description(self, pb: edgir.BlockLikeTypes, ref_map: Refable.RefMapType) -> None: @@ -454,10 +451,10 @@ def _populate_def_proto_description(self, pb: edgir.BlockLikeTypes, ref_map: Ref def _build_ref_map(self, ref_map: Refable.RefMapType, prefix: edgir.LocalPath) -> None: super()._build_ref_map(ref_map, prefix) ref_map[self.name()] = edgir.localpath_concat(prefix, edgir.NAME) - for name, param in self._parameters.items(): - param._build_ref_map(ref_map, edgir.localpath_concat(prefix, name)) - for name, port in self._ports.items(): - port._build_ref_map(ref_map, edgir.localpath_concat(prefix, name)) + for param in self._parameters: + param._build_ref_map(ref_map, edgir.localpath_concat(prefix, self._name_of(param))) + for port in self._ports: + port._build_ref_map(ref_map, edgir.localpath_concat(prefix, self._name_of(port))) def _bind_in_place(self, parent: Union[BaseBlock, Port]) -> None: self._parent = parent @@ -494,16 +491,22 @@ def check_subexpr(expr: Union[ConstraintExpr, BasePort]) -> None: # TODO rewrit def require(self, constraint: BoolLike, name: Optional[str] = None, *, _unchecked: bool = False) -> BoolExpr: constraint_typed = BoolExpr._to_expr_type(constraint) + if isinstance(constraint_typed.binding, ParamBinding): + # if this otherwise would return the raw parameter, wrap in an expression + # otherwise, this may redefine the name of the parameter + constraint_typed = constraint_typed == True + if not isinstance(name, (str, type(None))): raise EdgTypeError(f"require(...) name", name, (str, None)) if not _unchecked: # allow breaking structural / abstraction rules, used in some unit tests self._check_constraint(constraint_typed) - self._constraints.register(constraint_typed) + self._constraints.append(constraint_typed) + self._namer.add_expected(constraint_typed, SourceLocator.from_caller(), default_prefix="anon_constr") if name: # TODO unify naming API with everything else? - self.manager.add_element(name, constraint_typed) + self._namer.add_name(constraint_typed, name) return constraint_typed @@ -525,10 +528,11 @@ def assign( self._check_constraint(expr_value) constraint = AssignExpr()._bind(AssignBinding(target, expr_value)) - self._constraints.register(constraint) + self._constraints.append(constraint) + self._namer.add_expected(constraint, SourceLocator.from_caller(), default_prefix="anon_constr") if name: # TODO unify naming API with everything else? - self.manager.add_element(name, constraint) + self._namer.add_name(constraint, name) return constraint @@ -546,7 +550,8 @@ def Port(self, tpe: T, *, optional: BoolLike = False, doc: Optional[str] = None) raise EdgTypeError(f"param to Port(...)", tpe, BasePort) elt = tpe._bind(self) - self._ports.register(elt) + self._ports.append(elt) + self._namer.add_expected(elt, SourceLocator.from_caller()) if isinstance(optional, bool): if not optional: @@ -575,9 +580,11 @@ def Parameter(self, tpe: ConstraintType, *, doc: Optional[str] = None) -> Constr raise EdgTypeError(f"param to Parameter(...)", tpe, ConstraintExpr) elt = tpe._bind(ParamBinding(self)) - self._parameters.register(elt) + self._parameters.append(elt) + self._namer.add_expected(elt, SourceLocator.from_caller()) if elt.initializer is not None: - self._check_constraint(elt.initializer) + assign = self.assign(elt, elt.initializer) + self._namer.add_name(assign, lambda: f"(init){self._name_of(elt)}") if doc is not None: self._param_docs[elt] = doc @@ -604,7 +611,8 @@ def connect(self, *connects: Union[BasePort, Connection], flatten: bool = False) existing_connects = connects_connects + connects_ports_connects if not existing_connects: connect = Connection(self, flatten) - self._connects.register(connect) + self._connects.append(connect) + self._namer.add_expected(connect, SourceLocator.from_caller(), default_prefix="anon_link") else: # append to (potentially multiple) existing connect connect = existing_connects[0] # first one is authoritative connect_flattens = set([connect.flatten for connect in existing_connects]) diff --git a/edg/core/Core.py b/edg/core/Core.py index 388bdf11f..9229f3654 100644 --- a/edg/core/Core.py +++ b/edg/core/Core.py @@ -1,124 +1,138 @@ from __future__ import annotations +import inspect +from types import FrameType, FunctionType from typing import * from abc import abstractmethod from typing_extensions import override +from .HdlUserExceptions import ElementNamingError from .. import edgir from .Builder import builder from .IdentityDict import IdentityDict -from .IdentitySet import IdentitySet ElementType = TypeVar("ElementType") -class SubElementDict(Generic[ElementType]): - def __init__(self, anon_prefix: Optional[str] = None) -> None: - self.anons = IdentitySet[ElementType]() # TODO this should be order-preserving? - self.anon_prefix = anon_prefix - self.container: Dict[str, ElementType] = {} - self.names = IdentityDict[ElementType, str]() # TODO unify w/ container? - self.keys_list: List[str] = [] - self.closed = False - - def register(self, item: ElementType) -> ElementType: - assert not self.closed, "adding items after closed" - self.anons.add(item) - return item - - def add_element(self, name: str, item: Any) -> None: - assert not self.closed, "naming items after closed" - assert item in self.anons, f"attempted to add {name}={item}, but did not pre-register" - self.anons.remove(item) - assert name not in self.container, f"attempted to reassign {name}={item}" - - self.container[name] = item - self.names[item] = name - self.keys_list.append(name) - - # TODO should this be automatically called? - def finalize(self) -> None: - if self.anon_prefix is None: - assert not self.anons, f"can't have unnamed objects: {self.anons}" - else: - for id, elt in enumerate(self.anons): - name = f"{self.anon_prefix}_{id}" - assert name not in self.container, f"duplicate name {name}" - self.container[name] = elt - self.names[elt] = name - self.keys_list.append(name) - self.anons = IdentitySet[ElementType]() # TODO needs clear operation - self.closed = True - - def all_values_temp(self) -> Iterable[ElementType]: # TODO needs better API name, reconcile w/ values? - return list(self.container.values()) + list(self.anons) - - # TODO the below allows this to be used like a dict, is this a good idea? - # TODO should these make sure the dict is closed? - def items(self) -> ItemsView[str, ElementType]: - return self.container.items() - - def items_ordered(self) -> Iterable[Tuple[str, ElementType]]: - return [(key, self.container[key]) for key in self.keys_list] +class SourceLocator: + """Object that can be used to locate the source of a library element, for error reporting.""" - def keys_ordered(self) -> Iterable[str]: # TODO should this clone? - return self.keys_list - - def values(self) -> ValuesView[ElementType]: - return self.container.values() + def __init__(self, top_frame: Optional[FrameType], depth: int) -> None: + frame = top_frame + assert frame is not None + for i in range(depth): + frame = frame.f_back + assert frame is not None + self._filename = frame.f_code.co_filename + self._line = frame.f_lineno - def __getitem__(self, item: str) -> ElementType: - return self.container[item] + @classmethod + def from_caller(cls) -> SourceLocator: + # returns the source locator for the calling frame of the caller + return cls(inspect.currentframe(), 2) - def __contains__(self, item: str) -> bool: - return item in self.container + @override + def __repr__(self) -> str: + return f"{self._filename}:{self._line}" - def name_of(self, elt: ElementType) -> Optional[str]: - if elt in self.names: - return self.names[elt] - else: - return None +class NamespaceManager: + """Manages names of objects in a naming container. + Supports source locators via inspect, before an object is given a name. + Supports delayed naming via callables, allowing names to depend on other names that may not yet be assigned. + """ -class SubElementManager: - def __init__(self) -> None: - self.dicts: List[Tuple[Union[Type[Any], Tuple[Type[Any], ...]], SubElementDict[Any]]] = [] - self.aliases = IdentityDict[Any, Any]() + def __init__(self, parent: Any) -> None: + self._parent = parent + self._base_names = IdentityDict[Any, Union[SourceLocator, str, Callable[[], str]]]() + self._default_names = IdentityDict[Any, str]() + self._default_name_prefix_count: Dict[str, int] = {} + self._assigned_names: Set[str] = set() + self._aliases = IdentityDict[Any, Any]() + + def __contains__(self, item: Any) -> bool: + if item in self._aliases: + return self._aliases[item] in self + + return item in self._base_names + + def resolve_names(self) -> None: + # asserts that all names are given and valid and resolves anonymous names + missing_names: List[str] = [] + error_names: List[Any] = [] + for obj, name in self._base_names.items(): + if isinstance(name, SourceLocator): + default_name = self._default_names.get(obj, None) + if default_name is not None: + self._base_names.update(obj, default_name) + if default_name in self._assigned_names: + raise ElementNamingError(type(self._parent), f"duplicate name {default_name}") + self._assigned_names.add(default_name) + else: + missing_names.append(f"{obj} @ {name}") + else: + pass # has valid name - def new_dict( - self, filter_type: Union[Type[ElementType], Tuple[Type[ElementType], ...]], anon_prefix: Optional[str] = None - ) -> SubElementDict[ElementType]: - sub_dict = SubElementDict[ElementType](anon_prefix) - self.dicts.append((filter_type, sub_dict)) - return sub_dict + if missing_names: + raise ElementNamingError(type(self._parent), f"no name for objects {missing_names}") + if error_names: # errors may be due to missing names, so check after missing names are resolved + raise ElementNamingError(type(self._parent), f"error generating names for objects {error_names}") def add_alias(self, src: Any, target: Any) -> None: - self.aliases[src] = target - - def add_element(self, name: str, item: Any) -> None: - if isinstance(item, ElementDict): - item._set_parent((name, self)) + self._aliases[src] = target + + def add_expected(self, item: Any, source_locator: SourceLocator, *, default_prefix: Optional[str] = None) -> None: + self._base_names[item] = source_locator + if default_prefix: + count = self._default_name_prefix_count.get(default_prefix, 0) + self._default_name_prefix_count[default_prefix] = count + 1 + self._default_names[item] = f"{default_prefix}{count}" + + def add_name(self, item: Any, name: Union[str, Callable[[], str]]) -> None: + existing = self._base_names.get(item, None) + if existing is None: + if isinstance(item, ElementDict): + item._set_parent(self) + self._base_names[item] = name + else: + return # not expected, ignore + elif isinstance(existing, SourceLocator): + if name in self._assigned_names: + raise ElementNamingError(type(self._parent), f"duplicate name {name}") + if isinstance(name, str): + self._assigned_names.add(name) + + self._base_names.update(item, name) + elif existing is not None: + return # reassignments, eg for aliasing for deprecation, ignored; first one takes else: - assigned = [] - for tpe, dict in self.dicts: - if item in dict.anons: - dict.add_element(name, item) - assigned.append(dict) - else: # require not conflicting name, or direct reassignment - assert name not in dict.container or dict.names.get(item, None) == name, f"duplicate name {name}" - assert len(assigned) <= 1, f"assigned {item} to multiple SubElementDict {assigned}" - - def name_of(self, item: Any) -> Optional[str]: - if item in self.aliases: - item = self.aliases[item] - name_candidates = [sub_dict.name_of(item) for sub_dict_type, sub_dict in self.dicts] - name_candidates_filtered = [name_candidate for name_candidate in name_candidates if name_candidate is not None] - assert len(name_candidates_filtered) <= 1, f"more than 1 name candidates {name_candidates} for {item}" - if not name_candidates_filtered: - return None + raise ValueError("unexpected name entry type") + + def name(self, item: Any, *, debug: bool = False) -> str: + """Returns the name of an item. + If debug, returns a debugging name if the final name is not resolved, otherwise raises.""" + if item in self._aliases: + return self.name(self._aliases[item], debug=debug) + + name = self._base_names.get(item, None) + if isinstance(name, str): + return name + elif isinstance(name, FunctionType): + name = name() + assert isinstance(name, str) + self._base_names.update(item, name) + if name in self._assigned_names: + raise ElementNamingError(type(self._parent), f"duplicate name {name}") + self._assigned_names.add(name) + return name + elif isinstance(name, SourceLocator): + if debug: + return f"(unnamed {item}@{name})" + else: + raise ValueError(f"no name for {item}") else: - return name_candidates_filtered[0] + raise ValueError(f"no name for unexpected {item} (internal consistency error)") class ElementDict(Generic[ElementType]): @@ -127,16 +141,14 @@ class ElementDict(Generic[ElementType]): # TODO also have a KeyType? Perhaps enforce type bounds, eg must be str-able? def __init__(self) -> None: self.container: Dict[Union[str, int], ElementType] = {} - self._parent: Optional[Tuple[str, SubElementManager]] = None # name prefix, pointer to top level + self._parent: Optional[NamespaceManager] = None # name prefix, pointer to top level - def _set_parent(self, parent: Tuple[str, SubElementManager]) -> None: + def _set_parent(self, parent: NamespaceManager) -> None: self._parent = parent def __setitem__(self, key: Union[str, int], value: ElementType) -> None: assert self._parent is not None - self._parent[1].add_element( - f"{self._parent[0]}[{key}]", value - ) # TODO perhaps some kind of check to make sure this is required? + self._parent.add_name(value, lambda: f"{self._parent.name(self)}[{key}]") # type: ignore[union-attr] self.container[key] = value def __getitem__(self, item: Union[str, int]) -> ElementType: @@ -204,42 +216,38 @@ class LibraryElement(Refable): def __repr__(self) -> str: return "%s@%02x" % (self._get_def_name(), (id(self) // 4) & 0xFF) - def __init__(self) -> None: - self._parent: Optional[LibraryElement] = None # set by binding, None means not bound + BaseType = TypeVar("BaseType", bound="LibraryElement") + + @classmethod + def _get_bases_of(cls, base_type: Type[BaseType]) -> Tuple[List[Type[BaseType]], List[Type[BaseType]]]: + """Returns all the base classes of this class, as a list of direct superclasses (including through non_library + elements) and a list of additional (indirect) superclasses. Direct superclasses are in MRO order, indirect + superclasses order is not defined (but MRO in current practice). - self.manager = SubElementManager() - self.manager_ignored: Set[str] = set(["_parent"]) + mypy currently does not allow passing in abstract types, so generally calls to this need type: ignore.""" + direct_bases: Set[Type[LibraryElement.BaseType]] = set() - @override - 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) - super().__setattr__(name, value) + def process_direct_base(bcls: Type[LibraryElement.BaseType]) -> None: + if not issubclass(bcls, base_type): + return # ignore above base_type + if (bcls, NonLibraryProperty) in bcls._elt_properties: # non-library, recurse into parents + for bcls_base in bcls.__bases__: + process_direct_base(bcls_base) + else: # anything else, directly append if not existing + direct_bases.add(bcls) - def _name_of_child(self, subelt: Any, context: Any, allow_unknown: bool = False) -> str: - self_name = self.manager.name_of(subelt) - if self_name is not None: - return self_name - else: - if allow_unknown: - return f"(unknown {subelt.__class__.__name__})" - else: - raise ValueError(f"no name for {subelt}") + for bcls in cls.__bases__: + process_direct_base(bcls) - def _path_from(self, base: LibraryElement, allow_unknown: bool = False) -> List[str]: - if base is self: - return [] - else: - assert self._parent is not None, "can't get path / name for non-bound element" - return self._parent._path_from(base, allow_unknown) + [ - self._parent._name_of_child(self, base, allow_unknown) - ] + ordered_direct_bases: List[Type[LibraryElement.BaseType]] = [] + ordered_indirect_bases: List[Type[LibraryElement.BaseType]] = [] + for mro_base in cls.__mro__[1:]: # ignore self + if mro_base in direct_bases: + ordered_direct_bases.append(mro_base) + elif issubclass(mro_base, base_type) and (mro_base, NonLibraryProperty) not in mro_base._elt_properties: + ordered_indirect_bases.append(mro_base) - def _name_from(self, base: LibraryElement, allow_unknown: bool = False) -> str: - """Returns the path name to (inclusive) this element from some starting point. - allow_unknown allows elements that haven't been assigned a name yet to not crash, - this is useful when called from an error so the _name_from error doesn't stomp the real error.""" - return ".".join(self._path_from(base, allow_unknown)) + return ordered_direct_bases, ordered_indirect_bases @classmethod def _static_def_name(cls) -> str: @@ -255,67 +263,69 @@ def _get_def_name(self) -> str: def _def_to_proto(self) -> Union[edgir.PortTypes, edgir.BlockLikeTypes]: ... +class NamingContainer: + """Mixin class for objects that contain and name internal objects""" + + def __init__(self) -> None: + self._parent: Optional[NamingContainer] = None # set by binding, None means not bound + self._namer = NamespaceManager(self) + + @override + def __setattr__(self, name: str, value: Any) -> None: + if hasattr(self, "_namer"): # don't access before definition + self._namer.add_name(value, name) + super().__setattr__(name, value) + + def _path_of(self, elt: Any, *, debug: bool = False) -> List[str]: + """Returns the name path of an inner element. + If debug is true, will return the best-effort name even if it is not yet assigned. + If debug is false, raises an error for unassigned names.""" + if elt is self: + return [] + else: + assert elt._parent is not None, "can't get path / name for non-bound element" + return self._path_of(elt._parent, debug=debug) + [elt._parent._name_of(elt, debug=debug)] + + def _path_name_of(self, elt: Any, *, debug: bool = False) -> str: + """Convenience stringify wrapper around _path_of.""" + return ".".join(self._path_of(elt, debug=debug)) + + def _name_of(self, elt: Any, *, debug: bool = False) -> str: + """Returns the name of a contained element, non-recursively.""" + if elt not in self._namer: + if not debug: + raise ValueError(f"{self.__class__.__name__} has no {elt}") + else: + return f"(unknown {elt.__class__.__name__})" + else: + return self._namer.name(elt, debug=debug) + + @non_library -class HasMetadata(LibraryElement): +class HasMetadata(LibraryElement, NamingContainer): """A library element with the metadata dict-like field""" def __init__(self) -> None: super().__init__() - self._metadata: SubElementDict[Any] = self.manager.new_dict(Any) + self._metadata: List[Any] = [] - MetadataType = TypeVar( - "MetadataType", bound=Union[str, Mapping[str, Any], SubElementDict[Any], IdentityDict[Any, Any]] - ) + MetadataType = TypeVar("MetadataType", bound=Union[str, Mapping[str, Any], IdentityDict[Any, Any]]) def Metadata(self, value: MetadataType) -> MetadataType: """Adds a metadata field to this object. Reference to the value must not change, and reassignment will error. Value may be changed until proto generation. Utility method for library builders.""" - self._metadata.register(value) + self._metadata.append(value) + self._namer.add_expected(value, SourceLocator.from_caller()) return value - BaseType = TypeVar("BaseType", bound="HasMetadata") - - @classmethod - def _get_bases_of(cls, base_type: Type[BaseType]) -> Tuple[List[Type[BaseType]], List[Type[BaseType]]]: - """Returns all the base classes of this class, as a list of direct superclasses (including through non_library - elements) and a list of additional (indirect) superclasses. Direct superclasses are in MRO order, indirect - superclasses order is not defined (but MRO in current practice). - - mypy currently does not allow passing in abstract types, so generally calls to this need type: ignore.""" - direct_bases: Set[Type[HasMetadata.BaseType]] = set() - - def process_direct_base(bcls: Type[HasMetadata.BaseType]) -> None: - if not issubclass(bcls, base_type): - return # ignore above base_type - if (bcls, NonLibraryProperty) in bcls._elt_properties: # non-library, recurse into parents - for bcls_base in bcls.__bases__: - process_direct_base(bcls_base) - else: # anything else, directly append if not existing - direct_bases.add(bcls) - - for bcls in cls.__bases__: - process_direct_base(bcls) - - ordered_direct_bases: List[Type[HasMetadata.BaseType]] = [] - ordered_indirect_bases: List[Type[HasMetadata.BaseType]] = [] - for mro_base in cls.__mro__[1:]: # ignore self - if mro_base in direct_bases: - ordered_direct_bases.append(mro_base) - elif issubclass(mro_base, base_type) and (mro_base, NonLibraryProperty) not in mro_base._elt_properties: - ordered_indirect_bases.append(mro_base) - - return ordered_direct_bases, ordered_indirect_bases - def _populate_metadata(self, pb: edgir.Metadata, src: Any, ref_map: Refable.RefMapType) -> None: """Generate metadata from a given object.""" if isinstance(src, str): pb.text_leaf = src elif isinstance(src, bytes): pb.bin_leaf = src - elif isinstance(src, dict) or isinstance(src, SubElementDict) or isinstance(src, IdentityDict): - if isinstance(src, SubElementDict): # used at the top-level, for Metadata(...) - src.finalize() # TODO should this be here? + elif isinstance(src, dict) or isinstance(src, IdentityDict): for key, val in src.items(): assert isinstance(key, str), f"must overload _metadata_to_proto for non-str dict key {key}" self._populate_metadata(pb.members.node[key], val, ref_map) diff --git a/edg/core/DesignTop.py b/edg/core/DesignTop.py index 70de33375..fdb3b22f5 100644 --- a/edg/core/DesignTop.py +++ b/edg/core/DesignTop.py @@ -101,7 +101,7 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: else: raise TypeError assert isinstance(multipack_block, MultipackBlock) - multipack_name = self._name_of_child(multipack_block, self) + multipack_name = self._name_of(multipack_block) multipack_ref_base = edgir.LocalPath() multipack_ref_base.steps.add().name = multipack_name multipack_ref_map = multipack_block._create_ref_map(multipack_ref_base) @@ -113,9 +113,9 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: packed_ref_map = multipack_part_block._create_ref_map(packed_ref_base) if isinstance(multipack_part, Block): - part_name = multipack_block._name_of_child(multipack_part, self) + part_name = multipack_block._name_of(multipack_part) elif isinstance(multipack_part, PackedBlockAllocate): - part_name = multipack_block._name_of_child(multipack_part.parent, self) + part_name = multipack_block._name_of(multipack_part.parent) assert multipack_part.suggested_name, "multipack parts must have suggested name, for consistency" part_name += f"[{multipack_part.suggested_name}]" else: @@ -128,7 +128,7 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: packed_port_port = packed_port.port else: raise TypeError - packed_port_name = multipack_part_block._name_of_child(packed_port_port, self) + packed_port_name = multipack_part_block._name_of(packed_port_port) exported_tunnel = edgir.add_pair( pb.constraints, f"(packed){multipack_name}.{part_name}.{packed_port_name}" ).exportedTunnel @@ -141,14 +141,14 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: for multipack_param, packed_param in packing_rule.tunnel_assigns.items(): if isinstance(packed_param, ConstraintExpr): - packed_param_name = multipack_part_block._name_of_child(packed_param, self) + packed_param_name = multipack_part_block._name_of(packed_param) assign_tunnel = edgir.add_pair( pb.constraints, f"(packed){multipack_name}.{part_name}.{packed_param_name}" ).assignTunnel assign_tunnel.dst.CopyFrom(multipack_ref_map[multipack_param]) assign_tunnel.src.ref.CopyFrom(packed_ref_map[packed_param]) elif isinstance(packed_param, PackedBlockParamArray): - multipack_param_name = multipack_block._name_of_child(multipack_param, self) + multipack_param_name = multipack_block._name_of(multipack_param) constr_name = f"(packed){multipack_name}.{multipack_param_name}" packed_params.setdefault(constr_name, (multipack_ref_map[multipack_param], []))[1].append( packed_ref_map[packed_param.param] @@ -158,7 +158,7 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: for multipack_param, unpacked_param in packing_rule.tunnel_unpack_assigns.items(): if isinstance(unpacked_param, ConstraintExpr): - multipack_param_name = multipack_block._name_of_child(multipack_param, self) + multipack_param_name = multipack_block._name_of(multipack_param) # TODO need better constraint naming scheme assign_tunnel = edgir.add_pair( pb.constraints, f"(unpacked){multipack_name}.{part_name}.{multipack_param_name}" @@ -166,7 +166,7 @@ def _populate_def_proto_block_contents(self, pb: edgir.BlockLikeTypes, ref_map: assign_tunnel.dst.CopyFrom(packed_ref_map[unpacked_param]) assign_tunnel.src.ref.CopyFrom(multipack_ref_map[multipack_param]) elif isinstance(unpacked_param, PackedBlockParam): - multipack_param_name = multipack_block._name_of_child(multipack_param, self) + multipack_param_name = multipack_block._name_of(multipack_param) # TODO need better constraint naming scheme assign_tunnel = edgir.add_pair( pb.constraints, f"(unpacked){multipack_name}.{part_name}.{multipack_param_name}" @@ -206,5 +206,5 @@ def pack(self, multipack_part: Union[Block, PackedBlockAllocate], path: DesignPa else: raise TypeError assert isinstance(multipack_block, MultipackBlock), "block must be a part of a MultipackBlock" - assert self._blocks.name_of(multipack_block), "containing MultipackBlock must be a PackedBlock" + assert multipack_block._parent is self, "block must be in self" self._packed_blocks[multipack_part] = path diff --git a/edg/core/Generator.py b/edg/core/Generator.py index b1a0d6bf2..19e3fdde2 100644 --- a/edg/core/Generator.py +++ b/edg/core/Generator.py @@ -94,6 +94,8 @@ def generate(self) -> None: # @override def _def_to_proto(self) -> edgir.HierarchyBlock: + self._namer.resolve_names() + if ( self._elaboration_state != BlockElaborationState.post_generate ): # only write generator on the stub definition @@ -146,11 +148,11 @@ def _generated_def_to_proto( self.generate() except Exception as e: params_list_str = ", ".join( - f"{self._name_of_child(param, self, allow_unknown=True)}={self._generator_param_values[param]!r}" + f"{self._name_of(param, debug=True)}={self._generator_param_values[param]!r}" for param in self._generator_params_list ) raise RuntimeError( - f"Generator {self.__class__.__name__} with parameters {{{params_list_str}}} raised exception: {e!r}" + f"Generator {self.__class__.__name__} with parameters {params_list_str} raised exception: {e!r}" ) from e elif self._generator is not None: # legacy generator style fn_args = [ diff --git a/edg/core/HdlUserExceptions.py b/edg/core/HdlUserExceptions.py index b8ca0983e..c7bc7285f 100644 --- a/edg/core/HdlUserExceptions.py +++ b/edg/core/HdlUserExceptions.py @@ -57,3 +57,7 @@ def __init__(self, block_type: Type["BaseBlock"], exc: str, resolution: str = "" class ChainError(BlockDefinitionError): """Base error for bad elements in a chain connect""" + + +class ElementNamingError(BlockDefinitionError): + """Error for when an element has an invalid (missing, duplicated) name""" diff --git a/edg/core/HierarchyBlock.py b/edg/core/HierarchyBlock.py index 8de595f02..bdd8df545 100644 --- a/edg/core/HierarchyBlock.py +++ b/edg/core/HierarchyBlock.py @@ -27,7 +27,7 @@ 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 +from .Core import Refable, non_library, SourceLocator from .HdlUserExceptions import * from .IdentityDict import IdentityDict from .IdentitySet import IdentitySet @@ -335,19 +335,20 @@ def __init__(self) -> None: if hasattr(self, "_init_params"): # used to propagate params generated in the metaclass __init__ hook self._init_params: Dict[str, ConstraintExpr] for param_name, param in self._init_params.items(): - self._parameters.register(param) - self.manager.add_element(param_name, param) + self._parameters.append(param) + self._namer.add_expected(param, SourceLocator.from_caller()) + self._namer.add_name(param, param_name) delattr(self, "_init_params") self._mixins: List["BlockInterfaceMixin"] = [] - self._blocks = self.manager.new_dict(Block) - self._chains = self.manager.new_dict(ChainConnect, anon_prefix="anon_chain") + self._blocks: List[Block] = [] + self._chains: List[ChainConnect] = [] self._port_tags = IdentityDict[BasePort, Set[PortTag]]() def _get_ports_by_tag(self, tags: Set[PortTag]) -> List[BasePort]: out = [] - for block_port_name, block_port in self._ports.items(): + for block_port in self._ports: assert isinstance(block_port, BasePort) port_tags: Set[PortTag] = self._port_tags.get(block_port, set()) if port_tags.issuperset(tags): @@ -362,9 +363,8 @@ def _build_ref_map( for mixin in self._mixins: mixin._build_ref_map(ref_map, prefix) if not interface_only: - for name, block in self._blocks.items(): - assert isinstance(block, Block) - block._build_ref_map(ref_map, edgir.localpath_concat(prefix, name), interface_only=True) + for block in self._blocks: + block._build_ref_map(ref_map, edgir.localpath_concat(prefix, self._name_of(block)), interface_only=True) @override def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: @@ -372,19 +372,15 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: assert isinstance(pb, edgir.HierarchyBlock) # generate param defaults - for param_name, param in self._parameters.items(): + for param in self._parameters: if isinstance(param.binding, InitParamBinding) and param.binding.value is not None: # default values can't depend on anything so the ref_map is empty param_typed_value = param._to_expr_type(param.binding.value) - param_typed_value._populate_expr_proto(pb.param_defaults[param_name], IdentityDict()) + param_typed_value._populate_expr_proto(pb.param_defaults[self._name_of(param)], IdentityDict()) def _populate_def_proto_hierarchy(self, pb: edgir.HierarchyBlock, ref_map: Refable.RefMapType) -> None: - self._blocks.finalize() - self._connects.finalize() - self._chains.finalize() - - for name, block in self._blocks.items(): - new_block_lib = edgir.add_pair(pb.blocks, name).lib_elem + for block in self._blocks: + new_block_lib = edgir.add_pair(pb.blocks, self._name_of(block)).lib_elem new_block_lib.base.target.name = block._get_def_name() for mixin in block._mixins: new_block_lib.mixins.add().target.name = mixin._get_def_name() @@ -392,37 +388,42 @@ def _populate_def_proto_hierarchy(self, pb: edgir.HierarchyBlock, ref_map: Refab # actually generate the links and connects link_chain_names = IdentityDict[Connection, List[str]]() # prefer chain name where applicable # TODO generate into primary data structures - for name, chain in self._chains.items_ordered(): # TODO work with net join + for i, chain in enumerate(self._chains): # TODO work with net join + name = self._namer.name(chain) + if name is None: + name = f"anon_chain_{i}" for i, connect in enumerate(chain.links): link_chain_names.setdefault(connect, []).append(f"{name}_{i}") delegated_connects = self._all_delegated_connects() - for name, connect in self._connects.items_ordered(): + for i, connect in enumerate(self._connects): if connect in delegated_connects: continue - connect_names_opt = [self._connects.name_of(c) for c in self._all_connects_of(connect)] + + connect_elts = connect.make_connection() + + connect_names_opt = [self._namer.name(c) for c in self._all_connects_of(connect)] connect_names = [c for c in connect_names_opt if c is not None and not c.startswith("anon_")] if len(connect_names) > 1: - raise UnconnectableError(f"Multiple names {connect_names} for connect") + raise UnconnectableError(f"Multiple assigned names {connect_names} for connect") elif len(connect_names) == 1: name = connect_names[0] - - connect_elts = connect.make_connection() - - if name.startswith("anon_"): # infer a non-anon name if possible + else: # infer a better name if possible if connect in link_chain_names and not link_chain_names[connect][0].startswith("anon_"): name = link_chain_names[connect][0] # arbitrarily pick the first chain name elif isinstance(connect_elts, Connection.Export): - port_pathname = connect_elts.external_port._name_from(self).replace(".", "_") + port_pathname = "_".join(self._path_of(connect_elts.external_port)) name = f"_{port_pathname}_link" elif isinstance(connect_elts, Connection.ConnectedLink) and connect_elts.bridged_connects: - port_pathname = connect_elts.bridged_connects[0][0]._name_from(self).replace(".", "_") + port_pathname = "_".join(self._path_of(connect_elts.bridged_connects[0][0])) name = f"_{port_pathname}_link" elif isinstance(connect_elts, Connection.ConnectedLink) and connect_elts.link_connects: - port_pathname = connect_elts.link_connects[0][0]._name_from(self).replace(".", "_") + port_pathname = "_".join(self._path_of(connect_elts.link_connects[0][0])) name = f"_{port_pathname}_link" elif connect in link_chain_names: # if there's really nothing better, anon_chain is better than nothing name = link_chain_names[connect][0] + else: + name = connect_names_opt[0] if connect_elts is None: # single port net - effectively discard pass @@ -448,7 +449,7 @@ def _populate_def_proto_hierarchy(self, pb: edgir.HierarchyBlock, ref_map: Refab assert not connect_elts.is_link_array, "bridged arrays not supported" assert self_port.bridge_type is not None - port_name = self_port._name_from(self) + port_name = self._path_name_of(self_port) edgir.add_pair(pb.blocks, f"(bridge){port_name}").lib_elem.base.target.name = ( self_port.bridge_type._static_def_name() ) @@ -484,15 +485,15 @@ def _populate_def_proto_hierarchy(self, pb: edgir.HierarchyBlock, ref_map: Refab raise ValueError("unknown connect type") # generate block initializers - for block_name, block in self._blocks.items(): - all_block_params = dict(block._parameters.items()) + for block in self._blocks: + all_block_params = {block._name_of(param): param for param in block._parameters} for mixin in block._mixins: - all_block_params.update(mixin._parameters.items()) + all_block_params.update({mixin._name_of(param): param for param in mixin._parameters}) for block_param_name, block_param in all_block_params.items(): if isinstance(block_param.binding, InitParamBinding) and block_param.binding.value is not None: param_typed_value = block_param._to_expr_type(block_param.binding.value) AssignBinding.populate_assign_proto( - edgir.add_pair(pb.constraints, f"(init){block_name}.{block_param_name}"), + edgir.add_pair(pb.constraints, f"(init){self._name_of(block)}.{block_param_name}"), block_param, param_typed_value, ref_map, @@ -501,6 +502,8 @@ def _populate_def_proto_hierarchy(self, pb: edgir.HierarchyBlock, ref_map: Refab # TODO make this non-overriding? @override def _def_to_proto(self) -> edgir.HierarchyBlock: + self._namer.resolve_names() + assert not self._mixins # blocks with mixins can only be instantiated anonymously ref_map = self._create_ref_map() @@ -510,8 +513,6 @@ def _def_to_proto(self) -> edgir.HierarchyBlock: self._populate_def_proto_block_base(pb) self._populate_def_proto_port_init(pb, ref_map) - self._populate_def_proto_param_init(pb, ref_map) - self._populate_def_proto_hierarchy(pb, ref_map) self._populate_def_proto_block_contents(pb, ref_map) self._populate_def_proto_description(pb, ref_map) @@ -544,18 +545,18 @@ def with_mixin(self, tpe: MixinType) -> MixinType: assert self._parent is not None elt = tpe._bind(self._parent) - self._parent.manager.add_alias(elt, self) + self._parent._namer.add_alias(elt, self) self._mixins.append(elt) return elt def chain(self, *elts: Union[Connection, BasePort, Block]) -> Any: - if not elts: - return self._chains.register(ChainConnect([], [])) chain_blocks = [] chain_links = [] - if isinstance(elts[0], (BasePort, Connection)): + if not elts: + current_port: Optional[Union[BasePort, Connection]] = None + elif isinstance(elts[0], (BasePort, Connection)): current_port = elts[0] elif isinstance(elts[0], Block): outable_ports = elts[0]._get_ports_by_tag({Output}) + elts[0]._get_ports_by_tag({InOut}) @@ -570,6 +571,8 @@ def chain(self, *elts: Union[Connection, BasePort, Block]) -> Any: raise EdgTypeError(f"first element 0 to chain", elts[0], (BasePort, Connection, Block)) for i, elt in list(enumerate(elts))[1:-1]: + assert current_port is not None + if not isinstance(elt, Block): raise EdgTypeError(f"middle arguments elts[{i}] in chain", elt, Block) if elt._get_ports_by_tag({Input}) and elt._get_ports_by_tag({Output}): @@ -600,9 +603,13 @@ def chain(self, *elts: Union[Connection, BasePort, Block]) -> Any: type(self), f"element {i} to chain {type(elt)} has no Input and Output, or InOut ports" ) - if isinstance(elts[-1], (BasePort, Connection)): + if not elts: + pass + elif isinstance(elts[-1], (BasePort, Connection)): + assert current_port is not None chain_links.append(self.connect(current_port, elts[-1])) elif isinstance(elts[-1], Block): + assert current_port is not None inable_ports = elts[-1]._get_ports_by_tag({Input}) + elts[-1]._get_ports_by_tag({InOut}) if len(inable_ports) != 1: raise BlockDefinitionError( @@ -614,7 +621,10 @@ def chain(self, *elts: Union[Connection, BasePort, Block]) -> Any: else: raise EdgTypeError(f"last argument {len(elts) - 1} to chain", elts[-1], (BasePort, Connection, Block)) - return self._chains.register(ChainConnect(chain_blocks, chain_links)) + chain = ChainConnect(chain_blocks, chain_links) + self._chains.append(chain) + self._namer.add_expected(chain, SourceLocator.from_caller(), default_prefix="anon_chain") + return chain def implicit_connect(self, *implicits: ImplicitConnect) -> ImplicitScope: for implicit in implicits: @@ -634,22 +644,31 @@ def connect(self, *connects: Union[BasePort, Connection], flatten: bool = False) # type ignore is needed because IntLike overlaps BoolLike @overload def ArgParameter(self, param: BoolLike, *, doc: Optional[str] = None) -> BoolExpr: ... # type: ignore + @overload def ArgParameter(self, param: IntLike, *, doc: Optional[str] = None) -> IntExpr: ... # type: ignore + @overload def ArgParameter(self, param: FloatLike, *, doc: Optional[str] = None) -> FloatExpr: ... + @overload def ArgParameter(self, param: RangeLike, *, doc: Optional[str] = None) -> RangeExpr: ... # type: ignore + @overload def ArgParameter(self, param: StringLike, *, doc: Optional[str] = None) -> StringExpr: ... # type: ignore + @overload def ArgParameter(self, param: ArrayBoolLike, *, doc: Optional[str] = None) -> ArrayBoolExpr: ... # type: ignore + @overload def ArgParameter(self, param: ArrayIntLike, *, doc: Optional[str] = None) -> ArrayIntExpr: ... # type: ignore + @overload def ArgParameter(self, param: ArrayFloatLike, *, doc: Optional[str] = None) -> ArrayFloatExpr: ... + @overload def ArgParameter(self, param: ArrayRangeLike, *, doc: Optional[str] = None) -> ArrayRangeExpr: ... + @overload def ArgParameter(self, param: ArrayStringLike, *, doc: Optional[str] = None) -> ArrayStringExpr: ... @@ -742,7 +761,8 @@ def Block(self, tpe: BlockType) -> BlockType: raise TypeError(f"param to Block(...) may not be DesignTop") elt = tpe._bind(self) - self._blocks.register(elt) + self._blocks.append(elt) + self._namer.add_expected(elt, SourceLocator.from_caller()) return elt diff --git a/edg/core/Link.py b/edg/core/Link.py index 270e8f692..fc01e621f 100644 --- a/edg/core/Link.py +++ b/edg/core/Link.py @@ -42,27 +42,28 @@ def __init__(self) -> None: @override def _def_to_proto(self) -> edgir.Link: + self._namer.resolve_names() ref_map = self._create_ref_map() pb = edgir.Link() self._populate_def_proto_block_base(pb) - self._populate_def_proto_param_init(pb, ref_map) self._populate_def_proto_block_contents(pb, ref_map) self._populate_def_proto_description(pb, ref_map) # specifically ignore the port initializers # actually generate the links and connects - self._connects.finalize() delegated_connects = self._all_delegated_connects() - for name, connect in self._connects.items_ordered(): + for i, connect in enumerate(self._connects): if connect in delegated_connects: continue - connect_names_opt = [self._connects.name_of(c) for c in self._all_connects_of(connect)] + connect_names_opt = [self._namer.name(c) for c in self._all_connects_of(connect)] connect_names = [c for c in connect_names_opt if c is not None and not c.startswith("anon_")] if len(connect_names) > 1: raise UnconnectableError(f"Multiple names {connect_names} for connect") elif len(connect_names) == 1: name = connect_names[0] + else: + name = connect_names_opt[0] connect_elts = connect.make_connection() assert isinstance(connect_elts, Connection.ConnectedLink) @@ -80,7 +81,7 @@ def _def_to_proto(self) -> edgir.Link: ref_map[self_port.base] ) constraint_pb.exportedArray.exterior_port.map_extract.path.steps.add().name = ( - self_port.target._name_from(self_port.base._get_elt_sample()) + self_port.base._get_elt_sample()._name_of(self_port.target) ) constraint_pb.exportedArray.internal_block_port.ref.CopyFrom( edgir.localpath_concat(link_path, link_port_path) diff --git a/edg/core/MultipackBlock.py b/edg/core/MultipackBlock.py index 0b615504b..e0b64c466 100644 --- a/edg/core/MultipackBlock.py +++ b/edg/core/MultipackBlock.py @@ -10,7 +10,7 @@ from .Blocks import BlockElaborationState from .HdlUserExceptions import BlockDefinitionError, EdgTypeError from .IdentityDict import IdentityDict -from .Core import non_library, SubElementDict +from .Core import non_library, SourceLocator from .ConstraintExpr import ConstraintExpr, BoolExpr, IntExpr, FloatExpr, RangeExpr, StringExpr from .Ports import BasePort, Port from .HierarchyBlock import Block, BlockPrototype @@ -132,7 +132,7 @@ class MultipackBlock(Block): def __init__(self) -> None: super().__init__() - self._packed_blocks: SubElementDict[PackedBlockTypes] = self.manager.new_dict((Block, PackedBlockArray)) + # TODO should these be defined in terms of Refs? # packed block -> (exterior port -> packed block port) self._packed_connects_by_packed_block = IdentityDict[ @@ -163,7 +163,7 @@ def PackedPart(self, tpe: PackedPartType) -> PackedPartType: raise BlockDefinitionError(type(self), "can only define multipack in init") elt = tpe._bind(self) # TODO: does this actually need to be bound? - self._packed_blocks.register(elt) + self._namer.add_expected(elt, SourceLocator.from_caller()) self._packed_connects_by_packed_block[elt] = IdentityDict[BasePort, PackedPortTypes]() self._packed_assigns_by_packed_block[elt] = IdentityDict[ConstraintExpr, PackedParamTypes]() self._unpacked_assigns_by_packed_block[elt] = IdentityDict[ConstraintExpr, UnpackedParamTypes]() @@ -252,7 +252,8 @@ def PackedParameter(self, packed_param: PackedParamTypes) -> ConstraintExpr: else: raise TypeError() self.packed_assign(new_param, packed_param) - self._parameters.register(new_param) + self._parameters.append(new_param) + self._namer.add_expected(new_param, SourceLocator.from_caller()) return new_param def unpacked_assign(self, packed_param: UnpackedParamTypes, self_param: ConstraintExpr) -> None: @@ -274,7 +275,6 @@ def unpacked_assign(self, packed_param: UnpackedParamTypes, self_param: Constrai def _get_block_packing_rule(self, packed_part: Union[Block, PackedBlockAllocate]) -> MultipackPackingRule: """Internal API, returns the packing rules (tunnel exports and assigns) for a constituent PackedPart.""" - self._packed_blocks.finalize() if isinstance(packed_part, PackedBlockAllocate): packed_block: PackedBlockTypes = packed_part.parent else: diff --git a/edg/core/Ports.py b/edg/core/Ports.py index 0fba17ecd..2ca9d25d3 100644 --- a/edg/core/Ports.py +++ b/edg/core/Ports.py @@ -10,7 +10,7 @@ from .Binding import ParamBinding, IsConnectedBinding, NameBinding from .Builder import builder from .ConstraintExpr import ConstraintExpr, BoolExpr, StringExpr -from .Core import Refable, HasMetadata, SubElementDict, non_library +from .Core import Refable, HasMetadata, non_library, NamingContainer, SourceLocator from .HdlUserExceptions import * from .IdentityDict import IdentityDict from .. import edgir @@ -36,7 +36,7 @@ def __call__(cls, *args: Any, **kwargs: Any) -> Any: @non_library -class BasePort(HasMetadata, metaclass=InitializerContextMeta): +class BasePort(HasMetadata, NamingContainer, metaclass=InitializerContextMeta): SelfType = TypeVar("SelfType", bound="BasePort") def __init__(self) -> None: @@ -149,34 +149,29 @@ def __init__(self) -> None: self._bridge_instance: Optional[PortBridge] = None # internal only self._adapter_count: int = 0 - self._parameters: SubElementDict[ConstraintExpr] = self.manager.new_dict(ConstraintExpr) - self._ports: SubElementDict[Port] = self.manager.new_dict(Port) + self._parameters: List[ConstraintExpr] = [] + self._ports: List[Port] = [] - self.manager_ignored.update(["_is_connected", "_name"]) self._is_connected = BoolExpr()._bind(IsConnectedBinding(self)) self._name = StringExpr()._bind(NameBinding(self)) def _clear_initializers(self) -> None: - self._parameters.finalize() - for name, param in self._parameters.items(): + for param in self._parameters: param.initializer = None - self._ports.finalize() - for name, port in self._ports.items(): + for port in self._ports: port._clear_initializers() @override def _cloned_from(self: SelfType, other: SelfType) -> None: super()._cloned_from(other) - self._parameters.finalize() - for name, param in self._parameters.items(): - other_param = other._parameters[name] - assert isinstance(other_param, type(param)) - param.initializer = other_param.initializer - self._ports.finalize() - for name, port in self._ports.items(): - other_port = other._ports[name] - assert isinstance(other_port, type(port)) - port._cloned_from(other_port) + assert len(other._parameters) == len(self._parameters) + for self_param, other_param in zip(self._parameters, other._parameters): + assert type(other_param) is type(self_param) + self_param.initializer = other_param.initializer + assert len(other._ports) == len(self._ports) + for self_port, other_port in zip(self._ports, other._ports): + assert type(other_port) is type(self_port) + self_port._cloned_from(other_port) def init_from(self: SelfType, other: SelfType) -> None: assert self._parent is not None, "may only init_from on an bound port" @@ -214,8 +209,8 @@ def _convert(self, adapter: PortAdapter[ConvertTargetType]) -> ConvertTargetType adapter_inst = enclosing_block.Block(adapter) adapter_name_suffix = f"_{self._adapter_count}" if self._adapter_count > 0 else "" - enclosing_block.manager.add_element( - f"(adapter){self._name_from(enclosing_block)}{adapter_name_suffix}", adapter_inst + enclosing_block._namer.add_name( + adapter_inst, lambda: f"(adapter){enclosing_block._path_name_of(self)}{adapter_name_suffix}" ) enclosing_block.connect(self, adapter_inst.src) # we don't name it to avoid explicit name conflicts self._adapter_count += 1 @@ -227,8 +222,7 @@ def _populate_portlike_proto(self, pb: edgir.PortLike) -> None: @override def _def_to_proto(self) -> edgir.Port: - self._parameters.finalize() - self._ports.finalize() + self._namer.resolve_names() pb = edgir.Port() @@ -240,13 +234,13 @@ def _def_to_proto(self) -> edgir.Port: for cls in indirect_bases: pb.super_superclasses.add().target.name = cls._static_def_name() - for name, param in self._parameters.items(): - param._populate_decl_proto(edgir.add_pair(pb.params, name)) - for name, port in self._ports.items(): - port._populate_portlike_proto(edgir.add_pair(pb.ports, name)) - - self._populate_metadata(pb.meta, self._metadata, IdentityDict()) # TODO use ref map + for param in self._parameters: + param._populate_decl_proto(edgir.add_pair(pb.params, self._name_of(param))) + for port in self._ports: + port._populate_portlike_proto(edgir.add_pair(pb.ports, self._name_of(port))) + named_metadata = {self._name_of(metadata): metadata for metadata in self._metadata} + self._populate_metadata(pb.meta, named_metadata, IdentityDict()) return pb @override @@ -258,24 +252,22 @@ def _build_ref_map(self, ref_map: Refable.RefMapType, prefix: edgir.LocalPath) - super()._build_ref_map(ref_map, prefix) ref_map[self.is_connected()] = edgir.localpath_concat(prefix, edgir.IS_CONNECTED) ref_map[self.name()] = edgir.localpath_concat(prefix, edgir.NAME) - for name, param in self._parameters.items(): - param._build_ref_map(ref_map, edgir.localpath_concat(prefix, name)) - for name, port in self._ports.items(): - port._build_ref_map(ref_map, edgir.localpath_concat(prefix, name)) + for param in self._parameters: + param._build_ref_map(ref_map, edgir.localpath_concat(prefix, self._name_of(param))) + for port in self._ports: + port._build_ref_map(ref_map, edgir.localpath_concat(prefix, self._name_of(port))) if self._link_instance is not None: self._link_instance._build_ref_map(ref_map, edgir.localpath_concat(prefix, edgir.CONNECTED_LINK)) @override def _get_initializers(self, path_prefix: List[str]) -> List[Tuple[ConstraintExpr, List[str], ConstraintExpr]]: - self._parameters.finalize() initializers: List[Tuple[ConstraintExpr, List[str], ConstraintExpr]] = [ - (param, path_prefix + [name], param.initializer) - for (name, param) in self._parameters.items() + (param, path_prefix + [self._name_of(param)], param.initializer) + for param in self._parameters if param.initializer is not None ] - self._ports.finalize() - for name, port in self._ports.items(): - initializers.extend(port._get_initializers(path_prefix + [name])) + for port in self._ports: + initializers.extend(port._get_initializers(path_prefix + [self._name_of(port)])) return initializers def _with_elt_initializers(self: SelfType, replace_elts: dict[str, "Port"]) -> SelfType: @@ -285,7 +277,7 @@ def _with_elt_initializers(self: SelfType, replace_elts: dict[str, "Port"]) -> S cloned = self._clone() for name, replace_port in replace_elts.items(): assert replace_port._parent is None, "replace_elts must not be bound" - cloned_port = cloned._ports[name] + cloned_port = getattr(cloned, name) assert isinstance(replace_port, type(cloned_port)) assert not cloned_port._get_initializers([]), f"replace_elts sub-port {name} was not empty" cloned_port._cloned_from(replace_port) @@ -313,7 +305,8 @@ def link(self) -> PortLinkType: def Parameter(self, tpe: U) -> U: """Registers a parameter for this Port""" elt = tpe._bind(ParamBinding(self)) - self._parameters.register(elt) + self._parameters.append(elt) + self._namer.add_expected(elt, SourceLocator.from_caller()) return elt T = TypeVar("T", bound="Port") @@ -324,7 +317,8 @@ def Port(self, tpe: T, *, desc: Optional[str] = None) -> T: raise EdgTypeError(f"param to Port(...)", tpe, Port) elt = tpe._bind(self) - self._ports.register(elt) + self._ports.append(elt) + self._namer.add_expected(elt, SourceLocator.from_caller()) return elt diff --git a/edg/core/Util.py b/edg/core/Util.py index 9cff2f148..df81f86ec 100644 --- a/edg/core/Util.py +++ b/edg/core/Util.py @@ -12,8 +12,8 @@ def edg_to_dict(obj: Union[BasePort, BaseBlock]) -> Dict[str, Any]: if isinstance(obj, Port): result = { "_": edg_obj_name(obj), - "params": obj._parameters, - "fields": {k: edg_to_dict(v) for k, v in obj._ports.items()}, + "params": {obj._name_of(param): param for param in obj._parameters}, + "fields": {obj._name_of(port): edg_to_dict(port) for port in obj._ports}, } if obj._link_instance is not None: result["link"] = edg_to_dict(obj._link_instance) @@ -21,8 +21,8 @@ def edg_to_dict(obj: Union[BasePort, BaseBlock]) -> Dict[str, Any]: elif isinstance(obj, Block) or isinstance(obj, Link): return { "_": edg_obj_name(obj), - "params": obj._parameters, - "ports": {k: edg_to_dict(v) for k, v in obj._ports.items()}, + "params": {obj._name_of(param): param for param in obj._parameters}, + "ports": {obj._name_of(port): edg_to_dict(port) for port in obj._ports}, } else: raise ValueError diff --git a/edg/core/__init__.py b/edg/core/__init__.py index 8857e8a26..83582ea12 100644 --- a/edg/core/__init__.py +++ b/edg/core/__init__.py @@ -21,7 +21,7 @@ from .MultiBiDict import MultiBiDict # Features for library builders -from .Core import LibraryElement, SubElementDict, ElementDict, non_library +from .Core import LibraryElement, ElementDict, non_library from .Blocks import BasePort, BaseBlock from .Categories import InternalBlock from .Builder import builder diff --git a/edg/core/test_block_errors.py b/edg/core/test_block_errors.py index b853f421c..c379816fc 100644 --- a/edg/core/test_block_errors.py +++ b/edg/core/test_block_errors.py @@ -77,3 +77,38 @@ def contents(self) -> None: def test_ambiguous_join(self) -> None: with self.assertRaises(UnconnectableError): self.AmbiguousJoinBlock()._elaborated_def_to_proto() + + +class BadNamingTestCase(unittest.TestCase): + class UnnamedParamBlock(Block): + def __init__(self) -> None: + super().__init__() + self.Parameter(FloatExpr(0.0)) # unnamed param, should error + self.param = self.Parameter(IntExpr(2)) # frame continues, checks captured lineno is unaffected + + def test_unnamed_param(self) -> None: + with self.assertRaises(ElementNamingError) as exc: + self.UnnamedParamBlock()._elaborated_def_to_proto() + self.assertIn("86", str(exc.exception)) + + class DuplicateNameBlock(Block): + def __init__(self) -> None: + super().__init__() + self.duplicate_name = self.Parameter(FloatExpr()) + self.duplicate_name = self.Parameter(FloatExpr()) + + def test_duplicate_name(self) -> None: + with self.assertRaises(ElementNamingError) as exc: + self.DuplicateNameBlock()._elaborated_def_to_proto() + self.assertIn("duplicate_name", str(exc.exception)) + + class DuplicateAnonNameBlock(Block): + def __init__(self) -> None: + super().__init__() + self.require(True) + self.anon_constr0 = self.require(True) # duplicates a default unassigned name + + def test_duplicate_anon_name(self) -> None: + with self.assertRaises(ElementNamingError) as exc: + self.DuplicateAnonNameBlock()._elaborated_def_to_proto() + self.assertIn("anon_constr0", str(exc.exception)) diff --git a/edg/electronics_interfaces/AnalogPort.py b/edg/electronics_interfaces/AnalogPort.py index 182ebc26d..835049a9b 100644 --- a/edg/electronics_interfaces/AnalogPort.py +++ b/edg/electronics_interfaces/AnalogPort.py @@ -252,7 +252,7 @@ def from_supply( else: # generic default signal = supply_range - return AnalogSource(voltage=supply_range, signal=signal, current_limits=current_limits, impedance=impedance) # type: ignore + return AnalogSource(voltage=supply_range, signal=signal, current_limits=current_limits, impedance=impedance) @deprecated_param_remap(("voltage_out", "voltage"), ("signal_out", "signal")) def __init__( diff --git a/edg/electronics_interfaces/CircuitPackingBlock.py b/edg/electronics_interfaces/CircuitPackingBlock.py index ed52edaaf..1ff081716 100644 --- a/edg/electronics_interfaces/CircuitPackingBlock.py +++ b/edg/electronics_interfaces/CircuitPackingBlock.py @@ -12,7 +12,7 @@ @abstract_block class NetPackingBlock(InternalBlock, Block): def packed(self, elts: BasePort, merged: BasePort) -> None: - self.nets_packed = self.Metadata({"src": self._ports.name_of(elts), "dst": self._ports.name_of(merged)}) + self.nets_packed = self.Metadata({"src": self._name_of(elts), "dst": self._name_of(merged)}) class PackedPassive(InternalSubcircuit, GeneratorBlock): @@ -31,9 +31,7 @@ def generate(self) -> None: for request in self.get(self.elts.requested()): self.elts.append_elt(Passive(), request) - self.nets_packed = self.Metadata( - {"src": self._ports.name_of(self.elts), "dst": self._ports.name_of(self.merged)} - ) + self.nets_packed = self.Metadata({"src": self._name_of(self.elts), "dst": self._name_of(self.merged)}) class PackedGround(InternalSubcircuit, GeneratorBlock): diff --git a/edg/electronics_interfaces/DigitalPorts.py b/edg/electronics_interfaces/DigitalPorts.py index a7605f1c9..8c96e570f 100644 --- a/edg/electronics_interfaces/DigitalPorts.py +++ b/edg/electronics_interfaces/DigitalPorts.py @@ -357,13 +357,13 @@ def from_supply( else: output_threshold = (GroundLink._voltage_range(neg).upper(), VoltageLink._voltage_range(pos).lower()) - return DigitalSource(voltage=supply_range, current_limits=current_limits, output_thresholds=output_threshold) # type: ignore + return DigitalSource(voltage=supply_range, current_limits=current_limits, output_thresholds=output_threshold) @staticmethod def from_bidir(model: DigitalBidir) -> DigitalSource: model_is_empty = not model._get_initializers([]) if not model_is_empty: # DigitalSource has additional high_driver and low_driver fields - return DigitalSource( # type: ignore + return DigitalSource( model.voltage, model.current_limits, output_thresholds=model.output_thresholds, @@ -371,7 +371,7 @@ def from_bidir(model: DigitalBidir) -> DigitalSource: pulldown_capable=model.pulldown_capable, ) else: - return DigitalSource.empty() # type: ignore + return DigitalSource.empty() @deprecated_param_remap(("voltage_out", "voltage")) def __init__( @@ -579,7 +579,7 @@ def from_supply( else: # assumed ideal output_threshold = (neg_base, VoltageLink._voltage_range(pos).lower()) - return DigitalBidir( # type: ignore + return DigitalBidir( voltage_limits=voltage_limit, current_draw=current_draw, voltage=supply_range, diff --git a/edg/electronics_interfaces/VoltagePorts.py b/edg/electronics_interfaces/VoltagePorts.py index 329ebca0d..4131db299 100644 --- a/edg/electronics_interfaces/VoltagePorts.py +++ b/edg/electronics_interfaces/VoltagePorts.py @@ -19,7 +19,7 @@ class VoltageLink(Link): def _voltage_range(cls, port: Port[VoltageLink]) -> RangeExpr: """Returns the voltage for a Voltage port, either sink or source""" if isinstance(port, VoltageSource): - return port.voltage # type: ignore + return port.voltage elif isinstance(port, VoltageSink): return port.link().voltage else: diff --git a/edg/electronics_model/CircuitBlock.py b/edg/electronics_model/CircuitBlock.py index 8c9847f78..fa40b7c21 100644 --- a/edg/electronics_model/CircuitBlock.py +++ b/edg/electronics_model/CircuitBlock.py @@ -117,7 +117,7 @@ def footprint( else: pin_tuples = tuple(iter(pin_name)) for pin in pin_tuples: - pinning_array.append(f"{pin}={pin_port._name_from(self)}") + pinning_array.append(f"{pin}={self._path_name_of(pin_port)}") assert pin not in assigned_pins, f"duplicate pin name {pin} in footprint pinning" assigned_pins.add(pin) self.assign(self.fp_pinning, pinning_array) diff --git a/edg/electronics_model/PassivePort.py b/edg/electronics_model/PassivePort.py index 142654319..65bf22d1c 100644 --- a/edg/electronics_model/PassivePort.py +++ b/edg/electronics_model/PassivePort.py @@ -271,7 +271,8 @@ def adapt_to(self, that: AdaptTargetType) -> AdaptTargetType: # map initializers from that to constructor args adapter_init_kwargs = {} # make everything kwargs for simplicity - for param_name, param in that._parameters.items(): + for param in that._parameters: + param_name = that._name_of(param) assert param.initializer is not None, f"missing initializer for {param_name}" adapter_init_kwargs[param_name] = param.initializer diff --git a/edg/electronics_model/PinAssignmentUtil.py b/edg/electronics_model/PinAssignmentUtil.py index 1f02b3506..3c1bff516 100644 --- a/edg/electronics_model/PinAssignmentUtil.py +++ b/edg/electronics_model/PinAssignmentUtil.py @@ -16,7 +16,7 @@ def leaf_circuit_ports(prefix: str, port: Port) -> Iterable[Tuple[str, Passive]] return [(prefix, port)] # type: ignore else: assert port._ports, "must be Passive or Port bundle with sub-Ports" - return chain(*[leaf_circuit_ports(f"{prefix}.{name}", port) for (name, port) in port._ports.items()]) + return chain(*[leaf_circuit_ports(f"{prefix}.{port._name_of(subport)}", subport) for subport in port._ports]) class SpecialPin: diff --git a/edg/electronics_model/SubboardBlock.py b/edg/electronics_model/SubboardBlock.py index 3c9ffb94e..0721e7c98 100644 --- a/edg/electronics_model/SubboardBlock.py +++ b/edg/electronics_model/SubboardBlock.py @@ -49,7 +49,7 @@ def export_tap(self, exterior_port: BasePort, internal_port: BasePort) -> None: @override def _def_to_proto(self) -> edgir.HierarchyBlock: # create this assign after the block definition has run - self.assign(self.fp_external_blocks, [self._blocks.name_of(block) for block in self._external_blocks]) + self.assign(self.fp_external_blocks, [self._name_of(block) for block in self._external_blocks]) return super()._def_to_proto() @override @@ -57,7 +57,7 @@ def _populate_def_proto_hierarchy(self, pb: edgir.HierarchyBlock, ref_map: Refab super()._populate_def_proto_hierarchy(pb, ref_map) for exterior_port, internal_port in self._export_taps: - internal_port_name = internal_port._name_from(self).replace(".", "_") + internal_port_name = "_".join(self._path_of(internal_port)) constraint_pb = edgir.add_pair(pb.constraints, f"(export_tap){internal_port_name}") if isinstance(exterior_port, Vector): constraint_pb.exportedArray.exterior_port.ref.CopyFrom(ref_map[exterior_port]) diff --git a/examples/Multimeter/Multimeter.net.ref b/examples/Multimeter/Multimeter.net.ref index 0f9ec7cb4..e068c45e7 100644 --- a/examples/Multimeter/Multimeter.net.ref +++ b/examples/Multimeter/Multimeter.net.ref @@ -1431,24 +1431,24 @@ (node (ref R28) (pin 1)) (node (ref U19) (pin 5)) (node (ref C30) (pin 1))) -(net (code 6) (name "vcenter") - (node (ref U8) (pin 1)) - (node (ref U8) (pin 4)) - (node (ref U9) (pin 1))) -(net (code 7) (name "spk_chain_0") +(net (code 6) (name "spk_chain_0") (node (ref U6) (pin 36)) (node (ref R14) (pin 1))) -(net (code 8) (name "spk_chain_1") +(net (code 7) (name "spk_chain_1") (node (ref R14) (pin 2)) (node (ref C10) (pin 1)) (node (ref TP4) (pin 1)) (node (ref C13) (pin 2))) -(net (code 9) (name "spk_chain_2.a") +(net (code 8) (name "spk_chain_2.a") (node (ref U7) (pin 8)) (node (ref J4) (pin 1))) -(net (code 10) (name "spk_chain_2.b") +(net (code 9) (name "spk_chain_2.b") (node (ref U7) (pin 5)) (node (ref J4) (pin 2))) +(net (code 10) (name "vcenter") + (node (ref U8) (pin 1)) + (node (ref U8) (pin 4)) + (node (ref U9) (pin 1))) (net (code 11) (name "meas_chain_0") (node (ref R19) (pin 2)) (node (ref R20) (pin 1)) diff --git a/examples/Multimeter/Multimeter.svgpcb.js b/examples/Multimeter/Multimeter.svgpcb.js index 7c174973f..a823d916d 100644 --- a/examples/Multimeter/Multimeter.svgpcb.js +++ b/examples/Multimeter/Multimeter.svgpcb.js @@ -542,11 +542,11 @@ board.setNetlist([ {name: "v5v", pads: [["U3", "4"], ["C2", "1"], ["TP1", "1"], ["D3", "1"], ["U4", "1"], ["U4", "3"], ["C3", "1"], ["U5", "1"], ["U5", "3"], ["C5", "1"], ["U7", "1"], ["U7", "6"], ["C11", "1"], ["C12", "1"]]}, {name: "v3v3", pads: [["U4", "5"], ["C4", "1"], ["TP2", "1"], ["D4", "1"], ["U6", "28"], ["U6", "30"], ["C7", "1"], ["J2", "1"], ["D6", "2"], ["J3", "7"], ["R13", "1"], ["C9", "1"], ["R24", "1"]]}, {name: "vanalog", pads: [["U5", "5"], ["C6", "1"], ["TP3", "1"], ["D5", "1"], ["R17", "1"], ["U8", "5"], ["U8", "6"], ["C15", "1"], ["U9", "5"], ["C16", "1"], ["U10", "5"], ["C17", "1"], ["U11", "5"], ["C18", "1"], ["U12", "5"], ["C19", "1"], ["U13", "5"], ["U13", "6"], ["C20", "1"], ["R23", "1"], ["U15", "5"], ["U15", "6"], ["C26", "1"], ["U16", "5"], ["C27", "1"], ["U17", "5"], ["C28", "1"], ["U18", "5"], ["C29", "1"], ["R25", "1"], ["R26", "1"], ["R27", "1"], ["R28", "1"], ["U19", "5"], ["C30", "1"]]}, - {name: "vcenter", pads: [["U8", "1"], ["U8", "4"], ["U9", "1"]]}, {name: "spk_chain_0", pads: [["U6", "36"], ["R14", "1"]]}, {name: "spk_chain_1", pads: [["R14", "2"], ["C10", "1"], ["TP4", "1"], ["C13", "2"]]}, {name: "spk_chain_2.a", pads: [["U7", "8"], ["J4", "1"]]}, {name: "spk_chain_2.b", pads: [["U7", "5"], ["J4", "2"]]}, + {name: "vcenter", pads: [["U8", "1"], ["U8", "4"], ["U9", "1"]]}, {name: "meas_chain_0", pads: [["R19", "2"], ["R20", "1"], ["R21", "1"], ["R22", "1"], ["U13", "3"]]}, {name: "meas_chain_1", pads: [["U13", "1"], ["U13", "4"], ["TP5", "1"], ["U14", "5"]]}, {name: "data_usb.pwr", pads: [["J1", "A4"], ["J1", "A9"], ["J1", "B4"], ["J1", "B9"], ["U6", "32"], ["C8", "1"]]}, diff --git a/examples/PcbBot/PcbBot.net.ref b/examples/PcbBot/PcbBot.net.ref index d9b885d9f..be81dc67f 100644 --- a/examples/PcbBot/PcbBot.net.ref +++ b/examples/PcbBot/PcbBot.net.ref @@ -1491,17 +1491,9 @@ (node (ref J13) (pin 14)) (node (ref C25) (pin 1)) (node (ref R18) (pin 1))) -(net (code 6) (name "v2v5") - (node (ref U9) (pin 2)) - (node (ref C22) (pin 1)) - (node (ref J13) (pin 21))) -(net (code 7) (name "v1v2") - (node (ref U10) (pin 2)) - (node (ref C24) (pin 1)) - (node (ref J13) (pin 15))) -(net (code 8) (name "touch_0") +(net (code 6) (name "touch_0") (node (ref U4) (pin 7))) -(net (code 9) (name "i2c_chain_0.scl") +(net (code 7) (name "i2c_chain_0.scl") (node (ref U4) (pin 38)) (node (ref J4) (pin 3)) (node (ref J5) (pin 3)) @@ -1514,7 +1506,7 @@ (node (ref U7) (pin 14)) (node (ref J8) (pin 18)) (node (ref J13) (pin 20))) -(net (code 10) (name "i2c_chain_0.sda") +(net (code 8) (name "i2c_chain_0.sda") (node (ref U4) (pin 4)) (node (ref J4) (pin 4)) (node (ref J5) (pin 4)) @@ -1528,6 +1520,14 @@ (node (ref J8) (pin 19)) (node (ref J8) (pin 20)) (node (ref J13) (pin 22))) +(net (code 9) (name "v2v5") + (node (ref U9) (pin 2)) + (node (ref C22) (pin 1)) + (node (ref J13) (pin 21))) +(net (code 10) (name "v1v2") + (node (ref U10) (pin 2)) + (node (ref C24) (pin 1)) + (node (ref J13) (pin 15))) (net (code 11) (name "usb.usb.dp") (node (ref J1) (pin A6)) (node (ref J1) (pin B6)) diff --git a/examples/PcbBot/PcbBot.svgpcb.js b/examples/PcbBot/PcbBot.svgpcb.js index 5fbc6bd43..6669f3dd0 100644 --- a/examples/PcbBot/PcbBot.svgpcb.js +++ b/examples/PcbBot/PcbBot.svgpcb.js @@ -547,11 +547,11 @@ board.setNetlist([ {name: "vbatt", pads: [["Q1", "3"], ["D3", "1"], ["TP2", "1"], ["Q3", "3"], ["R16", "1"], ["J9", "2"], ["J10", "2"], ["J11", "2"], ["J12", "2"], ["D9", "1"], ["D10", "1"], ["D11", "1"], ["D12", "1"], ["D13", "1"], ["D14", "1"], ["D15", "1"], ["D16", "1"], ["D17", "1"], ["D18", "1"], ["D19", "1"], ["D20", "1"], ["D21", "1"], ["D22", "1"], ["D23", "1"], ["D24", "1"], ["D25", "1"], ["C20", "1"]]}, {name: "pwr", pads: [["D4", "1"], ["Q3", "2"], ["U2", "3"], ["C1", "1"], ["U9", "3"], ["C21", "1"], ["U10", "3"], ["C23", "1"]]}, {name: "v3v3", pads: [["U2", "2"], ["C2", "1"], ["D5", "1"], ["TP3", "1"], ["J3", "1"], ["U4", "2"], ["C5", "1"], ["C6", "1"], ["R8", "1"], ["J4", "1"], ["J4", "6"], ["J5", "1"], ["J6", "1"], ["J7", "1"], ["R10", "1"], ["R11", "1"], ["U5", "12"], ["U5", "5"], ["U5", "8"], ["C8", "1"], ["C9", "1"], ["U6", "13"], ["U6", "2"], ["U6", "4"], ["C10", "1"], ["U7", "16"], ["C13", "1"], ["D8", "2"], ["J8", "11"], ["J8", "6"], ["J8", "9"], ["C17", "1"], ["C18", "1"], ["J13", "14"], ["C25", "1"], ["R18", "1"]]}, - {name: "v2v5", pads: [["U9", "2"], ["C22", "1"], ["J13", "21"]]}, - {name: "v1v2", pads: [["U10", "2"], ["C24", "1"], ["J13", "15"]]}, {name: "touch_0", pads: [["U4", "7"]]}, {name: "i2c_chain_0.scl", pads: [["U4", "38"], ["J4", "3"], ["J5", "3"], ["J6", "3"], ["J7", "3"], ["R10", "2"], ["TP4", "1"], ["U5", "13"], ["U6", "1"], ["U7", "14"], ["J8", "18"], ["J13", "20"]]}, {name: "i2c_chain_0.sda", pads: [["U4", "4"], ["J4", "4"], ["J5", "4"], ["J6", "4"], ["J7", "4"], ["R11", "2"], ["TP5", "1"], ["U5", "14"], ["U6", "16"], ["U7", "15"], ["J8", "19"], ["J8", "20"], ["J13", "22"]]}, + {name: "v2v5", pads: [["U9", "2"], ["C22", "1"], ["J13", "21"]]}, + {name: "v1v2", pads: [["U10", "2"], ["C24", "1"], ["J13", "15"]]}, {name: "usb.usb.dp", pads: [["J1", "A6"], ["J1", "B6"], ["U1", "2"], ["U4", "14"]]}, {name: "usb.usb.dm", pads: [["J1", "A7"], ["J1", "B7"], ["U1", "1"], ["U4", "13"]]}, {name: "usb.conn.cc.cc1", pads: [["J1", "A5"], ["R1", "2"]]},