"""Transport selection and debug adapter configuration. OpenOCD supports multiple debug transports (JTAG, SWD, SWIM, etc.) and various adapter interfaces (CMSIS-DAP, ST-Link, J-Link, etc.). This module provides access to transport and adapter state. """ from __future__ import annotations import logging from typing import TYPE_CHECKING from openocd.errors import OpenOCDError if TYPE_CHECKING: from openocd.connection.base import Connection log = logging.getLogger(__name__) class Transport: """Query and configure the debug transport and adapter. Usage:: transport = Transport(conn) current = await transport.select() # e.g. "swd" available = await transport.list() # e.g. ["jtag", "swd"] speed = await transport.adapter_speed() # current kHz await transport.adapter_speed(4000) # set to 4 MHz """ def __init__(self, conn: Connection) -> None: self._conn = conn async def select(self) -> str: """Get the currently selected transport. Returns: Transport name string (e.g. "jtag", "swd", "swim"). Raises: OpenOCDError: If the command fails. """ response = await self._conn.send("transport select") response = response.strip() if not response: raise OpenOCDError("Empty response from 'transport select'") return response async def list(self) -> list[str]: """List transports available for the current adapter. Returns: List of transport name strings. Raises: OpenOCDError: If the command fails. """ response = await self._conn.send("transport list") response = response.strip() if not response: raise OpenOCDError("Empty response from 'transport list'") # OpenOCD may return a Tcl list like "jtag swd" or one per line transports: list[str] = [] for line in response.splitlines(): for token in line.split(): cleaned = token.strip("{}") if cleaned: transports.append(cleaned) return transports async def adapter_info(self) -> str: """Get adapter/interface information. Tries ``adapter name`` first (newer OpenOCD), falls back to ``adapter info`` for older versions. Returns: Adapter description string. """ # "adapter name" is the preferred command in OpenOCD >= 0.12 response = await self._conn.send("adapter name") response = response.strip() if not response or "invalid" in response.lower() or "error" in response.lower(): response = await self._conn.send("adapter info") response = response.strip() if not response: raise OpenOCDError("Could not determine adapter info") return response async def adapter_speed(self, khz: int | None = None) -> int: """Get or set the adapter clock speed. Args: khz: If provided, set the adapter speed to this value in kHz. If None, just query the current speed. Returns: The current (or newly set) adapter speed in kHz. Raises: OpenOCDError: If the command fails or response is not parseable. """ cmd = f"adapter speed {khz}" if khz is not None else "adapter speed" response = await self._conn.send(cmd) response = response.strip() # OpenOCD response is typically just a number, or # "adapter speed: 4000 kHz" depending on the interface speed = _parse_speed(response) if speed is None: raise OpenOCDError(f"Cannot parse adapter speed from: {response!r}") if khz is not None: log.info("Adapter speed set to %d kHz", speed) return speed # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _parse_speed(response: str) -> int | None: """Extract a numeric kHz value from an adapter speed response. Handles formats like: "4000" "adapter speed: 4000 kHz" "4000 kHz" """ # Try the whole thing as a plain integer try: return int(response) except ValueError: pass # Pull out the first integer-looking token for token in response.replace(":", " ").split(): try: return int(token) except ValueError: continue return None