|
1 |
| -from typing import Dict, Union, Optional, Tuple |
| 1 | +from typing import Dict, Union, Optional, Tuple, List |
2 | 2 |
|
3 | 3 | from py_hcl.core.expr import HclExpr
|
4 | 4 | from py_hcl.core.expr.error import ExprError
|
5 | 5 | from py_hcl.core.stmt.connect import VariableType
|
6 | 6 | from py_hcl.core.type import HclType
|
7 |
| -from py_hcl.core.type.bundle import Dir, BundleT |
| 7 | +from py_hcl.core.type.bundle import BundleDirection, BundleT |
8 | 8 | from py_hcl.core.utils import module_inherit_mro
|
9 | 9 | from py_hcl.utils.serialization import json_serialize
|
10 | 10 |
|
@@ -32,54 +32,52 @@ def __init__(self,
|
32 | 32 | self.module_name = module_name
|
33 | 33 |
|
34 | 34 |
|
35 |
| -@json_serialize |
36 |
| -class IONode(object): |
37 |
| - def __init__(self, io_holder: IOHolder, next_node: Optional["IOHolder"]): |
38 |
| - self.io_holder = io_holder |
39 |
| - if next_node is not None: |
40 |
| - self.next_node = next_node |
41 |
| - |
42 |
| - |
43 | 35 | @json_serialize(
|
44 |
| - json_fields=["id", "type", "hcl_type", "variable_type", "io_chain_head"]) |
| 36 | + json_fields=["id", "type", "hcl_type", "variable_type", "io_chain"]) |
45 | 37 | class IO(HclExpr):
|
46 |
| - def __init__(self, hcl_type: HclType, io_chain_head: IONode): |
| 38 | + def __init__(self, hcl_type: HclType, io_chain: List[IOHolder]): |
47 | 39 | self.type = 'io'
|
48 | 40 | self.hcl_type = hcl_type
|
49 | 41 | self.variable_type = VariableType.VALUE
|
50 |
| - self.io_chain_head = io_chain_head |
| 42 | + self.io_chain = io_chain |
51 | 43 |
|
52 | 44 |
|
53 | 45 | def io_extend(modules: Tuple[type]):
|
| 46 | + """ |
| 47 | +
|
| 48 | +
|
| 49 | + """ |
| 50 | + |
54 | 51 | modules = module_inherit_mro(modules)
|
55 | 52 |
|
56 | 53 | current_ports = {}
|
57 |
| - io_chain = None |
| 54 | + io_chain = [] |
58 | 55 | for m in modules[::-1]:
|
59 |
| - h = m.io.io_chain_head.io_holder |
| 56 | + h = m.io.io_chain[0] |
60 | 57 | current_ports.update(h.named_ports)
|
61 |
| - io_chain = IONode(h, io_chain) |
| 58 | + io_chain.insert(0, h) |
62 | 59 |
|
63 | 60 | def _(named_ports: Dict[str, Union[Input, Output]]):
|
64 | 61 | current_ports.update(named_ports)
|
65 |
| - io_chain_head = IONode(IOHolder(named_ports), io_chain) |
66 |
| - return IO(calc_type_from_ports(current_ports), io_chain_head) |
| 62 | + io_chain.insert(0, IOHolder(named_ports)) |
| 63 | + return IO(__build_bundle_type_from_ports(current_ports), io_chain) |
67 | 64 |
|
68 | 65 | return _
|
69 | 66 |
|
70 | 67 |
|
71 |
| -def calc_type_from_ports(named_ports: Dict[str, Union[Input, Output]]): |
72 |
| - types = {} |
| 68 | +def __build_bundle_type_from_ports( |
| 69 | + named_ports: Dict[str, Union[Input, Output]]) -> BundleT: |
| 70 | + fields = {} |
73 | 71 | for k, v in named_ports.items():
|
74 | 72 | if isinstance(v, Input):
|
75 |
| - types[k] = {"dir": Dir.SRC, "hcl_type": v.hcl_type} |
| 73 | + fields[k] = {"dir": BundleDirection.SOURCE, "hcl_type": v.hcl_type} |
76 | 74 | continue
|
77 | 75 |
|
78 | 76 | if isinstance(v, Output):
|
79 |
| - types[k] = {"dir": Dir.SINK, "hcl_type": v.hcl_type} |
| 77 | + fields[k] = {"dir": BundleDirection.SINK, "hcl_type": v.hcl_type} |
80 | 78 | continue
|
81 | 79 |
|
82 | 80 | raise ExprError.io_value_err(
|
83 | 81 | "type of '{}' is {}, not Input or Output".format(k, type(v)))
|
84 | 82 |
|
85 |
| - return BundleT(types) |
| 83 | + return BundleT(fields) |
0 commit comments