1from __future__
import annotations
3from pathlib
import Path
8from esiaccel.cosim.pytest
import cosim_test
10HW_DIR = Path(__file__).
resolve().parent.parent /
"hw"
13def run(acc: Accelerator) ->
None:
15 merge_a = acc.ports[
esi.AppID(
"merge_a")].write_port(
"data")
17 merge_b = acc.ports[
esi.AppID(
"merge_b")].write_port(
"data")
19 merge_x = acc.ports[
esi.AppID(
"merge_x")].read_port(
"data")
22 for i
in range(10, 15):
27 print(f
"merge_a: {i}, merge_b: {i + 10}, "
28 f
"merge_x 1: {x1}, merge_x 2: {x2}")
29 assert x1 == i + 10
or x1 == i
30 assert x2 == i + 10
or x2 == i
33 join_a = acc.ports[
esi.AppID(
"join_a")].write_port(
"data")
35 join_b = acc.ports[
esi.AppID(
"join_b")].write_port(
"data")
37 join_x = acc.ports[
esi.AppID(
"join_x")].read_port(
"data")
40 for i
in range(15, 27):
44 print(f
"join_a: {i}, join_b: {i + 10}, join_x: {x}")
45 assert x == (i + i + 10) & 0xFFFF
47 fork_a = acc.ports[
esi.AppID(
"fork_a")].write_port(
"data")
49 fork_x = acc.ports[
esi.AppID(
"fork_x")].read_port(
"data")
51 fork_y = acc.ports[
esi.AppID(
"fork_y")].read_port(
"data")
54 for i
in range(27, 33):
58 print(f
"fork_a: {i}, fork_x: {x}, fork_y: {y}")
62@cosim_test(HW_DIR / "esi_advanced.py")
67if __name__ ==
"__main__":
68 platform = sys.argv[1]
69 conn_str = sys.argv[2]
70 conn = esi.connect(platform, conn_str)
static void print(TypedAttr val, llvm::raw_ostream &os)
static mlir::Operation * resolve(Context &context, mlir::SymbolRefAttr sym)
None test_cosim_advanced(Accelerator accelerator)
None run(Accelerator acc)