forked from espressif/esp-idf
tools: Add python types hints
This commit is contained in:
@@ -9,21 +9,22 @@ import sys
|
||||
import threading
|
||||
import time
|
||||
from threading import Thread
|
||||
from typing import Any, Dict, List
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from click.core import Context
|
||||
from idf_py_actions.errors import FatalError
|
||||
from idf_py_actions.tools import ensure_build_directory
|
||||
from idf_py_actions.tools import PropertyDict, ensure_build_directory
|
||||
|
||||
PYTHON = sys.executable
|
||||
|
||||
|
||||
def action_extensions(base_actions, project_path):
|
||||
def action_extensions(base_actions: Dict, project_path: str) -> Dict:
|
||||
OPENOCD_OUT_FILE = 'openocd_out.txt'
|
||||
GDBGUI_OUT_FILE = 'gdbgui_out.txt'
|
||||
# Internal dictionary of currently active processes, threads and their output files
|
||||
processes = {'threads_to_join': [], 'openocd_issues': None}
|
||||
processes: Dict = {'threads_to_join': [], 'openocd_issues': None}
|
||||
|
||||
def _check_for_common_openocd_issues(file_name, print_all=True):
|
||||
def _check_for_common_openocd_issues(file_name: str, print_all: bool=True) -> Any:
|
||||
if processes['openocd_issues'] is not None:
|
||||
return processes['openocd_issues']
|
||||
try:
|
||||
@@ -39,7 +40,7 @@ def action_extensions(base_actions, project_path):
|
||||
processes['openocd_issues'] = message
|
||||
return message
|
||||
|
||||
def _check_openocd_errors(fail_if_openocd_failed, target, ctx):
|
||||
def _check_openocd_errors(fail_if_openocd_failed: Dict, target: str, ctx: Context) -> None:
|
||||
if fail_if_openocd_failed:
|
||||
if 'openocd' in processes and processes['openocd'] is not None:
|
||||
p = processes['openocd']
|
||||
@@ -62,7 +63,7 @@ def action_extensions(base_actions, project_path):
|
||||
# OpenOCD exited or error message detected -> print possible output and terminate
|
||||
raise FatalError('Action "{}" failed due to errors in OpenOCD:\n{}'.format(target, _check_for_common_openocd_issues(name)), ctx)
|
||||
|
||||
def _terminate_async_target(target):
|
||||
def _terminate_async_target(target: str) -> None:
|
||||
if target in processes and processes[target] is not None:
|
||||
try:
|
||||
if target + '_outfile' in processes:
|
||||
@@ -86,11 +87,11 @@ def action_extensions(base_actions, project_path):
|
||||
print('Failed to close/kill {}'.format(target))
|
||||
processes[target] = None # to indicate this has ended
|
||||
|
||||
def is_gdb_with_python(gdb):
|
||||
def is_gdb_with_python(gdb: str) -> bool:
|
||||
# execute simple python command to check is it supported
|
||||
return subprocess.run([gdb, '--batch-silent', '--ex', 'python import os'], stderr=subprocess.DEVNULL).returncode == 0
|
||||
|
||||
def create_local_gdbinit(gdb, gdbinit, elf_file):
|
||||
def create_local_gdbinit(gdb: str, gdbinit: str, elf_file: str) -> None:
|
||||
with open(gdbinit, 'w') as f:
|
||||
if is_gdb_with_python(gdb):
|
||||
f.write('python\n')
|
||||
@@ -107,7 +108,7 @@ def action_extensions(base_actions, project_path):
|
||||
f.write('thb app_main\n')
|
||||
f.write('c\n')
|
||||
|
||||
def debug_cleanup():
|
||||
def debug_cleanup() -> None:
|
||||
print('cleaning up debug targets')
|
||||
for t in processes['threads_to_join']:
|
||||
if threading.currentThread() != t:
|
||||
@@ -116,7 +117,7 @@ def action_extensions(base_actions, project_path):
|
||||
_terminate_async_target('gdbgui')
|
||||
_terminate_async_target('gdb')
|
||||
|
||||
def post_debug(action, ctx, args, **kwargs):
|
||||
def post_debug(action: str, ctx: Context, args: PropertyDict, **kwargs: str) -> None:
|
||||
""" Deal with asynchronous targets, such as openocd running in background """
|
||||
if kwargs['block'] == 1:
|
||||
for target in ['openocd', 'gdbgui']:
|
||||
@@ -143,7 +144,7 @@ def action_extensions(base_actions, project_path):
|
||||
_terminate_async_target('openocd')
|
||||
_terminate_async_target('gdbgui')
|
||||
|
||||
def get_project_desc(args, ctx):
|
||||
def get_project_desc(args: PropertyDict, ctx: Context) -> Any:
|
||||
desc_path = os.path.join(args.build_dir, 'project_description.json')
|
||||
if not os.path.exists(desc_path):
|
||||
ensure_build_directory(args, ctx.info_name)
|
||||
@@ -151,7 +152,7 @@ def action_extensions(base_actions, project_path):
|
||||
project_desc = json.load(f)
|
||||
return project_desc
|
||||
|
||||
def openocd(action, ctx, args, openocd_scripts, openocd_commands):
|
||||
def openocd(action: str, ctx: Context, args: PropertyDict, openocd_scripts: Optional[str], openocd_commands: str) -> None:
|
||||
"""
|
||||
Execute openocd as external tool
|
||||
"""
|
||||
@@ -188,14 +189,14 @@ def action_extensions(base_actions, project_path):
|
||||
processes['openocd_outfile_name'] = openocd_out_name
|
||||
print('OpenOCD started as a background task {}'.format(process.pid))
|
||||
|
||||
def get_gdb_args(gdbinit, project_desc: Dict[str, Any]) -> List[str]:
|
||||
def get_gdb_args(gdbinit: str, project_desc: Dict[str, Any]) -> List:
|
||||
args = ['-x={}'.format(gdbinit)]
|
||||
debug_prefix_gdbinit = project_desc.get('debug_prefix_map_gdbinit')
|
||||
if debug_prefix_gdbinit:
|
||||
args.append('-ix={}'.format(debug_prefix_gdbinit))
|
||||
return args
|
||||
|
||||
def gdbui(action, ctx, args, gdbgui_port, gdbinit, require_openocd):
|
||||
def gdbui(action: str, ctx: Context, args: PropertyDict, gdbgui_port: Optional[str], gdbinit: Optional[str], require_openocd: bool) -> None:
|
||||
"""
|
||||
Asynchronous GDB-UI target
|
||||
"""
|
||||
@@ -211,8 +212,8 @@ def action_extensions(base_actions, project_path):
|
||||
# - '"-x=foo -x=bar"', would return ['foo bar']
|
||||
# - '-x=foo', would return ['-x', 'foo'] and mess up the former option '--gdb-args'
|
||||
# so for one item, use extra double quotes. for more items, use no extra double quotes.
|
||||
gdb_args = get_gdb_args(gdbinit, project_desc)
|
||||
gdb_args = '"{}"'.format(' '.join(gdb_args)) if len(gdb_args) == 1 else ' '.join(gdb_args)
|
||||
gdb_args_list = get_gdb_args(gdbinit, project_desc)
|
||||
gdb_args = '"{}"'.format(' '.join(gdb_args_list)) if len(gdb_args_list) == 1 else ' '.join(gdb_args_list)
|
||||
args = ['gdbgui', '-g', gdb, '--gdb-args', gdb_args]
|
||||
print(args)
|
||||
|
||||
@@ -238,8 +239,8 @@ def action_extensions(base_actions, project_path):
|
||||
print('gdbgui started as a background task {}'.format(process.pid))
|
||||
_check_openocd_errors(fail_if_openocd_failed, action, ctx)
|
||||
|
||||
def global_callback(ctx, global_args, tasks):
|
||||
def move_to_front(task_name):
|
||||
def global_callback(ctx: Context, global_args: PropertyDict, tasks: List) -> None:
|
||||
def move_to_front(task_name: str) -> None:
|
||||
for index, task in enumerate(tasks):
|
||||
if task.name == task_name:
|
||||
tasks.insert(0, tasks.pop(index))
|
||||
@@ -264,18 +265,18 @@ def action_extensions(base_actions, project_path):
|
||||
if task.name in ('gdb', 'gdbgui', 'gdbtui'):
|
||||
task.action_args['require_openocd'] = True
|
||||
|
||||
def run_gdb(gdb_args):
|
||||
def run_gdb(gdb_args: List) -> int:
|
||||
p = subprocess.Popen(gdb_args)
|
||||
processes['gdb'] = p
|
||||
return p.wait()
|
||||
|
||||
def gdbtui(action, ctx, args, gdbinit, require_openocd):
|
||||
def gdbtui(action: str, ctx: Context, args: PropertyDict, gdbinit: str, require_openocd: bool) -> None:
|
||||
"""
|
||||
Synchronous GDB target with text ui mode
|
||||
"""
|
||||
gdb(action, ctx, args, 1, gdbinit, require_openocd)
|
||||
|
||||
def gdb(action, ctx, args, gdb_tui, gdbinit, require_openocd):
|
||||
def gdb(action: str, ctx: Context, args: PropertyDict, gdb_tui: Optional[int], gdbinit: Optional[str], require_openocd: bool) -> None:
|
||||
"""
|
||||
Synchronous GDB target
|
||||
"""
|
||||
|
Reference in New Issue
Block a user