#include "stm32h7xx_hal.h"
#include "lwip/netif.h"

#define ETH_RX_DESC_CNT   4U
#define ETH_TX_DESC_CNT   4U
#define PHY_ADDR          0x00U

ETH_DMADescTypeDef dma_rx[ETH_RX_DESC_CNT] __attribute__((section(".RxDecripSection")));
ETH_DMADescTypeDef dma_tx[ETH_TX_DESC_CNT] __attribute__((section(".TxDecripSection")));

static ETH_HandleTypeDef heth;
static ETH_TxPacketConfig tx_config;

HAL_StatusTypeDef eth_init(const uint8_t mac[6])
{
    heth.Instance = ETH;
    heth.Init.MACAddr = (uint8_t *)mac;
    heth.Init.MediaInterface = HAL_ETH_RMII_MODE;
    heth.Init.RxDesc = dma_rx;
    heth.Init.TxDesc = dma_tx;
    heth.Init.RxBuffLen = 1536;

    if (HAL_ETH_Init(&heth) != HAL_OK) {
        return HAL_ERROR;
    }

    tx_config.Attributes = ETH_TX_PACKETS_FEATURES_CSUM | ETH_TX_PACKETS_FEATURES_CRCPAD;
    tx_config.ChecksumCtrl = ETH_CHECKSUM_IPHDR_PAYLOAD_INSERT_PHDR_CALC;
    tx_config.CRCPadCtrl = ETH_CRC_PAD_INSERT;

    uint32_t bsr = 0;
    HAL_ETH_ReadPHYRegister(&heth, PHY_ADDR, PHY_BSR, &bsr);
    if ((bsr & PHY_LINKED_STATUS) == 0U) {
        return HAL_TIMEOUT;
    }
    return HAL_ETH_Start_IT(&heth);
}

void ETH_IRQHandler(void)
{
    HAL_ETH_IRQHandler(&heth);
}

void HAL_ETH_RxCpltCallback(ETH_HandleTypeDef *h)
{
    BaseType_t woken = pdFALSE;
    vTaskNotifyGiveFromISR(rx_task, &woken);
    portYIELD_FROM_ISR(woken);
}

"""Production test bench: flash, power, Ethernet link and throughput."""
import socket
import time

import pytest

from bench import Board, PowerSupply, Programmer

LIMITS = {"vdd_3v3": (3.25, 3.35), "idle_ma": (40, 95)}


@pytest.fixture(scope="module")
def board():
    psu = PowerSupply("/dev/ttyUSB0")
    psu.set(voltage=5.0, current_limit=0.5)
    with Board(psu, Programmer("stlink")) as dut:
        dut.flash("build/firmware.elf", verify=True)
        dut.reset()
        yield dut


@pytest.mark.parametrize("rail", ["vdd_3v3"])
def test_power_rails(board, rail):
    low, high = LIMITS[rail]
    assert low <= board.measure(rail) <= high


def test_idle_current(board):
    low, high = LIMITS["idle_ma"]
    assert low <= board.psu.current_ma(samples=20) <= high


def test_ethernet_link(board):
    deadline = time.monotonic() + 5
    while not board.link_up():
        assert time.monotonic() < deadline, "no PHY link after 5 s"
        time.sleep(0.1)


def test_udp_echo_throughput(board):
    payload = bytes(range(256)) * 4
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(0.5)
        start = time.perf_counter()
        for _ in range(1000):
            sock.sendto(payload, (board.ip, 7))
            assert sock.recv(2048) == payload
        mbps = len(payload) * 8 * 1000 / (time.perf_counter() - start) / 1e6
    assert mbps > 20

#include "stm32h7xx_hal.h"
#include "lwip/netif.h"

#define ETH_RX_DESC_CNT   4U
#define ETH_TX_DESC_CNT   4U
#define PHY_ADDR          0x00U

ETH_DMADescTypeDef dma_rx[ETH_RX_DESC_CNT] __attribute__((section(".RxDecripSection")));
ETH_DMADescTypeDef dma_tx[ETH_TX_DESC_CNT] __attribute__((section(".TxDecripSection")));

