ref: use esptool main instead other func (init commit)

pytest-embedded-idf/pytest_embedded_idf/serial.py

@@ -1,12 +1,12 @@
+import contextlib
 import hashlib
 import logging
 import os
 import tempfile
 from typing import Optional, TextIO, Union
 
 import esptool
-from pytest_embedded.log import live_print_call
-from pytest_embedded_serial_esp.serial import EspSerial, EsptoolArgs
+from pytest_embedded_serial_esp.serial import EspSerial
 
 from .app import IdfApp
 
@@ -73,7 +73,7 @@ def _start(self):
             if self.app.is_loadable_elf:
                 self.load_ram()
             else:
-                self.flash()
+                self.flash()  # ???
 
     def load_ram(self) -> None:
         if not self.app.is_loadable_elf:
@@ -84,30 +84,28 @@ def load_ram(self) -> None:
         if self.app.bin_file:
             bin_file = self.app.bin_file
         else:
-            live_print_call(
+            with contextlib.redirect_stdout(self._q):
+                esptool.main(
+                    [
+                        '--chip',
+                        self.app.target,
+                        'elf2image',
+                        self.app.elf_file,
+                        *self.app.write_flash_args,
+                    ]
+                )
+            bin_file = self.app.elf_file.replace('.elf', '.bin')
+
+        with contextlib.redirect_stdout(self._q):
+            esptool.main(
                 [
-                    'esptool.py',
                     '--chip',
                     self.app.target,
-                    'elf2image',
-                    self.app.elf_file,
-                    *self.app.write_flash_args,
+                    '--no-stub',
+                    'load_ram',
+                    bin_file,
                 ],
-                msg_queue=self._q,
             )
-            bin_file = self.app.elf_file.replace('.elf', '.bin')
-
-        live_print_call(
-            [
-                'esptool.py',
-                '--chip',
-                self.app.target,
-                '--no-stub',
-                'load_ram',
-                bin_file,
-            ],
-            msg_queue=self._q,
-        )
 
     @EspSerial.use_esptool()
     def flash(self) -> None:
@@ -122,8 +120,15 @@ def flash(self) -> None:
             logging.error('No flash settings detected. Skipping auto flash...')
             return
 
-        flash_files = [(file.offset, open(file.file_path, 'rb')) for file in self.app.flash_files if not file.encrypted]
-        encrypt_files = [(file.offset, open(file.file_path, 'rb')) for file in self.app.flash_files if file.encrypted]
+        flash_files = []
+        encrypt_files = []
+        for file in self.app.flash_files:
+            if file.encrypted:
+                encrypt_files.append(str(file.offset))
+                encrypt_files.append(str(file.file_path))
+            else:
+                flash_files.append(str(file.offset))
+                flash_files.append(str(file.file_path))
 
         nvs_file = None
         try:
@@ -136,12 +141,18 @@ def flash(self) -> None:
                     address = int(address, 0)
 
                 if self.app.flash_settings['encrypt']:
-                    encrypt_files.append((address, open(nvs_file.name, 'rb')))
+                    encrypt_files.append(str(address))
+                    encrypt_files.append(str(nvs_file.name))
                 else:
-                    flash_files.append((address, open(nvs_file.name, 'rb')))
+                    flash_files.append(str(address))
+                    flash_files.append(str(nvs_file.name))
 
             # write_flash expects the parameter encrypt_files to be None and not
             # an empty list, so perform the check here
+
+            addition_kwargs = self.app.flash_args.get('extra_esptool_args', {})
+            addition_kwargs.pop('stub')
+
             default_kwargs = {
                 'addr_filename': flash_files,
                 'encrypt_files': encrypt_files or None,
@@ -153,14 +164,31 @@ def flash(self) -> None:
                 'force': False,
             }
 
-            default_kwargs.update(self.app.flash_settings)
-            default_kwargs.update(self.app.flash_args.get('extra_esptool_args', {}))
-            args = EsptoolArgs(**default_kwargs)
+            for k in default_kwargs:
+                if k in addition_kwargs:
+                    default_kwargs[k] = addition_kwargs.pop(k)
+
+            _args = []
+
+            def _kwargs_extend_list(_items):
+                for _k, _v in _items:
+                    if not _v:
+                        continue
+                    _args.append(f'--{_k}')
+
+                    if isinstance(_v, bool):
+                        continue
+                    if isinstance(_v, list):
+                        _args.extend(_v)
+                        continue
+                    _args.append(_v.strip())
+
+            _kwargs_extend_list(addition_kwargs.items())
+            _args.extend(['write_flash', *default_kwargs.pop('addr_filename')])
+
+            _kwargs_extend_list((default_kwargs | self.app.flash_settings).items())
 
-            self.stub.change_baud(self.esptool_baud)
-            esptool.detect_flash_size(self.stub, args)
-            esptool.write_flash(self.stub, args)
-            self.stub.change_baud(self.baud)
+            esptool.main(['-b', str(self.esptool_baud), *_args, '--flash_size', 'detect'], esp=self.esp)
 
             if self._meta:
                 self._meta.set_port_app_cache(self.port, self.app)
@@ -171,10 +199,6 @@ def flash(self) -> None:
                     os.remove(nvs_file.name)
                 except OSError:
                     pass
-            for _, f in flash_files:
-                f.close()
-            for _, f in encrypt_files:
-                f.close()
 
     @EspSerial.use_esptool()
     def dump_flash(

pytest-embedded-serial-esp/pytest_embedded_serial_esp/serial.py

@@ -180,12 +180,10 @@ def wrapper(self, *args, **kwargs):
                     with contextlib.redirect_stdout(self._q):
                         settings = self.proc.get_settings()
                         try:
+                            self.esp = None
                             self.esp = esptool.detect_chip(self.proc, self.baud)
                             self.esp.connect('hard_reset')
 
-                            if not no_stub:
-                                self.stub = self.esp.run_stub()
-
                             ret = func(self, *args, **kwargs)
                         finally:
                             if hard_reset_after: