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>
diff --git a/scripts/west_commands/runners/nrfutil.py b/scripts/west_commands/runners/nrfutil.py
index 50db41a..a7ff512 100644
--- a/scripts/west_commands/runners/nrfutil.py
+++ b/scripts/west_commands/runners/nrfutil.py
@@ -10,8 +10,10 @@
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 @@
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'])