static ETH_HandleTypeDef heth;
static ETH_TxPacketConfig tx_config;

HAL_StatusTypeDef eth_init(const uint8_t mac[6])
{
    heth.Instance = ETH;
    heth.Init.MACAddr = (uint8_t *)mac;
    heth.Init.MediaInterface = HAL_ETH_RMII_MODE;
    heth.Init.RxDesc = dma_rx;
    heth.Init.TxDesc = dma_tx;
    heth.Init.RxBuffLen = 1536;

    if (HAL_ETH_Init(&heth) != HAL_OK) {
        return HAL_ERROR;
    }

    tx_config.Attributes = ETH_TX_PACKETS_FEATURES_CSUM | ETH_TX_PACKETS_FEATURES_CRCPAD;
    tx_config.ChecksumCtrl = ETH_CHECKSUM_IPHDR_PAYLOAD_INSERT_PHDR_CALC;
    tx_config.CRCPadCtrl = ETH_CRC_PAD_INSERT;

    uint32_t bsr = 0;
    HAL_ETH_ReadPHYRegister(&heth, PHY_ADDR, PHY_BSR, &bsr);
    if ((bsr & PHY_LINKED_STATUS) == 0U) {
        return HAL_TIMEOUT;
    }
    return HAL_ETH_Start_IT(&heth);
}

void ETH_IRQHandler(void)
{
    HAL_ETH_IRQHandler(&heth);
}

void HAL_ETH_RxCpltCallback(ETH_HandleTypeDef *h)
{
    BaseType_t woken = pdFALSE;
    vTaskNotifyGiveFromISR(rx_task, &woken);
    portYIELD_FROM_ISR(woken);
}

"""Production test bench: flash, power, Ethernet link and throughput."""
import socket
import time

import pytest

from bench import Board, PowerSupply, Programmer

LIMITS = {"vdd_3v3": (3.25, 3.35), "idle_ma": (40, 95)}


@pytest.fixture(scope="module")
def board():
    psu = PowerSupply("/dev/ttyUSB0")
    psu.set(voltage=5.0, current_limit=0.5)
    with Board(psu, Programmer("stlink")) as dut:
        dut.flash("build/firmware.elf", verify=True)
        dut.reset()
        yield dut


@pytest.mark.parametrize("rail", ["vdd_3v3"])
def test_power_rails(board, rail):
    low, high = LIMITS[rail]
    assert low <= board.measure(rail) <= high


def test_idle_current(board):
    low, high = LIMITS["idle_ma"]
    assert low <= board.psu.current_ma(samples=20) <= high


def test_ethernet_link(board):
    deadline = time.monotonic() + 5
    while not board.link_up():
        assert time.monotonic() < deadline, "no PHY link after 5 s"
        time.sleep(0.1)


def test_udp_echo_throughput(board):
    payload = bytes(range(256)) * 4
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(0.5)
        start = time.perf_counter()
        for _ in range(1000):
            sock.sendto(payload, (board.ip, 7))
            assert sock.recv(2048) == payload
        mbps = len(payload) * 8 * 1000 / (time.perf_counter() - start) / 1e6
    assert mbps > 20

"""Triage failing test reports with an LLM, then open a ticket per root cause."""
from dataclasses import dataclass
from pathlib import Path

import anthropic

client = anthropic.Anthropic()


@dataclass
class Failure:
    test: str
    log: str


def load_failures(report: Path) -> list[Failure]:
    failures = []
    for case in parse_junit(report):
        if case.failed:
            failures.append(Failure(case.name, case.output[-4000:]))
    return failures


def triage(failures: list[Failure]) -> dict[str, list[str]]:
    prompt = "Group these failures by probable root cause:\n\n" + "\n---\n".join(
        f"{f.test}\n{f.log}" for f in failures
    )
    answer = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        tools=[ROOT_CAUSE_SCHEMA],
        messages=[{"role": "user", "content": prompt}],
    )
    return extract_groups(answer)


