feat: purge battery_tester. [no changelog]
What changed, and why it matters
This commit simply deletes an entire internal testing tool called automatic_battery_tester from the repository. It removes Python scripts, configuration files, documentation, and dependency lists used only for hardware battery testing in the lab. No firmware code that ships to users was changed, and nothing in the commit suggests a security fix or vulnerability.
No security action needed. Treat as routine repository cleanup. If the tool is still used internally, ensure it is maintained in a separate repository or branch.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit purges the tools/automatic_battery_tester directory (25 files, ~2796 lines removed). The deleted code was a lab-only Python harness for controlling Trezor devices under test via serial port and relay boards, logging power-management reports, and analyzing battery temperature profiles. It also removes the corresponding path from tools/style.py.exclude. The change is a feature removal (‘feat: purge battery_tester. [no changelog]’) with no modifications to core firmware, bootloader, crypto, or communication stacks.
Changed components
tools/automatic_battery_tester/* (deleted)tools/style.py.excludeInspect captured patch +0 / −2796
diff --git a/tools/automatic_battery_tester/.gitignore b/tools/automatic_battery_tester/.gitignore
deleted file mode 100644
index 99e8e5a0..00000000
--- a/tools/automatic_battery_tester/.gitignore
+++ /dev/null
@@ -1,6 +0,0 @@
-
-
-venv/
-test_results/
-single_capture_test_results/
-test.log
diff --git a/tools/automatic_battery_tester/README.md b/tools/automatic_battery_tester/README.md
deleted file mode 100644
index c37633cd..00000000
--- a/tools/automatic_battery_tester/README.md
+++ /dev/null
@@ -1,105 +0,0 @@
-# Automated Battery Cycle Tester
-
-This project provides a framework for automated testing of charging and discharging cycles of batteries in connected devices (DUT - Device Under Test). It supports multiple test scenarios (Linear, Switching, Random Wonder) at various temperatures, logs detailed data for later analysis, and offers configurable user feedback (Slack notifications).
-
----
-
-## Project Structure
-
-```
-.
-├── hardware_ctl/ # Hardware control (relay, DUT, temperature)
-├── test_logic/ # Test scenario logic
-├── test_results/ # Output CSV logs
-├── main_tester.py # Main entry point
-├── test_config.toml # Configuration file
-├── requirements.txt # Python dependencies
-└── README.md # This file
-```
-
----
-
-## Installation
-
-1. **Python 3.9+ required**
-
-2. **Create and activate a virtual environment:**
- ```sh
- python -m venv .venv
- source .venv/bin/activate
- ```
-
-3. **Install dependencies:**
- ```sh
- pip install -r requirements.txt
- ```
-
-4. **Slack Webhook (optional):**
- If you want Slack notifications, create an [Incoming Webhook](https://api.slack.com/messaging/webhooks) and add the URL to `test_config.toml`.
-
----
-
-## Configuration
-
-Edit [`test_config.toml`](test_config.toml) to set up:
-
-- **[general]:** Output directory, logging interval, etc.
-- **[[duts]]:** List of DUTs (name, CPU ID, USB port, relay port).
-- **[relay]:** Relay board IP address.
-- **[test_plan]:** Temperatures, cycles, modes, and parameters for each mode.
-- **[notifications]:** Slack webhook and channel.
-
-Example:
-```toml
-[general]
-output_directory = "test_results"
-log_interval_seconds = 1
-
-[[duts]]
-name = "DUT1"
-cpu_id = "05001D000A50325557323120"
-usb_port = "/dev/ttyACM0"
-relay_port = 6
-
-[relay]
-ip_address = "192.168.1.10"
-
-[test_plan]
-temperatures_celsius = [15, 20, 25, 30, 35, 40, 45]
-cycles_per_temperature = 3
-test_modes = ["linear", "switching", "random_wonder"]
-```
-
----
-
-## Running the Test
-
-1. Activate the virtual environment:
- ```sh
- source .venv/bin/activate
- ```
-2. Edit `test_config.toml` as needed.
-3. Run the main script:
- ```sh
- python main_tester.py
- ```
-4. Follow console instructions or check log in "test.log" file.
-
----
-
-## Results
-
-- Results are saved in the `test_results/` directory as CSV files.
-- Each test mode and phase has its own file, e.g.:
- ```
- 274d.2506161536.linear.charged_relaxing.10.csv
- 274d.2506161536.linear.discharged_relaxing.10.csv
- ```
-
----
-
-## Authors
-
-- [Trezor Firmware Project Team](https://github.com/trezor/trezor-firmware)
-
----
diff --git a/tools/automatic_battery_tester/analysis/temperature_analysis.py b/tools/automatic_battery_tester/analysis/temperature_analysis.py
deleted file mode 100644
index 64817992..00000000
--- a/tools/automatic_battery_tester/analysis/temperature_analysis.py
+++ /dev/null
@@ -1,275 +0,0 @@
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-from typing import Any
-
-import matplotlib.pyplot as plt
-from InquirerPy import inquirer
-from InquirerPy.base import Choice
-from utils import load_measured_data
-
-default_dataset_dir = Path("../single_capture_test_results")
-
-battery_thermal_limit = 45.0 # Celsius
-case_thermal_limit = 41.0 # Celsius
-
-
-def select_waveforms(
- dataset_directory: Path = default_dataset_dir,
-) -> list[dict[str, Path]]:
- """
- Select waveforms from a given dataset directory.
-
- Args:
- dataset_directory (Path): The directory containing the dataset.
-
- Returns:
- list: A list of selected waveforms.
- """
-
- if not dataset_directory.exists():
- print(f"Dataset directory {dataset_directory} does not exist.")
- return []
-
- # glob all .csv files in the directory
- all_csv_files = list(dataset_directory.glob("*.csv"))
- if not all_csv_files:
- print(f"No CSV files found in {dataset_directory}.")
- return []
-
- external_temp_files = list(dataset_directory.glob("external_temp.*.csv"))
-
- # Filter out external temperature files
- waveform_files = [f for f in all_csv_files if f not in external_temp_files]
-
- choices = []
- for waveform_file in waveform_files:
- time_id = waveform_file.stem.split(".")[1]
-
- ch = Choice(
- name=f"{waveform_file.name}",
- value={"waveform": waveform_file, "external_temp": None},
- )
-
- for temp_file in external_temp_files:
- if time_id in temp_file.stem:
- ch.name += " (ext. temp available)"
- ch.value["external_temp"] = temp_file
- break
-
- choices.append(ch)
-
- try:
- selected = inquirer.fuzzy(
- message="Select one or more waveforms:",
- choices=choices,
- multiselect=True,
- instruction="(Use <tab> to select, <enter> to confirm)",
- ).execute()
-
- except KeyboardInterrupt:
- print("Selection cancelled by user.")
- sys.exit(0)
-
- except Exception as e:
- print(f"An error occurred during selection: {e}")
- sys.exit(1)
-
- return selected
-
-
-def colored_region_plot(
- axis: plt.Axes,
- time_vector: Any,
- data_vector: Any,
- mask: Any,
- color: str = "red",
- alpha: float = 0.5,
-) -> None:
-
- start = None
- in_region = False
- for i, val in enumerate(mask):
- if val and not in_region:
- start = i
- in_region = True
- elif not val and in_region:
- axis.plot(
- time_vector[start : i - 1],
- data_vector[start : i - 1],
- color=color,
- alpha=alpha,
- )
- in_region = False
-
- if in_region:
- axis.plot(
- time_vector[start : (i - 1)],
- data_vector[start : i - 1],
- color=color,
- alpha=alpha,
- )
-
-
-def colored_region_box(
- axis: plt.Axes,
- time_vector: Any,
- mask: Any,
- color: str = "orange",
- alpha: float = 0.5,
-) -> None:
-
- start = None
- in_region = False
- for i, val in enumerate(mask):
- if val and not in_region:
- start = i
- in_region = True
- elif not val and in_region:
- axis.axvspan(
- time_vector[start], time_vector[i - 1], color=color, alpha=alpha
- )
- in_region = False
-
- if in_region:
- axis.axvspan(time_vector[start], time_vector[-1], color=color, alpha=alpha)
-
-
-def sec_to_min(time_vector: Any) -> Any:
- return (time_vector - time_vector[0]) / 60.0
-
-
-def plot_temperature_profile(waveform_name: str, profile_data: Any) -> None:
-
- fig, ax = plt.subplots(2)
- fig.canvas.manager.set_window_title(waveform_name)
-
- ax[0].plot(
- sec_to_min(profile_data.time),
- profile_data.battery_temp,
- color="green",
- label="battery temeperature",
- )
- ax[0].axhline(y=battery_thermal_limit, color="green", linestyle="--")
-
- ax[0].plot(
- sec_to_min(profile_data.time),
- profile_data.pmic_die_temp,
- color="orange",
- label="pmic die temperature",
- )
-
- colored_region_plot(
- ax[0],
- sec_to_min(profile_data.time),
- profile_data.battery_temp,
- profile_data.battery_temp > battery_thermal_limit,
- color="red",
- alpha=1,
- )
-
- if profile_data.ext_temp is not None:
- ax[0].plot(
- sec_to_min(profile_data.ext_temp_time),
- profile_data.ext_temp,
- color="blue",
- label="case temperature",
- linestyle="--",
- )
- ax[0].axhline(y=case_thermal_limit, color="blue", linestyle="--")
-
- colored_region_plot(
- ax[0],
- sec_to_min(profile_data.ext_temp_time),
- profile_data.ext_temp,
- profile_data.ext_temp > case_thermal_limit,
- color="red",
- alpha=1,
- )
-
- ax[0].set_xlabel("Time (min)")
- ax[0].set_ylabel("Temperature (C)")
- ax[0].set_title("Temperature Profile: " + waveform_name)
- ax[0].set_xlim(
- left=sec_to_min(profile_data.time)[0], right=sec_to_min(profile_data.time)[-1]
- )
-
- ax[0].legend()
- ax[0].grid(True)
-
- def min_to_hr(x: float) -> float:
- return x / 60.0
-
- def hr_to_min(x: float) -> float:
- return x * 60.0
-
- secax = ax[0].secondary_xaxis("top", functions=(min_to_hr, hr_to_min))
- secax.set_xlabel("Time (hours)")
-
- # Change background color according to charging state
- usb_charging_mask = (profile_data.usb == "USB_connected") & (
- abs(profile_data.battery_current) > 0
- )
- wlc_charging_mask = (
- (profile_data.wlc == "WLC_connected")
- & ~usb_charging_mask
- & (abs(profile_data.battery_current) > 0)
- )
- colored_region_box(
- ax[0], sec_to_min(profile_data.time), usb_charging_mask, color="blue", alpha=0.2
- )
- colored_region_box(
- ax[0],
- sec_to_min(profile_data.time),
- wlc_charging_mask,
- color="green",
- alpha=0.2,
- )
-
- ax[1].plot(
- sec_to_min(profile_data.time),
- profile_data.battery_current,
- color="purple",
- label="battery current",
- )
- ax[1].set_xlabel("Time (min)")
- ax[1].set_ylabel("Current (mA)")
- ax[1].set_xlim(
- left=sec_to_min(profile_data.time)[0], right=sec_to_min(profile_data.time)[-1]
- )
- ax[1].grid(True)
- ax[1].legend()
-
- colored_region_box(
- ax[1], sec_to_min(profile_data.time), usb_charging_mask, color="blue", alpha=0.2
- )
- colored_region_box(
- ax[1],
- sec_to_min(profile_data.time),
- wlc_charging_mask,
- color="green",
- alpha=0.2,
- )
-
-
-def main() -> None:
-
- selected_waveforms = select_waveforms()
-
- for waveform in selected_waveforms:
- assert waveform["waveform"] is not None
- # Load data from files
- profile_data = load_measured_data(
- data_file_path=waveform["waveform"],
- extern_temp_file_path=waveform["external_temp"],
- )
-
- plot_temperature_profile(waveform["waveform"].name, profile_data)
-
- # Plot graphs
- plt.show()
-
-
-if __name__ == "__main__":
- main()
diff --git a/tools/automatic_battery_tester/analysis/utils/__init__.py b/tools/automatic_battery_tester/analysis/utils/__init__.py
deleted file mode 100644
index ebedaa6c..00000000
--- a/tools/automatic_battery_tester/analysis/utils/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from .data_convertor import BatteryAnalysisData, load_measured_data
-
-__all__ = ["BatteryAnalysisData", "load_measured_data"]
diff --git a/tools/automatic_battery_tester/analysis/utils/data_convertor.py b/tools/automatic_battery_tester/analysis/utils/data_convertor.py
deleted file mode 100644
index 51233e69..00000000
--- a/tools/automatic_battery_tester/analysis/utils/data_convertor.py
+++ /dev/null
@@ -1,78 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from pathlib import Path
-
-import numpy as np
-import pandas as pd
-
-
-@dataclass
-class BatteryAnalysisData:
- time: np.ndarray
- power_state: np.ndarray
- usb: np.ndarray
- wlc: np.ndarray
- battery_voltage: np.ndarray
- battery_current: np.ndarray
- battery_temp: np.ndarray
- battery_soc: np.ndarray
- battery_soc_latched: np.ndarray
- pmic_die_temp: np.ndarray
- wlc_voltage: np.ndarray
- wlc_current: np.ndarray
- wlc_die_temp: np.ndarray
- system_voltage: np.ndarray
- ext_temp_time: np.ndarray = None # Optional time vector for external temperature
- ext_temp: np.ndarray = None # Optional external temperature data
-
-
-def load_measured_data(
- data_file_path: Path, extern_temp_file_path: Path | None = None
-) -> BatteryAnalysisData:
-
- profile_data = pd.read_csv(data_file_path)
-
- # Extract data from the DataFrame
- time_vector = profile_data["time"].to_numpy()
- power_state_vector = profile_data["power_state"].to_numpy()
- usb_vector = profile_data["usb"].to_numpy()
- wlc_vector = profile_data["wlc"].to_numpy()
- battery_voltage_vector = profile_data["battery_voltage"].to_numpy()
- battery_current_vector = profile_data["battery_current"].to_numpy()
- battery_temp_vector = profile_data["battery_temp"].to_numpy()
- battery_soc_vector = profile_data["battery_soc"].to_numpy()
- battery_soc_latched_vector = profile_data["battery_soc_latched"].to_numpy()
- pmic_die_temp_vector = profile_data["pmic_die_temp"].to_numpy()
- wlc_voltage_vector = profile_data["wlc_voltage"].to_numpy()
- wlc_current_vector = profile_data["wlc_current"].to_numpy()
- wlc_die_temp_vector = profile_data["wlc_die_temp"].to_numpy()
- system_voltage_vector = profile_data["system_voltage"].to_numpy()
-
- if extern_temp_file_path is not None:
- # Load external temperature data if provided
- ext_temp_data = pd.read_csv(extern_temp_file_path)
- ext_temp_time_vector = ext_temp_data["time"].to_numpy()
- ext_temp_vector = ext_temp_data["temperature"].to_numpy()
- else:
- ext_temp_time_vector = None
- ext_temp_vector = None
-
- return BatteryAnalysisData(
- time=time_vector,
- power_state=power_state_vector,
- usb=usb_vector,
- wlc=wlc_vector,
- battery_voltage=battery_voltage_vector,
- battery_current=battery_current_vector,
- battery_temp=battery_temp_vector,
- battery_soc=battery_soc_vector,
- battery_soc_latched=battery_soc_latched_vector,
- pmic_die_temp=pmic_die_temp_vector,
- wlc_voltage=wlc_voltage_vector,
- wlc_current=wlc_current_vector,
- wlc_die_temp=wlc_die_temp_vector,
- system_voltage=system_voltage_vector,
- ext_temp_time=ext_temp_time_vector,
- ext_temp=ext_temp_vector,
- )
diff --git a/tools/automatic_battery_tester/dut/__init__.py b/tools/automatic_battery_tester/dut/__init__.py
deleted file mode 100644
index 76973c38..00000000
--- a/tools/automatic_battery_tester/dut/__init__.py
+++ /dev/null
@@ -1,4 +0,0 @@
-from .dut import Dut
-from .dut_controller import DutController
-
-__all__ = ["Dut", "DutController"]
diff --git a/tools/automatic_battery_tester/dut/dut.py b/tools/automatic_battery_tester/dut/dut.py
deleted file mode 100644
index d61f0b6e..00000000
--- a/tools/automatic_battery_tester/dut/dut.py
+++ /dev/null
@@ -1,406 +0,0 @@
-from __future__ import annotations
-
-import hashlib
-import logging
-import time
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-
-import serial
-from hardware_ctl.relay_controller import RelayController
-
-BAUDRATE_DEFAULT = 115200
-BYTESIZE_DEFAULT = 8
-PARITY_DEFAULT = serial.PARITY_NONE
-CMD_TIMEOUT = 10
-
-
-@dataclass
-class DutReportData:
- time: float = 0.0
- power_state: str = ""
- usb: str = ""
- wlc: str = ""
- battery_voltage: float = 0.0
- battery_current: float = 0.0
- battery_temp: float = 0.0
- battery_soc: int = 0
- battery_soc_latched: int = 0
- pmic_die_temp: float = 0.0
- wlc_voltage: float = 0.0
- wlc_current: float = 0.0
- wlc_die_temp: float = 0.0
- system_voltage: float = 0.0
-
-
-@dataclass
-class DutProdtestResponse:
- timestamp: float | None = None
- cmd: str | None = None
- trace: list = field(default_factory=list)
- data_entries: list = field(default_factory=list)
- OK: bool = False
-
-
-class Dut:
-
- def __init__(
- self,
- name: str,
- cpu_id: str | None = None,
- usb_port: str | None = None,
- relay_port: int | None = None,
- relay_ctl: RelayController | None = None,
- verbose: bool = False,
- ) -> None:
-
- self.name = name
- self.relay_ctl = relay_ctl
- self.verbose = verbose
- self.relay_port = relay_port
-
- # Power up the device with relay controller
- self.power_up()
-
- # Wait for device to boot up
- time.sleep(3)
-
- self.vcp = serial.Serial(
- port=usb_port,
- baudrate=BAUDRATE_DEFAULT,
- bytesize=BYTESIZE_DEFAULT,
- parity=PARITY_DEFAULT,
- )
-
- # Connect serial port
- if not self.vcp.is_open:
- self.init_error()
- raise RuntimeError(f"Failed to open serial port {usb_port} for DUT {name}")
-
- self.entry_interactive_mode()
- self.enable_charging()
- self.set_backlight(100)
-
- time.sleep(2) # Give some time to process te commands
-
- if not self.ping():
- self.init_error()
- raise RuntimeError(f"DUT {self.name} did not respond to ping command")
-
- self.cpu_id = self.get_cpuid()
- if self.cpu_id is None:
- self.init_error()
- raise RuntimeError(f"DUT {self.name} failed to retrieve CPU ID")
-
- logging.debug(f"DUT {self.name} initialized with CPU ID: {self.cpu_id}")
-
- self.cpu_id_hash = self.generate_id_hash(self.cpu_id)
-
- # cpu_id check
- if cpu_id is not None:
- if self.cpu_id != cpu_id:
- self.init_error()
- raise RuntimeError(
- f"DUT {self.name} CPU ID mismatch: expected {cpu_id}, got {self.cpu_id}"
- )
-
- logging.debug(f"DUT {self.name} ID hash: {self.cpu_id_hash}")
-
- # device should start charging
- report = self.read_report()
- if not report or not report.usb == "USB_connected":
- self.init_error()
- raise RuntimeError(
- f"{self.name} USB not connected. Check VCP and relay ports"
- )
-
- self.display_ok()
- self.disable_charging()
- self.power_down()
-
- def init_error(self) -> None:
- self.display_error()
- self.disable_charging()
- self.power_down()
-
- def display_error(self) -> None:
- self.display_bars("R")
- time.sleep(3)
-
- def display_ok(self) -> None:
- self.display_bars("G")
- time.sleep(3)
-
- def get_cpu_id_hash(self) -> str:
- return self.cpu_id_hash
-
- def get_relay_port(self) -> int | None:
- return self.relay_port
-
- def generate_id_hash(self, cpu_id: str | None) -> str:
- """
- Generate a unique ID hash for the DUT based on its CPU ID.
- :param cpu_id: The CPU ID of the DUT.
- :return: A unique ID hash string.
- """
- if cpu_id is None:
- raise ValueError("CPU ID cannot be None.")
-
- device_id_bytes = bytes.fromhex(cpu_id)
- digest = hashlib.sha256(device_id_bytes).digest()
- return digest[:2].hex()
-
- def set_verbose(self, verbose: bool) -> None:
- self.verbose = verbose
-
- def get_verbose(self) -> bool:
- return self.verbose
-
- def entry_interactive_mode(self) -> None:
- # Enter interactive mode
- self.send_command(".", skip_response=True)
-
- def power_up(self) -> None:
- """
- Power up the DUT by activating the relay.
- """
- if self.relay_port is None:
- logging.debug("Relay port not set for DUT, skipping power up.")
- return
- assert self.relay_ctl is not None
- self.relay_ctl.set_relay_on(self.relay_port)
-
- def power_down(self) -> None:
- """
- Power down the DUT by deactivating the relay.
- """
- if self.relay_port is None:
- logging.debug("Relay port not set for DUT, skipping power down.")
- return
- assert self.relay_ctl is not None
- self.relay_ctl.set_relay_off(self.relay_port)
-
- def display_bars(self, value: str) -> bool:
- """
- Display bars on the DUT's screen.
- :param value: A string representing the bars to display (e.g., "G" for green).
- :return: True if the command was successful, False otherwise.
- """
- response = self.send_command("display-bars", value)
- return response.OK
-
- def ping(self) -> bool:
- """
- Send a ping command to the DUT and wait for a response.
- Returns True if the DUT responds with "OK", False otherwise.
- """
- response = self.send_command("ping")
- return response.OK
-
- def enable_charging(self) -> bool:
-
- response = self.send_command("pm-charge-enable")
- return response.OK
-
- def disable_charging(self) -> bool:
-
- response = self.send_command("pm-charge-disable")
- return response.OK
-
- def set_soc_limit(self, soc_limit: int) -> bool:
- """
- Set the state of charge (SoC) limit for the DUT.
- :param soc_limit: The SoC limit to set (0-100).
- :return: True if the command was successful, False otherwise.
- """
- if not 0 <= soc_limit <= 100:
- raise ValueError("SoC limit must be between 0 and 100.")
-
- response = self.send_command("pm-set-soc-limit", soc_limit)
- return response.OK
-
- def set_backlight(self, value: int) -> bool:
-
- if not 0 <= value <= 255:
- raise ValueError("Backlight value must be between 0 and 255.")
-
- response = self.send_command("display-set-backlight", value)
-
- return response.OK
-
- def get_cpuid(self) -> str | None:
-
- response = self.send_command("get-cpuid")
- if not response.OK:
- return None
-
- if len(response.data_entries) == 0:
- # No cpuid in data entries
- return None
-
- return response.data_entries[0][0]
-
- def parse_report(self, response: DutProdtestResponse) -> DutReportData:
-
- data = DutReportData()
- assert response.timestamp is not None
- data.time = response.timestamp
- data.power_state = response.data_entries[0][0]
- data.usb = response.data_entries[0][1]
- data.wlc = response.data_entries[0][2] if response.data_entries else ""
- data.battery_voltage = float(response.data_entries[0][3])
- data.battery_current = float(response.data_entries[0][4])
- data.battery_temp = float(response.data_entries[0][5])
- data.battery_soc = int(float(response.data_entries[0][6]))
- data.battery_soc_latched = int(float(response.data_entries[0][7]))
- data.pmic_die_temp = float(response.data_entries[0][8])
- data.wlc_voltage = float(response.data_entries[0][9])
- data.wlc_current = float(response.data_entries[0][10])
- data.wlc_die_temp = float(response.data_entries[0][11])
- data.system_voltage = float(response.data_entries[0][12])
-
- return data
-
- def read_report(self) -> DutReportData | None:
- """
- Read the PM report from the DUT.
- Returns a ProdtestResponse object containing the report data.
- """
- response = self.send_command("pm-report")
- if not response.OK:
- logging.error(f"Failed to read PM report from {self.name}.")
- return None
-
- return self.parse_report(response)
-
- def send_command(
- self, cmd: str, *args: Any, skip_response: bool = False
- ) -> DutProdtestResponse:
-
- if self.vcp is None:
- raise RuntimeError("VPC not initalized")
-
- response = DutProdtestResponse()
- # assert(len == 0)
-
- # Assamble command
- response.cmd = cmd
- if args:
- response.cmd = response.cmd + " " + " ".join(str(k) for k in args)
- response.cmd = response.cmd + "\n"
-
- response.timestamp = time.time()
-
- # Flush serial
- self.vcp.flush()
-
- self._log_output(response.cmd.rstrip("\r\n"))
- self.vcp.write(response.cmd.encode())
-
- if skip_response:
- return response
-
- while True:
-
- line = self.vcp.readline().decode()
- self._log_input(line.strip("\r\n"))
-
- # Capture traces
- if line[:1] == "#":
- response.trace.append(line[2:])
-
- # Capture data
- if line[:8] == "PROGRESS":
- line = line.replace("\r\n", "")
- response.data_entries.append((line[9:].split(" ")))
-
- # Terminate
- if "OK" in line:
-
- response.OK = True
-
- # Check if there is any data comming along with OK
- line = line.replace("\r\n", "")
- response.data_entries.append((line[3:].split(" ")))
-
- break
-
- if "ERROR" in line:
- break
-
- return response
-
- def log_data(
- self,
- output_directory: Path,
- test_time_id: str,
- test_scenario: str,
- test_phase: str,
- temp: str,
- verbose: bool = False,
- ) -> None:
-
- # Log file name format:
- # > <device_id_hash>.<time_identifier>.<test_scenario>.<test><temperarture>.csv
- # Example: a8bf.2506091307.linear.charge.25_deg.csv
-
- file_path = (
- output_directory
- / f"{self.cpu_id_hash}.{test_time_id}.{test_scenario}.{test_phase}.{temp}.csv"
- )
-
- report = None
- try:
- report = self.send_command("pm-report")
- except Exception as e:
- logging.error(f"Failed to read PM report from {self.name}, skip log: {e}")
- return
-
- if not file_path.exists():
- # creat a file header
- with open(file_path, "w") as f:
- f.write(
- "time,power_state,usb,wlc,battery_voltage,battery_current,"
- "battery_temp,battery_soc,battery_soc_latched,pmic_die_temp,"
- "wlc_voltage,wlc_current,wlc_die_temp,system_voltage\n"
- )
-
- with open(file_path, "a") as f:
- f.write(
- str(report.timestamp) + "," + ",".join(report.data_entries[0]) + "\n"
- )
-
- if verbose:
- print(str(report.timestamp) + "," + ",".join(report.data_entries[0]))
-
- def _log_output(self, message: str) -> None:
- if self.verbose:
- prefix = f"\033[95m[{self.name}]\033[0m"
- logging.debug(prefix + " > " + message)
-
- def _log_input(self, message: str) -> None:
- if self.verbose:
- prefix = f"\033[95m[{self.name}]\033[0m"
- logging.debug(prefix + " < " + message)
-
- def close(self) -> None:
- """
- Close the DUT's serial port and clean up resources.
- """
- if self.vcp is not None and self.vcp.is_open:
- try:
- self.vcp.close()
- except Exception as e:
- logging.warning(f"Failed to close VCP for {self.name}: {e}")
- self.vcp = None
- self.name = None
- self.relay_ctl = None
- self.relay_port = None
-
- def __del__(self) -> None:
- try:
- if hasattr(self, "vcp") and self.vcp is not None and self.vcp.is_open:
- self.close()
- except Exception as e:
- logging.warning(f"Error during DUT cleanup: {e}")
diff --git a/tools/automatic_battery_tester/dut/dut_controller.py b/tools/automatic_battery_tester/dut/dut_controller.py
deleted file mode 100644
index a6a4eae1..00000000
--- a/tools/automatic_battery_tester/dut/dut_controller.py
+++ /dev/null
@@ -1,242 +0,0 @@
-from __future__ import annotations
-
-import logging
-import sys
-from dataclasses import dataclass
-from pathlib import Path
-
-from hardware_ctl.relay_controller import RelayController
-
-from .dut import Dut
-
-
-@dataclass
-class ProdtestPmReport:
- """pm-report command response data structure"""
-
- power_state: str = ""
- usb: str = ""
- wlc: str = ""
- battery_voltage: float = 0.0
- battery_current: float = 0.0
- battery_temp: float = 0.0
- battery_soc: int = 0
- battery_soc_latched: int = 0
- pmic_die_temp: float = 0.0
- wlc_voltage: float = 0.0
- wlc_current: float = 0.0
- wlc_die_temp: float = 0.0
- system_voltage: float = 0.0
-
- @classmethod
- def from_string_list(cls, data: list[str]) -> "ProdtestPmReport":
- """Parse a list of strings into a ProdtestPmReport instance."""
- try:
- return cls(
- power_state=str(data[0]),
- usb=str(data[1]),
- wlc=str(data[2]),
- battery_voltage=float(data[3]),
- battery_current=float(data[4]),
- battery_temp=float(data[5]),
- battery_soc=int(float(data[6])),
- battery_soc_latched=int(float(data[7])),
- pmic_die_temp=float(data[8]),
- wlc_voltage=float(data[9]),
- wlc_current=float(data[10]),
- wlc_die_temp=float(data[11]),
- system_voltage=float(data[12]),
- )
- except (IndexError, ValueError, TypeError) as e:
- logging.error(f"Failed to parse pm-report data: {data} ({e})")
- return cls()
-
-
-class DutController:
- """
- Device-under-test (DUT) controller.
- provides direct simultaneous control of configured DUTs
- """
-
- def __init__(
- self, duts: list[dict], relay_ctl: RelayController, verbose: bool = False
- ) -> None:
-
- self.duts = []
- self.relay_ctl = relay_ctl
-
- # Power off all DUTs before self test
- for d in duts:
- self.relay_ctl.set_relay_off(d["relay_port"])
-
- for d in duts:
-
- try:
- dut = Dut(
- name=d["name"],
- cpu_id=d["cpu_id"],
- usb_port=d["usb_port"],
- relay_port=d["relay_port"],
- relay_ctl=self.relay_ctl,
- verbose=verbose,
- )
- self.duts.append(dut)
- logging.info(f"Initialized {d['name']} on port {d['usb_port']}")
- logging.info(f" -- cpu_id hash : {dut.get_cpu_id_hash()}")
- logging.info(f" -- relay port : {dut.get_relay_port()}")
-
- except Exception as e:
- logging.critical(
- f"Failed to initialize DUT {d['name']} on port {d['usb_port']}: {e}"
- )
- sys.exit(1)
-
- if len(self.duts) == 0:
- logging.error("No DUTs initialized. Cannot proceed.")
- raise RuntimeError("No DUTs initialized. Check port configuration.")
-
- def power_up_all(self) -> None:
-
- for d in self.duts:
- d.power_up()
-
- def power_down_all(self) -> None:
-
- for d in self.duts:
- d.power_down()
-
- def enable_charging(self) -> None:
- """
- Enable charging on all DUTs.
- """
- for d in self.duts:
- try:
- d.enable_charging()
- except Exception as e:
- logging.error(f"Failed to enable charging on {d.name}: {e}")
-
- def disable_charging(self) -> None:
- """
- Disable charging on all DUTs.
- """
- for d in self.duts:
- try:
- d.disable_charging()
- except Exception as e:
- logging.error(f"Failed to disable charging on {d.name}: {e}")
-
- def set_soc_limit(self, soc_limit: int) -> None:
- """
- Set the state of charge (SoC) limit for all DUTs.
- :param soc_limit: The SoC limit to set (0-100).
- """
- for d in self.duts:
- try:
- d.set_soc_limit(soc_limit)
- except Exception as e:
- logging.error(f"Failed to set SoC limit on {d.name}: {e}")
-
- def all_duts_charged(self) -> bool:
-
- all_dut_charged = True
-
- for d in self.duts:
-
- # Read power report
- data = d.read_report()
-
- if data.battery_voltage >= 3.3 and abs(data.battery_current) < 0.1:
- # Charging completed
- d.disable_charging()
- else:
- all_dut_charged = False
-
- return all_dut_charged
-
- def all_duts_discharged(self) -> bool:
-
- all_dut_dischargerd = True
-
- for d in self.duts:
-
- # Read power report
- data = d.read_report()
-
- # Device start to shutdown, turn the power on.
- if data.power_state == "3":
-
- # Attach power
- d.disable_charging()
- d.power_up()
-
- elif data.usb == "USB_connected":
- # USB is connected, device already finish the discharge cycle
- continue
- else:
- # Discharging not completed
- all_dut_dischargerd = False
-
- return all_dut_dischargerd
-
- def any_dut_charged(self) -> bool:
-
- for d in self.duts:
- # Read power report
- data = d.read_report()
-
- if data.battery_voltage >= 3.3 and abs(data.battery_current) < 0.1:
- # Charging completed
- return True
-
- return False
-
- def any_dut_discharged(self) -> bool:
-
- for d in self.duts:
- # Read power report
- data = d.read_report()
-
- if data.power_state == "3":
- # Device start to shutdown, turn the power on.
- d.disable_charging()
- d.power_up()
- return True
- elif data.usb == "USB_connected":
- # USB is connected, device already finish the discharge cycle
- return True
-
- return False
-
- def set_backlight(self, value: int) -> None:
-
- for d in self.duts:
- try:
- d.set_backlight(value)
- except Exception as e:
- logging.error(f"Failed to set backlight on {d.name}: {e}")
-
- def log_data(
- self,
- output_directory: Path,
- test_time_id: str,
- test_scenario: str,
- test_phase: str,
- temp: float | int,
- ) -> None:
-
- # Log file name format:
- # > <device_id_hash>.<time_identifier>.<test_scenario>.<temperarture>.csv
- # Example: a8bf.2506091307.linear.charge.25_deg.csv
-
- for d in self.duts:
- d.log_data(output_directory, test_time_id, test_scenario, test_phase, temp)
-
- def close(self) -> None:
- for d in self.duts:
- try:
- d.close()
- except Exception as e:
- logging.error(f"Failed to close DUT {d.name}: {e}")
-
- def __del__(self) -> None:
- self.close()
diff --git a/tools/automatic_battery_tester/hardware_ctl/__init__.py b/tools/automatic_battery_tester/hardware_ctl/__init__.py
deleted file mode 100644
index a9532f93..00000000
--- a/tools/automatic_battery_tester/hardware_ctl/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from .relay_controller import RelayController
-
-__all__ = ["RelayController"]
diff --git a/tools/automatic_battery_tester/hardware_ctl/deditec/__init__.py b/tools/automatic_battery_tester/hardware_ctl/deditec/__init__.py
deleted file mode 100644
index 267cf241..00000000
--- a/tools/automatic_battery_tester/hardware_ctl/deditec/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from .bs_weu_16 import DeditecBsWeu16
-
-__all__ = ["DeditecBsWeu16"]
diff --git a/tools/automatic_battery_tester/hardware_ctl/deditec/bs_weu_16.py b/tools/automatic_battery_tester/hardware_ctl/deditec/bs_weu_16.py
deleted file mode 100644
index 436ead0f..00000000
--- a/tools/automatic_battery_tester/hardware_ctl/deditec/bs_weu_16.py
+++ /dev/null
@@ -1,172 +0,0 @@
-from __future__ import annotations
-
-import logging
-import signal
-import socket
-from typing import Any, List
-
-from typing_extensions import Self
-
-IP = "192.168.1.10" # default static IP address
-PORT = 9912
-PIN_COUNT = 16 # Total number of pins on the Deditec BS-WEU-16 board
-
-
-class DeditecBsWeu16:
-
- def __init__(
- self, ip: str = IP, port: int = PORT, timeout_seconds: int = 3
- ) -> None:
-
- self.ip = ip
- self.port = port
- self.timeout_seconds = max(1, timeout_seconds)
- self.socket: socket.socket | None = None
- self.pins_on_latched = []
-
- logging.debug(f"DeditecBsWeu16: instance created on {self.ip}:{self.port}")
-
- def connect(self) -> bool:
- if self.socket is not None:
- logging.warning(
- "DeditecBsWeu16: connect called, but socket already exists. Closing first."
- )
- self.close_connection()
-
- logging.debug(
- f"DeditecBsWeu16: connecting to device at {self.ip}:{self.port}..."
- )
- try:
- self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.socket.settimeout(self.timeout_seconds)
- self.socket.connect((self.ip, self.port))
- logging.debug("DeditecBsWeu16: connection established")
- return True
- except socket.timeout:
- logging.error(
- f"DeditecBsWeu16: connection timed out ({self.timeout_seconds}s)"
- )
- self.socket = None
- return False
- except Exception as e:
- logging.exception(f"DeditecBsWeu16: connection error > {e}")
- self.socket = None
- return False
-
- def send_command(self, command: bytes) -> bool:
-
- if self.socket is None:
- logging.error("DeditecBsWeu16: send_command called but not connected.")
- return False
-
- logging.debug(f"DeditecBsWeu16: sending command: {command!r}")
- try:
- self.socket.sendall(command)
- data = self.socket.recv(64)
- logging.debug(
- f"DeditecBsWeu16: received confirmation data (len={len(data)}): {data!r}"
- )
- return True
- except socket.timeout:
- logging.error(
- f"DeditecBsWeu16: socket timeout during send/recv ({self.timeout_seconds}s)"
- )
- return False
- except Exception as e:
- logging.exception(
- f"DeditecBsWeu16: error sending command or receiving confirmation: {e}"
- )
- return False
-
- def control_relay(self, pins_on: List[int], pins_off: List[int]) -> bool:
- """Turns on all pins specified in pins_on list and turns off all pins specified in pins_off list.
- Returns True if successful, False otherwise.
- """
-
- if not self.socket:
- logging.error("DeditecBsWeu16: Relay not connected.")
- return False
-
- for pin in pins_on:
- if not 1 <= pin <= PIN_COUNT:
- logging.error(f"DeditecBsWeu16: Invalid pin number {pin} in pins_on.")
- return False
-
- for pin in pins_off:
- if not 1 <= pin <= PIN_COUNT:
- logging.error(f"DeditecBsWeu16: Invalid pin number {pin} in pins_off.")
- return False
-
- # Update the list of
- self.pins_on_latched = list(set(self.pins_on_latched + pins_on) - set(pins_off))
- command = self.assabmle_command(self.pins_on_latched)
-
- if not self.send_command(command):
- logging.error("DeditecBsWeu16: Failed to send command to Deditec device.")
- return False
-
- logging.info(
- f"DeditecBsWeu16: Changed relay setup. Pins ON: {self.pins_on_latched}"
- )
-
- return True
-
- def assabmle_command(self, pins: List[int]) -> bytes:
- """Assembles the command to turn on specified pins on the Deditec BS-WEU-16 board."""
- command_prefix = b"\x63\x9a\x01\x01\x00\x0b\x57\x57\x00\x00"
-
- pin_mask_value = 0
- for pin in set(pins): # Ensure uniqueness
- if isinstance(pin, int) and 1 <= pin <= PIN_COUNT:
- pin_mask_value += 2 ** (pin - 1)
- else:
- logging.warning(
- f"DeditecBsWeu16: Invalid pin number provided to assabmle_command: {pin}. Ignoring."
- )
-
- command = command_prefix + pin_mask_value.to_bytes(2, byteorder="big")
- return command
-
- def close_connection(self) -> None:
- if self.socket:
- logging.debug("DeditecBsWeu16: closing connection")
- try:
- self.socket.close()
- except Exception as e:
- logging.error(f"DeditecBsWeu16: error closing socket: {e}")
- finally:
- self.socket = None
- else:
- logging.debug("Deditec:: close_connection called but already closed.")
-
- def __enter__(self) -> Self:
-
- try:
- signal.alarm(self.timeout_seconds + 1)
- except ValueError:
- logging.warning(
- "Cannot set SIGALRM handler (not on Unix main thread?), relying on socket timeout."
- )
- pass
-
- if not self.connect():
- signal.alarm(0)
- raise ConnectionError(
- f"Failed to connect to Deditec device at {self.ip}:{self.port}"
- )
-
- return self
-
- def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool:
-
- try:
- signal.alarm(0)
- except ValueError:
- pass
- self.close_connection()
-
- if exc_type:
- logging.error(
- f"Deditec:: An error occurred during 'with' block: {exc_type.__name__}: {exc_val}"
- )
- return False
diff --git a/tools/automatic_battery_tester/hardware_ctl/gdm8351/__init__.py b/tools/automatic_battery_tester/hardware_ctl/gdm8351/__init__.py
deleted file mode 100644
index 4158df2b..00000000
--- a/tools/automatic_battery_tester/hardware_ctl/gdm8351/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from .gdm8351 import GDM8351
-
-__all__ = ["GDM8351"]
diff --git a/tools/automatic_battery_tester/hardware_ctl/gdm8351/gdm8351.py b/tools/automatic_battery_tester/hardware_ctl/gdm8351/gdm8351.py
deleted file mode 100644
index c72b150c..00000000
--- a/tools/automatic_battery_tester/hardware_ctl/gdm8351/gdm8351.py
+++ /dev/null
@@ -1,130 +0,0 @@
-import time
-from pathlib import Path
-
-import pyvisa
-
-
-class GDM8351:
-
- def __init__(self) -> None:
-
- self.rm = pyvisa.ResourceManager()
-
- # List available resources, let the user to pick one from the list
- available_devices = {}
- device_count = 0
-
- print("Available devices:")
- for r_name in self.rm.list_resources():
- if "/dev/ttyACM" in r_name:
- device_count += 1
- available_devices[device_count] = r_name
- print(f" [{device_count}]: {r_name}")
-
- self.device_connected = False
- while not self.device_connected:
- input_device_id = input(
- "Digital multimeter GDM8351: Select VCP port number (or Q to quit the selection): "
- )
-
- if input_device_id.lower() == "q":
- print("Exiting device selection.")
- return
-
- for device_id, device_name in available_devices.items():
- if int(input_device_id) == device_id:
- print(f"Connecting to {device_name}...")
-
- try:
- self.device = self.rm.open_resource(device_name)
- self.device_id = self.device.query("*IDN?")
- if "GDM8351" in self.device_id:
- print("Device connected successfully.")
- else:
- self.device.close()
- print(
- "Connected device is not a GDM8351. Please check the device ID."
- )
- continue
-
- self.device_connected = True
- except Exception as e:
- print(f"Failed to connect to {device_name}: {e}")
-
- break
-
- def get_id(self) -> str:
- if self.device is None or not self.device_connected:
- raise Exception("Device not connected.")
-
- return self.device.query("*IDN?")
-
- def configure_temperature_sensing(
- self, sensor_type: str = "K", junction_temp_deg: float = 29.0
- ) -> None:
-
- if sensor_type not in ["K", "J", "T"]:
- raise ValueError("Invalid sensor type. Use 'K', 'J', or 'T'.")
-
- if junction_temp_deg < 0 or junction_temp_deg > 50:
- raise ValueError(
- "Junction temperature must be between 0 and 50 degrees Celsius."
- )
-
- try:
- junction_temp_deg = float(junction_temp_deg)
- except ValueError:
- raise ValueError("Junction temperature must be a number.")
-
- try:
- self.device.write(f"CONF:TEMP:TCO {sensor_type}")
- self.device.write(f"SENS:TEMP:RJUN:SIM {junction_temp_deg:.1f}")
- except Exception as e:
- raise Exception(f"Failed to configure temperature sensing: {e}")
-
- return
-
- def read_temperature(self) -> float:
-
- if self.device is None or not self.device_connected:
- raise Exception("Device not connected.")
-
- try:
- return float(self.device.query("MEAS:TEMP:TCO?"))
- except Exception as e:
- raise Exception(f"Failed to read temperature: {e}")
-
- def log_temperature(
- self, output_directory: Path, test_time_id: str, verbose: bool = False
- ) -> None:
-
- # Log file name format:
- # > external_temp.<time_identifier>.csv
- # Example: external_temp.2506091307.csv
-
- file_path = output_directory / f"external_temp.{test_time_id}.csv"
-
- try:
- temp = self.read_temperature()
- except Exception as e:
- print(f"Failed to read temperature: {e}")
- return
-
- if not file_path.exists():
- # creat a file header
- with open(file_path, "w") as f:
- f.write("time,temperature\n")
-
- with open(file_path, "a") as f:
- f.write(str(time.time()) + "," + str(temp) + "\n")
-
- if verbose:
- print(f"GDM8351 temperature: {temp}°C")
-
- def close(self) -> None:
- if self.device is not None and self.device_connected:
- try:
- self.device.close()
- print("GDM8351 connection closed.")
- except Exception as e:
- print(f"GDM8351 Failed to close connection: {e}")
diff --git a/tools/automatic_battery_tester/hardware_ctl/relay_controller.py b/tools/automatic_battery_tester/hardware_ctl/relay_controller.py
deleted file mode 100644
index 203722f8..00000000
--- a/tools/automatic_battery_tester/hardware_ctl/relay_controller.py
+++ /dev/null
@@ -1,115 +0,0 @@
-# hardware_ctl/relay_controller.py
-
-import logging
-import platform
-import subprocess
-import sys
-
-# Import Deditec drives
-from .deditec import DeditecBsWeu16
-
-
-class RelayController:
-
- DEDITEC_PORT = 9912
- MAX_PIN = 16 # Max PIN index (1-16)
-
- def __init__(self, ip_address: str) -> None:
- """
- Initilize relay controller.
-
- Args:
- ip_address: Deditec board IP addres.
- """
-
- self.ip_address = ip_address
- self.port = self.DEDITEC_PORT
-
- # Ping the device to check connectivity
- if not self.check_ping(self.ip_address):
- logging.warning(
- "Ping to Deditec relay board failed. Network issue possible, but attempting TCP check."
- )
-
- self.deditec = DeditecBsWeu16(ip=self.ip_address, port=self.port)
-
- # Connect to deditec relay board
- if not self.deditec.connect():
- logging.error(
- f"Failed to connect to Deditec relay board at {self.ip_address}:{self.DEDITEC_PORT}."
- )
- sys.exit(1)
-
- def check_ping(self, ip: str) -> bool:
- """Ping the given IP address"""
- logging.info(f"Pinging {ip}...")
- system = platform.system().lower()
- if system == "windows":
- command = ["ping", "-n", "1", "-w", "1000", ip]
- else: # Linux, macOS
- command = ["ping", "-c", "1", "-W", "1", ip]
-
- try:
-
- process = subprocess.Popen(
- command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
- )
- stdout, stderr = process.communicate(timeout=3)
- return_code = process.returncode
-
- logging.debug(f"Ping stdout:\n{stdout}")
-
- if stderr:
- logging.debug(f"Ping stderr:\n{stderr}")
-
- if return_code == 0:
-
- if (
- "unreachable" in stdout.lower()
- or "timed out" in stdout.lower()
- or "ttl expired" in stdout.lower()
- ):
- logging.error(
- f"Ping to {ip} technically succeeded (code 0) but output indicates failure."
- )
- return False
-
- logging.info(f"Ping to {ip} successful.")
- return True
- else:
-
- logging.error(f"Ping to {ip} failed (return code: {return_code}).")
- return False
-
- except FileNotFoundError:
- logging.error(
- "Ping command not found. Install ping or check PATH. Skipping ping check."
- )
- return True
-
- except subprocess.TimeoutExpired:
- logging.error(f"Ping process to {ip} timed out.")
- return False
-
- except Exception as e:
- logging.error(f"Unknown error during ping check: {e}")
- return False
-
- def set_relay_off(self, pin: int) -> bool:
-
- if not self.deditec:
- logging.error("RelayController: Deditec not initialized.")
- return False
-
- return self.deditec.control_relay(pins_on=[], pins_off=[pin])
-
- def set_relay_on(self, pin: int) -> bool:
-
- if not self.deditec:
- logging.error("RelayController: Deditec not initialized.")
- return False
-
- return self.deditec.control_relay(pins_on=[pin], pins_off=[])
-
- def close(self) -> None:
- pass
diff --git a/tools/automatic_battery_tester/main_tester.py b/tools/automatic_battery_tester/main_tester.py
deleted file mode 100644
index 7d3f5881..00000000
--- a/tools/automatic_battery_tester/main_tester.py
+++ /dev/null
@@ -1,344 +0,0 @@
-# main_tester.py
-from __future__ import annotations
-
-import logging
-import sys
-import time
-from pathlib import Path
-from typing import Any, Dict, Optional
-
-import toml
-from dut import DutController
-from hardware_ctl import RelayController
-from notifications import send_slack_message
-from test_logic import LinearScenario, RandomWonderScenario, SwitchingScenario
-
-# Configure logging
-log_formatter = log_formatter = logging.Formatter(
- "[%(levelname).1s %(asctime)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
-)
-log_file = "test.log"
-
-logger = logging.getLogger()
-logger.setLevel(logging.DEBUG)
-
-# Clear all existing handlers to avoid duplicates
-if logger.hasHandlers():
- logger.handlers.clear()
-
-# File handler
-try:
- file_handler = logging.FileHandler(log_file, mode="w", encoding="utf-8")
- file_handler.setFormatter(log_formatter)
- file_handler.setLevel(logging.INFO)
- logger.addHandler(file_handler)
-except Exception as log_e:
- print(f"WARNING: Failed to create file log handler for {log_file}: {log_e}")
-
-# Console handler
-console_handler = logging.StreamHandler(sys.stdout)
-console_handler.setFormatter(log_formatter)
-console_handler.setLevel(logging.INFO)
-logger.addHandler(console_handler)
-
-
-def load_config(config_path: str = "test_config.toml") -> Optional[Dict[str, Any]]:
- """Load test configuration from TOML config file ."""
- config_file = config_path
- logging.info(f"Loading configuration file: {config_file}")
- try:
- config = toml.load(config_file)
-
- # Validate required sections in the config file
- required_sections = ["general", "relay", "duts", "test_plan", "notifications"]
-
- for section in required_sections:
- if section not in config:
- raise ValueError(
- f"Missing required section '{section}' in config file."
- )
-
- logging.info("Config file loading successfully.")
- return config
-
- except FileNotFoundError:
- logging.error(f"Configuration file not found at {config_file}")
- return None
-
- except toml.TomlDecodeError as e:
- logging.error(f"Error decoding TOML configuration file: {e}")
- return None
-
- except Exception as e:
- logging.error(f"Error loading configuration file: {e}")
- return None
-
-
-def run_test_cycle(
- config: dict[str, Any],
- temp_c: float,
- cycle_num: int,
- test_mode: str,
- relay_ctl: RelayController,
- dut_ctl: DutController,
-) -> bool:
- """Run single test cycle for given test mode on all DUTs at specified temperature"""
-
- # Select test scenario
- if test_mode == "linear":
-
- test_scenario = LinearScenario(
- discharge_load=config["test_plan"]["linear_discharge_load"],
- relaxation_time_min=config["test_plan"]["linear_relaxation_time_min"],
- )
-
- elif test_mode == "switching":
-
- test_scenario = SwitchingScenario(
- discharge_switch_cycle_min=config["test_plan"][
- "switching_discharge_switch_cycle_min"
- ],
- relaxation_time_min=config["test_plan"]["switching_relaxation_time_min"],
- )
-
- elif test_mode == "random_wonder":
-
- test_scenario = RandomWonderScenario(
- core_test_time=config["test_plan"]["random_wonder_core_test_time_min"],
- relaxation_time_min=config["test_plan"][
- "random_wonder_relaxation_time_min"
- ],
- )
-
- else:
-
- logging.error(f"Unknown test mode: {test_mode}. Cannot run test cycle.")
- return False
-
- # Setup
- test_scenario.setup(dut_controller=dut_ctl)
-
- # Test loop
- while True:
-
- # Call run function in loop to execute the test scenario.
- finished = test_scenario.run(dut_controller=dut_ctl)
-
- # Read power manager report and log them to file
- test_scenario.log_data(
- dut_controller=dut_ctl,
- output_directory=Path(config["general"]["output_directory"]),
- temp=temp_c,
- )
-
- if finished:
- break # Exit test loop
-
- time.sleep(1)
-
- # Tear down test scenario
- test_scenario.teardown(dut_controller=dut_ctl)
-
- return True
-
-
-def main() -> None:
-
- logging.info("==============================================")
- logging.info(" Starting Automated Battery Cycle Tester ")
- logging.info("==============================================")
-
- config = load_config()
- if config is None:
- logging.critical("Failed to load configuration. Exiting.")
- sys.exit(1)
-
- # Initialize hardware controllers
- logging.info("Initializing hardware controllers...")
- relay_ctl = None
- dut_ctl = None
-
- logging.info("==============================================")
- logging.info(" Initializing Peripherals ")
- logging.info("==============================================")
-
- try:
-
- # Initialize relay controller (Deditec board)
- relay_ctl = RelayController(ip_address=config["relay"]["ip_address"])
-
- # Initialize DUTs
- from dut.dut_controller import DutController
-
- dut_ctl = DutController(config["duts"], relay_ctl=relay_ctl, verbose=False)
-
- except Exception as e:
- logging.exception(f"Failed to intialize peripherals: {e}")
- exit(1)
-
- # Create output data directory
- output_data_dir = Path(config["general"]["output_directory"])
- try:
- output_data_dir.mkdir(parents=True, exist_ok=True)
- logging.info(f"Test results will be saved in: {output_data_dir.resolve()}")
- except OSError as e:
- logging.critical(
- f"Failed to create output directory {output_data_dir}: {e}. Exiting."
- )
- if relay_ctl:
- relay_ctl.close()
- if dut_ctl:
- dut_ctl.close()
- sys.exit(1)
-
- ############################################################################
- # LOAD TEST PLAN
- ############################################################################
-
- logging.info("==============================================")
- logging.info(" TEST PLAN LOADING ")
- logging.info("==============================================")
-
- temperatures = config["test_plan"].get("temperatures_celsius", [25])
- test_modes_to_run = config["test_plan"].get("test_modes", ["linear"])
- cycles_per_temp = config["test_plan"].get("cycles_per_temperature", 1)
- total_runs = len(temperatures) * cycles_per_temp * len(test_modes_to_run)
- completed_runs = 0
-
- logger.info(f" + Tested temperatures : {temperatures}")
- logger.info(f" + Cycles per temperature : {cycles_per_temp}")
- logger.info(f" + Test modes in every cycle : {test_modes_to_run}")
- logger.info(f" + Total runs planned : {total_runs}")
-
- start_time = time.time()
- test_aborted = False
-
- ############################################################################
- # MAIN TEST LOOP
- ############################################################################
- try:
-
- # Run a full test cycle for each temperature setting.
- # since the temperature chamber is still controlled only manually, every
- # test will notify the user to set the temperature manually before it
- # start next iteration.
- for temp_c in temperatures:
-
- # Set temperature in temperature chamber
- logging.info("==============================================")
- logging.info(f" {temp_c} °C TEMP TEST ")
- logging.info("==============================================")
- logging.info(
- f"Set the temperature chamber to {temp_c} °C and wait for stabilization."
- )
-
- try:
- if config["notifications"]["notification_channel"] == "slack":
- send_slack_message(
- config["notifications"]["slack_webhook_url"],
- f"""Set the temperature chamber to {temp_c} °C and confirm to continue with the test.""",
- )
- except Exception as e:
- logging.error(f"Failed to send Slack notification: {e}")
-
- while True:
- user_input = input(f"Confirm temperature is set to {temp_c} °C (Y)?")
- if user_input.lower() == "y":
- break
-
- for cycle_num in range(cycles_per_temp):
-
- for test_mode in test_modes_to_run:
-
- run_start_time = time.time()
- logging.info(f"Running Test Mode: '{test_mode}'")
-
- success = run_test_cycle(
- config, temp_c, cycle_num, test_mode, relay_ctl, dut_ctl
- )
-
- run_end_time = time.time()
- run_duration_m = (run_end_time - run_start_time) / 60
-
- if success:
- completed_runs += 1
- logging.info(
- f"Test Mode '{test_mode}' finished successfully in {run_duration_m:.1f} minutes."
- )
- else:
- logging.error(
- f"Test Mode '{test_mode}' failed after {run_duration_m:.1f} minutes."
- )
- if config.get("general", {}).get("fail_fast", False):
- logging.warning(
- "Fail fast enabled. Aborting entire test plan."
- )
- test_aborted = True
- break
- else:
- logging.info(
- "Continuing with the next mode/cycle/temperature."
- )
-
- logging.info(
- f"Progress: {completed_runs}/{total_runs} total runs completed."
- )
-
- if test_aborted:
- break
-
- if test_aborted:
- break
-
- logging.info("==============================================")
- logging.info(" TEST FINISHED ")
- logging.info("==============================================")
-
- except KeyboardInterrupt:
- logging.warning("Test execution interrupted by user (Ctrl+C)")
- test_aborted = True
- except Exception as e:
- logging.exception(f"FATAL ERROR during test execution: {e}")
- test_aborted = True
- finally:
-
- logging.info("Performing final cleanup...")
- if relay_ctl:
- try:
- logging.info("Ensuring all relays are OFF...")
- dut_ctl.power_down_all() # Power down all DUTs
- relay_ctl.close()
- except Exception as e_relay:
- logging.error(f"Error during relay cleanup: {e_relay}")
-
- if dut_ctl:
- try:
- dut_ctl.close()
- except Exception as e_dut:
- logging.error(f"Error during DUT controller cleanup: {e_dut}")
-
- logging.info("==================== TEST SUMMARY ====================")
- end_time = time.time()
- total_duration_s = end_time - start_time
- total_duration_h = total_duration_s / 3600
- status = (
- "ABORTED"
- if test_aborted
- else (
- "COMPLETED" if completed_runs == total_runs else "PARTIALLY COMPLETED"
- )
- )
- logging.info("-" * 60)
- logging.info(f"Test execution {status}.")
- logging.info(f"Total runs completed: {completed_runs}/{total_runs}")
- logging.info(
- f"Total duration: {total_duration_s:.0f} seconds ({total_duration_h:.2f} hours)."
- )
- logging.info("==================== TEST END ====================")
-
-
-if __name__ == "__main__":
- # Ensure at least one handler is set up
- if not logger.hasHandlers():
- logger.addHandler(logging.StreamHandler(sys.stdout))
- main()
diff --git a/tools/automatic_battery_tester/notifications.py b/tools/automatic_battery_tester/notifications.py
deleted file mode 100644
index afefcb5b..00000000
--- a/tools/automatic_battery_tester/notifications.py
+++ /dev/null
@@ -1,86 +0,0 @@
-# notifications.py
-
-import json
-import logging
-from typing import Optional
-
-import requests
-
-logger = logging.getLogger(__name__)
-
-
-def send_slack_message(
- webhook_url: Optional[str],
- message: str,
- fallback_text: str = "Notification from Battery Tester",
-) -> bool:
- """
- Send slack message using Incoming Webhook URL
-
- Args:
- webhook_url: Slack webhook URL.
- message: Message text (may consist of markdown).
- fallback_text: Text, který se zobrazí v notifikacích.
-
- Returns:
- True if the message was successfully sent (status code 2xx).
- """
- if not webhook_url:
- logger.error("Slack Error: Webhook URL is not configured.")
- return False
- if not message:
- logger.warning("Slack Warning: Attempting to send an empty message.")
-
- slack_data = {
- "text": fallback_text,
- "blocks": [{"type": "section", "text": {"type": "mrkdwn", "text": message}}],
- }
-
- try:
-
- payload_json_string = json.dumps(slack_data)
-
- post_data = {"payload": payload_json_string}
-
- logger.info(
- "Sending Slack notification via webhook (using payload parameter)..."
- )
- logger.debug(f"Slack Webhook URL: {webhook_url}")
- logger.debug(f"Slack Payload (JSON String): {payload_json_string}")
-
- timeout_seconds = 15
- response = requests.post(
- webhook_url,
- data=post_data,
- # headers={'Content-Type': 'application/x-www-form-urlencoded'}
- timeout=timeout_seconds,
- )
-
- if response.status_code == 200 and response.text.lower() == "ok":
- logger.info("Slack notification request sent successfully.")
- return True
- else:
- error_detail = (
- f"Status: {response.status_code}, Response: '{response.text[:500]}...'"
- )
- logger.error(f"Slack request failed. {error_detail}")
- if response.status_code == 400 and "invalid_payload" in response.text:
- logger.error(
- "Slack Error Detail: The JSON payload structure might be incorrect."
- )
- elif response.status_code == 403:
- logger.error(
- "Slack Error Detail: Forbidden - Check webhook URL validity or permissions."
- )
- elif response.status_code == 404:
- logger.error(
- "Slack Error Detail: Not Found - The webhook URL might be incorrect or deactivated."
- )
- return False
-
- except requests.exceptions.RequestException as e:
- logger.error(f"Slack request failed (RequestException): {e}")
- return False
- except Exception as e:
- logger.exception(f"An unexpected error occurred during Slack notification: {e}")
- return False
diff --git a/tools/automatic_battery_tester/requirements.txt b/tools/automatic_battery_tester/requirements.txt
deleted file mode 100644
index cba02805..00000000
--- a/tools/automatic_battery_tester/requirements.txt
+++ /dev/null
@@ -1,36 +0,0 @@
-argcomplete==3.6.2
-certifi==2025.6.15
-charset-normalizer==3.4.2
-click==8.2.1
-contourpy==1.3.2
-cycler==0.12.1
-idna==3.10
-ifaddr==0.2.0
-inquirerpy==0.3.4
-kiwisolver==1.4.8
-matplotlib==3.10.3
-numpy==2.3.1
-packaging==25.0
-pandas==2.3.0
-pfzy==0.3.4
-pillow==12.2.0
-prompt_toolkit==3.0.51
-psutil==7.0.0
-pyparsing==3.2.3
-PyQt5==5.15.11
-PyQt5-Qt5==5.15.17
-PyQt5_sip==12.17.0
-pyserial==3.5
-python-dateutil==2.9.0.post0
-pytz==2025.2
-PyVISA==1.15.0
-PyVISA-py==0.8.0
-requests==2.33.0
-scipy==1.16.0
-six==1.17.0
-toml==0.10.2
-typing_extensions==4.14.0
-tzdata==2025.2
-urllib3==2.7.0
-wcwidth==0.2.13
-zeroconf==0.147.0
diff --git a/tools/automatic_battery_tester/single_capture.py b/tools/automatic_battery_tester/single_capture.py
deleted file mode 100644
index ee692df0..00000000
--- a/tools/automatic_battery_tester/single_capture.py
+++ /dev/null
@@ -1,140 +0,0 @@
-import sys
-import time
-from pathlib import Path
-
-from dut import Dut
-from hardware_ctl.gdm8351 import GDM8351
-from serial.tools import list_ports
-
-output_directory = Path("single_capture_test_results")
-test_description = "non_specified_test"
-temp_description = "ambient"
-
-
-"""
-This script will connect to a signle DUT over VCP port and will run a continous
-log of the power manager data (continously calls pm-report command) into the
-log file. User can also select to log the temepertature readings from an
-external thermocouple sensor connected to the GDM8351 multimeter.
-"""
-
-
-def main() -> None:
-
- print("**********************************************************")
- print(" DUT port selection ")
- print("**********************************************************")
-
- ports = list_ports.comports()
-
- available_ports = {}
- port_count = 0
- print("Available VCP ports:")
- for port in ports:
- if "ACM" in port.device:
- port_count += 1
- available_ports[port_count] = port.device
- print(f" [{port_count}]: {port.device} - {port.description}")
-
- if port_count == 0:
- print("No device conneceted. Exiting.")
- return
-
- dut_port_selection = input("Select VCP port number (or Q to quit the selection): ")
-
- if dut_port_selection.lower() == "q":
- print("Exiting script.")
- sys.exit(0)
-
- selected_port = None
- for port_id, port_name in available_ports.items():
- if int(dut_port_selection) == port_id:
- selected_port = port_name
- break
-
- try:
- dut = Dut(name="Trezor", usb_port=selected_port)
- except Exception as e:
- print(f"Failed to initialize DUT on port {selected_port}: {e}")
- sys.exit(1)
- # Initialize DUT
-
- print("**********************************************************")
- print(" GDM8351 port selection (temp measurement) ")
- print("**********************************************************")
-
- # Initialize the GDM8351 multimeter
- gdm8351 = GDM8351()
-
- # Get the device ID to confirm connection
- try:
- device_id = gdm8351.get_id()
- print(f"Connected to device: {device_id}")
- except Exception as e:
- print(f"Error getting device ID: {e}")
- return
-
- # Configure temperature sensing
- try:
- gdm8351.configure_temperature_sensing(sensor_type="K", junction_temp_deg=29.0)
- print("Temperature sensing configured successfully.")
- except ValueError as ve:
- print(f"Configuration error: {ve}")
- except Exception as e:
- print(f"Error configuring temperature sensing: {e}")
-
- # Creat test time ID
- test_time_id = f"{time.strftime('%y%m%d%H%M')}"
-
- # Create output data directory
- try:
- output_directory.mkdir(parents=True, exist_ok=True)
- except OSError as e:
- print("Failed to create output directory:", e)
- sys.exit(1)
-
- #########################################################################
- # Test setup section
- #########################################################################
-
- dut.set_soc_limit(100)
- dut.set_backlight(100)
- dut.enable_charging()
-
- #########################################################################
- # Main test loop
- #########################################################################
- try:
-
- while True:
-
- dut.log_data(
- output_directory=output_directory,
- test_time_id=test_time_id,
- test_scenario="single_capture",
- test_phase=test_description,
- temp=temp_description,
- verbose=True,
- )
-
- # Read temperature from GDM8351
- gdm8351.log_temperature(
- output_directory=output_directory,
- test_time_id=test_time_id,
- verbose=True,
- )
-
- time.sleep(1)
-
- except KeyboardInterrupt:
- print("Test execution interrupted by user (Ctrl+C)")
- except Exception as e:
- print(f"FATAL ERROR during test execution: {e}")
- finally:
-
- dut.close()
- gdm8351.close()
-
-
-if __name__ == "__main__":
- main()
diff --git a/tools/automatic_battery_tester/test_config.toml b/tools/automatic_battery_tester/test_config.toml
deleted file mode 100644
index 13be85a5..00000000
--- a/tools/automatic_battery_tester/test_config.toml
+++ /dev/null
@@ -1,67 +0,0 @@
-
-[general]
-
-output_directory = "test_results"
-log_interval_seconds = 1
-
-################################################################################
-# DEVICE UNDER TEST (DUT) LIST
-################################################################################
-
-[[duts]]
-name = "DUT1"
-cpu_id = "05001D000A50325557323120"
-usb_port = "/dev/ttyACM0"
-relay_port = 1
-
-#[[duts]]
-#name = "DUT2"
-#cpu_id = "1C0023000A50325557323120"
-#usb_port = "/dev/ttyACM0"
-#relay_port = 1
-
-# Uncomment and edit to add more DUTs
-#[[duts]]
-#name = "DUT3"
-#usb_port = "/dev/ttyACM2"
-#relay_port = 3
-
-################################################################################
-# RELAY BOADR (DEDITEC) SETTINGS
-################################################################################
-
-[relay]
-ip_address = "192.168.1.10" # Deditec board IP address
-
-################################################################################
-# TEST PLAN
-################################################################################
-
-[test_plan]
-
-temperatures_celsius = [15, 20, 25, 30, 35, 40, 45] # Temp scenario lists
-cycles_per_temperature = 3
-test_modes = ["linear", "switching", "random_wonder"] # Test modes run in every cycle
-
-# Linear test mode parameters
-linear_discharge_load = 0 # could be set in range 0-255
-linear_relaxation_time_min = 60
-
-# Switching test mode parameters
-switching_discharge_switch_cycle_min = 5 # Time interval between switching load step
-switching_relaxation_time_min = 60
-
-# Random wonder test mode parameters
-random_wonder_core_test_time_min = 120
-random_wonder_relaxation_time_min = 30
-
-################################################################################
-# NOTIFICATION SETTINGS
-################################################################################
-
-[notifications]
-notification_channel = "slack"
-
-slack_webhook_url = "<Fill your webhook URL>"
-
-
diff --git a/tools/automatic_battery_tester/test_logic/__init__.py b/tools/automatic_battery_tester/test_logic/__init__.py
deleted file mode 100644
index 3b37d7f6..00000000
--- a/tools/automatic_battery_tester/test_logic/__init__.py
+++ /dev/null
@@ -1,5 +0,0 @@
-from .linear_scenario import LinearScenario
-from .random_wonder_scenario import RandomWonderScenario
-from .switching_scenario import SwitchingScenario
-
-__all__ = ["LinearScenario", "RandomWonderScenario", "SwitchingScenario"]
diff --git a/tools/automatic_battery_tester/test_logic/linear_scenario.py b/tools/automatic_battery_tester/test_logic/linear_scenario.py
deleted file mode 100644
index 441084f2..00000000
--- a/tools/automatic_battery_tester/test_logic/linear_scenario.py
+++ /dev/null
@@ -1,144 +0,0 @@
-import enum
-import logging
-import time
-from pathlib import Path
-
-from dut.dut_controller import DutController
-
-from .test_scenario import TestScenario
-
-SKIP_CHARGING = False
-SKIP_RELAXING = False
-SKIP_DISCHARGING = False
-
-
-class ScenarioPhase(enum.Enum):
- NOT_STARTED = 0
- CHARGING = 1
- CHARGED_RELAXING = 2
- DISCHARGING = 3
- DISCHARGED_RELAXING = 4
- DONE = 5
-
-
-class LinearScenario(TestScenario):
-
- def __init__(
- self, discharge_load: int = 100, relaxation_time_min: int = 60
- ) -> None:
-
- # DUT use display backlight intensity to change its load (discharge
- # current). Backlight intensity could be set in range of 0-255, but
- # there is no direct translation into discharge current.
- # Typically the discharge current is ~aprx 80mA with backlight set to 0
- # and 220mA when set to max 225.
- self.discharge_load = discharge_load
- self.relaxation_time_min = relaxation_time_min
- self.phase_start = time.time()
- self.remaining_time = time.time()
- self.scenario_phase = ScenarioPhase.NOT_STARTED
- self.time_id = "0000000000"
- self.previous_phase = ScenarioPhase.NOT_STARTED
-
- def setup(self, dut_controller: DutController) -> None:
-
- # Start with charging phase first, so connect the charger with relay
- # and enable the charging.
- self.scenario_phase = ScenarioPhase.CHARGING
- dut_controller.power_up_all() # Power up all DUTs
-
- # Wait for DUTs to power up
- time.sleep(3)
-
- dut_controller.set_backlight(150)
- dut_controller.set_soc_limit(100)
- dut_controller.enable_charging()
- time.sleep(1) # Give some time for the command to be processed
-
- self.phase_start = time.time()
- self.test_time_id = f"{time.strftime('%y%m%d%H%M')}"
-
- def run(self, dut_controller: DutController) -> bool:
-
- if self.previous_phase != self.scenario_phase:
- logging.info(f"Linear scenario entered {self.scenario_phase} phase.")
- self.previous_phase = self.scenario_phase
-
- # Charge until all DuT is above certain voltage threshold and charging
- # state is IDLE (means that PMIC automatically stopped charging)
- # Move to next phase when all DUTs are charged.
- if self.scenario_phase == ScenarioPhase.CHARGING:
-
- if dut_controller.all_duts_charged() or SKIP_CHARGING:
- self.scenario_phase = ScenarioPhase.CHARGED_RELAXING
- self.phase_start = time.time()
-
- # Relax
- elif self.scenario_phase == ScenarioPhase.CHARGED_RELAXING:
-
- # Relaxation time is set in minutes, so convert to seconds
- if (time.time() - self.phase_start) >= (
- self.relaxation_time_min * 60
- ) or SKIP_RELAXING:
-
- dut_controller.power_down_all()
- dut_controller.set_backlight(self.discharge_load)
- self.scenario_phase = ScenarioPhase.DISCHARGING
- self.phase_start = time.time()
-
- else:
-
- elapsed_min = int((time.time() - self.phase_start) / 60)
-
- if self.remaining_time != self.relaxation_time_min - elapsed_min:
- # Update remaining time only if it changed
- self.remaining_time = self.relaxation_time_min - elapsed_min
- logging.info(
- f"Relaxing for {self.relaxation_time_min} minutes, remaining: {self.remaining_time} minutes"
- )
-
- elif self.scenario_phase == ScenarioPhase.DISCHARGING:
-
- if dut_controller.all_duts_discharged() or SKIP_DISCHARGING:
-
- self.scenario_phase = ScenarioPhase.DISCHARGED_RELAXING
- self.phase_start = time.time()
-
- elif self.scenario_phase == ScenarioPhase.DISCHARGED_RELAXING:
-
- # Relaxation time is set in minutes, so convert to seconds
- if (time.time() - self.phase_start) >= (
- self.relaxation_time_min * 60
- ) or SKIP_RELAXING:
-
- self.scenario_phase = ScenarioPhase.DONE
- logging.info("Scenario completed successfully.")
- return True
-
- else:
- elapsed_min = int((time.time() - self.phase_start) / 60)
-
- if self.remaining_time != self.relaxation_time_min - elapsed_min:
- # Update remaining time only if it changed
- self.remaining_time = self.relaxation_time_min - elapsed_min
- logging.info(
- f"Relaxing for {self.relaxation_time_min} minutes, remaining: {self.remaining_time} minutes"
- )
-
- # Relax
- return False
-
- def log_data(
- self, dut_controller: DutController, output_directory: Path, temp: float
- ) -> None:
-
- dut_controller.log_data(
- output_directory,
- self.test_time_id,
- "linear",
- self.scenario_phase.name.lower(),
- temp,
- )
-
- def teardown(self, dut_controller: DutController) -> None:
- pass
diff --git a/tools/automatic_battery_tester/test_logic/random_wonder_scenario.py b/tools/automatic_battery_tester/test_logic/random_wonder_scenario.py
deleted file mode 100644
index 74e304fb..00000000
--- a/tools/automatic_battery_tester/test_logic/random_wonder_scenario.py
+++ /dev/null
@@ -1,240 +0,0 @@
-from __future__ import annotations
-
-import enum
-import logging
-import random
-import time
-from pathlib import Path
-
-from dut.dut_controller import DutController
-
-from .test_scenario import TestScenario
-
-SKIP_CHARGING = False
-SKIP_RELAXING = False
-SKIP_RANDOM_WONDER = False
-SKIP_DISCHARGING = False
-
-
-class ScenarioPhase(enum.Enum):
- NOT_STARTED = 0
- CHARGING = 1
- CHARGED_RELAXING = 2
- RANDOM_WONDER = 3
- DISCHARGING = 4
- DISCHARGED_RELAXING = 5
- DONE = 6
-
-
-class RandomWonderScenario(TestScenario):
-
- def __init__(self, core_test_time: int = 60, relaxation_time_min: int = 60) -> None:
-
- self.relaxation_time_min = relaxation_time_min
- self.phase_start = time.time()
- self.core_test_time = core_test_time
- self.random_cycle_time_min = 10
- self.random_direction_up = True
- self.random_discharge_load = 100
- self.random_cycle_start = 0
- self.remaining_time = time.time()
- self.scenario_phase = ScenarioPhase.NOT_STARTED
- self.time_id = "0000000000"
- self.previous_phase = ScenarioPhase.NOT_STARTED
-
- def setup(self, dut_controller: DutController) -> None:
-
- # Start with charging phase first, so connect the charger with relay
- # and enable the charging.
- self.scenario_phase = ScenarioPhase.CHARGING
- dut_controller.power_up_all() # Power up all DUTs
-
- # Wait for DUTs to power up
- time.sleep(3)
-
- dut_controller.set_backlight(150)
- dut_controller.set_soc_limit(100)
- dut_controller.enable_charging()
-
- time.sleep(1) # Give some time for the command to be processed
-
- self.phase_start = time.time()
- self.test_time_id = f"{time.strftime('%y%m%d%H%M')}"
-
- def run(self, dut_controller: DutController) -> bool:
-
- if self.previous_phase != self.scenario_phase:
- logging.info(f"Random Wonder scenario entered {self.scenario_phase} phase.")
- self.previous_phase = self.scenario_phase
-
- # Charge until all DuT is above certain voltage threshold and charging
- # state is IDLE (means that PMIC automatically stopped charging)
- # Move to next phase when all DUTs are charged.
- if self.scenario_phase == ScenarioPhase.CHARGING:
-
- if dut_controller.all_duts_charged() or SKIP_CHARGING:
- self.scenario_phase = ScenarioPhase.CHARGED_RELAXING
- self.phase_start = time.time()
-
- # Relax
- elif self.scenario_phase == ScenarioPhase.CHARGED_RELAXING:
-
- # Relaxation time is set in minutes, so convert to seconds
- if (time.time() - self.phase_start) >= (
- self.relaxation_time_min * 60
- ) or SKIP_RELAXING:
-
- dut_controller.power_down_all()
- self.scenario_phase = ScenarioPhase.RANDOM_WONDER
- self.phase_start = time.time()
-
- else:
-
- elapsed_min = int((time.time() - self.phase_start) / 60)
- if self.remaining_time != self.relaxation_time_min - elapsed_min:
- # Update remaining time only if it changed
- self.remaining_time = self.relaxation_time_min - elapsed_min
- logging.info(
- f"Relaxing for {self.relaxation_time_min} minutes, remaining: {self.remaining_time} minutes"
- )
-
- elif self.scenario_phase == ScenarioPhase.RANDOM_WONDER:
-
- # Random Wonder phase, switch the backlight intensity to change
- # the discharge current.
- if (time.time() - self.phase_start) >= (
- self.core_test_time * 60
- ) or SKIP_RANDOM_WONDER:
-
- # Random wonder test over
- dut_controller.power_down_all()
- dut_controller.set_backlight(100)
- self.scenario_phase = ScenarioPhase.DISCHARGING
-
- # Reset phase start time
- self.phase_start = time.time()
-
- elif self.random_direction_up and dut_controller.any_dut_charged():
-
- # One of the DUTs is charged, bounce back to discharging.
- logging.info("One of the DUTs is charged, switching back to relaxing.")
- self.random_direction_up = False
-
- dut_controller.power_down_all()
- dut_controller.disable_charging()
-
- self.random_cycle_time_min = random.randint(
- 5, 15
- ) # Random cycle time between 5 and 15 minutes
- self.random_discharge_load = random.randint(
- 0, 225
- ) # Random discharge load between 0 and 225
- dut_controller.set_backlight(
- self.random_discharge_load
- ) # Set backlight to 0 to stop discharge
-
- self.random_cycle_start = time.time()
-
- elif not self.random_direction_up and dut_controller.any_dut_discharged():
-
- # One of the DUTs got discharged, bounce back to charging.
-
- logging.info(
- "One of the DUTs got discharged, switching back to charging."
- )
- self.random_direction_up = True
-
- dut_controller.power_up_all() # Power up all DUTs
- dut_controller.enable_charging() # Enable charging
-
- self.random_cycle_time_min = random.randint(
- 5, 15
- ) # Random cycle time between 5 and 15 minutes
- self.random_cycle_start = time.time()
-
- else:
-
- if (time.time() - self.random_cycle_start) >= (
- self.random_cycle_time_min * 60
- ):
-
- # Randomize following section
- self.random_direction_up = random.choice([True, False])
- self.random_cycle_time_min = random.randint(
- 5, 15
- ) # Random cycle time between 5 and 15 minutes
-
- if self.random_direction_up:
-
- # Enable charging
- dut_controller.power_up_all() # Power up all DUTs
- dut_controller.enable_charging()
-
- else:
-
- self.random_discharge_load = random.randint(
- 0, 225
- ) # Random discharge load between 0 and 225
-
- dut_controller.disable_charging() # Disable charging
- dut_controller.power_down_all() # Power down all DUTs
- dut_controller.set_backlight(
- self.random_discharge_load
- ) # Set backlight to 0 to stop discharge
-
- # Reset phase start time
- self.random_cycle_start = time.time()
-
- logging.info(
- f"Random Wonder cycle changed: "
- f"direction={'up' if self.random_direction_up else 'down'}, "
- f"cycle_time={self.random_cycle_time_min}min, "
- f"discharge_load={self.random_discharge_load}"
- )
-
- # Continue in random wonder phase
-
- elif self.scenario_phase == ScenarioPhase.DISCHARGING:
-
- if dut_controller.all_duts_discharged() or SKIP_DISCHARGING:
- self.scenario_phase = ScenarioPhase.DISCHARGED_RELAXING
- self.phase_start = time.time()
-
- elif self.scenario_phase == ScenarioPhase.DISCHARGED_RELAXING:
-
- # Relaxation time is set in minutes, so convert to seconds
- if (time.time() - self.phase_start) >= (
- self.relaxation_time_min * 60
- ) or SKIP_RELAXING:
-
- self.scenario_phase = ScenarioPhase.DONE
- logging.info("Scenario completed successfully.")
- return True
-
- else:
- elapsed_min = int((time.time() - self.phase_start) / 60)
-
- if self.remaining_time != self.relaxation_time_min - elapsed_min:
- # Update remaining time only if it changed
- self.remaining_time = self.relaxation_time_min - elapsed_min
- logging.info(
- f"Relaxing for {self.relaxation_time_min} minutes, remaining: {self.remaining_time} minutes"
- )
-
- # Relax
- return False
-
- def log_data(
- self, dut_controller: DutController, output_directory: Path, temp: float | int
- ) -> None:
-
- dut_controller.log_data(
- output_directory,
- self.test_time_id,
- "random_wonder",
- self.scenario_phase.name.lower(),
- temp,
- )
-
- def teardown(self, dut_controller: DutController) -> None:
- pass
diff --git a/tools/automatic_battery_tester/test_logic/switching_scenario.py b/tools/automatic_battery_tester/test_logic/switching_scenario.py
deleted file mode 100644
index 6699550a..00000000
--- a/tools/automatic_battery_tester/test_logic/switching_scenario.py
+++ /dev/null
@@ -1,160 +0,0 @@
-from __future__ import annotations
-
-import enum
-import logging
-import time
-from pathlib import Path
-
-from dut.dut_controller import DutController
-
-from .test_scenario import TestScenario
-
-SKIP_CHARGING = False
-SKIP_RELAXING = False
-SKIP_DISCHARGING = False
-
-
-class ScenarioPhase(enum.Enum):
- NOT_STARTED = 0
- CHARGING = 1
- CHARGED_RELAXING = 2
- DISCHARGING = 3
- DISCHARGED_RELAXING = 4
- DONE = 5
-
-
-class SwitchingScenario(TestScenario):
-
- def __init__(
- self, discharge_switch_cycle_min: int = 5, relaxation_time_min: int = 60
- ) -> None:
-
- # DUT use display backlight intensity to change its load (discharge
- # current). Backlight intensity could be set in range of 0-255, but
- # there is no direct translation into discharge current.
- # Typically the discharge current is ~aprx 80mA with backlight set to 0
- # and 220mA when set to max 225.
- self.discharge_switch_cycle_min = discharge_switch_cycle_min
- self.relaxation_time_min = relaxation_time_min
- self.discharge_load = 100 # initial discharge load
- self.phase_start = time.time()
- self.remaining_time = time.time()
- self.scenario_phase = ScenarioPhase.NOT_STARTED
- self.test_time_id = "0000000000"
- self.previous_phase = ScenarioPhase.NOT_STARTED
-
- def setup(self, dut_controller: DutController) -> None:
-
- # Start with charging phase first, so connect the charger with relay
- # and enable the charging.
- self.scenario_phase = ScenarioPhase.CHARGING
- dut_controller.power_up_all() # Power up all DUTs
-
- # Wait for DUTs to power up
- time.sleep(3)
-
- dut_controller.set_backlight(150)
- dut_controller.set_soc_limit(100)
- dut_controller.enable_charging()
-
- time.sleep(1) # Give some time for the command to be processed
-
- self.phase_start = time.time()
- self.test_time_id = f"{time.strftime('%y%m%d%H%M')}"
-
- def run(self, dut_controller: DutController) -> bool:
-
- if self.previous_phase != self.scenario_phase:
- logging.info(f"Switching scenario entered {self.scenario_phase} phase.")
- self.previous_phase = self.scenario_phase
-
- # Charge until all DuT is above certain voltage threshold and charging
- # state is IDLE (means that PMIC automatically stopped charging)
- # Move to next phase when all DUTs are charged.
- if self.scenario_phase == ScenarioPhase.CHARGING:
-
- if dut_controller.all_duts_charged() or SKIP_CHARGING:
- self.scenario_phase = ScenarioPhase.CHARGED_RELAXING
- self.phase_start = time.time()
-
- # Relax
- elif self.scenario_phase == ScenarioPhase.CHARGED_RELAXING:
-
- # Relaxation time is set in minutes, so convert to seconds
- if (time.time() - self.phase_start) >= (
- self.relaxation_time_min * 60
- ) or SKIP_RELAXING:
-
- dut_controller.power_down_all()
- self.scenario_phase = ScenarioPhase.DISCHARGING
- self.phase_start = time.time()
- else:
-
- elapsed_min = int((time.time() - self.phase_start) / 60)
-
- if self.remaining_time != self.relaxation_time_min - elapsed_min:
- # Update remaining time only if it changed
- self.remaining_time = self.relaxation_time_min - elapsed_min
- logging.info(
- f"Relaxing for {self.relaxation_time_min} minutes, remaining: {self.remaining_time} minutes"
- )
-
- elif self.scenario_phase == ScenarioPhase.DISCHARGING:
-
- if dut_controller.all_duts_discharged() or SKIP_DISCHARGING:
- self.scenario_phase = ScenarioPhase.DISCHARGED_RELAXING
- self.phase_start = time.time()
- elif (time.time() - self.phase_start) >= (
- self.discharge_switch_cycle_min * 60
- ):
-
- # Change discharge load
- self.discharge_load += 50
- if self.discharge_load > 225:
- self.discharge_load = 0
-
- dut_controller.set_backlight(self.discharge_load)
-
- self.phase_start = time.time()
- logging.info(
- f"Switched discharge cycle to {self.discharge_load}, next change in {self.discharge_switch_cycle_min} minutes."
- )
-
- elif self.scenario_phase == ScenarioPhase.DISCHARGED_RELAXING:
-
- # Relaxation time is set in minutes, so convert to seconds
- if (time.time() - self.phase_start) >= (
- self.relaxation_time_min * 60
- ) or SKIP_RELAXING:
-
- self.scenario_phase = ScenarioPhase.DONE
- logging.info("Scenario completed successfully.")
- return True
-
- else:
- elapsed_min = int((time.time() - self.phase_start) / 60)
-
- if self.remaining_time != self.relaxation_time_min - elapsed_min:
- # Update remaining time only if it changed
- self.remaining_time = self.relaxation_time_min - elapsed_min
- logging.info(
- f"Relaxing for {self.relaxation_time_min} minutes, remaining: {self.remaining_time} minutes"
- )
-
- # Relax
- return False
-
- def log_data(
- self, dut_controller: DutController, output_directory: Path, temp: float | int
- ) -> None:
-
- dut_controller.log_data(
- output_directory,
- self.test_time_id,
- "switching",
- self.scenario_phase.name.lower(),
- temp,
- )
-
- def teardown(self, dut_controller: DutController) -> None:
- pass
diff --git a/tools/automatic_battery_tester/test_logic/test_scenario.py b/tools/automatic_battery_tester/test_logic/test_scenario.py
deleted file mode 100644
index 39da91aa..00000000
--- a/tools/automatic_battery_tester/test_logic/test_scenario.py
+++ /dev/null
@@ -1,28 +0,0 @@
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from pathlib import Path
-
-from dut.dut_controller import DutController
-
-
-class TestScenario(ABC):
- """Parent class for test scenarios."""
-
- @abstractmethod
- def setup(self, dut_controller: DutController) -> None:
- pass
-
- @abstractmethod
- def run(self, dut_controller: DutController) -> bool:
- pass
-
- @abstractmethod
- def teardown(self, dut_controller: DutController) -> None:
- pass
-
- @abstractmethod
- def log_data(
- self, dut_controller: DutController, output_directory: Path, temp: float | int
- ) -> None:
- pass
diff --git a/tools/style.py.exclude b/tools/style.py.exclude
index 7c835bee..bbbdabc8 100644
--- a/tools/style.py.exclude
+++ b/tools/style.py.exclude
@@ -2,4 +2,3 @@
^legacy/firmware/protob/options_pb2\.py
^legacy/firmware/protob/messages_nem_pb2\.py
^legacy/vendor
-^tools/automatic_battery_tester/.venv/
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.