diff --git a/esphome/__main__.py b/esphome/__main__.py
index 54c1aa112a..0715650fc2 100644
--- a/esphome/__main__.py
+++ b/esphome/__main__.py
@@ -6,6 +6,8 @@ import os
 import re
 import sys
 import time
+import hashlib
+import gzip
 from datetime import datetime
 
 import argcomplete
@@ -37,7 +39,7 @@ from esphome.const import (
     SECRETS_FILES,
 )
 from esphome.core import CORE, EsphomeError, coroutine
-from esphome.helpers import indent, is_ip_address
+from esphome.helpers import indent, is_ip_address, copy_file_if_changed
 from esphome.util import (
     run_external_command,
     run_external_process,
@@ -220,7 +222,25 @@ def compile_program(args, config):
     if rc != 0:
         return rc
     idedata = platformio_api.get_idedata(config)
-    return 0 if idedata is not None else 1
+    if idedata is None:
+        return 1
+
+    with open(CORE.firmware_bin, "rb") as firmware_bin:
+        bin_content = firmware_bin.read()
+
+    md5 = hashlib.md5(bin_content).hexdigest()
+    with open(CORE.firmware_md5, "w", encoding="ascii") as firmware_md5:
+        firmware_md5.write(md5)
+
+    gz_content = gzip.compress(bin_content, compresslevel=9)
+    with open(CORE.firmware_gz, "wb") as firmware_gz:
+        firmware_gz.write(gz_content)
+
+    md5_gz = hashlib.md5(gz_content).hexdigest()
+    with open(CORE.firmware_gz_md5, "w", encoding="ascii") as firmware_gz_md5:
+        firmware_gz_md5.write(md5_gz)
+
+    return 0
 
 
 def upload_using_esptool(config, port, file):
@@ -687,6 +707,22 @@ def command_rename(args, config):
     print()
     return 0
 
+def command_publish(args, config):
+    files = [ CORE.firmware_bin, CORE.firmware_md5, CORE.firmware_gz, CORE.firmware_gz_md5]
+    if not all(map(lambda f: os.path.isfile(f), files)):
+        exit_code = compile_program(args, config)
+        if exit_code != 0:
+            return exit_code
+
+    topdir = os.path.join(args.directory, CORE.name)
+    _LOGGER.info(f"Copying firmware files to {topdir}...")
+    for src_file in files:
+        dst_file = os.path.join(args.directory, CORE.name, os.path.basename(src_file))
+        copy_file_if_changed(src_file, dst_file)
+
+
+    return 0
+
 
 PRE_CONFIG_ACTIONS = {
     "wizard": command_wizard,
@@ -708,6 +744,7 @@ POST_CONFIG_ACTIONS = {
     "idedata": command_idedata,
     "rename": command_rename,
     "discover": command_discover,
+    "publish": command_publish,
 }
 
 
@@ -808,6 +845,19 @@ def parse_args(argv):
         "configuration", help="Your YAML configuration file.", nargs=1
     )
 
+    parser_publish = subparsers.add_parser(
+        "publish",
+        help="Publish compiled binary. (currently copy binary to directory)",
+    )
+    parser_publish.add_argument(
+        "configuration", help="Your YAML configuration file(s).", nargs="+"
+    )
+    parser_publish.add_argument(
+        "--directory",
+        default=os.path.join(os.path.sep, "config", "www"),
+        help="Destination directory",
+    )
+
     parser_run = subparsers.add_parser(
         "run",
         help="Validate the configuration, create a binary, upload it, and start logs.",
diff --git a/esphome/core/__init__.py b/esphome/core/__init__.py
index 58ae23e139..247a71796f 100644
--- a/esphome/core/__init__.py
+++ b/esphome/core/__init__.py
@@ -627,6 +627,18 @@ class EsphomeCore:
             return self.relative_pioenvs_path(self.name, "firmware.uf2")
         return self.relative_pioenvs_path(self.name, "firmware.bin")
 
+    @property
+    def firmware_md5(self):
+        return self.firmware_bin + '.md5'
+
+    @property
+    def firmware_gz(self):
+        return self.firmware_bin + '.gz'
+
+    @property
+    def firmware_gz_md5(self):
+        return self.firmware_gz + '.md5'
+
     @property
     def target_platform(self):
         return self.data[KEY_CORE][KEY_TARGET_PLATFORM]
diff --git a/esphome/dashboard/web_server.py b/esphome/dashboard/web_server.py
index 9ee2312781..24bc395ec7 100644
--- a/esphome/dashboard/web_server.py
+++ b/esphome/dashboard/web_server.py
@@ -375,6 +375,11 @@ class EsphomeRunHandler(EsphomePortCommandWebSocket):
         """Build the command to run."""
         return await self.build_device_command(["run"], json_message)
 
+class EsphomePublishHandler(EsphomePortCommandWebSocket):
+    async def build_command(self, json_message: dict[str, Any]) -> list[str]:
+        config_file = settings.rel_path(json_message["configuration"])
+        return [*DASHBOARD_COMMAND, "publish", config_file]
+
 
 class EsphomeCompileHandler(EsphomeCommandWebSocket):
     async def build_command(self, json_message: dict[str, Any]) -> list[str]:
@@ -1038,6 +1043,8 @@ def get_base_frontend_path() -> str:
     if not static_path.endswith("/"):
         static_path += "/"
 
+    _LOGGER.warning("esphome_dashboard %s" % os.path.abspath(os.path.join(os.getcwd(), static_path, "esphome_dashboard")))
+
     # This path can be relative, so resolve against the root or else templates don't work
     return os.path.abspath(os.path.join(os.getcwd(), static_path, "esphome_dashboard"))
 
@@ -1118,6 +1125,7 @@ def make_app(debug=get_bool_env(ENV_DEV)) -> tornado.web.Application:
             (f"{rel}logs", EsphomeLogsHandler),
             (f"{rel}upload", EsphomeUploadHandler),
             (f"{rel}run", EsphomeRunHandler),
+            (f"{rel}publish", EsphomePublishHandler),
             (f"{rel}compile", EsphomeCompileHandler),
             (f"{rel}validate", EsphomeValidateHandler),
             (f"{rel}clean-mqtt", EsphomeCleanMqttHandler),
diff --git a/esphome/espota2.py b/esphome/espota2.py
index 580536153a..1f1e22ae70 100644
--- a/esphome/espota2.py
+++ b/esphome/espota2.py
@@ -8,8 +8,9 @@ import random
 import socket
 import sys
 import time
+from typing import Optional
 
-from esphome.core import EsphomeError
+from esphome.core import EsphomeError, CORE
 from esphome.helpers import is_ip_address, resolve_ip_address
 
 RESPONSE_OK = 0x00
@@ -194,11 +195,11 @@ def send_check(sock, data, msg):
 
 
 def perform_ota(
-    sock: socket.socket, password: str, file_handle: io.IOBase, filename: str
+    sock: socket.socket, password: str, filename: Optional[str] = None
 ) -> None:
-    file_contents = file_handle.read()
-    file_size = len(file_contents)
-    _LOGGER.info("Uploading %s (%s bytes)", filename, file_size)
+    #file_contents = file_handle.read()
+    #file_size = len(file_contents)
+    #_LOGGER.info("Uploading %s (%s bytes)", filename, file_size)
 
     # Enable nodelay, we need it for phase 1
     sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
@@ -219,10 +220,28 @@ def perform_ota(
     )[0]
 
     if features == RESPONSE_SUPPORTS_COMPRESSION:
-        upload_contents = gzip.compress(file_contents, compresslevel=9)
-        _LOGGER.info("Compressed to %s bytes", len(upload_contents))
+        filename = CORE.firmware_gz
+        filename_md5 = CORE.firmware_gz_md5
+        #upload_contents = gzip.compress(file_contents, compresslevel=9)
+        #_LOGGER.info("Compressed to %s bytes", len(upload_contents))
+        _LOGGER.info("Using compressed image")
     else:
-        upload_contents = file_contents
+        _LOGGER.info("Not using compressed image")
+        filename = CORE.firmware_bin
+        filename_md5 = CORE.firmware_md5
+        #upload_contents = file_contents
+
+    try:
+        with open(filename, "rb") as file_handle:
+            file_contents = file_handle.read()
+    except IOError as e:
+        raise OTAError(f"Error on {filename}: {str(e)}")
+
+    try:
+        with open(filename_md5, "r", encoding='ascii') as file_md5_handle:
+            md5 = file_md5_handle.read()
+    except IOError as e:
+        raise OTAError(f"Error on {filename_md5}: {str(e)}")
 
     (auth,) = receive_exactly(
         sock, 1, "auth", [RESPONSE_REQUEST_AUTH, RESPONSE_AUTH_OK]
@@ -249,7 +268,7 @@ def perform_ota(
         send_check(sock, result, "auth result")
         receive_exactly(sock, 1, "auth result", RESPONSE_AUTH_OK)
 
-    upload_size = len(upload_contents)
+    upload_size = len(file_contents)
     upload_size_encoded = [
         (upload_size >> 24) & 0xFF,
         (upload_size >> 16) & 0xFF,
@@ -259,10 +278,9 @@ def perform_ota(
     send_check(sock, upload_size_encoded, "binary size")
     receive_exactly(sock, 1, "binary size", RESPONSE_UPDATE_PREPARE_OK)
 
-    upload_md5 = hashlib.md5(upload_contents).hexdigest()
-    _LOGGER.debug("MD5 of upload is %s", upload_md5)
+    _LOGGER.debug("MD5 of upload is %s", md5)
 
-    send_check(sock, upload_md5, "file checksum")
+    send_check(sock, md5, "file checksum")
     receive_exactly(sock, 1, "file checksum", RESPONSE_BIN_MD5_OK)
 
     # Disable nodelay for transfer
@@ -278,7 +296,7 @@ def perform_ota(
     offset = 0
     progress = ProgressBar()
     while True:
-        chunk = upload_contents[offset : offset + UPLOAD_BLOCK_SIZE]
+        chunk = file_contents[offset : offset + UPLOAD_BLOCK_SIZE]
         if not chunk:
             break
         offset += len(chunk)
@@ -310,7 +328,7 @@ def perform_ota(
     time.sleep(1)
 
 
-def run_ota_impl_(remote_host, remote_port, password, filename):
+def run_ota_impl_(remote_host, remote_port, password, filename=None):
     if is_ip_address(remote_host):
         _LOGGER.info("Connecting to %s", remote_host)
         ip = remote_host
@@ -339,21 +357,20 @@ def run_ota_impl_(remote_host, remote_port, password, filename):
         _LOGGER.error("Connecting to %s:%s failed: %s", remote_host, remote_port, err)
         return 1
 
-    with open(filename, "rb") as file_handle:
-        try:
-            perform_ota(sock, password, file_handle, filename)
-        except OTAError as err:
-            _LOGGER.error(str(err))
-            return 1
-        finally:
-            sock.close()
+    try:
+        perform_ota(sock, password, filename=filename)
+    except OTAError as err:
+        _LOGGER.error(str(err))
+        return 1
+    finally:
+        sock.close()
 
     return 0
 
 
-def run_ota(remote_host, remote_port, password, filename):
+def run_ota(remote_host, remote_port, password, filename=None):
     try:
-        return run_ota_impl_(remote_host, remote_port, password, filename)
+        return run_ota_impl_(remote_host, remote_port, password, filename=filename)
     except OTAError as err:
         _LOGGER.error(err)
         return 1