def main() -> None:
    failures = load_failures(Path("reports/nightly.xml"))
    for cause, tests in triage(failures).items():
        tracker.open_issue(title=cause, body="\n".join(tests), labels=["nightly"])
    print(f"{len(failures)} failures triaged")


if __name__ == "__main__":
    main()

/* Lock-free single-producer, single-consumer ring buffer. */
#include <stdint.h>
#include <stdbool.h>

#define RB_SIZE 256U  /* power of two */

typedef struct {
    volatile uint32_t head;
    volatile uint32_t tail;
    uint8_t data[RB_SIZE];
} ring_t;

static inline bool rb_push(ring_t *rb, uint8_t byte)
{
    uint32_t next = (rb->head + 1U) & (RB_SIZE - 1U);
    if (next == rb->tail) {
        return false;            /* full: drop, never block in an ISR */
    }
    rb->data[rb->head] = byte;
    __DMB();
    rb->head = next;
    return true;
}

static inline bool rb_pop(ring_t *rb, uint8_t *byte)
{
    if (rb->tail == rb->head) {
        return false;
    }
    *byte = rb->data[rb->tail];
    __DMB();
    rb->tail = (rb->tail + 1U) & (RB_SIZE - 1U);
    return true;
}

void USART3_IRQHandler(void)
{
    if (USART3->ISR & USART_ISR_RXNE_RXFNE) {
        rb_push(&uart_rx, (uint8_t)USART3->RDR);
    }
    if (USART3->ISR & USART_ISR_ORE) {
        USART3->ICR = USART_ICR_ORECF;
        stats.overruns++;
    }
}

"""Triage failing test reports with an LLM, then open a ticket per root cause."""
from dataclasses import dataclass
from pathlib import Path

import anthropic

client = anthropic.Anthropic()


@dataclass
class Failure:
    test: str
    log: str


def load_failures(report: Path) -> list[Failure]:
    failures = []
    for case in parse_junit(report):
        if case.failed:
            failures.append(Failure(case.name, case.output[-4000:]))
    return failures


def triage(failures: list[Failure]) -> dict[str, list[str]]:
    prompt = "Group these failures by probable root cause:\n\n" + "\n---\n".join(
        f"{f.test}\n{f.log}" for f in failures
    )
    answer = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        tools=[ROOT_CAUSE_SCHEMA],
        messages=[{"role": "user", "content": prompt}],
    )
    return extract_groups(answer)


def main() -> None:
    failures = load_failures(Path("reports/nightly.xml"))
    for cause, tests in triage(failures).items():
        tracker.open_issue(title=cause, body="\n".join(tests), labels=["nightly"])
    print(f"{len(failures)} failures triaged")


if __name__ == "__main__":
    main()

/* Lock-free single-producer, single-consumer ring buffer. */
#include <stdint.h>
#include <stdbool.h>

#define RB_SIZE 256U  /* power of two */

typedef struct {
    volatile uint32_t head;
    volatile uint32_t tail;
    uint8_t data[RB_SIZE];
} ring_t;

static inline bool rb_push(ring_t *rb, uint8_t byte)
{
    uint32_t next = (rb->head + 1U) & (RB_SIZE - 1U);
    if (next == rb->tail) {
        return false;            /* full: drop, never block in an ISR */
    }
    rb->data[rb->head] = byte;
    __DMB();
    rb->head = next;
    return true;
}

static inline bool rb_pop(ring_t *rb, uint8_t *byte)
{
    if (rb->tail == rb->head) {
        return false;
    }
    *byte = rb->data[rb->tail];
    __DMB();
    rb->tail = (rb->tail + 1U) & (RB_SIZE - 1U);
    return true;
}

void USART3_IRQHandler(void)
{
    if (USART3->ISR & USART_ISR_RXNE_RXFNE) {
        rb_push(&uart_rx, (uint8_t)USART3->RDR);
    }
    if (USART3->ISR & USART_ISR_ORE) {
        USART3->ICR = USART_ICR_ORECF;
        stats.overruns++;
    }
}

