1# Copyright (c) 2020-2022, Niklas Hauser
  2# Copyright (c) 2023, Auterion AG
  3# SPDX-License-Identifier: BSD-3-Clause
  4
  5"""
  6.. include:: openocd.md
  7"""
  8
  9from __future__ import annotations
 10import os
 11import time
 12import signal
 13import logging
 14import tempfile
 15import platform
 16import subprocess
 17from pathlib import Path
 18
 19from . import utils
 20from .backend import ProbeBackend
 21
 22LOGGER = logging.getLogger("debug:openocd")
 23
 24# -----------------------------------------------------------------------------
 25class OpenOcdBackend(ProbeBackend):
 26    """
 27    OpenOCD specific debug backend implementation. Starts `openocd` in
 28    its own subprocess and connects GDB to port `3333` of the localhost.
 29
 30    See `call()` for additional information.
 31    """
 32    def __init__(self, commands: list[str] = None, config: list[Path] = None,
 33                 search: list[Path] = None, serial: str = None):
 34        """
 35        :param commands: list of commands to execute on launch.
 36        :param config: list of configuration files to execute on launch.
 37        :param search: list of directories to search configuration files in.
 38        :param serial: serial number of debug probe.
 39        """
 40        super().__init__(":3333")
 41        self.commands = utils.listify(commands)
 42        self.config = utils.listify(config)
 43        self.search = utils.listify(search)
 44        self.serial = serial
 45        self.process = None
 46        self.name = "openocd"
 47
 48    def start(self):
 49        self.process = call(self.commands, self.config, self.search,
 50                            blocking=False, log_output=False, serial=self.serial)
 51        LOGGER.info(f"Starting {self.process.pid}...")
 52
 53    def stop(self):
 54        if self.process is not None:
 55            LOGGER.info(f"Stopping {self.process.pid}.")
 56            os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
 57            os.waitpid(os.getpgid(self.process.pid), 0)
 58            self.process = None
 59
 60
 61def call(commands: list[str] = None, config: list[Path] = None,
 62         search: list[Path] = None, log_output: Path | bool = None,
 63         flags: str = None, blocking: bool = True, serial: str = None) -> "int | subprocess.Popen":
 64    """
 65    Starts `openocd` and connects to the microcontroller without resetting the
 66    device. You can overwrite the default binary by exporting an alternative in
 67    your environment:
 68
 69    ```sh
 70    export PX4_OPENOCD=path/to/openocd
 71    ```
 72
 73    :param commands: list of commands to execute on launch.
 74    :param config: list of configuration files to execute on launch.
 75    :param search: list of directories to search configuration files in.
 76    :param log_output: Redirect OpenOCD stdout output to a file, or disable
 77                       entirely via `False`.
 78    :param flags: Additional flags
 79    :param blocking: Run in current process as a blocking call.
 80                     Set to `False` to run in a new subprocess.
 81    :param serial: serial number of debug probe.
 82
 83    :return: The process return code if `blocking` or the Popen object.
 84    """
 85    if log_output == False: log_output = "/dev/null"
 86    cmds = [f"log_output {log_output}"] if log_output is not None else []
 87    cmds += [f"adapter serial {serial}"] if serial is not None else []
 88    cmds += ["init"] + utils.listify(commands)
 89    config = utils.listify(config)
 90    search = utils.listify(search)
 91
 92
 93    # Provide additional search paths via the OPENOCD_SCRIPTS environment variable
 94    # See http://openocd.org/doc/html/Running.html
 95    # os.environ.get("OPENOCD_SCRIPTS", "")
 96
 97    binary = os.environ.get("PX4_OPENOCD", "openocd")
 98
 99    command_openocd = "{} {} {} {} {}".format(
100        binary, flags or "",
101        " ".join(map('-s "{}"'.format, search)),
102        " ".join(map('-f "{}"'.format, config)),
103        " ".join(map('-c "{}"'.format, cmds))
104    )
105    LOGGER.debug(command_openocd)
106
107    kwargs = {"cwd": os.getcwd(), "shell": True}
108    if blocking:
109        return subprocess.call(command_openocd, **kwargs)
110
111    # We have to start openocd in its own session ID, so that Ctrl-C in GDB
112    # does not kill OpenOCD.
113    kwargs["start_new_session"] = True
114    return subprocess.Popen(command_openocd, **kwargs)
115
116
117# -----------------------------------------------------------------------------
118def itm(backend, fcpu: int, baudrate: int = None) -> int:
119    """
120    Launches `openocd` and configured ITM output on the SWO pin.
121    The data is written into a temporary file, which is `tail`'ed for display.
122
123    :param backend: A OpenOCD backend.
124    :param fcpu: CPU frequency of the target.
125    :param baudrate: optional frequency of the SWO connection.
126
127    :return: the process return code
128    """
129    if not fcpu:
130        raise ValueError("fcpu must be the CPU/HCLK frequency!")
131
132    with tempfile.NamedTemporaryFile() as tmpfile:
133        backend.commands += [
134            "tpiu create itm.tpiu -dap [dap names] -ap-num 0 -protocol uart",
135           f"itm.tpiu configure -traceclk {fcpu} -pin-freq {baudrate or 2000000} -output {tmpfile}",
136            "itm.tpiu enable",
137            "tpiu init",
138            "itm port 0 on",
139        ]
140        # Start OpenOCD in the background
141        with backend.scope():
142            # Start a blocking call to monitor the log file
143            # TODO: yield out new log lines in the future
144            try:
145                subprocess.call("tail -f {}".format(tmpfile.name),
146                                cwd=os.getcwd(), shell=True)
147            except KeyboardInterrupt:
148                pass
149    return 0
150
151def rtt(backend, channel: int = 0) -> int:
152    """
153    Launches the backend in the background and connects a telnet client to the
154    Real-Time Transfer process. You can disconnect with Ctrl+D.
155
156    :param backend: A OpenOCD backend object.
157    :param channel: The RTT channel to connect to.
158
159    :return: the process return code
160    """
161    backend.commands += [
162        "rtt setup 0x20000000 256000",
163        "rtt start",
164        "rtt polling_interval 1",
165       f"rtt server start {9090 + channel} {channel}",
166    ]
167    import telnetlib
168    # Start OpenOCD in the background
169    with backend.scope():
170        with telnetlib.Telnet("localhost", 9090+channel) as tn:
171            try:
172                tn.interact()
173            except KeyboardInterrupt:
174                pass
175    return 0
176
177
178# -----------------------------------------------------------------------------
179def program(source: Path, commands: list[str] = None, config: list[Path] = None,
180            search: list[Path] = None, serial: str = None) -> int:
181    """
182    Loads the source file into the microcontroller and resets the device.
183
184    :param commands: list of commands to execute on launch.
185    :param config: list of configuration files to execute on launch.
186    :param search: list of directories to search configuration files in.
187    :param source: path to a `.elf` file to upload.
188    :param serial: serial number of debug probe.
189
190    :return: the process return code of openocd
191    """
192    commands = utils.listify(commands) + \
193        [f"program {Path(source).absolute()} verify reset exit"]
194    return call(commands=commands, config=config, search=search)
195
196    # Unfortunately, some older OpenOCD versions seems to erase Flash sector 0
197    # even if the ELF file has an offset. This overwrites the bootloader and
198    # bricks the FMU, so we must use GDB instead.
199    # from .gdb import call as gdb_call
200    # backend = OpenOcdBackend(commands, config, search, serial)
201    # gdb_cmds = ["monitor reset halt", "load", "monitor reset run", "quit"]
202    # return gdb_call(backend, source, ui="batch", commands=gdb_cmds, with_python=False)
203
204
205def reset(commands: list[str] = None, config: list[Path] = None,
206          search: list[Path] = None, serial: str = None) -> int:
207    """
208    Resets the device via OpenOCD.
209
210    :param commands: list of commands to execute on launch.
211    :param config: list of configuration files to execute on launch.
212    :param search: list of directories to search configuration files in.
213    :param serial: serial number of debug probe.
214
215    :return: the process return code of OpenOCD
216    """
217    commands = utils.listify(commands) + ["reset", "shutdown"]
218    return call(commands, config, search, serial=serial)
219
220
221# -----------------------------------------------------------------------------
222def _add_subparser(subparser):
223    parser = subparser.add_parser("openocd", help="Use OpenOCD as Backend.")
224    parser.add_argument(
225        "-f",
226        dest="oconfig",
227        action="append",
228        help="Use these OpenOCD config files.")
229    parser.add_argument(
230        "-s",
231        dest="osearch",
232        action="append",
233        help="Search in these paths for config files.")
234    parser.add_argument(
235        "-c",
236        dest="ocommands",
237        action="append",
238        help="Extra OpenOCD commands.")
239    parser.add_argument(
240        "--speed",
241        dest="ospeed",
242        type=int,
243        default=8000,
244        choices=[24000, 8000, 3300, 1000, 200, 50, 5],
245        help="SWD baudrate in kHz.")
246    parser.add_argument(
247        "--serial",
248        dest="oserial",
249        default=None,
250        help="Serial number of debug probe.")
251    parser.set_defaults(backend=lambda args:
252        OpenOcdBackend([f"adapter speed {args.ospeed}"] +
253                       utils.listify(args.ocommands),
254                       args.oconfig, args.osearch, args.oserial))
255    return parser
256
257
258# -----------------------------------------------------------------------------
259if __name__ == "__main__":
260    import argparse
261    import emdbg
262
263    parser = argparse.ArgumentParser(
264        description="OpenOCD debug probe: Upload, reset, and logging")
265    parser.add_argument(
266        "-f",
267        dest="config",
268        action="append",
269        help="Use these OpenOCD config files.")
270    parser.add_argument(
271        "-s",
272        dest="searchdirs",
273        action="append",
274        help="Search in these paths for config files.")
275    parser.add_argument(
276        "-c",
277        dest="commands",
278        action="append",
279        help="Extra OpenOCD commands.")
280    parser.add_argument(
281        "--speed",
282        type=int,
283        default=8000,
284        choices=[24000, 8000, 3300, 1000, 200, 50, 5],
285        help="SWD baudrate in kHz.")
286    parser.add_argument(
287        "--serial",
288        default=None,
289        help="Serial number of debug probe.")
290    parser.add_argument(
291        "-v",
292        dest="verbosity",
293        action="count",
294        help="Verbosity level.")
295
296    subparsers = parser.add_subparsers(title="Command", dest="command")
297
298    subparsers.add_parser("reset", help="Reset the device.")
299
300    subparsers.add_parser("run", help="Run OpenOCD.")
301
302    upload_parser = subparsers.add_parser("upload", help="Upload firmware.")
303    upload_parser.add_argument(
304        "--source",
305        required=True,
306        help="The firmware to upload: `.elf` file")
307
308    args = parser.parse_args()
309    emdbg.logger.configure(args.verbosity)
310
311    commands = [f"adapter speed {args.speed}"] + utils.listify(args.commands)
312    if not (config := args.config):
313        config = ["interface/stlink.cfg", "target/stm32f7x.cfg"]
314
315    if args.command == "reset":
316        exit(reset(commands, config, args.searchdirs, args.serial))
317
318    if args.command == "run":
319        exit(call(commands, config, args.searchdirs, blocking=True, serial=args.serial))
320
321    if args.command == "upload":
322        exit(program(args.source, commands, config, args.searchdirs, args.serial))
323
324    LOGGER.error("Unknown command!")
325    exit(1)