()
| 96 | |
| 97 | |
| 98 | def test_replace_opr(): |
| 99 | |
| 100 | a = Tensor([1, 2]) |
| 101 | b = Tensor([3, 4]) |
| 102 | |
| 103 | @trace(symbolic=True, capture_as_const=True) |
| 104 | def fwd(a, b): |
| 105 | return (a + b) * 2 |
| 106 | |
| 107 | fwd(a, b) |
| 108 | orig_model = io.BytesIO() |
| 109 | fwd.dump( |
| 110 | orig_model, arg_names=["a", "b"], output_names="o", optimize_for_inference=False |
| 111 | ) |
| 112 | orig_model.seek(0) |
| 113 | |
| 114 | graph = Net.load(orig_model) |
| 115 | vara = graph.var_filter.name("a").as_unique() |
| 116 | varb = graph.var_filter.name("b").as_unique() |
| 117 | |
| 118 | out1 = F.mul(vara, varb) |
| 119 | out1 = F.relu(out1) |
| 120 | out1 += 2 |
| 121 | out1 *= 3 |
| 122 | out1 = graph.add_dep_oprs(out1) |
| 123 | orig_opr = graph.opr_filter.has_input(vara).as_unique() |
| 124 | |
| 125 | new_opr = out1[0].owner |
| 126 | repl_dict = {orig_opr: new_opr} |
| 127 | graph.replace_oprs(repl_dict) |
| 128 | |
| 129 | var_out = orig_opr.outputs |
| 130 | |
| 131 | for idx, node in enumerate(var_out): |
| 132 | assert node.owner is new_opr |
| 133 | assert node.owner.outputs[idx] is node |
| 134 | |
| 135 | modified_model1 = io.BytesIO() |
| 136 | graph.dump(modified_model1) |
| 137 | modified_model1.seek(0) |
| 138 | |
| 139 | load_graph = GraphInference(modified_model1) |
| 140 | out = load_graph.run(a, b) |
| 141 | np.testing.assert_equal(out["o"], [30, 60]) |
| 142 | |
| 143 | |
| 144 | def test_splice_network(): |
nothing calls this directly
no test coverage detected