/* Lock-free single-producer, single-consumer ring buffer. */
#include <stdint.h>
#include <stdbool.h>

#define RB_SIZE 256U  /* power of two */

typedef struct {
    volatile uint32_t head;
    volatile uint32_t tail;
    uint8_t data[RB_SIZE];
} ring_t;

static inline bool rb_push(ring_t *rb, uint8_t byte)
{
    uint32_t next = (rb->head + 1U) & (RB_SIZE - 1U);
    if (next == rb->tail) {
        return false;            /* full: drop, never block in an ISR */
    }
    rb->data[rb->head] = byte;
    __DMB();
    rb->head = next;
    return true;
}

static inline bool rb_pop(ring_t *rb, uint8_t *byte)
{
    if (rb->tail == rb->head) {
        return false;
    }
    *byte = rb->data[rb->tail];
    __DMB();
    rb->tail = (rb->tail + 1U) & (RB_SIZE - 1U);
    return true;
}

void USART3_IRQHandler(void)
{
    if (USART3->ISR & USART_ISR_RXNE_RXFNE) {
        rb_push(&uart_rx, (uint8_t)USART3->RDR);
    }
    if (USART3->ISR & USART_ISR_ORE) {
        USART3->ICR = USART_ICR_ORECF;
        stats.overruns++;
    }
}

#include "stm32h7xx_hal.h"
#include "lwip/netif.h"

#define ETH_RX_DESC_CNT   4U
#define ETH_TX_DESC_CNT   4U
#define PHY_ADDR          0x00U

ETH_DMADescTypeDef dma_rx[ETH_RX_DESC_CNT] __attribute__((section(".RxDecripSection")));
ETH_DMADescTypeDef dma_tx[ETH_TX_DESC_CNT] __attribute__((section(".TxDecripSection")));

static ETH_HandleTypeDef heth;
static ETH_TxPacketConfig tx_config;

HAL_StatusTypeDef eth_init(const uint8_t mac[6])
{
    heth.Instance = ETH;
    heth.Init.MACAddr = (uint8_t *)mac;
    heth.Init.MediaInterface = HAL_ETH_RMII_MODE;
    heth.Init.RxDesc = dma_rx;
    heth.Init.TxDesc = dma_tx;
    heth.Init.RxBuffLen = 1536;

    if (HAL_ETH_Init(&heth) != HAL_OK) {
        return HAL_ERROR;
    }

    tx_config.Attributes = ETH_TX_PACKETS_FEATURES_CSUM | ETH_TX_PACKETS_FEATURES_CRCPAD;
    tx_config.ChecksumCtrl = ETH_CHECKSUM_IPHDR_PAYLOAD_INSERT_PHDR_CALC;
    tx_config.CRCPadCtrl = ETH_CRC_PAD_INSERT;

    uint32_t bsr = 0;
    HAL_ETH_ReadPHYRegister(&heth, PHY_ADDR, PHY_BSR, &bsr);
    if ((bsr & PHY_LINKED_STATUS) == 0U) {
        return HAL_TIMEOUT;
    }
    return HAL_ETH_Start_IT(&heth);
}

void ETH_IRQHandler(void)
{
    HAL_ETH_IRQHandler(&heth);
}

void HAL_ETH_RxCpltCallback(ETH_HandleTypeDef *h)
{
    BaseType_t woken = pdFALSE;
    vTaskNotifyGiveFromISR(rx_task, &woken);
    portYIELD_FROM_ISR(woken);
}

/* Lock-free single-producer, single-consumer ring buffer. */
#include <stdint.h>
#include <stdbool.h>

#define RB_SIZE 256U  /* power of two */

typedef struct {
    volatile uint32_t head;
    volatile uint32_t tail;
    uint8_t data[RB_SIZE];
} ring_t;

static inline bool rb_push(ring_t *rb, uint8_t byte)
{
    uint32_t next = (rb->head + 1U) & (RB_SIZE - 1U);
    if (next == rb->tail) {
        return false;            /* full: drop, never block in an ISR */
    }
    rb->data[rb->head] = byte;
    __DMB();
    rb->head = next;
    return true;
}

