#!/usr/bin/env python3

# Copyright (C) 2020  Max Regan

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

import pytest
import serial
import serial.tools.list_ports
import logging
import pylink
import time
import os
import sys
import re
import subprocess

PYTEST_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_FW_DIR = os.path.abspath(PYTEST_DIR + "../../../../firmware/")
SAMPLE_TIME_MS = 2500
SAMPLES_PER_SEC = 12000000
DRIVER = "fx2lafw"

TEST_START_TEXT = b"TEST_BEGIN\r\n"
TEST_PASS_TEXT = b"TEST_PASS\r\n"
TEST_FAIL_TEXT = b"TEST_FAIL\r\n"


@pytest.fixture
def logger():
    logging.basicConfig()
    return logging.getLogger(__name__)


@pytest.fixture
def context_factory():
    def create_context(
        fw_rel_path: str,
        mcu: str = "STM32L010C6",
        addr: int = 0x8000000,
        leave_halted: bool = False,
    ):
        ports = [
            p
            for p in serial.tools.list_ports.comports()
            if p.product == "FT232R USB UART"
        ]

        if len(ports) == 0:
            raise RuntimeError("No serial devices found")
        if len(ports) > 1:
            raise RuntimeError(
                "Too many serial devices, not sure which to use. "
                "This should be made configurable"
            )

        jlink = pylink.jlink.JLink()
        jlink.open()
        jlink.disable_dialog_boxes()
        jlink.set_tif(pylink.enums.JLinkInterfaces.SWD)
        jlink.connect(mcu)
        fw = DEFAULT_FW_DIR + "/" + fw_rel_path
        logging.info("Flashing {}...".format(fw))
        jlink.flash_file(fw, addr)
        logging.info("Flashing done.")
        assert jlink.halted()
        jlink.reset(halt=True)

        serial_dev = serial.Serial(port=ports[0].device, baudrate=115200, timeout=2)
        if leave_halted:
            return serial_dev, jlink

        # Start test test, and check that it started correctly over serial
        jlink.reset(halt=False)
        while True:
            try:
                logging.info("Waiting for firmware to start...")
                assert serial_dev.read_until(TEST_START_TEXT).endswith(
                    TEST_START_TEXT
                ), "Timed out starting test firmware application"
                logging.debug("Test execution started")
            except serial.serialutil.SerialException:
                continue
            break
        return serial_dev, jlink

    return create_context


def measure_frequency(
    period: float,
    pin_name: str,
    executable: str = "sigrok-cli",
    driver_name: str = "fx2lafw",
    trigger: str = "r",
):

    cmd = [
        executable,
        "-C",
        pin_name,
        "-d",
        driver_name,
        "-c",
        "samplerate=1M",
        "--time",
        "{}ms".format(int(period * 1000)),
        "-t",
        "{}={}".format(pin_name, trigger),
        "-P",
        "timing:data={}".format(pin_name),
        "-A",
        "timing=time",
    ]

    print("sigrok-cli cmd {}".format(cmd))
    proc = subprocess.run(cmd, capture_output=True, check=True)
    lines = proc.stdout.decode("utf-8").split("\n")
    reg = re.compile(".*:\\W(\\d+.\\d+)\\W(\\w+)")
    periods = []
    for line in lines:
        m = reg.match(line)
        if not m:
            break
        num = float(m.groups(1)[0])
        units = m.groups(1)[1]
        if units == "s":
            periods.append(num)
        elif units == "ms":
            periods.append(num / 1000)
        elif units == "μs":
            periods.append(num / 1000000)
        else:
            assert False, "Couldnt find units in line '{}', units were '{}'".format(
                line, units
            )

    return periods[::2], periods[1:][::2]


