Add Owon power supply library, and test cases

This commit is contained in:
Hosam-Eldin.mostafa 2025-10-24 23:24:54 +02:00
parent b988cdaae5
commit e552e9a8e9
28 changed files with 1010 additions and 33 deletions

View File

@ -15,6 +15,8 @@ Python-based ECU testing framework built on pytest, with a pluggable LIN communi
- Using the framework (common runs, markers, CI, Pi): `docs/12_using_the_framework.md` - Using the framework (common runs, markers, CI, Pi): `docs/12_using_the_framework.md`
- Plugin overview (reporting, hooks, artifacts): `docs/11_conftest_plugin_overview.md` - Plugin overview (reporting, hooks, artifacts): `docs/11_conftest_plugin_overview.md`
- Power supply (Owon) usage and troubleshooting: `docs/14_power_supply.md`
- Report properties cheatsheet (standard keys): `docs/15_report_properties_cheatsheet.md`
## TL;DR quick start (copy/paste) ## TL;DR quick start (copy/paste)
@ -203,6 +205,64 @@ The `ecu_framework/lin/babylin.py` implementation uses the official `BabyLIN_lib
- Permission errors in PowerShell: run the venv's full Python path or adjust ExecutionPolicy for scripts. - Permission errors in PowerShell: run the venv's full Python path or adjust ExecutionPolicy for scripts.
- Import errors: activate the venv and reinstall `requirements.txt`. - Import errors: activate the venv and reinstall `requirements.txt`.
## Owon Power Supply (SCPI) — library, config, tests, and tryout
We provide a reusable pyserial-based library, a hardware test integrated with the central config,
and a minimal tryout script.
- Library: `ecu_framework/power/owon_psu.py` (class `OwonPSU`, `SerialParams`, `scan_ports`)
- Central config: `config/test_config.yaml` (`power_supply` section)
- Optionally merge `config/owon_psu.yaml` or set `OWON_PSU_CONFIG` to a YAML path
- Hardware test: `tests/hardware/test_owon_psu.py` (skips unless `power_supply.enabled` is true)
- Tryout: `vendor/Owon/tryout.py` (reads `OWON_PSU_CONFIG` or `config/owon_psu.yaml`)
Quick setup (Windows PowerShell):
```powershell
# Ensure dependencies
pip install -r .\requirements.txt
# Option A: configure centrally in test_config.yaml
# Edit config\test_config.yaml and set:
# power_supply.enabled: true
# power_supply.port: COM4
# Option B: use a separate machine-specific YAML
copy .\config\owon_psu.example.yaml .\config\owon_psu.yaml
# edit COM port and options in .\config\owon_psu.yaml
# Run the hardware PSU test (skips if disabled or missing port)
pytest -k test_owon_psu_idn_and_optional_set -m hardware -q
# Run the tryout script
python .\vendor\Owon\tryout.py
```
YAML keys supported by `power_supply`:
```yaml
power_supply:
enabled: true
port: COM4 # or /dev/ttyUSB0
baudrate: 115200
timeout: 1.0
eol: "\n" # or "\r\n"
parity: N # N|E|O
stopbits: 1 # 1|2
xonxoff: false
rtscts: false
dsrdtr: false
idn_substr: OWON
do_set: false
set_voltage: 5.0
set_current: 0.1
```
Troubleshooting:
- If `*IDN?` is empty, confirm port, parity/stopbits, and `eol` (try `\r\n`).
- On Windows, if COM>9, use `\\.\COM10` style in some tools; here plain `COM10` usually works.
- Ensure only one program opens the COM port at a time.
## Next steps ## Next steps
- Replace `HexFlasher` with a production flashing routine (UDS) - Replace `HexFlasher` with a production flashing routine (UDS)

View File

@ -304,6 +304,41 @@ The enhanced HTML report includes:
- YAML configuration loading: ✅ Working - YAML configuration loading: ✅ Working
- Environment variable override: ✅ Working - Environment variable override: ✅ Working
- BabyLIN SDF/schedule configuration: ✅ Working - BabyLIN SDF/schedule configuration: ✅ Working
- Power supply (PSU) configuration: ✅ Working (see `config/test_config.yaml``power_supply`)
## Owon Power Supply (PSU) Integration
The framework includes a serial SCPI controller for Owon PSUs and a hardware test wired to the central config.
- Library: `ecu_framework/power/owon_psu.py` (pyserial)
- Config: `config/test_config.yaml` (`power_supply` section)
- Optionally merge machine-specific settings from `config/owon_psu.yaml` or env `OWON_PSU_CONFIG`
- Hardware test: `tests/hardware/test_owon_psu.py` (skips unless `power_supply.enabled` and `port` present)
- Tryout: `vendor/Owon/tryout.py`
Quick run:
```powershell
pip install -r .\requirements.txt
copy .\config\owon_psu.example.yaml .\config\owon_psu.yaml
# edit COM port in .\config\owon_psu.yaml
pytest -k test_owon_psu_idn_and_optional_set -m hardware -q
python .\vendor\Owon\tryout.py
```
Common config keys:
```yaml
power_supply:
enabled: true
port: COM4
baudrate: 115200
timeout: 1.0
eol: "\n"
parity: N
stopbits: 1
idn_substr: OWON
```
## Next Steps ## Next Steps

View File

@ -0,0 +1,18 @@
# Example configuration for Owon PSU hardware test
# Copy to config/owon_psu.yaml and adjust values for your setup
port: COM4 # e.g., COM4 on Windows, /dev/ttyUSB0 on Linux
baudrate: 115200 # default 115200
timeout: 1.0 # seconds
# eol: "\n" # write/query line termination (default "\n"); use "\r\n" if required
# parity: N # N|E|O (default N)
# stopbits: 1 # 1 or 2 (default 1)
# xonxoff: false
# rtscts: false
# dsrdtr: false
# Optional assertions/behavior
# idn_substr: OWON # require this substring in *IDN?
# do_set: true # briefly set V/I and toggle output
# set_voltage: 1.0 # volts when do_set is true
# set_current: 0.1 # amps when do_set is true

18
config/owon_psu.yaml Normal file
View File

@ -0,0 +1,18 @@
# Example configuration for Owon PSU hardware test
# Copy to config/owon_psu.yaml and adjust values for your setup
port: COM4 # e.g., COM4 on Windows, /dev/ttyUSB0 on Linux
baudrate: 115200 # default 115200
timeout: 1.0 # seconds
eol: "\n" # write/query line termination (default "\n"); use "\r\n" if required
parity: N # N|E|O (default N)
stopbits: 1 # 1 or 2 (default 1)
xonxoff: false
rtscts: false
dsrdtr: false
# Optional assertions/behavior
idn_substr: OWON # require this substring in *IDN?
do_set: true # briefly set V/I and toggle output
set_voltage: 10.0 # volts when do_set is true
set_current: 0.1 # amps when do_set is true

View File

@ -5,3 +5,21 @@ interface:
flash: flash:
enabled: false enabled: false
hex_path: hex_path:
# Optional: central power supply config used by hardware tests/demos
# You can also place machine-specific values in config/owon_psu.yaml or set OWON_PSU_CONFIG
power_supply:
enabled: true
# port: COM4
baudrate: 115200
timeout: 1.0
eol: "\n"
parity: N
stopbits: 1
xonxoff: false
rtscts: false
dsrdtr: false
# idn_substr: OWON
do_set: false
set_voltage: 1.0
set_current: 0.1

27
conftest.py Normal file
View File

@ -0,0 +1,27 @@
"""
Pytest configuration for this repository.
Purpose:
- Optionally register the local plugin in `conftest_plugin.py` if present.
- Avoid hard failures on environments where that file isn't available.
"""
from __future__ import annotations
import importlib
import sys
from typing import Any
def pytest_configure(config: Any) -> None:
try:
plugin = importlib.import_module("conftest_plugin")
except Exception as e:
# Soft warning only; tests can still run without the extra report features.
sys.stderr.write(f"[pytest] conftest_plugin not loaded: {e}\n")
return
# Register the plugin module so its hooks are active.
try:
config.pluginmanager.register(plugin, name="conftest_plugin")
except Exception as reg_err:
sys.stderr.write(f"[pytest] failed to register conftest_plugin: {reg_err}\n")

View File

@ -27,6 +27,7 @@ sequenceDiagram
participant T as Test Discovery (tests/*) participant T as Test Discovery (tests/*)
participant F as Fixtures (conftest.py) participant F as Fixtures (conftest.py)
participant C as Config Loader (ecu_framework/config.py) participant C as Config Loader (ecu_framework/config.py)
participant PS as Power Supply (optional)
participant L as LIN Adapter (mock/BabyLIN SDK) participant L as LIN Adapter (mock/BabyLIN SDK)
participant X as HexFlasher (optional) participant X as HexFlasher (optional)
participant R as Reports (HTML/JUnit) participant R as Reports (HTML/JUnit)
@ -45,6 +46,9 @@ sequenceDiagram
F->>X: HexFlasher(lin).flash_hex(hex_path) F->>X: HexFlasher(lin).flash_hex(hex_path)
X-->>F: Flash result (ok/fail) X-->>F: Flash result (ok/fail)
end end
opt power_supply.enabled and port provided
Note over PS: Tests/tryouts may open PSU via ecu_framework.power.owon_psu
end
loop for each test loop for each test
P->>PL: runtest_makereport(item, call) P->>PL: runtest_makereport(item, call)
Note over PL: Parse docstring and attach metadata Note over PL: Parse docstring and attach metadata
@ -68,6 +72,7 @@ Session fixture: config()
→ calls ecu_framework.config.load_config(workspace_root) → calls ecu_framework.config.load_config(workspace_root)
→ determines config file path by precedence → determines config file path by precedence
→ merges YAML + overrides into dataclasses (EcuTestConfig) → merges YAML + overrides into dataclasses (EcuTestConfig)
→ optionally merges config/owon_psu.yaml (or OWON_PSU_CONFIG) into power_supply
Session fixture: lin(config) Session fixture: lin(config)
→ chooses interface by config.interface.type → chooses interface by config.interface.type

View File

@ -25,6 +25,15 @@ From highest to lowest precedence:
- `flash: FlashConfig` - `flash: FlashConfig`
- `enabled`: whether to flash before tests - `enabled`: whether to flash before tests
- `hex_path`: path to HEX file - `hex_path`: path to HEX file
- `power_supply: PowerSupplyConfig`
- `enabled`: whether PSU features/tests are active
- `port`: Serial device (e.g., `COM4`, `/dev/ttyUSB0`)
- `baudrate`, `timeout`, `eol`: line settings (e.g., `"\n"` or `"\r\n"`)
- `parity`: `N|E|O`
- `stopbits`: `1` or `2`
- `xonxoff`, `rtscts`, `dsrdtr`: flow control flags
- `idn_substr`: optional substring to assert in `*IDN?`
- `do_set`, `set_voltage`, `set_current`: optional demo/test actions
## YAML examples ## YAML examples
@ -52,6 +61,26 @@ interface:
flash: flash:
enabled: true enabled: true
hex_path: "firmware/ecu_firmware.hex" hex_path: "firmware/ecu_firmware.hex"
Power supply configuration (either inline or merged from a dedicated YAML):
```yaml
power_supply:
enabled: true
port: COM4 # or /dev/ttyUSB0 on Linux
baudrate: 115200
timeout: 1.0
eol: "\n" # or "\r\n" if your device requires CRLF
parity: N # N|E|O
stopbits: 1 # 1|2
xonxoff: false
rtscts: false
dsrdtr: false
idn_substr: OWON
do_set: false
set_voltage: 5.0
set_current: 0.1
```
``` ```
## Load flow ## Load flow
@ -64,6 +93,14 @@ tests/conftest.py: config() fixture
→ else use defaults → else use defaults
→ convert dicts to EcuTestConfig dataclasses → convert dicts to EcuTestConfig dataclasses
→ provide to other fixtures/tests → provide to other fixtures/tests
Additionally, if present, a dedicated PSU YAML is merged into `power_supply`:
- Environment variable `OWON_PSU_CONFIG` (path to YAML), else
- `config/owon_psu.yaml` under the workspace root
This lets you keep machine-specific serial settings separate while still having
central defaults in `config/test_config.yaml`.
``` ```
## How tests and adapters consume config ## How tests and adapters consume config
@ -72,6 +109,10 @@ tests/conftest.py: config() fixture
- Mock adapter uses `bitrate` and `channel` to simulate timing/behavior - Mock adapter uses `bitrate` and `channel` to simulate timing/behavior
- BabyLIN adapter (SDK wrapper) uses `sdf_path`, `schedule_nr`, `channel` to open the device, load the SDF, and start a schedule. `bitrate` is informational unless explicitly applied via commands/SDF. - BabyLIN adapter (SDK wrapper) uses `sdf_path`, `schedule_nr`, `channel` to open the device, load the SDF, and start a schedule. `bitrate` is informational unless explicitly applied via commands/SDF.
- `flash_ecu` uses `flash.enabled` and `flash.hex_path` - `flash_ecu` uses `flash.enabled` and `flash.hex_path`
- PSU-related tests or utilities read `config.power_supply` for serial parameters
and optional actions (IDN assertions, on/off toggle, set/measure). The reference
implementation is `ecu_framework/power/owon_psu.py`, with a hardware test in
`tests/hardware/test_owon_psu.py` and a tryout script in `vendor/Owon/tryout.py`.
## Tips ## Tips
@ -79,3 +120,5 @@ tests/conftest.py: config() fixture
- Check path validity for `sdf_path` and `hex_path` before running hardware tests - Check path validity for `sdf_path` and `hex_path` before running hardware tests
- Ensure `vendor/BabyLIN_library.py` and the platform-specific libraries from the SDK are available on `PYTHONPATH` - Ensure `vendor/BabyLIN_library.py` and the platform-specific libraries from the SDK are available on `PYTHONPATH`
- Use environment-specific YAML files for labs vs. CI - Use environment-specific YAML files for labs vs. CI
- For PSU, prefer `OWON_PSU_CONFIG` or `config/owon_psu.yaml` to avoid committing
local COM port settings. Central defaults can live in `config/test_config.yaml`.

View File

@ -85,3 +85,25 @@ Declared in `pytest.ini` and used via `@pytest.mark.<name>` in tests. They also
- Add more columns to HTML by updating `pytest_html_results_table_header/row` - Add more columns to HTML by updating `pytest_html_results_table_header/row`
- Persist full metadata (steps, expected) to a JSON file after the run for audit trails - Persist full metadata (steps, expected) to a JSON file after the run for audit trails
- Populate requirement coverage map by scanning markers and aggregating results - Populate requirement coverage map by scanning markers and aggregating results
## Runtime properties (record_property) and the `rp` helper fixture
Beyond static docstrings, you can attach dynamic key/value properties during a test.
- Built-in: `record_property("key", value)` in any test
- Convenience: use the shared `rp` fixture which wraps `record_property` and also prints a short line to captured output for quick scanning.
Example usage:
```python
def test_example(rp):
rp("device", "mock")
rp("tx_id", "0x12")
rp("rx_present", True)
```
Where they show up:
- HTML report: expand a test row to see a Properties table listing all recorded key/value pairs
- Captured output: look for lines like `[prop] key=value` emitted by the `rp` helper
Suggested standardized keys across suites live in `docs/15_report_properties_cheatsheet.md`.

View File

@ -142,6 +142,8 @@ Expected Result:
""" """
``` ```
Tip: For runtime properties in reports, prefer the shared `rp` fixture (wrapper around `record_property`) and use standardized keys from `docs/15_report_properties_cheatsheet.md`.
## Continuous Integration (CI) ## Continuous Integration (CI)
- Run `pytest` with your preferred markers in your pipeline. - Run `pytest` with your preferred markers in your pipeline.
@ -170,3 +172,17 @@ Running tests headless via systemd typically involves:
- No BabyLIN devices found: check USB connection, drivers, and permissions. - No BabyLIN devices found: check USB connection, drivers, and permissions.
- Timeouts on receive: increase `timeout` or verify schedule activity and SDF correctness. - Timeouts on receive: increase `timeout` or verify schedule activity and SDF correctness.
- Missing reports: ensure `pytest.ini` includes the HTML/JUnit plugins and the custom plugin is loaded. - Missing reports: ensure `pytest.ini` includes the HTML/JUnit plugins and the custom plugin is loaded.
## Power supply (Owon) hardware test
Enable `power_supply` in your config and set the serial port, then run the dedicated test or the tryout script.
```powershell
copy .\config\owon_psu.example.yaml .\config\owon_psu.yaml
# edit COM port in .\config\owon_psu.yaml or set values in config\test_config.yaml
pytest -k test_owon_psu_idn_and_optional_set -m hardware -q
python .\vendor\Owon\tryout.py
```
See also: `docs/14_power_supply.md` for details and troubleshooting.

103
docs/14_power_supply.md Normal file
View File

@ -0,0 +1,103 @@
# Power Supply (Owon) — control, configuration, tests, and tryout
This guide covers using the Owon bench power supply via SCPI over serial with the framework.
- Library: `ecu_framework/power/owon_psu.py`
- Hardware test: `tests/hardware/test_owon_psu.py`
- Tryout script: `vendor/Owon/tryout.py`
- Configuration: `config/test_config.yaml` (`power_supply`), optionally merged from `config/owon_psu.yaml` or env `OWON_PSU_CONFIG`
## Install dependencies
```powershell
pip install -r .\requirements.txt
```
## Configure
You can keep PSU settings centrally or in a machine-specific YAML.
- Central: `config/test_config.yaml``power_supply` section
- Separate: `config/owon_psu.yaml` (or `OWON_PSU_CONFIG` env var)
Supported keys:
```yaml
power_supply:
enabled: true
port: COM4 # e.g., COM4 (Windows) or /dev/ttyUSB0 (Linux)
baudrate: 115200
timeout: 1.0
eol: "\n" # or "\r\n" if required
parity: N # N|E|O
stopbits: 1 # 1|2
xonxoff: false
rtscts: false
dsrdtr: false
idn_substr: OWON
do_set: false
set_voltage: 5.0
set_current: 0.1
```
The central config loader automatically merges `config/owon_psu.yaml` (or the path in `OWON_PSU_CONFIG`) into `power_supply`.
## Run the hardware test
Skips unless `power_supply.enabled` is true and `port` is set.
```powershell
pytest -k test_owon_psu_idn_and_optional_set -m hardware -q
```
What it does:
- Opens serial with your configured line params
- Queries `*IDN?` (checks `idn_substr` if provided)
- If `do_set` is true, sets voltage/current, enables output briefly, then disables
## Use the library programmatically
```python
from ecu_framework.power import OwonPSU, SerialParams
params = SerialParams(baudrate=115200, timeout=1.0)
with OwonPSU("COM4", params, eol="\n") as psu:
print(psu.idn())
psu.set_voltage(1, 5.0)
psu.set_current(1, 0.1)
psu.set_output(True)
# ... measure, etc.
psu.set_output(False)
```
Notes:
- Commands use newline-terminated writes; reads use `readline()`
- SCPI forms: `SOUR:VOLT`, `SOUR:CURR`, `MEAS:VOLT?`, `MEAS:CURR?`, `output 0/1`, `output?`
## Tryout script
The tryout reads `OWON_PSU_CONFIG` or `config/owon_psu.yaml` and performs a small sequence.
```powershell
python .\vendor\Owon\tryout.py
```
It also scans ports with `*IDN?` using `scan_ports()`.
## Troubleshooting
- Empty `*IDN?` or timeouts:
- Verify COM port and exclusivity (no other program holding it)
- Try `eol: "\r\n"`
- Adjust `parity` and `stopbits` per your device manual
- Windows COM > 9:
- Most Python code accepts `COM10` directly; if needed in other tools, use `\\.\\COM10`
- Flow control:
- Keep `xonxoff`, `rtscts`, `dsrdtr` false unless required
## Related files
- `ecu_framework/power/owon_psu.py` — PSU controller (pyserial)
- `tests/hardware/test_owon_psu.py` — Hardware test using central config
- `vendor/Owon/tryout.py` — Quick demo runner
- `config/owon_psu.example.yaml` — Example machine-specific YAML

View File

@ -0,0 +1,53 @@
# Report properties cheatsheet (record_property / rp)
Use these standardized keys when calling `record_property("key", value)` or the `rp("key", value)` helper.
This keeps reports consistent and easy to scan across suites.
## General
- test_phase: setup | call | teardown (if you want to distinguish)
- environment: local | ci | lab
- config_source: defaults | file | env | env+overrides (already used in unit tests)
## LIN (common)
- lin_type: mock | babylin
- tx_id: hex string or int (e.g., "0x12")
- tx_data: list of ints (bytes)
- rx_present: bool
- rx_id: hex string or int
- rx_data: list of ints
- timeout_s: float seconds
## BabyLIN specifics
- sdf_path: string
- schedule_nr: int
- receive_result: frame | timeout
- wrapper: mock_bl | _MockBytesOnly | real (for future)
## Mock-specific
- expected_data: list of ints
## Power supply (PSU)
- psu_idn: string from `*IDN?`
- output_status_before: bool
- output_status_after: bool
- set_voltage: float (V)
- set_current: float (A)
- measured_voltage: float (V)
- measured_current: float (A)
- psu_port: e.g., COM4 or /dev/ttyUSB0 (if helpful)
## Flashing
- hex_path: string
- sent_count: int (frames sent by stub/mock)
- flash_result: ok | fail (for future real flashing)
## Configuration highlights
- interface_type: mock | babylin
- interface_channel: int
- flash_enabled: bool
## Tips
- Prefer simple, lowercase snake_case keys
- Use lists for byte arrays so they render clearly in JSON and HTML
- Log both expected and actual when asserting patterns (e.g., deterministic responses)
- Keep units in the key name when helpful (voltage/current include V/A in the name)

View File

@ -16,8 +16,11 @@ A guided tour of the ECU testing framework. Start here:
11. `10_build_custom_image.md` — Build a custom Raspberry Pi OS image with the framework baked in 11. `10_build_custom_image.md` — Build a custom Raspberry Pi OS image with the framework baked in
12. `12_using_the_framework.md` — Practical usage: local, hardware, CI, and Pi 12. `12_using_the_framework.md` — Practical usage: local, hardware, CI, and Pi
13. `13_unit_testing_guide.md` — Unit tests layout, markers, coverage, and tips 13. `13_unit_testing_guide.md` — Unit tests layout, markers, coverage, and tips
14. `14_power_supply.md` — Owon PSU control, configuration, tests, and tryout script
15. `15_report_properties_cheatsheet.md` — Standardized keys for record_property/rp across suites
Related references: Related references:
- Root project guide: `../README.md` - Root project guide: `../README.md`
- Full framework guide: `../TESTING_FRAMEWORK_GUIDE.md` - Full framework guide: `../TESTING_FRAMEWORK_GUIDE.md`
- BabyLIN placement and integration: `../vendor/README.md` - BabyLIN placement and integration: `../vendor/README.md`
- PSU tryout and scripts: `../vendor/Owon/`

View File

@ -55,6 +55,39 @@ class EcuTestConfig:
interface: InterfaceConfig = field(default_factory=InterfaceConfig) interface: InterfaceConfig = field(default_factory=InterfaceConfig)
flash: FlashConfig = field(default_factory=FlashConfig) flash: FlashConfig = field(default_factory=FlashConfig)
# Serial power supply (e.g., Owon) configuration
# Test code can rely on these values to interact with PSU if enabled
power_supply: "PowerSupplyConfig" = field(default_factory=lambda: PowerSupplyConfig())
@dataclass
class PowerSupplyConfig:
"""Serial power supply configuration (e.g., Owon PSU).
enabled: Whether PSU tests/features should be active.
port: Serial device (e.g., COM4 on Windows, /dev/ttyUSB0 on Linux).
baudrate/timeout/eol: Basic line settings; eol often "\n" or "\r\n".
parity: One of "N", "E", "O".
stopbits: 1 or 2.
xonxoff/rtscts/dsrdtr: Flow control flags.
idn_substr: Optional substring to assert in *IDN? responses.
do_set/set_voltage/set_current: Optional demo/test actions.
"""
enabled: bool = False
port: Optional[str] = None
baudrate: int = 115200
timeout: float = 1.0
eol: str = "\n"
parity: str = "N"
stopbits: float = 1.0
xonxoff: bool = False
rtscts: bool = False
dsrdtr: bool = False
idn_substr: Optional[str] = None
do_set: bool = False
set_voltage: float = 1.0
set_current: float = 0.1
DEFAULT_CONFIG_RELATIVE = pathlib.Path("config") / "test_config.yaml" # Default config path relative to repo root DEFAULT_CONFIG_RELATIVE = pathlib.Path("config") / "test_config.yaml" # Default config path relative to repo root
@ -83,6 +116,7 @@ def _to_dataclass(cfg: Dict[str, Any]) -> EcuTestConfig:
""" """
iface = cfg.get("interface", {}) # Sub-config for interface iface = cfg.get("interface", {}) # Sub-config for interface
flash = cfg.get("flash", {}) # Sub-config for flashing flash = cfg.get("flash", {}) # Sub-config for flashing
psu = cfg.get("power_supply", {}) # Sub-config for power supply
return EcuTestConfig( return EcuTestConfig(
interface=InterfaceConfig( interface=InterfaceConfig(
type=str(iface.get("type", "mock")).lower(), # Normalize to lowercase type=str(iface.get("type", "mock")).lower(), # Normalize to lowercase
@ -98,6 +132,22 @@ def _to_dataclass(cfg: Dict[str, Any]) -> EcuTestConfig:
enabled=bool(flash.get("enabled", False)), # Coerce to bool enabled=bool(flash.get("enabled", False)), # Coerce to bool
hex_path=flash.get("hex_path"), # Optional hex path hex_path=flash.get("hex_path"), # Optional hex path
), ),
power_supply=PowerSupplyConfig(
enabled=bool(psu.get("enabled", False)),
port=psu.get("port"),
baudrate=int(psu.get("baudrate", 115200)),
timeout=float(psu.get("timeout", 1.0)),
eol=str(psu.get("eol", "\n")),
parity=str(psu.get("parity", "N")),
stopbits=float(psu.get("stopbits", 1.0)),
xonxoff=bool(psu.get("xonxoff", False)),
rtscts=bool(psu.get("rtscts", False)),
dsrdtr=bool(psu.get("dsrdtr", False)),
idn_substr=psu.get("idn_substr"),
do_set=bool(psu.get("do_set", False)),
set_voltage=float(psu.get("set_voltage", 1.0)),
set_current=float(psu.get("set_current", 0.1)),
),
) )
@ -121,6 +171,22 @@ def load_config(workspace_root: Optional[str] = None, overrides: Optional[Dict[s
"enabled": False, "enabled": False,
"hex_path": None, "hex_path": None,
}, },
"power_supply": {
"enabled": False,
"port": None,
"baudrate": 115200,
"timeout": 1.0,
"eol": "\n",
"parity": "N",
"stopbits": 1.0,
"xonxoff": False,
"rtscts": False,
"dsrdtr": False,
"idn_substr": None,
"do_set": False,
"set_voltage": 1.0,
"set_current": 0.1,
},
} }
cfg_path: Optional[pathlib.Path] = None # Resolved configuration file path cfg_path: Optional[pathlib.Path] = None # Resolved configuration file path
@ -145,6 +211,23 @@ def load_config(workspace_root: Optional[str] = None, overrides: Optional[Dict[s
if isinstance(file_cfg, dict): # Only merge dicts if isinstance(file_cfg, dict): # Only merge dicts
_deep_update(base, file_cfg) _deep_update(base, file_cfg)
# Optionally merge a dedicated PSU YAML if present (or env var path)
# This allows users to keep sensitive or machine-specific serial settings separate
psu_env = os.getenv("OWON_PSU_CONFIG")
psu_default = None
if workspace_root:
candidate = pathlib.Path(workspace_root) / "config" / "owon_psu.yaml"
if candidate.is_file():
psu_default = candidate
psu_path: Optional[pathlib.Path] = pathlib.Path(psu_env) if psu_env else psu_default
if psu_path and psu_path.is_file():
with open(psu_path, "r", encoding="utf-8") as f:
psu_cfg = yaml.safe_load(f) or {}
if isinstance(psu_cfg, dict):
base.setdefault("power_supply", {})
# Merge PSU YAML into power_supply section
base["power_supply"] = _deep_update(base["power_supply"], psu_cfg)
# 1) In-memory overrides always win # 1) In-memory overrides always win
if overrides: if overrides:
_deep_update(base, overrides) _deep_update(base, overrides)

View File

@ -0,0 +1,13 @@
"""Power control helpers for ECU tests.
Currently includes Owon PSU serial SCPI controller.
"""
from .owon_psu import SerialParams, OwonPSU, scan_ports, auto_detect
__all__ = [
"SerialParams",
"OwonPSU",
"scan_ports",
"auto_detect",
]

View File

@ -0,0 +1,193 @@
"""Owon PSU SCPI control over raw serial (pyserial).
This module provides a small, programmatic API suitable for tests:
- OwonPSU: context-manageable controller class
- scan_ports(): find devices responding to *IDN?
- auto_detect(): select the first matching device by IDN substring
Behavior follows the working tryout example (serial):
- Both commands and queries are terminated with a newline ("\n" by default).
- Queries use readline() to fetch a single-line response.
- Command set uses: 'output 0/1', 'output?', 'SOUR:VOLT <V>', 'SOUR:CURR <A>', 'MEAS:VOLT?', 'MEAS:CURR?', '*IDN?'
"""
from __future__ import annotations
from dataclasses import dataclass
from time import sleep
from typing import Iterable, Optional
import serial
from serial import Serial
from serial.tools import list_ports
@dataclass
class SerialParams:
baudrate: int = 115200
timeout: float = 1.0 # seconds
bytesize: int = serial.EIGHTBITS
parity: str = serial.PARITY_NONE
stopbits: float = serial.STOPBITS_ONE
xonxoff: bool = False
rtscts: bool = False
dsrdtr: bool = False
write_timeout: float = 1.0 # seconds
class OwonPSU:
def __init__(self, port: str, params: SerialParams | None = None, eol: str = "\n") -> None:
self.port = port
self.params = params or SerialParams()
self.eol = eol
self._ser: Optional[Serial] = None
def open(self) -> None:
if self._ser and self._ser.is_open:
return
ser = Serial()
ser.port = self.port
ser.baudrate = self.params.baudrate
ser.bytesize = self.params.bytesize
ser.parity = self.params.parity
ser.stopbits = self.params.stopbits
ser.xonxoff = self.params.xonxoff
ser.rtscts = self.params.rtscts
ser.dsrdtr = self.params.dsrdtr
ser.timeout = self.params.timeout
ser.write_timeout = self.params.write_timeout
ser.open()
self._ser = ser
def close(self) -> None:
if self._ser and self._ser.is_open:
try:
self._ser.close()
finally:
self._ser = None
def __enter__(self) -> "OwonPSU":
self.open()
return self
def __exit__(self, exc_type, exc, tb) -> None:
self.close()
@property
def is_open(self) -> bool:
return bool(self._ser and self._ser.is_open)
# ---- low-level ops ----
def write(self, cmd: str) -> None:
"""Write a SCPI command (append eol)."""
if not self._ser:
raise RuntimeError("Port is not open")
data = (cmd + self.eol).encode("ascii", errors="ignore")
self._ser.write(data)
self._ser.flush()
def query(self, q: str) -> str:
"""Send a query with terminator and return a single-line response using readline()."""
if not self._ser:
raise RuntimeError("Port is not open")
# clear buffers to avoid stale data
try:
self._ser.reset_input_buffer()
self._ser.reset_output_buffer()
except Exception:
pass
self._ser.write((q + self.eol).encode("ascii", errors="ignore"))
self._ser.flush()
line = self._ser.readline().strip()
return line.decode("ascii", errors="ignore")
# ---- high-level ops ----
def idn(self) -> str:
return self.query("*IDN?")
def set_voltage(self, channel: int, volts: float) -> None:
# Using SOUR:VOLT <V> per working example
self.write(f"SOUR:VOLT {volts:.3f}")
def set_current(self, channel: int, amps: float) -> None:
# Using SOUR:CURR <A> per working example
self.write(f"SOUR:CURR {amps:.3f}")
def set_output(self, on: bool) -> None:
# Using 'output 1/0' per working example
self.write("output 1" if on else "output 0")
def output_status(self) -> str:
return self.query("output?")
def measure_voltage(self) -> str:
return self.query("MEAS:VOLT?")
def measure_current(self) -> str:
return self.query("MEAS:CURR?")
# ------- discovery helpers -------
def try_idn_on_port(port: str, params: SerialParams) -> str:
dev: Optional[Serial] = None
try:
dev = Serial()
dev.port = port
dev.baudrate = params.baudrate
dev.bytesize = params.bytesize
dev.parity = params.parity
dev.stopbits = params.stopbits
dev.xonxoff = params.xonxoff
dev.rtscts = params.rtscts
dev.dsrdtr = params.dsrdtr
dev.timeout = params.timeout
dev.write_timeout = params.write_timeout
dev.open()
# Query with newline terminator and read a single line
dev.reset_input_buffer(); dev.reset_output_buffer()
dev.write(b"*IDN?\n"); dev.flush()
line = dev.readline().strip()
return line.decode("ascii", errors="ignore")
except Exception:
return ""
finally:
if dev and dev.is_open:
try:
dev.close()
except Exception:
pass
def scan_ports(params: SerialParams | None = None) -> list[tuple[str, str]]:
"""Return [(port, idn_response), ...] for ports that responded."""
params = params or SerialParams()
results: list[tuple[str, str]] = []
for p in list_ports.comports():
dev = p.device
resp = try_idn_on_port(dev, params)
if resp:
results.append((dev, resp))
return results
def auto_detect(params: SerialParams | None = None, idn_substr: str | None = None) -> Optional[str]:
"""Return the first port whose *IDN? contains idn_substr (case-insensitive), else first responder."""
params = params or SerialParams()
matches = scan_ports(params)
if not matches:
return None
if idn_substr:
isub = idn_substr.lower()
for port, idn in matches:
if isub in idn.lower():
return port
return matches[0][0]
__all__ = [
"SerialParams",
"OwonPSU",
"scan_ports",
"auto_detect",
]

View File

@ -5,11 +5,13 @@
# --html=... → Generate a human-friendly HTML report after each run. # --html=... → Generate a human-friendly HTML report after each run.
# --self-contained-html → Inline CSS/JS in the HTML report for easy sharing. # --self-contained-html → Inline CSS/JS in the HTML report for easy sharing.
# --tb=short → Short tracebacks to keep logs readable. # --tb=short → Short tracebacks to keep logs readable.
# -p conftest_plugin → Load our custom plugin (conftest_plugin.py) that: # Plugin note: We no longer force-load via `-p conftest_plugin` to avoid ImportError
# on environments where the file might be missing. Instead, `conftest.py` will
# register the plugin if present. The plugin:
# - extracts Title/Description/Requirements/Steps from test docstrings # - extracts Title/Description/Requirements/Steps from test docstrings
# - adds custom columns to the HTML report # - adds custom columns to the HTML report
# - writes requirements_coverage.json and summary.md in reports/ # - writes requirements_coverage.json and summary.md in reports/
addopts = -ra --junitxml=reports/junit.xml --html=reports/report.html --self-contained-html --tb=short -p conftest_plugin --cov=ecu_framework --cov-report=term-missing addopts = -ra --junitxml=reports/junit.xml --html=reports/report.html --self-contained-html --tb=short --cov=ecu_framework --cov-report=term-missing
# markers: Document all custom markers so pytest doesn't warn and so usage is clear. # markers: Document all custom markers so pytest doesn't warn and so usage is clear.
# Use with: pytest -m "markername" # Use with: pytest -m "markername"

View File

@ -1,6 +1,7 @@
# Core testing and utilities # Core testing and utilities
pytest>=8,<9 # Test runner and framework (parametrize, fixtures, markers) pytest>=8,<9 # Test runner and framework (parametrize, fixtures, markers)
pyyaml>=6,<7 # Parse YAML config files under ./config/ pyyaml>=6,<7 # Parse YAML config files under ./config/
pyserial>=3,<4 # Serial communication for Owon PSU and hardware tests
# BabyLIN SDK wrapper requires 'six' on some platforms # BabyLIN SDK wrapper requires 'six' on some platforms
six>=1.16,<2 six>=1.16,<2

View File

@ -62,3 +62,22 @@ def flash_ecu(config: EcuTestConfig, lin: LinInterface) -> None:
ok = flasher.flash_hex(config.flash.hex_path) ok = flasher.flash_hex(config.flash.hex_path)
if not ok: if not ok:
pytest.fail("ECU flashing failed") pytest.fail("ECU flashing failed")
@pytest.fixture
def rp(record_property: "pytest.RecordProperty"):
"""Convenience reporter: attaches a key/value as a test property and echoes to captured output.
Usage in tests:
def test_something(rp):
rp("key", value)
"""
def _rp(key: str, value):
# Attach property (pytest-html will show in Properties table)
record_property(str(key), value)
# Echo to captured output for quick scanning in report details
try:
print(f"[prop] {key}={value}")
except Exception:
pass
return _rp

View File

@ -0,0 +1,102 @@
import time
import pytest
import serial
from ecu_framework.power import OwonPSU, SerialParams
from ecu_framework.config import EcuTestConfig
pytestmark = [pytest.mark.hardware]
def test_owon_psu_idn_and_optional_set(config: EcuTestConfig, rp):
"""
Title: Owon PSU - IDN, Output Status, Set/Measure Verification
Description:
Validates serial SCPI control of an Owon PSU: IDN retrieval, output status query,
and optional set/measure cycle using values from central configuration.
Test Steps:
1. Load PSU config from EcuTestConfig.power_supply
2. Open serial connection and query *IDN?
3. Query output status (output?) and record initial state
4. If configured, set voltage/current, enable output briefly, measure V/I, then disable output
5. Record IDN, output status before/after, set values, and measured values in the report
Expected Result:
*IDN? returns a non-empty string (containing idn_substr if configured), serial operations succeed,
and, when enabled, the output toggles on then off with measurements returned.
"""
psu_cfg = config.power_supply
if not psu_cfg.enabled:
pytest.skip("Power supply tests disabled in config.power_supply.enabled")
if not psu_cfg.port:
pytest.skip("No power supply 'port' configured (config.power_supply.port)")
# Serial params (with sensible defaults via central config)
baud = int(psu_cfg.baudrate)
timeout = float(psu_cfg.timeout)
parity = psu_cfg.parity or "N"
stopbits = psu_cfg.stopbits or 1
xonxoff = bool(psu_cfg.xonxoff)
rtscts = bool(psu_cfg.rtscts)
dsrdtr = bool(psu_cfg.dsrdtr)
eol = psu_cfg.eol or "\n"
ps = SerialParams(
baudrate=baud,
timeout=timeout,
parity={"N": serial.PARITY_NONE, "E": serial.PARITY_EVEN, "O": serial.PARITY_ODD}.get(str(parity).upper(), serial.PARITY_NONE),
stopbits={1: serial.STOPBITS_ONE, 2: serial.STOPBITS_TWO}.get(int(float(stopbits)), serial.STOPBITS_ONE),
xonxoff=xonxoff,
rtscts=rtscts,
dsrdtr=dsrdtr,
)
want_substr = psu_cfg.idn_substr
do_set = bool(psu_cfg.do_set)
set_v = float(psu_cfg.set_voltage)
set_i = float(psu_cfg.set_current)
port = str(psu_cfg.port).strip()
with OwonPSU(port, ps, eol=eol) as psu:
# Step 2: IDN
idn = psu.idn()
rp("psu_idn", idn)
print(f"PSU IDN: {idn}")
assert isinstance(idn, str)
assert idn != "", "*IDN? returned empty response"
if want_substr:
assert str(want_substr).lower() in idn.lower(), f"IDN does not contain expected substring: {want_substr}. Got: {idn}"
# Step 3: Output status before
out_before = psu.output_status()
rp("output_status_before", str(out_before))
print(f"Output status (before): {out_before}")
if do_set:
# Step 4: Set and measure
rp("set_voltage", set_v)
rp("set_current", set_i)
print(f"Setting: voltage={set_v}V, current={set_i}A")
psu.set_voltage(1, set_v)
psu.set_current(1, set_i)
psu.set_output(True)
time.sleep(1.0) # allow settling
try:
mv = psu.measure_voltage()
mi = psu.measure_current()
rp("measured_voltage", mv)
rp("measured_current", mi)
print(f"Measured: voltage={mv}V, current={mi}A")
finally:
psu.set_output(False)
out_after = psu.output_status()
rp("output_status_after", str(out_after))
print(f"Output status (after): {out_after}")

View File

@ -8,7 +8,7 @@ pytestmark = [pytest.mark.hardware, pytest.mark.babylin, pytest.mark.smoke]
WORKSPACE_ROOT = pathlib.Path(__file__).resolve().parents[1] WORKSPACE_ROOT = pathlib.Path(__file__).resolve().parents[1]
def test_babylin_sdk_example_flow(config, lin): def test_babylin_sdk_example_flow(config, lin, rp):
""" """
Title: BabyLIN SDK Example Flow - Open, Load SDF, Start Schedule, Rx Timeout Title: BabyLIN SDK Example Flow - Open, Load SDF, Start Schedule, Rx Timeout
@ -37,9 +37,12 @@ def test_babylin_sdk_example_flow(config, lin):
# Step 1: Ensure config is set for hardware with SDK wrapper # Step 1: Ensure config is set for hardware with SDK wrapper
assert config.interface.type == "babylin" assert config.interface.type == "babylin"
assert config.interface.sdf_path is not None assert config.interface.sdf_path is not None
rp("sdf_path", str(config.interface.sdf_path))
rp("schedule_nr", int(config.interface.schedule_nr))
# Step 3: Attempt a short receive to validate RX path while schedule runs # Step 3: Attempt a short receive to validate RX path while schedule runs
rx = lin.receive(timeout=0.2) rx = lin.receive(timeout=0.2)
rp("receive_result", "timeout" if rx is None else "frame")
# Step 4: Accept timeout or a valid frame object depending on bus activity # Step 4: Accept timeout or a valid frame object depending on bus activity
assert rx is None or hasattr(rx, "id") assert rx is None or hasattr(rx, "id")

View File

@ -4,7 +4,7 @@ import pytest
pytestmark = [pytest.mark.hardware, pytest.mark.babylin] pytestmark = [pytest.mark.hardware, pytest.mark.babylin]
def test_babylin_connect_receive_timeout(lin): def test_babylin_connect_receive_timeout(lin, rp):
""" """
Title: BabyLIN Hardware Smoke - Connect and Timed Receive Title: BabyLIN Hardware Smoke - Connect and Timed Receive
@ -28,6 +28,7 @@ def test_babylin_connect_receive_timeout(lin):
""" """
# Step 2: Perform a short receive to verify operability # Step 2: Perform a short receive to verify operability
rx = lin.receive(timeout=0.2) rx = lin.receive(timeout=0.2)
rp("receive_result", "timeout" if rx is None else "frame")
# Step 3: Accept either a timeout (None) or a frame-like object # Step 3: Accept either a timeout (None) or a frame-like object
assert rx is None or hasattr(rx, "id") assert rx is None or hasattr(rx, "id")

View File

@ -39,7 +39,7 @@ class _MockBytesOnly:
@pytest.mark.babylin @pytest.mark.babylin
@pytest.mark.smoke @pytest.mark.smoke
@pytest.mark.req_001 @pytest.mark.req_001
def test_babylin_sdk_adapter_with_mock_wrapper(): def test_babylin_sdk_adapter_with_mock_wrapper(rp):
""" """
Title: SDK Adapter - Send/Receive with Mock Wrapper Title: SDK Adapter - Send/Receive with Mock Wrapper
@ -62,6 +62,7 @@ def test_babylin_sdk_adapter_with_mock_wrapper():
""" """
# Step 1-2: Create adapter with wrapper injection and connect # Step 1-2: Create adapter with wrapper injection and connect
lin = BabyLinInterface(sdf_path="./vendor/Example.sdf", schedule_nr=0, wrapper_module=mock_bl) lin = BabyLinInterface(sdf_path="./vendor/Example.sdf", schedule_nr=0, wrapper_module=mock_bl)
rp("wrapper", "mock_bl")
lin.connect() lin.connect()
try: try:
# Step 3: Transmit a known payload on a chosen ID # Step 3: Transmit a known payload on a chosen ID
@ -70,6 +71,9 @@ def test_babylin_sdk_adapter_with_mock_wrapper():
# Step 4: Receive from the mock's RX queue (loopback) # Step 4: Receive from the mock's RX queue (loopback)
rx = lin.receive(timeout=0.1) rx = lin.receive(timeout=0.1)
rp("tx_id", f"0x{tx.id:02X}")
rp("tx_data", list(tx.data))
rp("rx_present", rx is not None)
# Step 5: Validate ID and payload integrity # Step 5: Validate ID and payload integrity
assert rx is not None, "Expected a frame from mock loopback" assert rx is not None, "Expected a frame from mock loopback"
@ -87,7 +91,7 @@ def test_babylin_sdk_adapter_with_mock_wrapper():
(mock_bl, True), # length signature available: expect deterministic pattern (mock_bl, True), # length signature available: expect deterministic pattern
(_MockBytesOnly, False), # bytes-only signature: expect zeros of requested length (_MockBytesOnly, False), # bytes-only signature: expect zeros of requested length
]) ])
def test_babylin_master_request_with_mock_wrapper(wrapper, expect_pattern): def test_babylin_master_request_with_mock_wrapper(wrapper, expect_pattern, rp):
""" """
Title: SDK Adapter - Master Request using Mock Wrapper Title: SDK Adapter - Master Request using Mock Wrapper
@ -112,11 +116,14 @@ def test_babylin_master_request_with_mock_wrapper(wrapper, expect_pattern):
""" """
# Step 1-2: Initialize mock-backed adapter # Step 1-2: Initialize mock-backed adapter
lin = BabyLinInterface(wrapper_module=wrapper) lin = BabyLinInterface(wrapper_module=wrapper)
rp("wrapper", getattr(wrapper, "__name__", str(wrapper)))
lin.connect() lin.connect()
try: try:
# Step 3: Request 4 bytes for ID 0x22 # Step 3: Request 4 bytes for ID 0x22
req_id = 0x22 req_id = 0x22
length = 4 length = 4
rp("req_id", f"0x{req_id:02X}")
rp("req_len", length)
rx = lin.request(id=req_id, length=length, timeout=0.1) rx = lin.request(id=req_id, length=length, timeout=0.1)
# Step 4-5: Validate response # Step 4-5: Validate response
@ -124,9 +131,15 @@ def test_babylin_master_request_with_mock_wrapper(wrapper, expect_pattern):
assert rx.id == req_id assert rx.id == req_id
if expect_pattern: if expect_pattern:
# length-signature mock returns deterministic pattern # length-signature mock returns deterministic pattern
assert rx.data == bytes(((req_id + i) & 0xFF) for i in range(length)) expected = bytes(((req_id + i) & 0xFF) for i in range(length))
rp("expected_data", list(expected))
rp("rx_data", list(rx.data))
assert rx.data == expected
else: else:
# bytes-only mock returns exactly the bytes we sent (zeros of requested length) # bytes-only mock returns exactly the bytes we sent (zeros of requested length)
assert rx.data == bytes([0] * length) expected = bytes([0] * length)
rp("expected_data", list(expected))
rp("rx_data", list(rx.data))
assert rx.data == expected
finally: finally:
lin.disconnect() lin.disconnect()

View File

@ -20,7 +20,7 @@ class TestMockLinInterface:
@pytest.mark.smoke @pytest.mark.smoke
@pytest.mark.req_001 @pytest.mark.req_001
@pytest.mark.req_003 @pytest.mark.req_003
def test_mock_send_receive_echo(self, lin): def test_mock_send_receive_echo(self, lin, rp):
""" """
Title: Mock LIN Interface - Send/Receive Echo Test Title: Mock LIN Interface - Send/Receive Echo Test
@ -41,23 +41,30 @@ class TestMockLinInterface:
- Received frame ID matches transmitted frame ID (0x12) - Received frame ID matches transmitted frame ID (0x12)
- Received frame data payload matches transmitted data [1, 2, 3] - Received frame data payload matches transmitted data [1, 2, 3]
""" """
# Step 1: Create test frame with known ID and payload # Step 1: Create test frame with known ID and payload
test_frame = LinFrame(id=0x12, data=bytes([1, 2, 3])) test_frame = LinFrame(id=0x12, data=bytes([1, 2, 3]))
rp("lin_type", "mock")
rp("tx_id", f"0x{test_frame.id:02X}")
rp("tx_data", list(test_frame.data))
# Step 2: Transmit frame via mock interface (mock will enqueue to RX) # Step 2: Transmit frame via mock interface (mock will enqueue to RX)
lin.send(test_frame) lin.send(test_frame)
# Step 3: Receive echoed frame with ID filtering and timeout # Step 3: Receive echoed frame with ID filtering and timeout
received_frame = lin.receive(id=0x12, timeout=0.5) received_frame = lin.receive(id=0x12, timeout=0.5)
rp("rx_present", received_frame is not None)
if received_frame is not None:
rp("rx_id", f"0x{received_frame.id:02X}")
rp("rx_data", list(received_frame.data))
# Step 4: Validate echo functionality and payload integrity # Step 4: Validate echo functionality and payload integrity
assert received_frame is not None, "Mock interface should echo transmitted frames" assert received_frame is not None, "Mock interface should echo transmitted frames"
assert received_frame.id == test_frame.id, f"Expected ID {test_frame.id:#x}, got {received_frame.id:#x}" assert received_frame.id == test_frame.id, f"Expected ID {test_frame.id:#x}, got {received_frame.id:#x}"
assert received_frame.data == test_frame.data, f"Expected data {test_frame.data!r}, got {received_frame.data!r}" assert received_frame.data == test_frame.data, f"Expected data {test_frame.data!r}, got {received_frame.data!r}"
@pytest.mark.smoke @pytest.mark.smoke
@pytest.mark.req_002 @pytest.mark.req_002
def test_mock_request_synthesized_response(self, lin): def test_mock_request_synthesized_response(self, lin, rp):
""" """
Title: Mock LIN Interface - Master Request Response Test Title: Mock LIN Interface - Master Request Response Test
@ -80,27 +87,32 @@ class TestMockLinInterface:
- Response data length equals requested length (4 bytes) - Response data length equals requested length (4 bytes)
- Response data follows deterministic pattern: [id+0, id+1, id+2, id+3] - Response data follows deterministic pattern: [id+0, id+1, id+2, id+3]
""" """
# Step 1: Issue master request with specific parameters # Step 1: Issue master request with specific parameters
request_id = 0x21 request_id = 0x21
requested_length = 4 requested_length = 4
# Step 2: Execute request operation; mock synthesizes deterministic bytes # Step 2: Execute request operation; mock synthesizes deterministic bytes
rp("lin_type", "mock")
rp("req_id", f"0x{request_id:02X}")
rp("req_len", requested_length)
response_frame = lin.request(id=request_id, length=requested_length, timeout=0.5) response_frame = lin.request(id=request_id, length=requested_length, timeout=0.5)
# Step 3: Validate response generation # Step 3: Validate response generation
assert response_frame is not None, "Mock interface should generate response for master requests" assert response_frame is not None, "Mock interface should generate response for master requests"
# Step 4: Verify response frame properties (ID and length) # Step 4: Verify response frame properties (ID and length)
assert response_frame.id == request_id, f"Response ID {response_frame.id:#x} should match request ID {request_id:#x}" assert response_frame.id == request_id, f"Response ID {response_frame.id:#x} should match request ID {request_id:#x}"
assert len(response_frame.data) == requested_length, f"Response length {len(response_frame.data)} should match requested length {requested_length}" assert len(response_frame.data) == requested_length, f"Response length {len(response_frame.data)} should match requested length {requested_length}"
# Step 5: Validate deterministic response pattern # Step 5: Validate deterministic response pattern
expected_data = bytes((request_id + i) & 0xFF for i in range(requested_length)) expected_data = bytes((request_id + i) & 0xFF for i in range(requested_length))
rp("rx_data", list(response_frame.data) if response_frame else None)
rp("expected_data", list(expected_data))
assert response_frame.data == expected_data, f"Response data {response_frame.data!r} should follow deterministic pattern {expected_data!r}" assert response_frame.data == expected_data, f"Response data {response_frame.data!r} should follow deterministic pattern {expected_data!r}"
@pytest.mark.smoke @pytest.mark.smoke
@pytest.mark.req_004 @pytest.mark.req_004
def test_mock_receive_timeout_behavior(self, lin): def test_mock_receive_timeout_behavior(self, lin, rp):
""" """
Title: Mock LIN Interface - Receive Timeout Test Title: Mock LIN Interface - Receive Timeout Test
@ -120,14 +132,18 @@ class TestMockLinInterface:
- Operation completes within specified timeout period - Operation completes within specified timeout period
- No exceptions or errors during timeout scenario - No exceptions or errors during timeout scenario
""" """
# Step 1: Attempt to receive frame with ID that hasn't been transmitted # Step 1: Attempt to receive frame with ID that hasn't been transmitted
non_existent_id = 0xFF non_existent_id = 0xFF
short_timeout = 0.1 # 100ms timeout short_timeout = 0.1 # 100ms timeout
# Step 2: Execute receive with timeout (should return None quickly) # Step 2: Execute receive with timeout (should return None quickly)
rp("lin_type", "mock")
rp("rx_id", f"0x{non_existent_id:02X}")
rp("timeout_s", short_timeout)
result = lin.receive(id=non_existent_id, timeout=short_timeout) result = lin.receive(id=non_existent_id, timeout=short_timeout)
rp("rx_present", result is not None)
# Step 3: Verify proper timeout behavior (no exceptions, returns None) # Step 3: Verify proper timeout behavior (no exceptions, returns None)
assert result is None, "Receive operation should return None when no matching frames available" assert result is None, "Receive operation should return None when no matching frames available"
@pytest.mark.boundary @pytest.mark.boundary
@ -139,7 +155,7 @@ class TestMockLinInterface:
(0x20, bytes([0x01, 0x02, 0x03, 0x04, 0x05])), (0x20, bytes([0x01, 0x02, 0x03, 0x04, 0x05])),
(0x15, bytes([0xFF, 0x00, 0xCC, 0x33, 0xF0, 0x0F, 0xA5, 0x5A])), (0x15, bytes([0xFF, 0x00, 0xCC, 0x33, 0xF0, 0x0F, 0xA5, 0x5A])),
]) ])
def test_mock_frame_validation_boundaries(self, lin, frame_id, data_payload): def test_mock_frame_validation_boundaries(self, lin, rp, frame_id, data_payload):
""" """
Title: Mock LIN Interface - Frame Validation Boundaries Test Title: Mock LIN Interface - Frame Validation Boundaries Test
@ -158,14 +174,17 @@ class TestMockLinInterface:
- All valid frame configurations are properly echoed - All valid frame configurations are properly echoed
- Frame ID and data integrity preserved across echo operation - Frame ID and data integrity preserved across echo operation
""" """
# Step 1: Create frame with parameterized values # Step 1: Create frame with parameterized values
test_frame = LinFrame(id=frame_id, data=data_payload) test_frame = LinFrame(id=frame_id, data=data_payload)
rp("lin_type", "mock")
rp("tx_id", f"0x{frame_id:02X}")
rp("tx_len", len(data_payload))
# Step 2: Send and receive frame # Step 2: Send and receive frame
lin.send(test_frame) lin.send(test_frame)
received_frame = lin.receive(id=frame_id, timeout=0.5) received_frame = lin.receive(id=frame_id, timeout=0.5)
# Step 3: Validate frame integrity across IDs and payload sizes # Step 3: Validate frame integrity across IDs and payload sizes
assert received_frame is not None, f"Frame with ID {frame_id:#x} should be echoed" assert received_frame is not None, f"Frame with ID {frame_id:#x} should be echoed"
assert received_frame.id == frame_id, f"Frame ID should be preserved: expected {frame_id:#x}" assert received_frame.id == frame_id, f"Frame ID should be preserved: expected {frame_id:#x}"
assert received_frame.data == data_payload, f"Frame data should be preserved for ID {frame_id:#x}" assert received_frame.data == data_payload, f"Frame data should be preserved for ID {frame_id:#x}"

View File

@ -7,7 +7,7 @@ from ecu_framework.config import load_config
@pytest.mark.unit @pytest.mark.unit
def test_config_precedence_env_overrides(monkeypatch, tmp_path): def test_config_precedence_env_overrides(monkeypatch, tmp_path, rp):
# Create a YAML file to use via env var # Create a YAML file to use via env var
yaml_path = tmp_path / "cfg.yaml" yaml_path = tmp_path / "cfg.yaml"
yaml_path.write_text("interface:\n type: babylin\n channel: 7\n") yaml_path.write_text("interface:\n type: babylin\n channel: 7\n")
@ -17,6 +17,9 @@ def test_config_precedence_env_overrides(monkeypatch, tmp_path):
# Apply overrides on top # Apply overrides on top
cfg = load_config(workspace_root=str(tmp_path), overrides={"interface": {"channel": 9}}) cfg = load_config(workspace_root=str(tmp_path), overrides={"interface": {"channel": 9}})
rp("config_source", "env+overrides")
rp("interface_type", cfg.interface.type)
rp("interface_channel", cfg.interface.channel)
# Env file applied # Env file applied
assert cfg.interface.type == "babylin" assert cfg.interface.type == "babylin"
@ -25,10 +28,13 @@ def test_config_precedence_env_overrides(monkeypatch, tmp_path):
@pytest.mark.unit @pytest.mark.unit
def test_config_defaults_when_no_file(monkeypatch): def test_config_defaults_when_no_file(monkeypatch, rp):
# Ensure no env path # Ensure no env path
monkeypatch.delenv("ECU_TESTS_CONFIG", raising=False) monkeypatch.delenv("ECU_TESTS_CONFIG", raising=False)
cfg = load_config(workspace_root=None) cfg = load_config(workspace_root=None)
rp("config_source", "defaults")
rp("interface_type", cfg.interface.type)
rp("flash_enabled", cfg.flash.enabled)
assert cfg.interface.type == "mock" assert cfg.interface.type == "mock"
assert cfg.flash.enabled is False assert cfg.flash.enabled is False

View File

@ -17,7 +17,7 @@ class _StubLin:
@pytest.mark.unit @pytest.mark.unit
def test_hex_flasher_sends_basic_sequence(tmp_path): def test_hex_flasher_sends_basic_sequence(tmp_path, rp):
# Minimal valid Intel HEX file (EOF record) # Minimal valid Intel HEX file (EOF record)
hex_path = tmp_path / "fw.hex" hex_path = tmp_path / "fw.hex"
hex_path.write_text(":00000001FF\n") hex_path.write_text(":00000001FF\n")
@ -25,6 +25,8 @@ def test_hex_flasher_sends_basic_sequence(tmp_path):
lin = _StubLin() lin = _StubLin()
flasher = HexFlasher(lin) flasher = HexFlasher(lin)
flasher.flash_hex(str(hex_path)) flasher.flash_hex(str(hex_path))
rp("hex_path", str(hex_path))
rp("sent_count", len(lin.sent))
# Placeholder assertion; refine as the flasher gains functionality # Placeholder assertion; refine as the flasher gains functionality
assert isinstance(lin.sent, list) assert isinstance(lin.sent, list)

View File

@ -3,19 +3,23 @@ from ecu_framework.lin.base import LinFrame
@pytest.mark.unit @pytest.mark.unit
def test_linframe_accepts_valid_ranges(): def test_linframe_accepts_valid_ranges(record_property: "pytest.RecordProperty"): # type: ignore[name-defined]
f = LinFrame(id=0x3F, data=bytes([0] * 8)) f = LinFrame(id=0x3F, data=bytes([0] * 8))
record_property("valid_id", f"0x{f.id:02X}")
record_property("data_len", len(f.data))
assert f.id == 0x3F and len(f.data) == 8 assert f.id == 0x3F and len(f.data) == 8
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.parametrize("bad_id", [-1, 0x40]) @pytest.mark.parametrize("bad_id", [-1, 0x40])
def test_linframe_invalid_id_raises(bad_id): def test_linframe_invalid_id_raises(bad_id, record_property: "pytest.RecordProperty"): # type: ignore[name-defined]
record_property("bad_id", bad_id)
with pytest.raises(ValueError): with pytest.raises(ValueError):
LinFrame(id=bad_id, data=b"\x00") LinFrame(id=bad_id, data=b"\x00")
@pytest.mark.unit @pytest.mark.unit
def test_linframe_too_long_raises(): def test_linframe_too_long_raises(record_property: "pytest.RecordProperty"): # type: ignore[name-defined]
record_property("data_len", 9)
with pytest.raises(ValueError): with pytest.raises(ValueError):
LinFrame(id=0x01, data=bytes(range(9))) LinFrame(id=0x01, data=bytes(range(9)))

95
vendor/Owon/tryout.py vendored Normal file
View File

@ -0,0 +1,95 @@
"""Owon PSU quick demo (optimized to use ecu_framework.power.owon_psu).
This script reads configuration from OWON_PSU_CONFIG (YAML) or ./config/owon_psu.yaml,
prints discovered ports responding to *IDN?, then connects to the configured port
and performs a small sequence (IDN, optional V/I set, toggle output, measure V/I).
No CLI flags; edit YAML to change behavior.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
import yaml
try:
from ecu_framework.power import OwonPSU, SerialParams, scan_ports
except ModuleNotFoundError:
# Ensure repository root is on sys.path when running this file directly
import sys
repo_root = Path(__file__).resolve().parents[2]
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
from ecu_framework.power import OwonPSU, SerialParams, scan_ports
def _load_yaml_config() -> dict:
cfg_path = str(Path("config") / "owon_psu.yaml")
p = Path(cfg_path).resolve()
print("Using config path:", str(p))
if not p.is_file():
return {}
with p.open("r", encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
return data if isinstance(data, dict) else {}
def run_demo() -> int:
cfg = _load_yaml_config()
if not cfg or "port" not in cfg:
print("Config not found or missing 'port'. Set OWON_PSU_CONFIG or create ./config/owon_psu.yaml")
return 2
print("Scanning ports (responding to *IDN?):")
for dev, idn in scan_ports(SerialParams(baudrate=int(cfg.get("baudrate", 115200)), timeout=float(cfg.get("timeout", 1.0)))):
print(f" {dev} -> {idn}")
# Serial params
baud = int(cfg.get("baudrate", 115200))
timeout = float(cfg.get("timeout", 1.0))
eol = cfg.get("eol", "\n")
from serial import PARITY_NONE, PARITY_EVEN, PARITY_ODD, STOPBITS_ONE, STOPBITS_TWO
parity = {"N": PARITY_NONE, "E": PARITY_EVEN, "O": PARITY_ODD}.get(str(cfg.get("parity", "N")).upper(), PARITY_NONE)
stopbits = {1: STOPBITS_ONE, 2: STOPBITS_TWO}.get(int(float(cfg.get("stopbits", 1))), STOPBITS_ONE)
xonxoff = bool(cfg.get("xonxoff", False))
rtscts = bool(cfg.get("rtscts", False))
dsrdtr = bool(cfg.get("dsrdtr", False))
ps = SerialParams(
baudrate=baud,
timeout=timeout,
parity=parity,
stopbits=stopbits,
xonxoff=xonxoff,
rtscts=rtscts,
dsrdtr=dsrdtr,
)
port = str(cfg["port"]).strip()
do_set = bool(cfg.get("do_set", False))
set_v = float(cfg.get("set_voltage", 1.0))
set_i = float(cfg.get("set_current", 0.1))
with OwonPSU(port, ps, eol=eol) as psu:
idn = psu.idn()
print(f"IDN: {idn}")
print(f"Output status: {psu.output_status()}")
if do_set:
psu.set_output(True)
time.sleep(0.5)
psu.set_voltage(1, set_v)
psu.set_current(1, set_i)
time.sleep(0.5)
print(f"Measured V: {psu.measure_voltage()} V")
print(f"Measured I: {psu.measure_current()} A")
time.sleep(0.5)
psu.set_output(False)
return 0
if __name__ == "__main__":
raise SystemExit(run_demo())