static inline bool rb_pop(ring_t *rb, uint8_t *byte)
{
    if (rb->tail == rb->head) {
        return false;
    }
    *byte = rb->data[rb->tail];
    __DMB();
    rb->tail = (rb->tail + 1U) & (RB_SIZE - 1U);
    return true;
}

void USART3_IRQHandler(void)
{
    if (USART3->ISR & USART_ISR_RXNE_RXFNE) {
        rb_push(&uart_rx, (uint8_t)USART3->RDR);
    }
    if (USART3->ISR & USART_ISR_ORE) {
        USART3->ICR = USART_ICR_ORECF;
        stats.overruns++;
    }
}

#include "stm32h7xx_hal.h"
#include "lwip/netif.h"

#define ETH_RX_DESC_CNT   4U
#define ETH_TX_DESC_CNT   4U
#define PHY_ADDR          0x00U

ETH_DMADescTypeDef dma_rx[ETH_RX_DESC_CNT] __attribute__((section(".RxDecripSection")));
ETH_DMADescTypeDef dma_tx[ETH_TX_DESC_CNT] __attribute__((section(".TxDecripSection")));

static ETH_HandleTypeDef heth;
static ETH_TxPacketConfig tx_config;

HAL_StatusTypeDef eth_init(const uint8_t mac[6])
{
    heth.Instance = ETH;
    heth.Init.MACAddr = (uint8_t *)mac;
    heth.Init.MediaInterface = HAL_ETH_RMII_MODE;
    heth.Init.RxDesc = dma_rx;
    heth.Init.TxDesc = dma_tx;
    heth.Init.RxBuffLen = 1536;

    if (HAL_ETH_Init(&heth) != HAL_OK) {
        return HAL_ERROR;
    }

    tx_config.Attributes = ETH_TX_PACKETS_FEATURES_CSUM | ETH_TX_PACKETS_FEATURES_CRCPAD;
    tx_config.ChecksumCtrl = ETH_CHECKSUM_IPHDR_PAYLOAD_INSERT_PHDR_CALC;
    tx_config.CRCPadCtrl = ETH_CRC_PAD_INSERT;

    uint32_t bsr = 0;
    HAL_ETH_ReadPHYRegister(&heth, PHY_ADDR, PHY_BSR, &bsr);
    if ((bsr & PHY_LINKED_STATUS) == 0U) {
        return HAL_TIMEOUT;
    }
    return HAL_ETH_Start_IT(&heth);
}

void ETH_IRQHandler(void)
{
    HAL_ETH_IRQHandler(&heth);
}

void HAL_ETH_RxCpltCallback(ETH_HandleTypeDef *h)
{
    BaseType_t woken = pdFALSE;
    vTaskNotifyGiveFromISR(rx_task, &woken);
    portYIELD_FROM_ISR(woken);
}

"""Production test bench: flash, power, Ethernet link and throughput."""
import socket
import time

import pytest

from bench import Board, PowerSupply, Programmer

LIMITS = {"vdd_3v3": (3.25, 3.35), "idle_ma": (40, 95)}


@pytest.fixture(scope="module")
def board():
    psu = PowerSupply("/dev/ttyUSB0")
    psu.set(voltage=5.0, current_limit=0.5)
    with Board(psu, Programmer("stlink")) as dut:
        dut.flash("build/firmware.elf", verify=True)
        dut.reset()
        yield dut


@pytest.mark.parametrize("rail", ["vdd_3v3"])
def test_power_rails(board, rail):
    low, high = LIMITS[rail]
    assert low <= board.measure(rail) <= high


def test_idle_current(board):
    low, high = LIMITS["idle_ma"]
    assert low <= board.psu.current_ma(samples=20) <= high


def test_ethernet_link(board):
    deadline = time.monotonic() + 5
    while not board.link_up():
        assert time.monotonic() < deadline, "no PHY link after 5 s"
        time.sleep(0.1)


