sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-10-30 22:53:59 +00:00
Code Issues Releases Wiki Activity

Merge branch 'integration' into memory_api

Browse Source
This commit is contained in:
J. Nick Koston
2025-09-08 10:05:14 -05:00
parent f37223c756 6b73cdbafb
commit 4c9fd2b800
4 changed files with 57 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
esphome/__main__.py
View File

@@ -398,28 +398,29 @@ def check_permissions(port: str):
def upload_program(
config: ConfigType, args: ArgsProtocol, devices: list[str]
) -> int | str:
) -> tuple[int, str | None]:
host = devices[0]
try:
module = importlib.import_module("esphome.components." + CORE.target_platform)
if getattr(module, "upload_program")(config, args, host):
return 0
return 0, host
except AttributeError:
pass
if get_port_type(host) == "SERIAL":
check_permissions(host)
exit_code = 1
if CORE.target_platform in (PLATFORM_ESP32, PLATFORM_ESP8266):
file = getattr(args, "file", None)
return upload_using_esptool(config, host, file, args.upload_speed)
exit_code = upload_using_esptool(config, host, file, args.upload_speed)
elif CORE.target_platform == PLATFORM_RP2040 or CORE.is_libretiny:
exit_code = upload_using_platformio(config, host)
else:
# Unknown target platform
pass
if CORE.target_platform in (PLATFORM_RP2040):
return upload_using_platformio(config, host)
if CORE.is_libretiny:
return upload_using_platformio(config, host)
return 1 # Unknown target platform
return exit_code, host if exit_code == 0 else None
ota_conf = {}
for ota_item in config.get(CONF_OTA, []):
@@ -560,7 +561,7 @@ def command_upload(args: ArgsProtocol, config: ConfigType) -> int | None:
purpose="uploading",
)
exit_code = upload_program(config, args, devices)
exit_code, successful_device = upload_program(config, args, devices)
if exit_code == 0:
_LOGGER.info("Successfully uploaded program.")
else:
@@ -617,19 +618,11 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None:
purpose="uploading",
)
# Try each device for upload until one succeeds
successful_device: str | None = None
for device in devices:
_LOGGER.info("Uploading to %s", device)
exit_code = upload_program(config, args, device)
if exit_code == 0:
_LOGGER.info("Successfully uploaded program.")
successful_device = device
break
if len(devices) > 1:
_LOGGER.warning("Failed to upload to %s", device)
if successful_device is None:
exit_code, successful_device = upload_program(config, args, devices)
if exit_code == 0:
_LOGGER.info("Successfully uploaded program.")
else:
_LOGGER.warning("Failed to upload to %s", devices)
return exit_code
if args.no_logs:

44
esphome/components/esp32/gpio.cpp
View File

