scripts: west_commands: runners: nrfutil: provide live feedback

nrfutil runner uses the batch-mode, so no live feedback is provided to
the user. However, batch-mode reports batch progress containing
human-readable strings of the operation being done. This patch changes
the _exec() implementation to parse the subprocess output in real-time,
logging to info the 'batch_update' reports. Note that only the first
batch update of a sequence (percentage = 0) is logged because first,
percentage resolution seems to be pretty bad, and, because logging
messages cannot be easily _appended_.

Signed-off-by: Gerard Marull-Paretas <gerard@teslabs.com>
This commit is contained in:
Gerard Marull-Paretas 2024-03-18 11:42:50 +01:00 committed by Carles Cufí
commit acc0a43fb9

View file

@ -10,8 +10,10 @@ from pathlib import Path
import sys
import subprocess
from runners.core import _DRY_RUN
from runners.nrf_common import NrfBinaryRunner
class NrfUtilBinaryRunner(NrfBinaryRunner):
'''Runner front-end for nrfutil.'''
@ -40,27 +42,34 @@ class NrfUtilBinaryRunner(NrfBinaryRunner):
recover=args.recover)
def _exec(self, args):
try:
out = self.check_output(['nrfutil', '--json', 'device'] + args)
except subprocess.CalledProcessError as e:
# https://docs.python.org/3/reference/compound_stmts.html#except-clause
cpe = e
out = cpe.stdout
else:
cpe = None
finally:
# https://github.com/ndjson/ndjson-spec
out = [json.loads(s) for s in
out.decode(sys.getdefaultencoding()).split('\n') if len(s)]
self.logger.debug(f'output: {out}')
if cpe:
if 'execute-batch' in args:
for o in out:
if o['type'] == 'batch_end' and o['data']['error']:
cpe.returncode = o['data']['error']['code']
raise cpe
jout_all = []
return out
cmd = ['nrfutil', '--json', 'device'] + args
self._log_cmd(cmd)
if _DRY_RUN:
return {}
with subprocess.Popen(cmd, stdout=subprocess.PIPE) as p:
for line in iter(p.stdout.readline, b''):
# https://github.com/ndjson/ndjson-spec
jout = json.loads(line.decode(sys.getdefaultencoding()))
jout_all.append(jout)
if 'x-execute-batch' in args:
if jout['type'] == 'batch_update':
pld = jout['data']['data']
if (
pld['type'] == 'task_progress' and
pld['data']['progress']['progressPercentage'] == 0
):
self.logger.info(pld['data']['progress']['description'])
elif jout['type'] == 'batch_end' and jout['data']['error']:
raise subprocess.CalledProcessError(
jout['data']['error']['code'], cmd
)
return jout_all
def do_get_boards(self):
out = self._exec(['list'])