def test_meta_pass(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/pass.bin")
    text = serial_dev.read_until(TEST_PASS_TEXT)
    print("Got serial output: {}".format(text))
    assert text.endswith(TEST_PASS_TEXT)


def test_meta_fail(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/fail.bin")
    text = serial_dev.read_until(TEST_PASS_TEXT)
    print("Got serial output: {}".format(text))
    assert not text.endswith(TEST_PASS_TEXT)


def test_meta_timeout(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/timeout.bin")
    text = serial_dev.read_until(TEST_PASS_TEXT)
    assert not text.endswith(TEST_PASS_TEXT)


def test_meta_nostart(context_factory, logger):
    with pytest.raises(AssertionError):
        serial_dev, jlink = context_factory("Test/Apps/no_start.bin")


def test_watch(context_factory, logger):
    serial_dev, jlink = context_factory("Application/main.bin", leave_halted=True)
    jlink.reset(halt=False)


def test_set_time(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/set_time.bin")
    text = serial_dev.read_until(TEST_PASS_TEXT)
    print("Text:", text.decode())
    assert text.endswith(TEST_PASS_TEXT)


def test_periodic_alarms(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/periodic_alarms.bin")
    serial_dev.timeout = 6
    text = serial_dev.read_until(TEST_PASS_TEXT)
    print("Text:", text.decode())
    assert text.endswith(TEST_PASS_TEXT)


def test_button_slow(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/button.bin")
    serial_dev.timeout = 0.3
    ASSERTED = True

    serial_dev.dtr = not ASSERTED
    while (line := serial_dev.readline()) is not None and len(line) > 0:
        pass

    for _ in range(5):
        serial_dev.dtr = ASSERTED
        press_line = serial_dev.readline()
        serial_dev.dtr = not ASSERTED
        release_line = serial_dev.readline()
        assert press_line == b"up:pressed\r\n"
        assert release_line == b"up:released\r\n"


def test_button_fast(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/button.bin")
    serial_dev.timeout = 0.3
    ASSERTED = True

    serial_dev.dtr = not ASSERTED
    time.sleep(0.3)
    serial_dev.timeout = 0
    while (line := serial_dev.readline()) is not None and len(line) > 0:
        pass

    for _ in range(25):
        serial_dev.dtr = ASSERTED
        time.sleep(0.075)
        serial_dev.dtr = not ASSERTED
        time.sleep(0.075)

    serial_dev.timeout = 0.3
    for _ in range(25):
        press_line = serial_dev.readline()
        release_line = serial_dev.readline()
        assert press_line == b"up:pressed\r\n"
        assert release_line == b"up:released\r\n"

    serial_dev.timeout = 0
    assert serial_dev.readline() == b""


def test_button_lowpower(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/button_lowpower.bin")
    serial_dev.timeout = 0.3
    ASSERTED = True

    serial_dev.dtr = not ASSERTED
    while (line := serial_dev.readline()) is not None and len(line) > 0:
        pass

    for _ in range(5):
        serial_dev.dtr = ASSERTED
        press_line = serial_dev.readline()
        serial_dev.dtr = not ASSERTED
        release_line = serial_dev.readline()
        assert press_line == b"up:pressed\r\n"
        assert release_line == b"up:released\r\n"


def test_clock(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/clock.bin")
    EXPECTED_RUNTIME = 10
    TOLERANCE = 0.2

    serial_dev.timeout = EXPECTED_RUNTIME * 1.2

    START_MARKER = b"GO\r\n"
    END_MARKER = b"STOP\r\n"

    start_text = serial_dev.read_until(START_MARKER)
    start = time.monotonic()
    print("Start text:", start_text)
    assert start_text.endswith(START_MARKER)

    end_text = serial_dev.read_until(END_MARKER)
    end = time.monotonic()
    print("Time:", end - start)
    assert end_text.endswith(END_MARKER)

    delta = end - start
    logger.info("Serial text: {}".format(start_text))
    logger.info("Serial text: {}".format(end_text))
    logger.info("Start time: {}".format(start))
    logger.info("End time: {}".format(end))
    logger.info("Delta time: {}".format(delta))

    # TODO: Using a single pin, instead of UART, would make this more
    # accurate. Add support via sigrok.
    assert abs(delta - EXPECTED_RUNTIME) < TOLERANCE


def test_wakeup_irq(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/wakeup_irq.bin")
    serial_dev.timeout = 65

    pattern = re.compile("Requested=(\\d*) Actual=(\\d*) Wakeups=(\\d*)")
    while True:
        line = serial_dev.readline()
        if line == TEST_PASS_TEXT:
            break

        line = line.decode("ascii", errors="ignore")
        print(line.strip())
        match = pattern.match(line)
        assert match
        req = int(match.group(1))
        actual = int(match.group(2))
        wakeups = int(match.group(3))
        delta = req - actual

        assert wakeups == 1, "Expected one wakeup per line of test output"
        if req < 32000:
            assert abs(delta < req * (2.0 / 100.0)) or (delta <= 1)
        else:
            # Delays > 32sec have reduced resolution (1 sec)
            assert abs(delta) < 1000


def test_lptim(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/lptim.bin")
    state0_periods, state1_periods = measure_frequency(1, "D0")
    num_periods = min(len(state0_periods), len(state1_periods))
    periods = [state0_periods[i] + state1_periods[i] for i in range(num_periods)]
    freqs = list(map(lambda x: 1 / x, periods))
    assert (
        periods
    ), "No LPTIM state changes detected, is the right analyzer being used? Is the device connected?"

    min_f = min(freqs)
    max_f = max(freqs)
    avg_f = sum(freqs) / len(freqs)
    print("min_f:{}, max_f:{}, avg_f:{}".format(min_f, max_f, avg_f))
    assert abs(avg_f - 50) < 0.25
    assert min_f > 49
    assert max_f < 51


def test_app_lowpower(context_factory, logger):
    serial_dev, jlink = context_factory("Application/main.bin", leave_halted=True)
    jlink.reset(halt=False)
    state0_periods, state1_periods = measure_frequency(10, "D0")
    num_periods = min(len(state0_periods), len(state1_periods))
    periods = [state0_periods[i] + state1_periods[i] for i in range(num_periods)]
    freqs = list(map(lambda x: 1 / x, periods))
    assert (
        periods
    ), "No debug pin state changes detected, is the right analyzer being used? Is the device connected?"

    min_f = min(freqs)
    max_f = max(freqs)
    avg_f = sum(freqs) / len(freqs)
    pct_sleep = sum(state1_periods) * 100 / sum(state0_periods + state1_periods)
    print(
        "min_f:{}, max_f:{}, avg_f:{}, pct_sleep:{}".format(
            min_f, max_f, avg_f, pct_sleep
        )
    )
    assert len(periods) >= 5
    assert pct_sleep > 99.95, "Spent too much time awake"


def test_stop(context_factory, logger):
    serial_dev, jlink = context_factory("Test/Apps/stop.bin")
    serial_dev.timeout = 70

    pattern = re.compile("Requested=(\\d*) Actual=(\\d*)")
    while True:
        line = serial_dev.readline()
        if line == TEST_PASS_TEXT:
            break

        line = line.decode("ascii", errors="ignore")
        print(line.strip())
        match = pattern.match(line)
        assert match
        req = int(match.group(1))
        actual = int(match.group(2))
        delta = req - actual

        if req < 32000:
            assert abs(delta < req * (2.0 / 100.0)) or (delta <= 1)
        else:
            # Delays > 32sec have reduced resolution (1 sec)
            assert abs(delta) < 1000


def main():
    pytest.main(sys.argv)


if __name__ == "__main__":
    main()