def test_udp_echo_throughput(board):
    payload = bytes(range(256)) * 4
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(0.5)
        start = time.perf_counter()
        for _ in range(1000):
            sock.sendto(payload, (board.ip, 7))
            assert sock.recv(2048) == payload
        mbps = len(payload) * 8 * 1000 / (time.perf_counter() - start) / 1e6
    assert mbps > 20

"""Triage failing test reports with an LLM, then open a ticket per root cause."""
from dataclasses import dataclass
from pathlib import Path

import anthropic

client = anthropic.Anthropic()


@dataclass
class Failure:
    test: str
    log: str


def load_failures(report: Path) -> list[Failure]:
    failures = []
    for case in parse_junit(report):
        if case.failed:
            failures.append(Failure(case.name, case.output[-4000:]))
    return failures


def triage(failures: list[Failure]) -> dict[str, list[str]]:
    prompt = "Group these failures by probable root cause:\n\n" + "\n---\n".join(
        f"{f.test}\n{f.log}" for f in failures
    )
    answer = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        tools=[ROOT_CAUSE_SCHEMA],
        messages=[{"role": "user", "content": prompt}],
    )
    return extract_groups(answer)


def main() -> None:
    failures = load_failures(Path("reports/nightly.xml"))
    for cause, tests in triage(failures).items():
        tracker.open_issue(title=cause, body="\n".join(tests), labels=["nightly"])
    print(f"{len(failures)} failures triaged")


if __name__ == "__main__":
    main()

"""Production test bench: flash, power, Ethernet link and throughput."""
import socket
import time

import pytest

from bench import Board, PowerSupply, Programmer

LIMITS = {"vdd_3v3": (3.25, 3.35), "idle_ma": (40, 95)}


@pytest.fixture(scope="module")
def board():
    psu = PowerSupply("/dev/ttyUSB0")
    psu.set(voltage=5.0, current_limit=0.5)
    with Board(psu, Programmer("stlink")) as dut:
        dut.flash("build/firmware.elf", verify=True)
        dut.reset()
        yield dut


@pytest.mark.parametrize("rail", ["vdd_3v3"])
def test_power_rails(board, rail):
    low, high = LIMITS[rail]
    assert low <= board.measure(rail) <= high


def test_idle_current(board):
    low, high = LIMITS["idle_ma"]
    assert low <= board.psu.current_ma(samples=20) <= high


def test_ethernet_link(board):
    deadline = time.monotonic() + 5
    while not board.link_up():
        assert time.monotonic() < deadline, "no PHY link after 5 s"
        time.sleep(0.1)


def test_udp_echo_throughput(board):
    payload = bytes(range(256)) * 4
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(0.5)
        start = time.perf_counter()
        for _ in range(1000):
            sock.sendto(payload, (board.ip, 7))
            assert sock.recv(2048) == payload
        mbps = len(payload) * 8 * 1000 / (time.perf_counter() - start) / 1e6
    assert mbps > 20

"""Triage failing test reports with an LLM, then open a ticket per root cause."""
from dataclasses import dataclass
from pathlib import Path

import anthropic

client = anthropic.Anthropic()


@dataclass
class Failure:
    test: str
    log: str


def load_failures(report: Path) -> list[Failure]:
    failures = []
    for case in parse_junit(report):
        if case.failed:
            failures.append(Failure(case.name, case.output[-4000:]))
    return failures


def triage(failures: list[Failure]) -> dict[str, list[str]]:
    prompt = "Group these failures by probable root cause:\n\n" + "\n---\n".join(
        f"{f.test}\n{f.log}" for f in failures
    )
    answer = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        tools=[ROOT_CAUSE_SCHEMA],
        messages=[{"role": "user", "content": prompt}],
    )
    return extract_groups(answer)


def main() -> None:
    failures = load_failures(Path("reports/nightly.xml"))
    for cause, tests in triage(failures).items():
        tracker.open_issue(title=cause, body="\n".join(tests), labels=["nightly"])
    print(f"{len(failures)} failures triaged")


if __name__ == "__main__":
    main()