Always upload using esptool (#2433)

esphome/__main__.py

@@ -184,12 +184,30 @@ def compile_program(args, config):
 
 
 def upload_using_esptool(config, port):
-    path = CORE.firmware_bin
+    from esphome import platformio_api
+
     first_baudrate = config[CONF_ESPHOME][CONF_PLATFORMIO_OPTIONS].get(
         "upload_speed", 460800
     )
 
     def run_esptool(baud_rate):
+        idedata = platformio_api.get_idedata(config)
+
+        firmware_offset = "0x10000" if CORE.is_esp32 else "0x0"
+        flash_images = [
+            platformio_api.FlashImage(
+                path=idedata.firmware_bin_path,
+                offset=firmware_offset,
+            ),
+            *idedata.extra_flash_images,
+        ]
+
+        mcu = "esp8266"
+        if CORE.is_esp32:
+            from esphome.components.esp32 import get_esp32_variant
+
+            mcu = get_esp32_variant().lower()
+
         cmd = [
             "esptool.py",
             "--before",
@@ -198,14 +216,15 @@ def upload_using_esptool(config, port):
             "hard_reset",
             "--baud",
             str(baud_rate),
-            "--chip",
-            "esp8266",
             "--port",
             port,
+            "--chip",
+            mcu,
             "write_flash",
-            "0x0",
+            "-z",
-            path,
         ]
+        for img in flash_images:
+            cmd += [img.offset, img.path]
 
         if os.environ.get("ESPHOME_USE_SUBPROCESS") is None:
             import esptool
@@ -229,11 +248,7 @@ def upload_using_esptool(config, port):
 def upload_program(config, args, host):
     # if upload is to a serial port use platformio, otherwise assume ota
     if get_port_type(host) == "SERIAL":
-        from esphome import platformio_api
-
-        if CORE.is_esp8266:
         return upload_using_esptool(config, host)
-        return platformio_api.run_upload(config, CORE.verbose, host)
 
     from esphome import espota2
 

esphome/core/__init__.py

@@ -542,6 +542,9 @@ class EsphomeCore:
         path_ = os.path.expanduser(os.path.join(*path))
         return os.path.join(self.config_dir, path_)
 
+    def relative_internal_path(self, *path: str) -> str:
+        return self.relative_config_path(".esphome", *path)
+
     def relative_build_path(self, *path):
         # pylint: disable=no-value-for-parameter
         path_ = os.path.expanduser(os.path.join(*path))

esphome/platformio_api.py

@@ -1,12 +1,15 @@
+from dataclasses import dataclass
 import json
-from typing import Union
+from typing import List, Union
+from pathlib import Path
 
 import logging
 import os
 import re
 import subprocess
 
-from esphome.core import CORE
+from esphome.const import KEY_CORE
+from esphome.core import CORE, EsphomeError
 from esphome.util import run_external_command, run_external_process
 
 _LOGGER = logging.getLogger(__name__)
@@ -96,36 +99,56 @@ def run_compile(config, verbose):
     return run_platformio_cli_run(config, verbose)
 
 
-def run_upload(config, verbose, port):
+def _run_idedata(config):
-    return run_platformio_cli_run(
-        config, verbose, "-t", "upload", "--upload-port", port
-    )
-
-
-def run_idedata(config):
     args = ["-t", "idedata"]
     stdout = run_platformio_cli_run(config, False, *args, capture_stdout=True)
     match = re.search(r'{\s*".*}', stdout)
     if match is None:
-        _LOGGER.debug("Could not match IDEData for %s", stdout)
+        _LOGGER.error("Could not match idedata, please report this error")
-        return IDEData(None)
+        _LOGGER.error("Stdout: %s", stdout)
+        raise EsphomeError
+
     try:
-        return IDEData(json.loads(match.group()))
+        return json.loads(match.group())
     except ValueError:
-        _LOGGER.debug("Could not load IDEData for %s", stdout, exc_info=1)
+        _LOGGER.error("Could not parse idedata", exc_info=True)
-        return IDEData(None)
+        _LOGGER.error("Stdout: %s", stdout)
+        raise
 
 
-IDE_DATA = None
+def _load_idedata(config):
+    platformio_ini = Path(CORE.relative_build_path("platformio.ini"))
+    temp_idedata = Path(CORE.relative_internal_path(CORE.name, "idedata.json"))
+
+    changed = False
+    if not platformio_ini.is_file() or not temp_idedata.is_file():
+        changed = True
+    elif platformio_ini.stat().st_mtime >= temp_idedata.stat().st_mtime:
+        changed = True
+
+    if not changed:
+        try:
+            return json.loads(temp_idedata.read_text(encoding="utf-8"))
+        except ValueError:
+            pass
+
+    temp_idedata.parent.mkdir(exist_ok=True, parents=True)
+
+    data = _run_idedata(config)
+
+    temp_idedata.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
+    return data
 
 
-def get_idedata(config):
+KEY_IDEDATA = "idedata"
-    global IDE_DATA
 
-    if IDE_DATA is None:
+
-        _LOGGER.info("Need to fetch platformio IDE-data, please stand by")
+def get_idedata(config) -> "IDEData":
-        IDE_DATA = run_idedata(config)
+    if KEY_IDEDATA in CORE.data[KEY_CORE]:
-    return IDE_DATA
+        return CORE.data[KEY_CORE][KEY_IDEDATA]
+    idedata = IDEData(_load_idedata(config))
+    CORE.data[KEY_CORE][KEY_IDEDATA] = idedata
+    return idedata
 
 
 # ESP logs stack trace decoder, based on https://github.com/me-no-dev/EspExceptionDecoder
@@ -261,37 +284,42 @@ def process_stacktrace(config, line, backtrace_state):
     return backtrace_state
 
 
+@dataclass
+class FlashImage:
+    path: str
+    offset: str
+
+
 class IDEData:
     def __init__(self, raw):
-        if not isinstance(raw, dict):
-            self.raw = {}
-        else:
         self.raw = raw
 
     @property
     def firmware_elf_path(self):
-        return self.raw.get("prog_path")
+        return self.raw["prog_path"]
 
     @property
-    def flash_extra_images(self):
+    def firmware_bin_path(self) -> str:
+        return str(Path(self.firmware_elf_path).with_suffix(".bin"))
+
+    @property
+    def extra_flash_images(self) -> List[FlashImage]:
         return [
-            (x["path"], x["offset"]) for x in self.raw.get("flash_extra_images", [])
+            FlashImage(path=entry["path"], offset=entry["offset"])
+            for entry in self.raw["extra"]["flash_images"]
         ]
 
     @property
-    def cc_path(self):
+    def cc_path(self) -> str:
         # For example /Users/<USER>/.platformio/packages/toolchain-xtensa32/bin/xtensa-esp32-elf-gcc
-        return self.raw.get("cc_path")
+        return self.raw["cc_path"]
 
     @property
-    def addr2line_path(self):
+    def addr2line_path(self) -> str:
-        cc_path = self.cc_path
-        if cc_path is None:
-            return None
         # replace gcc at end with addr2line
 
         # Windows
-        if cc_path.endswith(".exe"):
+        if self.cc_path.endswith(".exe"):
-            return f"{cc_path[:-7]}addr2line.exe"
+            return f"{self.cc_path[:-7]}addr2line.exe"
 
-        return f"{cc_path[:-3]}addr2line"
+        return f"{self.cc_path[:-3]}addr2line"

esphome/storage_json.py

@@ -16,7 +16,7 @@ _LOGGER = logging.getLogger(__name__)
 
 
 def storage_path():  # type: () -> str
-    return CORE.relative_config_path(".esphome", f"{CORE.config_filename}.json")
+    return CORE.relative_internal_path(f"{CORE.config_filename}.json")
 
 
 def ext_storage_path(base_path, config_filename):  # type: (str, str) -> str