copapy/tests/test_compile_armv7.py

import os
import subprocess
import warnings

import pytest

import copapy as cp
import copapy.backend as backend
from copapy import NumLike, _binwrite
from copapy.backend import Store, add_read_value_remote, compile_to_dag

if os.name == "nt":
    # On Windows wsl and qemu-user is required:
    # sudo apt install qemu-user
    qemu_command = ["wsl", "qemu-arm"]
else:
    qemu_command = ["qemu-arm"]


def run_command(command: list[str]) -> str:
    result = subprocess.run(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding="utf8",
        check=False,
    )
    assert result.returncode != 11, (
        f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}"
    )
    assert result.returncode == 0, (
        f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}"
    )
    return result.stdout


def check_for_qemu() -> bool:
    command = qemu_command + ["--version"]
    try:
        result = subprocess.run(
            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False
        )
    except Exception:
        return False
    return result.returncode == 0


def function(c1: NumLike, c2: NumLike) -> tuple[NumLike, ...]:
    i1 = c1 // 3.3 + 5
    i2 = c2 * 5 + c1
    r1 = i1 + i2 * 55 / 4
    r2 = 4 * i2 + 5

    return i1, i2, r1, r2


@pytest.mark.runner
def test_compile():
    t1 = cp.vector([10, 11, 12]) + cp.vector(cp.value(v) for v in range(3))
    t2 = t1.sum()

    t3 = cp.vector(cp.value(1 / (v + 1)) for v in range(3))
    t4 = ((t3 * t1) * 2).sum()
    t5 = ((t3 * t1) * 2).magnitude()

    ret = (t2, t4, t5)

    out = [Store(r) for r in ret]

    sdb = backend.stencil_db_from_package("armv7")
    il, variables = compile_to_dag(out, sdb)

    # run program command
    il.write_com(_binwrite.Command.RUN_PROG)
    # il.write_com(_binwrite.Command.DUMP_CODE)

    for v in ret:
        assert isinstance(v, cp.value)
        add_read_value_remote(il, variables, v.net)

    il.write_com(_binwrite.Command.END_COM)

    # print('* Data to runner:')
    # il.print()

    il.to_file("build/runner/test-armv7.copapy")

    if not check_for_qemu():
        warnings.warn("qemu-armv7 not found, test skipped!", UserWarning)
    elif "wsl" in qemu_command:
        warnings.warn("qemu-armv7 seams not work on wsl1, test skipped!", UserWarning)
    elif not os.path.isfile("build/runner/coparun-armv7"):
        warnings.warn("armv7 runner not found, aarch64 test skipped!", UserWarning)
    else:
        command = [
            "build/runner/coparun-armv7",
            "build/runner/test-armv7.copapy",
            "build/runner/test-armv7.copapy.bin",
        ]
        result = run_command(qemu_command + command)
        print("* Output from runner:\n--")
        print(result)
        print("--")

        assert "Return value: 1" in result

        # Compare to x86_64 reference results
        assert " size=4 data=24 00 00 00" in result
        assert " size=4 data=56 55 25 42" in result
        assert " size=4 data=B4 F9 C8 41" in result


if __name__ == "__main__":
    # test_example()
    test_compile()