@@ -56,7 +56,7 @@ ISRInternalGPIOPin ESP32InternalGPIOPin::to_isr() const {
auto *arg = new ISRPinArg{}; // NOLINT(cppcoreguidelines-owning-memory)
arg->pin = this->get_pin_num();
arg->flags = gpio::FLAG_NONE;
arg->inverted = pin_flags_.inverted;
arg->inverted = this->pin_flags_.inverted;
#if defined(USE_ESP32_VARIANT_ESP32)
arg->use_rtc = rtc_gpio_is_valid_gpio(this->get_pin_num());
if (arg->use_rtc)
@@ -69,23 +69,23 @@ void ESP32InternalGPIOPin::attach_interrupt(void (*func)(void *), void *arg, gpi
gpio_int_type_t idf_type = GPIO_INTR_ANYEDGE;
switch (type) {
case gpio::INTERRUPT_RISING_EDGE:
idf_type = pin_flags_.inverted ? GPIO_INTR_NEGEDGE : GPIO_INTR_POSEDGE;
idf_type = this->pin_flags_.inverted ? GPIO_INTR_NEGEDGE : GPIO_INTR_POSEDGE;
break;
case gpio::INTERRUPT_FALLING_EDGE:
idf_type = pin_flags_.inverted ? GPIO_INTR_POSEDGE : GPIO_INTR_NEGEDGE;
idf_type = this->pin_flags_.inverted ? GPIO_INTR_POSEDGE : GPIO_INTR_NEGEDGE;
break;
case gpio::INTERRUPT_ANY_EDGE:
idf_type = GPIO_INTR_ANYEDGE;
break;
case gpio::INTERRUPT_LOW_LEVEL:
idf_type = pin_flags_.inverted ? GPIO_INTR_HIGH_LEVEL : GPIO_INTR_LOW_LEVEL;
idf_type = this->pin_flags_.inverted ? GPIO_INTR_HIGH_LEVEL : GPIO_INTR_LOW_LEVEL;
break;
case gpio::INTERRUPT_HIGH_LEVEL:
idf_type = pin_flags_.inverted ? GPIO_INTR_LOW_LEVEL : GPIO_INTR_HIGH_LEVEL;
idf_type = this->pin_flags_.inverted ? GPIO_INTR_LOW_LEVEL : GPIO_INTR_HIGH_LEVEL;
break;
}
gpio_set_intr_type(get_pin_num(), idf_type);
gpio_intr_enable(get_pin_num());
gpio_set_intr_type(this->get_pin_num(), idf_type);
gpio_intr_enable(this->get_pin_num());
if (!isr_service_installed) {
auto res = gpio_install_isr_service(ESP_INTR_FLAG_LEVEL3);
if (res != ESP_OK) {
@@ -94,31 +94,31 @@ void ESP32InternalGPIOPin::attach_interrupt(void (*func)(void *), void *arg, gpi
}
isr_service_installed = true;
}
gpio_isr_handler_add(get_pin_num(), func, arg);
gpio_isr_handler_add(this->get_pin_num(), func, arg);
}
std::string ESP32InternalGPIOPin::dump_summary() const {
char buffer[32];
snprintf(buffer, sizeof(buffer), "GPIO%" PRIu32, static_cast<uint32_t>(pin_));
snprintf(buffer, sizeof(buffer), "GPIO%" PRIu32, static_cast<uint32_t>(this->pin_));
return buffer;
}
void ESP32InternalGPIOPin::setup() {
gpio_config_t conf{};
conf.pin_bit_mask = 1ULL << static_cast<uint32_t>(pin_);
conf.mode = flags_to_mode(flags_);
conf.pull_up_en = flags_ & gpio::FLAG_PULLUP ? GPIO_PULLUP_ENABLE : GPIO_PULLUP_DISABLE;
conf.pull_down_en = flags_ & gpio::FLAG_PULLDOWN ? GPIO_PULLDOWN_ENABLE : GPIO_PULLDOWN_DISABLE;
conf.pin_bit_mask = 1ULL << static_cast<uint32_t>(this->pin_);
conf.mode = flags_to_mode(this->flags_);
conf.pull_up_en = this->flags_ & gpio::FLAG_PULLUP ? GPIO_PULLUP_ENABLE : GPIO_PULLUP_DISABLE;
conf.pull_down_en = this->flags_ & gpio::FLAG_PULLDOWN ? GPIO_PULLDOWN_ENABLE : GPIO_PULLDOWN_DISABLE;
conf.intr_type = GPIO_INTR_DISABLE;
gpio_config(&conf);
if (flags_ & gpio::FLAG_OUTPUT) {
gpio_set_drive_capability(get_pin_num(), get_drive_strength());
if (this->flags_ & gpio::FLAG_OUTPUT) {
gpio_set_drive_capability(this->get_pin_num(), this->get_drive_strength());
}
}
void ESP32InternalGPIOPin::pin_mode(gpio::Flags flags) {
// can't call gpio_config here because that logs in esp-idf which may cause issues
gpio_set_direction(get_pin_num(), flags_to_mode(flags));
gpio_set_direction(this->get_pin_num(), flags_to_mode(flags));
gpio_pull_mode_t pull_mode = GPIO_FLOATING;
if ((flags & gpio::FLAG_PULLUP) && (flags & gpio::FLAG_PULLDOWN)) {
pull_mode = GPIO_PULLUP_PULLDOWN;
@@ -127,14 +127,16 @@ void ESP32InternalGPIOPin::pin_mode(gpio::Flags flags) {
} else if (flags & gpio::FLAG_PULLDOWN) {
pull_mode = GPIO_PULLDOWN_ONLY;
}
gpio_set_pull_mode(get_pin_num(), pull_mode);
gpio_set_pull_mode(this->get_pin_num(), pull_mode);
}
bool ESP32InternalGPIOPin::digital_read() { return bool(gpio_get_level(get_pin_num())) != pin_flags_.inverted; }
void ESP32InternalGPIOPin::digital_write(bool value) {
gpio_set_level(get_pin_num(), value != pin_flags_.inverted ? 1 : 0);
bool ESP32InternalGPIOPin::digital_read() {
return bool(gpio_get_level(this->get_pin_num())) != this->pin_flags_.inverted;
}
void ESP32InternalGPIOPin::detach_interrupt() const { gpio_intr_disable(get_pin_num()); }
void ESP32InternalGPIOPin::digital_write(bool value) {
gpio_set_level(this->get_pin_num(), value != this->pin_flags_.inverted ? 1 : 0);
}
void ESP32InternalGPIOPin::detach_interrupt() const { gpio_intr_disable(this->get_pin_num()); }
} // namespace esp32

20
esphome/components/esp32/gpio.h
View File

@@ -13,12 +13,12 @@ static_assert(GPIO_DRIVE_CAP_MAX <= 4, "gpio_drive_cap_t has too many values for
class ESP32InternalGPIOPin : public InternalGPIOPin {
public:
void set_pin(gpio_num_t pin) { pin_ = static_cast<uint8_t>(pin); }
void set_inverted(bool inverted) { pin_flags_.inverted = inverted; }
void set_pin(gpio_num_t pin) { this->pin_ = static_cast<uint8_t>(pin); }
void set_inverted(bool inverted) { this->pin_flags_.inverted = inverted; }
void set_drive_strength(gpio_drive_cap_t drive_strength) {
pin_flags_.drive_strength = static_cast<uint8_t>(drive_strength);
this->pin_flags_.drive_strength = static_cast<uint8_t>(drive_strength);
}
void set_flags(gpio::Flags flags) { flags_ = flags; }
void set_flags(gpio::Flags flags) { this->flags_ = flags; }
void setup() override;
void pin_mode(gpio::Flags flags) override;
@@ -27,11 +27,11 @@ class ESP32InternalGPIOPin : public InternalGPIOPin {
std::string dump_summary() const override;
void detach_interrupt() const override;
ISRInternalGPIOPin to_isr() const override;
uint8_t get_pin() const override { return pin_; }
gpio::Flags get_flags() const override { return flags_; }
bool is_inverted() const override { return pin_flags_.inverted; }
gpio_num_t get_pin_num() const { return static_cast<gpio_num_t>(pin_); }
gpio_drive_cap_t get_drive_strength() const { return static_cast<gpio_drive_cap_t>(pin_flags_.drive_strength); }
uint8_t get_pin() const override { return this->pin_; }
gpio::Flags get_flags() const override { return this->flags_; }
bool is_inverted() const override { return this->pin_flags_.inverted; }
gpio_num_t get_pin_num() const { return static_cast<gpio_num_t>(this->pin_); }
gpio_drive_cap_t get_drive_strength() const { return static_cast<gpio_drive_cap_t>(this->pin_flags_.drive_strength); }
protected:
void attach_interrupt(void (*func)(void *), void *arg, gpio::InterruptType type) const override;
@@ -40,7 +40,7 @@ class ESP32InternalGPIOPin : public InternalGPIOPin {
// - 3 bytes for members below
// - 1 byte padding for alignment
// - 4 bytes for vtable pointer
uint8_t pin_; // GPIO pin number (0-255, actual max ~48 on ESP32)
uint8_t pin_; // GPIO pin number (0-255, actual max ~54 on ESP32)
gpio::Flags flags_; // GPIO flags (1 byte)
struct PinFlags {
uint8_t inverted : 1; // Invert pin logic (1 bit)

13
esphome/espota2.py
View File

@@ -310,7 +310,7 @@ def perform_ota(
def run_ota_impl_(
remote_host: str | list[str], remote_port: int, password: str, filename: str
) -> int:
) -> tuple[int, str | None]:
# Handle both single host and list of hosts
try:
# Resolve all hosts at once for parallel DNS resolution
@@ -344,21 +344,22 @@ def run_ota_impl_(
perform_ota(sock, password, file_handle, filename)
except OTAError as err:
_LOGGER.error(str(err))
return 1
return 1, None
finally:
sock.close()
return 0
# Successfully uploaded to sa[0]
return 0, sa[0]
_LOGGER.error("Connection failed.")
return 1
return 1, None
def run_ota(
remote_host: str | list[str], remote_port: int, password: str, filename: str
) -> int:
) -> tuple[int, str | None]:
try:
return run_ota_impl_(remote_host, remote_port, password, filename)
except OTAError as err:
_LOGGER.error(err)
return 1
return 1, None