From eb89d99999941a6de6872d3bea01d2ef3c684924 Mon Sep 17 00:00:00 2001
From: Keith Burzinski <kbx81x@gmail.com>
Date: Mon, 22 Apr 2024 23:47:03 -0500
Subject: [PATCH] Add valve component (#6447)

Co-authored-by: Jesse Hills <3060199+jesserockz@users.noreply.github.com>
---
 CODEOWNERS                                    |   1 +
 esphome/components/api/api.proto              |  51 ++++
 esphome/components/api/api_connection.cpp     |  42 +++
 esphome/components/api/api_connection.h       |   5 +
 esphome/components/api/api_pb2.cpp            | 247 ++++++++++++++++++
 esphome/components/api/api_pb2.h              |  57 ++++
 esphome/components/api/api_pb2_service.cpp    |  42 +++
 esphome/components/api/api_pb2_service.h      |  15 ++
 esphome/components/api/api_server.cpp         |   9 +
 esphome/components/api/api_server.h           |   3 +
 esphome/components/api/list_entities.cpp      |   3 +
 esphome/components/api/list_entities.h        |   3 +
 esphome/components/api/subscribe_state.cpp    |   3 +
 esphome/components/api/subscribe_state.h      |   3 +
 esphome/components/mqtt/__init__.py           |   1 +
 esphome/components/mqtt/mqtt_valve.cpp        |  90 +++++++
 esphome/components/mqtt/mqtt_valve.h          |  41 +++
 esphome/components/template/valve/__init__.py | 118 +++++++++
 .../template/valve/template_valve.cpp         | 131 ++++++++++
 .../template/valve/template_valve.h           |  60 +++++
 esphome/components/valve/__init__.py          | 186 +++++++++++++
 esphome/components/valve/automation.h         | 129 +++++++++
 esphome/components/valve/valve.cpp            | 179 +++++++++++++
 esphome/components/valve/valve.h              | 152 +++++++++++
 esphome/components/valve/valve_traits.h       |  27 ++
 .../components/web_server/list_entities.cpp   |   9 +
 esphome/components/web_server/list_entities.h |   3 +
 esphome/components/web_server/web_server.cpp  |  74 ++++++
 esphome/components/web_server/web_server.h    |  10 +
 esphome/core/application.h                    |  19 ++
 esphome/core/component_iterator.cpp           |  15 ++
 esphome/core/component_iterator.h             |   6 +
 esphome/core/controller.cpp                   |   6 +
 esphome/core/controller.h                     |   6 +
 esphome/core/defines.h                        |   1 +
 script/ci-custom.py                           |   1 +
 tests/components/template/test.all.yaml       |  17 ++
 37 files changed, 1765 insertions(+)
 create mode 100644 esphome/components/mqtt/mqtt_valve.cpp
 create mode 100644 esphome/components/mqtt/mqtt_valve.h
 create mode 100644 esphome/components/template/valve/__init__.py
 create mode 100644 esphome/components/template/valve/template_valve.cpp
 create mode 100644 esphome/components/template/valve/template_valve.h
 create mode 100644 esphome/components/valve/__init__.py
 create mode 100644 esphome/components/valve/automation.h
 create mode 100644 esphome/components/valve/valve.cpp
 create mode 100644 esphome/components/valve/valve.h
 create mode 100644 esphome/components/valve/valve_traits.h

diff --git a/CODEOWNERS b/CODEOWNERS
index 9f770d4efc..28ce8e7b6a 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -390,6 +390,7 @@ esphome/components/ufire_ec/* @pvizeli
 esphome/components/ufire_ise/* @pvizeli
 esphome/components/ultrasonic/* @OttoWinter
 esphome/components/uponor_smatrix/* @kroimon
+esphome/components/valve/* @esphome/core
 esphome/components/vbus/* @ssieb
 esphome/components/veml3235/* @kbx81
 esphome/components/veml7700/* @latonita
diff --git a/esphome/components/api/api.proto b/esphome/components/api/api.proto
index 8d5459e717..12b7ef0958 100644
--- a/esphome/components/api/api.proto
+++ b/esphome/components/api/api.proto
@@ -43,6 +43,7 @@ service APIConnection {
   rpc select_command (SelectCommandRequest) returns (void) {}
   rpc button_command (ButtonCommandRequest) returns (void) {}
   rpc lock_command (LockCommandRequest) returns (void) {}
+  rpc valve_command (ValveCommandRequest) returns (void) {}
   rpc media_player_command (MediaPlayerCommandRequest) returns (void) {}
   rpc date_command (DateCommandRequest) returns (void) {}
   rpc time_command (TimeCommandRequest) returns (void) {}
@@ -1700,3 +1701,53 @@ message TimeCommandRequest {
   uint32 minute = 3;
   uint32 second = 4;
 }
+
+
+// ==================== VALVE ====================
+message ListEntitiesValveResponse {
+  option (id) = 109;
+  option (source) = SOURCE_SERVER;
+  option (ifdef) = "USE_VALVE";
+
+  string object_id = 1;
+  fixed32 key = 2;
+  string name = 3;
+  string unique_id = 4;
+
+  string icon = 5;
+  bool disabled_by_default = 6;
+  EntityCategory entity_category = 7;
+  string device_class = 8;
+
+  bool assumed_state = 9;
+  bool supports_position = 10;
+  bool supports_stop = 11;
+}
+
+enum ValveOperation {
+  VALVE_OPERATION_IDLE = 0;
+  VALVE_OPERATION_IS_OPENING = 1;
+  VALVE_OPERATION_IS_CLOSING = 2;
+}
+message ValveStateResponse {
+  option (id) = 110;
+  option (source) = SOURCE_SERVER;
+  option (ifdef) = "USE_VALVE";
+  option (no_delay) = true;
+
+  fixed32 key = 1;
+  float position = 2;
+  ValveOperation current_operation = 3;
+}
+
+message ValveCommandRequest {
+  option (id) = 111;
+  option (source) = SOURCE_CLIENT;
+  option (ifdef) = "USE_VALVE";
+  option (no_delay) = true;
+
+  fixed32 key = 1;
+  bool has_position = 2;
+  float position = 3;
+  bool stop = 4;
+}
diff --git a/esphome/components/api/api_connection.cpp b/esphome/components/api/api_connection.cpp
index e51fa8c154..47136fff98 100644
--- a/esphome/components/api/api_connection.cpp
+++ b/esphome/components/api/api_connection.cpp
@@ -915,6 +915,48 @@ void APIConnection::lock_command(const LockCommandRequest &msg) {
 }
 #endif
 
+#ifdef USE_VALVE
+bool APIConnection::send_valve_state(valve::Valve *valve) {
+  if (!this->state_subscription_)
+    return false;
+
+  ValveStateResponse resp{};
+  resp.key = valve->get_object_id_hash();
+  resp.position = valve->position;
+  resp.current_operation = static_cast<enums::ValveOperation>(valve->current_operation);
+  return this->send_valve_state_response(resp);
+}
+bool APIConnection::send_valve_info(valve::Valve *valve) {
+  auto traits = valve->get_traits();
+  ListEntitiesValveResponse msg;
+  msg.key = valve->get_object_id_hash();
+  msg.object_id = valve->get_object_id();
+  if (valve->has_own_name())
+    msg.name = valve->get_name();
+  msg.unique_id = get_default_unique_id("valve", valve);
+  msg.icon = valve->get_icon();
+  msg.disabled_by_default = valve->is_disabled_by_default();
+  msg.entity_category = static_cast<enums::EntityCategory>(valve->get_entity_category());
+  msg.device_class = valve->get_device_class();
+  msg.assumed_state = traits.get_is_assumed_state();
+  msg.supports_position = traits.get_supports_position();
+  msg.supports_stop = traits.get_supports_stop();
+  return this->send_list_entities_valve_response(msg);
+}
+void APIConnection::valve_command(const ValveCommandRequest &msg) {
+  valve::Valve *valve = App.get_valve_by_key(msg.key);
+  if (valve == nullptr)
+    return;
+
+  auto call = valve->make_call();
+  if (msg.has_position)
+    call.set_position(msg.position);
+  if (msg.stop)
+    call.set_command_stop();
+  call.perform();
+}
+#endif
+
 #ifdef USE_MEDIA_PLAYER
 bool APIConnection::send_media_player_state(media_player::MediaPlayer *media_player) {
   if (!this->state_subscription_)
diff --git a/esphome/components/api/api_connection.h b/esphome/components/api/api_connection.h
index 5c0a78015d..af2dd9e681 100644
--- a/esphome/components/api/api_connection.h
+++ b/esphome/components/api/api_connection.h
@@ -101,6 +101,11 @@ class APIConnection : public APIServerConnection {
   bool send_lock_info(lock::Lock *a_lock);
   void lock_command(const LockCommandRequest &msg) override;
 #endif
+#ifdef USE_VALVE
+  bool send_valve_state(valve::Valve *valve);
+  bool send_valve_info(valve::Valve *valve);
+  void valve_command(const ValveCommandRequest &msg) override;
+#endif
 #ifdef USE_MEDIA_PLAYER
   bool send_media_player_state(media_player::MediaPlayer *media_player);
   bool send_media_player_info(media_player::MediaPlayer *media_player);
diff --git a/esphome/components/api/api_pb2.cpp b/esphome/components/api/api_pb2.cpp
index 884396bda3..508947ba24 100644
--- a/esphome/components/api/api_pb2.cpp
+++ b/esphome/components/api/api_pb2.cpp
@@ -537,6 +537,20 @@ template<> const char *proto_enum_to_string<enums::TextMode>(enums::TextMode val
   }
 }
 #endif
+#ifdef HAS_PROTO_MESSAGE_DUMP
+template<> const char *proto_enum_to_string<enums::ValveOperation>(enums::ValveOperation value) {
+  switch (value) {
+    case enums::VALVE_OPERATION_IDLE:
+      return "VALVE_OPERATION_IDLE";
+    case enums::VALVE_OPERATION_IS_OPENING:
+      return "VALVE_OPERATION_IS_OPENING";
+    case enums::VALVE_OPERATION_IS_CLOSING:
+      return "VALVE_OPERATION_IS_CLOSING";
+    default:
+      return "UNKNOWN";
+  }
+}
+#endif
 bool HelloRequest::decode_varint(uint32_t field_id, ProtoVarInt value) {
   switch (field_id) {
     case 2: {
@@ -7695,6 +7709,239 @@ void TimeCommandRequest::dump_to(std::string &out) const {
   out.append("}");
 }
 #endif
+bool ListEntitiesValveResponse::decode_varint(uint32_t field_id, ProtoVarInt value) {
+  switch (field_id) {
+    case 6: {
+      this->disabled_by_default = value.as_bool();
+      return true;
+    }
+    case 7: {
+      this->entity_category = value.as_enum<enums::EntityCategory>();
+      return true;
+    }
+    case 9: {
+      this->assumed_state = value.as_bool();
+      return true;
+    }
+    case 10: {
+      this->supports_position = value.as_bool();
+      return true;
+    }
+    case 11: {
+      this->supports_stop = value.as_bool();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+bool ListEntitiesValveResponse::decode_length(uint32_t field_id, ProtoLengthDelimited value) {
+  switch (field_id) {
+    case 1: {
+      this->object_id = value.as_string();
+      return true;
+    }
+    case 3: {
+      this->name = value.as_string();
+      return true;
+    }
+    case 4: {
+      this->unique_id = value.as_string();
+      return true;
+    }
+    case 5: {
+      this->icon = value.as_string();
+      return true;
+    }
+    case 8: {
+      this->device_class = value.as_string();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+bool ListEntitiesValveResponse::decode_32bit(uint32_t field_id, Proto32Bit value) {
+  switch (field_id) {
+    case 2: {
+      this->key = value.as_fixed32();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+void ListEntitiesValveResponse::encode(ProtoWriteBuffer buffer) const {
+  buffer.encode_string(1, this->object_id);
+  buffer.encode_fixed32(2, this->key);
+  buffer.encode_string(3, this->name);
+  buffer.encode_string(4, this->unique_id);
+  buffer.encode_string(5, this->icon);
+  buffer.encode_bool(6, this->disabled_by_default);
+  buffer.encode_enum<enums::EntityCategory>(7, this->entity_category);
+  buffer.encode_string(8, this->device_class);
+  buffer.encode_bool(9, this->assumed_state);
+  buffer.encode_bool(10, this->supports_position);
+  buffer.encode_bool(11, this->supports_stop);
+}
+#ifdef HAS_PROTO_MESSAGE_DUMP
+void ListEntitiesValveResponse::dump_to(std::string &out) const {
+  __attribute__((unused)) char buffer[64];
+  out.append("ListEntitiesValveResponse {\n");
+  out.append("  object_id: ");
+  out.append("'").append(this->object_id).append("'");
+  out.append("\n");
+
+  out.append("  key: ");
+  sprintf(buffer, "%" PRIu32, this->key);
+  out.append(buffer);
+  out.append("\n");
+
+  out.append("  name: ");
+  out.append("'").append(this->name).append("'");
+  out.append("\n");
+
+  out.append("  unique_id: ");
+  out.append("'").append(this->unique_id).append("'");
+  out.append("\n");
+
+  out.append("  icon: ");
+  out.append("'").append(this->icon).append("'");
+  out.append("\n");
+
+  out.append("  disabled_by_default: ");
+  out.append(YESNO(this->disabled_by_default));
+  out.append("\n");
+
+  out.append("  entity_category: ");
+  out.append(proto_enum_to_string<enums::EntityCategory>(this->entity_category));
+  out.append("\n");
+
+  out.append("  device_class: ");
+  out.append("'").append(this->device_class).append("'");
+  out.append("\n");
+
+  out.append("  assumed_state: ");
+  out.append(YESNO(this->assumed_state));
+  out.append("\n");
+
+  out.append("  supports_position: ");
+  out.append(YESNO(this->supports_position));
+  out.append("\n");
+
+  out.append("  supports_stop: ");
+  out.append(YESNO(this->supports_stop));
+  out.append("\n");
+  out.append("}");
+}
+#endif
+bool ValveStateResponse::decode_varint(uint32_t field_id, ProtoVarInt value) {
+  switch (field_id) {
+    case 3: {
+      this->current_operation = value.as_enum<enums::ValveOperation>();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+bool ValveStateResponse::decode_32bit(uint32_t field_id, Proto32Bit value) {
+  switch (field_id) {
+    case 1: {
+      this->key = value.as_fixed32();
+      return true;
+    }
+    case 2: {
+      this->position = value.as_float();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+void ValveStateResponse::encode(ProtoWriteBuffer buffer) const {
+  buffer.encode_fixed32(1, this->key);
+  buffer.encode_float(2, this->position);
+  buffer.encode_enum<enums::ValveOperation>(3, this->current_operation);
+}
+#ifdef HAS_PROTO_MESSAGE_DUMP
+void ValveStateResponse::dump_to(std::string &out) const {
+  __attribute__((unused)) char buffer[64];
+  out.append("ValveStateResponse {\n");
+  out.append("  key: ");
+  sprintf(buffer, "%" PRIu32, this->key);
+  out.append(buffer);
+  out.append("\n");
+
+  out.append("  position: ");
+  sprintf(buffer, "%g", this->position);
+  out.append(buffer);
+  out.append("\n");
+
+  out.append("  current_operation: ");
+  out.append(proto_enum_to_string<enums::ValveOperation>(this->current_operation));
+  out.append("\n");
+  out.append("}");
+}
+#endif
+bool ValveCommandRequest::decode_varint(uint32_t field_id, ProtoVarInt value) {
+  switch (field_id) {
+    case 2: {
+      this->has_position = value.as_bool();
+      return true;
+    }
+    case 4: {
+      this->stop = value.as_bool();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+bool ValveCommandRequest::decode_32bit(uint32_t field_id, Proto32Bit value) {
+  switch (field_id) {
+    case 1: {
+      this->key = value.as_fixed32();
+      return true;
+    }
+    case 3: {
+      this->position = value.as_float();
+      return true;
+    }
+    default:
+      return false;
+  }
+}
+void ValveCommandRequest::encode(ProtoWriteBuffer buffer) const {
+  buffer.encode_fixed32(1, this->key);
+  buffer.encode_bool(2, this->has_position);
+  buffer.encode_float(3, this->position);
+  buffer.encode_bool(4, this->stop);
+}
+#ifdef HAS_PROTO_MESSAGE_DUMP
+void ValveCommandRequest::dump_to(std::string &out) const {
+  __attribute__((unused)) char buffer[64];
+  out.append("ValveCommandRequest {\n");
+  out.append("  key: ");
+  sprintf(buffer, "%" PRIu32, this->key);
+  out.append(buffer);
+  out.append("\n");
+
+  out.append("  has_position: ");
+  out.append(YESNO(this->has_position));
+  out.append("\n");
+
+  out.append("  position: ");
+  sprintf(buffer, "%g", this->position);
+  out.append(buffer);
+  out.append("\n");
+
+  out.append("  stop: ");
+  out.append(YESNO(this->stop));
+  out.append("\n");
+  out.append("}");
+}
+#endif
 
 }  // namespace api
 }  // namespace esphome
diff --git a/esphome/components/api/api_pb2.h b/esphome/components/api/api_pb2.h
index 2ae6fd2bb6..950ffcdc88 100644
--- a/esphome/components/api/api_pb2.h
+++ b/esphome/components/api/api_pb2.h
@@ -216,6 +216,11 @@ enum TextMode : uint32_t {
   TEXT_MODE_TEXT = 0,
   TEXT_MODE_PASSWORD = 1,
 };
+enum ValveOperation : uint32_t {
+  VALVE_OPERATION_IDLE = 0,
+  VALVE_OPERATION_IS_OPENING = 1,
+  VALVE_OPERATION_IS_CLOSING = 2,
+};
 
 }  // namespace enums
 
@@ -1969,6 +1974,58 @@ class TimeCommandRequest : public ProtoMessage {
   bool decode_32bit(uint32_t field_id, Proto32Bit value) override;
   bool decode_varint(uint32_t field_id, ProtoVarInt value) override;
 };
+class ListEntitiesValveResponse : public ProtoMessage {
+ public:
+  std::string object_id{};
+  uint32_t key{0};
+  std::string name{};
+  std::string unique_id{};
+  std::string icon{};
+  bool disabled_by_default{false};
+  enums::EntityCategory entity_category{};
+  std::string device_class{};
+  bool assumed_state{false};
+  bool supports_position{false};
+  bool supports_stop{false};
+  void encode(ProtoWriteBuffer buffer) const override;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  void dump_to(std::string &out) const override;
+#endif
+
+ protected:
+  bool decode_32bit(uint32_t field_id, Proto32Bit value) override;
+  bool decode_length(uint32_t field_id, ProtoLengthDelimited value) override;
+  bool decode_varint(uint32_t field_id, ProtoVarInt value) override;
+};
+class ValveStateResponse : public ProtoMessage {
+ public:
+  uint32_t key{0};
+  float position{0.0f};
+  enums::ValveOperation current_operation{};
+  void encode(ProtoWriteBuffer buffer) const override;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  void dump_to(std::string &out) const override;
+#endif
+
+ protected:
+  bool decode_32bit(uint32_t field_id, Proto32Bit value) override;
+  bool decode_varint(uint32_t field_id, ProtoVarInt value) override;
+};
+class ValveCommandRequest : public ProtoMessage {
+ public:
+  uint32_t key{0};
+  bool has_position{false};
+  float position{0.0f};
+  bool stop{false};
+  void encode(ProtoWriteBuffer buffer) const override;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  void dump_to(std::string &out) const override;
+#endif
+
+ protected:
+  bool decode_32bit(uint32_t field_id, Proto32Bit value) override;
+  bool decode_varint(uint32_t field_id, ProtoVarInt value) override;
+};
 
 }  // namespace api
 }  // namespace esphome
diff --git a/esphome/components/api/api_pb2_service.cpp b/esphome/components/api/api_pb2_service.cpp
index 7a97df1ce0..4b8b8cf5ae 100644
--- a/esphome/components/api/api_pb2_service.cpp
+++ b/esphome/components/api/api_pb2_service.cpp
@@ -557,6 +557,24 @@ bool APIServerConnectionBase::send_time_state_response(const TimeStateResponse &
 #endif
 #ifdef USE_DATETIME_TIME
 #endif
+#ifdef USE_VALVE
+bool APIServerConnectionBase::send_list_entities_valve_response(const ListEntitiesValveResponse &msg) {
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  ESP_LOGVV(TAG, "send_list_entities_valve_response: %s", msg.dump().c_str());
+#endif
+  return this->send_message_<ListEntitiesValveResponse>(msg, 109);
+}
+#endif
+#ifdef USE_VALVE
+bool APIServerConnectionBase::send_valve_state_response(const ValveStateResponse &msg) {
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  ESP_LOGVV(TAG, "send_valve_state_response: %s", msg.dump().c_str());
+#endif
+  return this->send_message_<ValveStateResponse>(msg, 110);
+}
+#endif
+#ifdef USE_VALVE
+#endif
 bool APIServerConnectionBase::read_message(uint32_t msg_size, uint32_t msg_type, uint8_t *msg_data) {
   switch (msg_type) {
     case 1: {
@@ -1019,6 +1037,17 @@ bool APIServerConnectionBase::read_message(uint32_t msg_size, uint32_t msg_type,
       ESP_LOGVV(TAG, "on_voice_assistant_audio: %s", msg.dump().c_str());
 #endif
       this->on_voice_assistant_audio(msg);
+#endif
+      break;
+    }
+    case 111: {
+#ifdef USE_VALVE
+      ValveCommandRequest msg;
+      msg.decode(msg_data, msg_size);
+#ifdef HAS_PROTO_MESSAGE_DUMP
+      ESP_LOGVV(TAG, "on_valve_command_request: %s", msg.dump().c_str());
+#endif
+      this->on_valve_command_request(msg);
 #endif
       break;
     }
@@ -1282,6 +1311,19 @@ void APIServerConnection::on_lock_command_request(const LockCommandRequest &msg)
   this->lock_command(msg);
 }
 #endif
+#ifdef USE_VALVE
+void APIServerConnection::on_valve_command_request(const ValveCommandRequest &msg) {
+  if (!this->is_connection_setup()) {
+    this->on_no_setup_connection();
+    return;
+  }
+  if (!this->is_authenticated()) {
+    this->on_unauthenticated_access();
+    return;
+  }
+  this->valve_command(msg);
+}
+#endif
 #ifdef USE_MEDIA_PLAYER
 void APIServerConnection::on_media_player_command_request(const MediaPlayerCommandRequest &msg) {
   if (!this->is_connection_setup()) {
diff --git a/esphome/components/api/api_pb2_service.h b/esphome/components/api/api_pb2_service.h
index 095ce51b0f..9f1d711257 100644
--- a/esphome/components/api/api_pb2_service.h
+++ b/esphome/components/api/api_pb2_service.h
@@ -279,6 +279,15 @@ class APIServerConnectionBase : public ProtoService {
 #endif
 #ifdef USE_DATETIME_TIME
   virtual void on_time_command_request(const TimeCommandRequest &value){};
+#endif
+#ifdef USE_VALVE
+  bool send_list_entities_valve_response(const ListEntitiesValveResponse &msg);
+#endif
+#ifdef USE_VALVE
+  bool send_valve_state_response(const ValveStateResponse &msg);
+#endif
+#ifdef USE_VALVE
+  virtual void on_valve_command_request(const ValveCommandRequest &value){};
 #endif
  protected:
   bool read_message(uint32_t msg_size, uint32_t msg_type, uint8_t *msg_data) override;
@@ -331,6 +340,9 @@ class APIServerConnection : public APIServerConnectionBase {
 #ifdef USE_LOCK
   virtual void lock_command(const LockCommandRequest &msg) = 0;
 #endif
+#ifdef USE_VALVE
+  virtual void valve_command(const ValveCommandRequest &msg) = 0;
+#endif
 #ifdef USE_MEDIA_PLAYER
   virtual void media_player_command(const MediaPlayerCommandRequest &msg) = 0;
 #endif
@@ -423,6 +435,9 @@ class APIServerConnection : public APIServerConnectionBase {
 #ifdef USE_LOCK
   void on_lock_command_request(const LockCommandRequest &msg) override;
 #endif
+#ifdef USE_VALVE
+  void on_valve_command_request(const ValveCommandRequest &msg) override;
+#endif
 #ifdef USE_MEDIA_PLAYER
   void on_media_player_command_request(const MediaPlayerCommandRequest &msg) override;
 #endif
diff --git a/esphome/components/api/api_server.cpp b/esphome/components/api/api_server.cpp
index 4c809126e6..457eeb1229 100644
--- a/esphome/components/api/api_server.cpp
+++ b/esphome/components/api/api_server.cpp
@@ -300,6 +300,15 @@ void APIServer::on_lock_update(lock::Lock *obj) {
 }
 #endif
 
+#ifdef USE_VALVE
+void APIServer::on_valve_update(valve::Valve *obj) {
+  if (obj->is_internal())
+    return;
+  for (auto &c : this->clients_)
+    c->send_valve_state(obj);
+}
+#endif
+
 #ifdef USE_MEDIA_PLAYER
 void APIServer::on_media_player_update(media_player::MediaPlayer *obj) {
   if (obj->is_internal())
diff --git a/esphome/components/api/api_server.h b/esphome/components/api/api_server.h
index 8a9c26af73..d64643b961 100644
--- a/esphome/components/api/api_server.h
+++ b/esphome/components/api/api_server.h
@@ -81,6 +81,9 @@ class APIServer : public Component, public Controller {
 #ifdef USE_LOCK
   void on_lock_update(lock::Lock *obj) override;
 #endif
+#ifdef USE_VALVE
+  void on_valve_update(valve::Valve *obj) override;
+#endif
 #ifdef USE_MEDIA_PLAYER
   void on_media_player_update(media_player::MediaPlayer *obj) override;
 #endif
diff --git a/esphome/components/api/list_entities.cpp b/esphome/components/api/list_entities.cpp
index 18685ee4d1..d6ff8e5557 100644
--- a/esphome/components/api/list_entities.cpp
+++ b/esphome/components/api/list_entities.cpp
@@ -38,6 +38,9 @@ bool ListEntitiesIterator::on_text_sensor(text_sensor::TextSensor *text_sensor)
 #ifdef USE_LOCK
 bool ListEntitiesIterator::on_lock(lock::Lock *a_lock) { return this->client_->send_lock_info(a_lock); }
 #endif
+#ifdef USE_VALVE
+bool ListEntitiesIterator::on_valve(valve::Valve *valve) { return this->client_->send_valve_info(valve); }
+#endif
 
 bool ListEntitiesIterator::on_end() { return this->client_->send_list_info_done(); }
 ListEntitiesIterator::ListEntitiesIterator(APIConnection *client) : client_(client) {}
diff --git a/esphome/components/api/list_entities.h b/esphome/components/api/list_entities.h
index 95a09fc25b..5d0c243f4a 100644
--- a/esphome/components/api/list_entities.h
+++ b/esphome/components/api/list_entities.h
@@ -61,6 +61,9 @@ class ListEntitiesIterator : public ComponentIterator {
 #ifdef USE_LOCK
   bool on_lock(lock::Lock *a_lock) override;
 #endif
+#ifdef USE_VALVE
+  bool on_valve(valve::Valve *valve) override;
+#endif
 #ifdef USE_MEDIA_PLAYER
   bool on_media_player(media_player::MediaPlayer *media_player) override;
 #endif
diff --git a/esphome/components/api/subscribe_state.cpp b/esphome/components/api/subscribe_state.cpp
index 5eb40cfa7e..7aa8e8ffac 100644
--- a/esphome/components/api/subscribe_state.cpp
+++ b/esphome/components/api/subscribe_state.cpp
@@ -59,6 +59,9 @@ bool InitialStateIterator::on_select(select::Select *select) {
 #ifdef USE_LOCK
 bool InitialStateIterator::on_lock(lock::Lock *a_lock) { return this->client_->send_lock_state(a_lock, a_lock->state); }
 #endif
+#ifdef USE_VALVE
+bool InitialStateIterator::on_valve(valve::Valve *valve) { return this->client_->send_valve_state(valve); }
+#endif
 #ifdef USE_MEDIA_PLAYER
 bool InitialStateIterator::on_media_player(media_player::MediaPlayer *media_player) {
   return this->client_->send_media_player_state(media_player);
diff --git a/esphome/components/api/subscribe_state.h b/esphome/components/api/subscribe_state.h
index 447f1707d2..8d50e0d89a 100644
--- a/esphome/components/api/subscribe_state.h
+++ b/esphome/components/api/subscribe_state.h
@@ -58,6 +58,9 @@ class InitialStateIterator : public ComponentIterator {
 #ifdef USE_LOCK
   bool on_lock(lock::Lock *a_lock) override;
 #endif
+#ifdef USE_VALVE
+  bool on_valve(valve::Valve *valve) override;
+#endif
 #ifdef USE_MEDIA_PLAYER
   bool on_media_player(media_player::MediaPlayer *media_player) override;
 #endif
diff --git a/esphome/components/mqtt/__init__.py b/esphome/components/mqtt/__init__.py
index b2c03c1546..72ee81dbbc 100644
--- a/esphome/components/mqtt/__init__.py
+++ b/esphome/components/mqtt/__init__.py
@@ -119,6 +119,7 @@ MQTTTextComponent = mqtt_ns.class_("MQTTTextComponent", MQTTComponent)
 MQTTSelectComponent = mqtt_ns.class_("MQTTSelectComponent", MQTTComponent)
 MQTTButtonComponent = mqtt_ns.class_("MQTTButtonComponent", MQTTComponent)
 MQTTLockComponent = mqtt_ns.class_("MQTTLockComponent", MQTTComponent)
+MQTTValveComponent = mqtt_ns.class_("MQTTValveComponent", MQTTComponent)
 
 MQTTDiscoveryUniqueIdGenerator = mqtt_ns.enum("MQTTDiscoveryUniqueIdGenerator")
 MQTT_DISCOVERY_UNIQUE_ID_GENERATOR_OPTIONS = {
diff --git a/esphome/components/mqtt/mqtt_valve.cpp b/esphome/components/mqtt/mqtt_valve.cpp
new file mode 100644
index 0000000000..07eeca08d6
--- /dev/null
+++ b/esphome/components/mqtt/mqtt_valve.cpp
@@ -0,0 +1,90 @@
+#include "mqtt_valve.h"
+#include "esphome/core/log.h"
+
+#include "mqtt_const.h"
+
+#ifdef USE_MQTT
+#ifdef USE_VALVE
+
+namespace esphome {
+namespace mqtt {
+
+static const char *const TAG = "mqtt.valve";
+
+using namespace esphome::valve;
+
+MQTTValveComponent::MQTTValveComponent(Valve *valve) : valve_(valve) {}
+void MQTTValveComponent::setup() {
+  auto traits = this->valve_->get_traits();
+  this->valve_->add_on_state_callback([this]() { this->publish_state(); });
+  this->subscribe(this->get_command_topic_(), [this](const std::string &topic, const std::string &payload) {
+    auto call = this->valve_->make_call();
+    call.set_command(payload.c_str());
+    call.perform();
+  });
+  if (traits.get_supports_position()) {
+    this->subscribe(this->get_position_command_topic(), [this](const std::string &topic, const std::string &payload) {
+      auto value = parse_number<float>(payload);
+      if (!value.has_value()) {
+        ESP_LOGW(TAG, "Invalid position value: '%s'", payload.c_str());
+        return;
+      }
+      auto call = this->valve_->make_call();
+      call.set_position(*value / 100.0f);
+      call.perform();
+    });
+  }
+}
+
+void MQTTValveComponent::dump_config() {
+  ESP_LOGCONFIG(TAG, "MQTT valve '%s':", this->valve_->get_name().c_str());
+  auto traits = this->valve_->get_traits();
+  bool has_command_topic = traits.get_supports_position();
+  LOG_MQTT_COMPONENT(true, has_command_topic)
+  if (traits.get_supports_position()) {
+    ESP_LOGCONFIG(TAG, "  Position State Topic: '%s'", this->get_position_state_topic().c_str());
+    ESP_LOGCONFIG(TAG, "  Position Command Topic: '%s'", this->get_position_command_topic().c_str());
+  }
+}
+void MQTTValveComponent::send_discovery(JsonObject root, mqtt::SendDiscoveryConfig &config) {
+  if (!this->valve_->get_device_class().empty())
+    root[MQTT_DEVICE_CLASS] = this->valve_->get_device_class();
+
+  auto traits = this->valve_->get_traits();
+  if (traits.get_is_assumed_state()) {
+    root[MQTT_OPTIMISTIC] = true;
+  }
+  if (traits.get_supports_position()) {
+    root[MQTT_POSITION_TOPIC] = this->get_position_state_topic();
+    root[MQTT_SET_POSITION_TOPIC] = this->get_position_command_topic();
+  }
+}
+
+std::string MQTTValveComponent::component_type() const { return "valve"; }
+const EntityBase *MQTTValveComponent::get_entity() const { return this->valve_; }
+
+bool MQTTValveComponent::send_initial_state() { return this->publish_state(); }
+bool MQTTValveComponent::publish_state() {
+  auto traits = this->valve_->get_traits();
+  bool success = true;
+  if (traits.get_supports_position()) {
+    std::string pos = value_accuracy_to_string(roundf(this->valve_->position * 100), 0);
+    if (!this->publish(this->get_position_state_topic(), pos))
+      success = false;
+  }
+  const char *state_s = this->valve_->current_operation == VALVE_OPERATION_OPENING   ? "opening"
+                        : this->valve_->current_operation == VALVE_OPERATION_CLOSING ? "closing"
+                        : this->valve_->position == VALVE_CLOSED                     ? "closed"
+                        : this->valve_->position == VALVE_OPEN                       ? "open"
+                        : traits.get_supports_position()                             ? "open"
+                                                                                     : "unknown";
+  if (!this->publish(this->get_state_topic_(), state_s))
+    success = false;
+  return success;
+}
+
+}  // namespace mqtt
+}  // namespace esphome
+
+#endif
+#endif  // USE_MQTT
diff --git a/esphome/components/mqtt/mqtt_valve.h b/esphome/components/mqtt/mqtt_valve.h
new file mode 100644
index 0000000000..63a0462193
--- /dev/null
+++ b/esphome/components/mqtt/mqtt_valve.h
@@ -0,0 +1,41 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#include "mqtt_component.h"
+
+#ifdef USE_MQTT
+#ifdef USE_VALVE
+
+#include "esphome/components/valve/valve.h"
+
+namespace esphome {
+namespace mqtt {
+
+class MQTTValveComponent : public mqtt::MQTTComponent {
+ public:
+  explicit MQTTValveComponent(valve::Valve *valve);
+
+  void setup() override;
+  void send_discovery(JsonObject root, mqtt::SendDiscoveryConfig &config) override;
+
+  MQTT_COMPONENT_CUSTOM_TOPIC(position, command)
+  MQTT_COMPONENT_CUSTOM_TOPIC(position, state)
+
+  bool send_initial_state() override;
+
+  bool publish_state();
+
+  void dump_config() override;
+
+ protected:
+  std::string component_type() const override;
+  const EntityBase *get_entity() const override;
+
+  valve::Valve *valve_;
+};
+
+}  // namespace mqtt
+}  // namespace esphome
+
+#endif
+#endif  // USE_MQTT
diff --git a/esphome/components/template/valve/__init__.py b/esphome/components/template/valve/__init__.py
new file mode 100644
index 0000000000..89d776dfdd
--- /dev/null
+++ b/esphome/components/template/valve/__init__.py
@@ -0,0 +1,118 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome import automation
+from esphome.components import valve
+from esphome.const import (
+    CONF_ASSUMED_STATE,
+    CONF_CLOSE_ACTION,
+    CONF_CURRENT_OPERATION,
+    CONF_ID,
+    CONF_LAMBDA,
+    CONF_OPEN_ACTION,
+    CONF_OPTIMISTIC,
+    CONF_POSITION,
+    CONF_POSITION_ACTION,
+    CONF_RESTORE_MODE,
+    CONF_STATE,
+    CONF_STOP_ACTION,
+)
+from .. import template_ns
+
+TemplateValve = template_ns.class_("TemplateValve", valve.Valve, cg.Component)
+
+TemplateValveRestoreMode = template_ns.enum("TemplateValveRestoreMode")
+RESTORE_MODES = {
+    "NO_RESTORE": TemplateValveRestoreMode.VALVE_NO_RESTORE,
+    "RESTORE": TemplateValveRestoreMode.VALVE_RESTORE,
+    "RESTORE_AND_CALL": TemplateValveRestoreMode.VALVE_RESTORE_AND_CALL,
+}
+
+CONF_HAS_POSITION = "has_position"
+CONF_TOGGLE_ACTION = "toggle_action"
+
+CONFIG_SCHEMA = valve.VALVE_SCHEMA.extend(
+    {
+        cv.GenerateID(): cv.declare_id(TemplateValve),
+        cv.Optional(CONF_LAMBDA): cv.returning_lambda,
+        cv.Optional(CONF_OPTIMISTIC, default=False): cv.boolean,
+        cv.Optional(CONF_ASSUMED_STATE, default=False): cv.boolean,
+        cv.Optional(CONF_HAS_POSITION, default=False): cv.boolean,
+        cv.Optional(CONF_OPEN_ACTION): automation.validate_automation(single=True),
+        cv.Optional(CONF_CLOSE_ACTION): automation.validate_automation(single=True),
+        cv.Optional(CONF_STOP_ACTION): automation.validate_automation(single=True),
+        cv.Optional(CONF_TOGGLE_ACTION): automation.validate_automation(single=True),
+        cv.Optional(CONF_POSITION_ACTION): automation.validate_automation(single=True),
+        cv.Optional(CONF_RESTORE_MODE, default="NO_RESTORE"): cv.enum(
+            RESTORE_MODES, upper=True
+        ),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+async def to_code(config):
+    var = await valve.new_valve(config)
+    await cg.register_component(var, config)
+    if lambda_config := config.get(CONF_LAMBDA):
+        template_ = await cg.process_lambda(
+            lambda_config, [], return_type=cg.optional.template(float)
+        )
+        cg.add(var.set_state_lambda(template_))
+    if open_action_config := config.get(CONF_OPEN_ACTION):
+        await automation.build_automation(
+            var.get_open_trigger(), [], open_action_config
+        )
+    if close_action_config := config.get(CONF_CLOSE_ACTION):
+        await automation.build_automation(
+            var.get_close_trigger(), [], close_action_config
+        )
+    if stop_action_config := config.get(CONF_STOP_ACTION):
+        await automation.build_automation(
+            var.get_stop_trigger(), [], stop_action_config
+        )
+        cg.add(var.set_has_stop(True))
+    if toggle_action_config := config.get(CONF_TOGGLE_ACTION):
+        await automation.build_automation(
+            var.get_toggle_trigger(), [], toggle_action_config
+        )
+        cg.add(var.set_has_toggle(True))
+    if position_action_config := config.get(CONF_POSITION_ACTION):
+        await automation.build_automation(
+            var.get_position_trigger(), [(float, "pos")], position_action_config
+        )
+        cg.add(var.set_has_position(True))
+    else:
+        cg.add(var.set_has_position(config[CONF_HAS_POSITION]))
+    cg.add(var.set_optimistic(config[CONF_OPTIMISTIC]))
+    cg.add(var.set_assumed_state(config[CONF_ASSUMED_STATE]))
+    cg.add(var.set_restore_mode(config[CONF_RESTORE_MODE]))
+
+
+@automation.register_action(
+    "valve.template.publish",
+    valve.ValvePublishAction,
+    cv.Schema(
+        {
+            cv.Required(CONF_ID): cv.use_id(valve.Valve),
+            cv.Exclusive(CONF_STATE, "pos"): cv.templatable(valve.validate_valve_state),
+            cv.Exclusive(CONF_POSITION, "pos"): cv.templatable(cv.percentage),
+            cv.Optional(CONF_CURRENT_OPERATION): cv.templatable(
+                valve.validate_valve_operation
+            ),
+        }
+    ),
+)
+async def valve_template_publish_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    var = cg.new_Pvariable(action_id, template_arg, paren)
+    if state_config := config.get(CONF_STATE):
+        template_ = await cg.templatable(state_config, args, float)
+        cg.add(var.set_position(template_))
+    if (position_config := config.get(CONF_POSITION)) is not None:
+        template_ = await cg.templatable(position_config, args, float)
+        cg.add(var.set_position(template_))
+    if current_operation_config := config.get(CONF_CURRENT_OPERATION):
+        template_ = await cg.templatable(
+            current_operation_config, args, valve.ValveOperation
+        )
+        cg.add(var.set_current_operation(template_))
+    return var
diff --git a/esphome/components/template/valve/template_valve.cpp b/esphome/components/template/valve/template_valve.cpp
new file mode 100644
index 0000000000..f943e19da9
--- /dev/null
+++ b/esphome/components/template/valve/template_valve.cpp
@@ -0,0 +1,131 @@
+#include "template_valve.h"
+#include "esphome/core/log.h"
+
+namespace esphome {
+namespace template_ {
+
+using namespace esphome::valve;
+
+static const char *const TAG = "template.valve";
+
+TemplateValve::TemplateValve()
+    : open_trigger_(new Trigger<>()),
+      close_trigger_(new Trigger<>),
+      stop_trigger_(new Trigger<>()),
+      toggle_trigger_(new Trigger<>()),
+      position_trigger_(new Trigger<float>()) {}
+
+void TemplateValve::setup() {
+  ESP_LOGCONFIG(TAG, "Setting up template valve '%s'...", this->name_.c_str());
+  switch (this->restore_mode_) {
+    case VALVE_NO_RESTORE:
+      break;
+    case VALVE_RESTORE: {
+      auto restore = this->restore_state_();
+      if (restore.has_value())
+        restore->apply(this);
+      break;
+    }
+    case VALVE_RESTORE_AND_CALL: {
+      auto restore = this->restore_state_();
+      if (restore.has_value()) {
+        restore->to_call(this).perform();
+      }
+      break;
+    }
+  }
+}
+
+void TemplateValve::loop() {
+  bool changed = false;
+
+  if (this->state_f_.has_value()) {
+    auto s = (*this->state_f_)();
+    if (s.has_value()) {
+      auto pos = clamp(*s, 0.0f, 1.0f);
+      if (pos != this->position) {
+        this->position = pos;
+        changed = true;
+      }
+    }
+  }
+
+  if (changed)
+    this->publish_state();
+}
+
+void TemplateValve::set_optimistic(bool optimistic) { this->optimistic_ = optimistic; }
+void TemplateValve::set_assumed_state(bool assumed_state) { this->assumed_state_ = assumed_state; }
+void TemplateValve::set_state_lambda(std::function<optional<float>()> &&f) { this->state_f_ = f; }
+float TemplateValve::get_setup_priority() const { return setup_priority::HARDWARE; }
+
+Trigger<> *TemplateValve::get_open_trigger() const { return this->open_trigger_; }
+Trigger<> *TemplateValve::get_close_trigger() const { return this->close_trigger_; }
+Trigger<> *TemplateValve::get_stop_trigger() const { return this->stop_trigger_; }
+Trigger<> *TemplateValve::get_toggle_trigger() const { return this->toggle_trigger_; }
+
+void TemplateValve::dump_config() {
+  LOG_VALVE("", "Template Valve", this);
+  ESP_LOGCONFIG(TAG, "  Has position: %s", YESNO(this->has_position_));
+  ESP_LOGCONFIG(TAG, "  Optimistic: %s", YESNO(this->optimistic_));
+}
+
+void TemplateValve::control(const ValveCall &call) {
+  if (call.get_stop()) {
+    this->stop_prev_trigger_();
+    this->stop_trigger_->trigger();
+    this->prev_command_trigger_ = this->stop_trigger_;
+    this->publish_state();
+  }
+  if (call.get_toggle().has_value()) {
+    this->stop_prev_trigger_();
+    this->toggle_trigger_->trigger();
+    this->prev_command_trigger_ = this->toggle_trigger_;
+    this->publish_state();
+  }
+  if (call.get_position().has_value()) {
+    auto pos = *call.get_position();
+    this->stop_prev_trigger_();
+
+    if (pos == VALVE_OPEN) {
+      this->open_trigger_->trigger();
+      this->prev_command_trigger_ = this->open_trigger_;
+    } else if (pos == VALVE_CLOSED) {
+      this->close_trigger_->trigger();
+      this->prev_command_trigger_ = this->close_trigger_;
+    } else {
+      this->position_trigger_->trigger(pos);
+    }
+
+    if (this->optimistic_) {
+      this->position = pos;
+    }
+  }
+
+  this->publish_state();
+}
+
+ValveTraits TemplateValve::get_traits() {
+  auto traits = ValveTraits();
+  traits.set_is_assumed_state(this->assumed_state_);
+  traits.set_supports_stop(this->has_stop_);
+  traits.set_supports_toggle(this->has_toggle_);
+  traits.set_supports_position(this->has_position_);
+  return traits;
+}
+
+Trigger<float> *TemplateValve::get_position_trigger() const { return this->position_trigger_; }
+
+void TemplateValve::set_has_stop(bool has_stop) { this->has_stop_ = has_stop; }
+void TemplateValve::set_has_toggle(bool has_toggle) { this->has_toggle_ = has_toggle; }
+void TemplateValve::set_has_position(bool has_position) { this->has_position_ = has_position; }
+
+void TemplateValve::stop_prev_trigger_() {
+  if (this->prev_command_trigger_ != nullptr) {
+    this->prev_command_trigger_->stop_action();
+    this->prev_command_trigger_ = nullptr;
+  }
+}
+
+}  // namespace template_
+}  // namespace esphome
diff --git a/esphome/components/template/valve/template_valve.h b/esphome/components/template/valve/template_valve.h
new file mode 100644
index 0000000000..5e3fb6aff3
--- /dev/null
+++ b/esphome/components/template/valve/template_valve.h
@@ -0,0 +1,60 @@
+#pragma once
+
+#include "esphome/core/component.h"
+#include "esphome/core/automation.h"
+#include "esphome/components/valve/valve.h"
+
+namespace esphome {
+namespace template_ {
+
+enum TemplateValveRestoreMode {
+  VALVE_NO_RESTORE,
+  VALVE_RESTORE,
+  VALVE_RESTORE_AND_CALL,
+};
+
+class TemplateValve : public valve::Valve, public Component {
+ public:
+  TemplateValve();
+
+  void set_state_lambda(std::function<optional<float>()> &&f);
+  Trigger<> *get_open_trigger() const;
+  Trigger<> *get_close_trigger() const;
+  Trigger<> *get_stop_trigger() const;
+  Trigger<> *get_toggle_trigger() const;
+  Trigger<float> *get_position_trigger() const;
+  void set_optimistic(bool optimistic);
+  void set_assumed_state(bool assumed_state);
+  void set_has_stop(bool has_stop);
+  void set_has_position(bool has_position);
+  void set_has_toggle(bool has_toggle);
+  void set_restore_mode(TemplateValveRestoreMode restore_mode) { restore_mode_ = restore_mode; }
+
+  void setup() override;
+  void loop() override;
+  void dump_config() override;
+
+  float get_setup_priority() const override;
+
+ protected:
+  void control(const valve::ValveCall &call) override;
+  valve::ValveTraits get_traits() override;
+  void stop_prev_trigger_();
+
+  TemplateValveRestoreMode restore_mode_{VALVE_NO_RESTORE};
+  optional<std::function<optional<float>()>> state_f_;
+  bool assumed_state_{false};
+  bool optimistic_{false};
+  Trigger<> *open_trigger_;
+  Trigger<> *close_trigger_;
+  bool has_stop_{false};
+  bool has_toggle_{false};
+  Trigger<> *stop_trigger_;
+  Trigger<> *toggle_trigger_;
+  Trigger<> *prev_command_trigger_{nullptr};
+  Trigger<float> *position_trigger_;
+  bool has_position_{false};
+};
+
+}  // namespace template_
+}  // namespace esphome
diff --git a/esphome/components/valve/__init__.py b/esphome/components/valve/__init__.py
new file mode 100644
index 0000000000..22d617cc36
--- /dev/null
+++ b/esphome/components/valve/__init__.py
@@ -0,0 +1,186 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome import automation
+from esphome.automation import maybe_simple_id, Condition
+from esphome.components import mqtt
+from esphome.const import (
+    CONF_DEVICE_CLASS,
+    CONF_ID,
+    CONF_MQTT_ID,
+    CONF_ON_OPEN,
+    CONF_POSITION,
+    CONF_POSITION_COMMAND_TOPIC,
+    CONF_POSITION_STATE_TOPIC,
+    CONF_STATE,
+    CONF_STOP,
+    CONF_TRIGGER_ID,
+)
+from esphome.core import CORE, coroutine_with_priority
+from esphome.cpp_helpers import setup_entity
+
+IS_PLATFORM_COMPONENT = True
+
+CODEOWNERS = ["@esphome/core"]
+
+valve_ns = cg.esphome_ns.namespace("valve")
+
+Valve = valve_ns.class_("Valve", cg.EntityBase)
+
+VALVE_OPEN = valve_ns.VALVE_OPEN
+VALVE_CLOSED = valve_ns.VALVE_CLOSED
+
+VALVE_STATES = {
+    "OPEN": VALVE_OPEN,
+    "CLOSED": VALVE_CLOSED,
+}
+validate_valve_state = cv.enum(VALVE_STATES, upper=True)
+
+ValveOperation = valve_ns.enum("ValveOperation")
+VALVE_OPERATIONS = {
+    "IDLE": ValveOperation.VALVE_OPERATION_IDLE,
+    "OPENING": ValveOperation.VALVE_OPERATION_OPENING,
+    "CLOSING": ValveOperation.VALVE_OPERATION_CLOSING,
+}
+validate_valve_operation = cv.enum(VALVE_OPERATIONS, upper=True)
+
+# Actions
+OpenAction = valve_ns.class_("OpenAction", automation.Action)
+CloseAction = valve_ns.class_("CloseAction", automation.Action)
+StopAction = valve_ns.class_("StopAction", automation.Action)
+ToggleAction = valve_ns.class_("ToggleAction", automation.Action)
+ControlAction = valve_ns.class_("ControlAction", automation.Action)
+ValvePublishAction = valve_ns.class_("ValvePublishAction", automation.Action)
+ValveIsOpenCondition = valve_ns.class_("ValveIsOpenCondition", Condition)
+ValveIsClosedCondition = valve_ns.class_("ValveIsClosedCondition", Condition)
+
+# Triggers
+ValveOpenTrigger = valve_ns.class_("ValveOpenTrigger", automation.Trigger.template())
+ValveClosedTrigger = valve_ns.class_(
+    "ValveClosedTrigger", automation.Trigger.template()
+)
+
+CONF_ON_CLOSED = "on_closed"
+
+VALVE_SCHEMA = cv.ENTITY_BASE_SCHEMA.extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA).extend(
+    {
+        cv.GenerateID(): cv.declare_id(Valve),
+        cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTValveComponent),
+        cv.Optional(CONF_POSITION_COMMAND_TOPIC): cv.All(
+            cv.requires_component("mqtt"), cv.subscribe_topic
+        ),
+        cv.Optional(CONF_POSITION_STATE_TOPIC): cv.All(
+            cv.requires_component("mqtt"), cv.subscribe_topic
+        ),
+        cv.Optional(CONF_ON_OPEN): automation.validate_automation(
+            {
+                cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(ValveOpenTrigger),
+            }
+        ),
+        cv.Optional(CONF_ON_CLOSED): automation.validate_automation(
+            {
+                cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(ValveClosedTrigger),
+            }
+        ),
+    }
+)
+
+
+async def setup_valve_core_(var, config):
+    await setup_entity(var, config)
+
+    if device_class_config := config.get(CONF_DEVICE_CLASS):
+        cg.add(var.set_device_class(device_class_config))
+
+    for conf in config.get(CONF_ON_OPEN, []):
+        trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
+        await automation.build_automation(trigger, [], conf)
+    for conf in config.get(CONF_ON_CLOSED, []):
+        trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
+        await automation.build_automation(trigger, [], conf)
+
+    if mqtt_id_config := config.get(CONF_MQTT_ID):
+        mqtt_ = cg.new_Pvariable(mqtt_id_config, var)
+        await mqtt.register_mqtt_component(mqtt_, config)
+
+        if position_state_topic_config := config.get(CONF_POSITION_STATE_TOPIC):
+            cg.add(mqtt_.set_custom_position_state_topic(position_state_topic_config))
+        if position_command_topic_config := config.get(CONF_POSITION_COMMAND_TOPIC):
+            cg.add(
+                mqtt_.set_custom_position_command_topic(position_command_topic_config)
+            )
+
+
+async def register_valve(var, config):
+    if not CORE.has_id(config[CONF_ID]):
+        var = cg.Pvariable(config[CONF_ID], var)
+    cg.add(cg.App.register_valve(var))
+    await setup_valve_core_(var, config)
+
+
+async def new_valve(config, *args):
+    var = cg.new_Pvariable(config[CONF_ID], *args)
+    await register_valve(var, config)
+    return var
+
+
+VALVE_ACTION_SCHEMA = maybe_simple_id(
+    {
+        cv.Required(CONF_ID): cv.use_id(Valve),
+    }
+)
+
+
+@automation.register_action("valve.open", OpenAction, VALVE_ACTION_SCHEMA)
+async def valve_open_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    return cg.new_Pvariable(action_id, template_arg, paren)
+
+
+@automation.register_action("valve.close", CloseAction, VALVE_ACTION_SCHEMA)
+async def valve_close_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    return cg.new_Pvariable(action_id, template_arg, paren)
+
+
+@automation.register_action("valve.stop", StopAction, VALVE_ACTION_SCHEMA)
+async def valve_stop_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    return cg.new_Pvariable(action_id, template_arg, paren)
+
+
+@automation.register_action("valve.toggle", ToggleAction, VALVE_ACTION_SCHEMA)
+def valve_toggle_to_code(config, action_id, template_arg, args):
+    paren = yield cg.get_variable(config[CONF_ID])
+    yield cg.new_Pvariable(action_id, template_arg, paren)
+
+
+VALVE_CONTROL_ACTION_SCHEMA = cv.Schema(
+    {
+        cv.Required(CONF_ID): cv.use_id(Valve),
+        cv.Optional(CONF_STOP): cv.templatable(cv.boolean),
+        cv.Exclusive(CONF_STATE, "pos"): cv.templatable(validate_valve_state),
+        cv.Exclusive(CONF_POSITION, "pos"): cv.templatable(cv.percentage),
+    }
+)
+
+
+@automation.register_action("valve.control", ControlAction, VALVE_CONTROL_ACTION_SCHEMA)
+async def valve_control_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    var = cg.new_Pvariable(action_id, template_arg, paren)
+    if stop_config := config.get(CONF_STOP):
+        template_ = await cg.templatable(stop_config, args, bool)
+        cg.add(var.set_stop(template_))
+    if state_config := config.get(CONF_STATE):
+        template_ = await cg.templatable(state_config, args, float)
+        cg.add(var.set_position(template_))
+    if (position_config := config.get(CONF_POSITION)) is not None:
+        template_ = await cg.templatable(position_config, args, float)
+        cg.add(var.set_position(template_))
+    return var
+
+
+@coroutine_with_priority(100.0)
+async def to_code(config):
+    cg.add_define("USE_VALVE")
+    cg.add_global(valve_ns.using)
diff --git a/esphome/components/valve/automation.h b/esphome/components/valve/automation.h
new file mode 100644
index 0000000000..24c94a5570
--- /dev/null
+++ b/esphome/components/valve/automation.h
@@ -0,0 +1,129 @@
+#pragma once
+
+#include "esphome/core/component.h"
+#include "esphome/core/automation.h"
+#include "valve.h"
+
+namespace esphome {
+namespace valve {
+
+template<typename... Ts> class OpenAction : public Action<Ts...> {
+ public:
+  explicit OpenAction(Valve *valve) : valve_(valve) {}
+
+  void play(Ts... x) override { this->valve_->make_call().set_command_open().perform(); }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class CloseAction : public Action<Ts...> {
+ public:
+  explicit CloseAction(Valve *valve) : valve_(valve) {}
+
+  void play(Ts... x) override { this->valve_->make_call().set_command_close().perform(); }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class StopAction : public Action<Ts...> {
+ public:
+  explicit StopAction(Valve *valve) : valve_(valve) {}
+
+  void play(Ts... x) override { this->valve_->make_call().set_command_stop().perform(); }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class ToggleAction : public Action<Ts...> {
+ public:
+  explicit ToggleAction(Valve *valve) : valve_(valve) {}
+
+  void play(Ts... x) override { this->valve_->make_call().set_command_toggle().perform(); }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class ControlAction : public Action<Ts...> {
+ public:
+  explicit ControlAction(Valve *valve) : valve_(valve) {}
+
+  TEMPLATABLE_VALUE(bool, stop)
+  TEMPLATABLE_VALUE(float, position)
+
+  void play(Ts... x) override {
+    auto call = this->valve_->make_call();
+    if (this->stop_.has_value())
+      call.set_stop(this->stop_.value(x...));
+    if (this->position_.has_value())
+      call.set_position(this->position_.value(x...));
+    call.perform();
+  }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class ValvePublishAction : public Action<Ts...> {
+ public:
+  ValvePublishAction(Valve *valve) : valve_(valve) {}
+  TEMPLATABLE_VALUE(float, position)
+  TEMPLATABLE_VALUE(ValveOperation, current_operation)
+
+  void play(Ts... x) override {
+    if (this->position_.has_value())
+      this->valve_->position = this->position_.value(x...);
+    if (this->current_operation_.has_value())
+      this->valve_->current_operation = this->current_operation_.value(x...);
+    this->valve_->publish_state();
+  }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class ValveIsOpenCondition : public Condition<Ts...> {
+ public:
+  ValveIsOpenCondition(Valve *valve) : valve_(valve) {}
+  bool check(Ts... x) override { return this->valve_->is_fully_open(); }
+
+ protected:
+  Valve *valve_;
+};
+
+template<typename... Ts> class ValveIsClosedCondition : public Condition<Ts...> {
+ public:
+  ValveIsClosedCondition(Valve *valve) : valve_(valve) {}
+  bool check(Ts... x) override { return this->valve_->is_fully_closed(); }
+
+ protected:
+  Valve *valve_;
+};
+
+class ValveOpenTrigger : public Trigger<> {
+ public:
+  ValveOpenTrigger(Valve *a_valve) {
+    a_valve->add_on_state_callback([this, a_valve]() {
+      if (a_valve->is_fully_open()) {
+        this->trigger();
+      }
+    });
+  }
+};
+
+class ValveClosedTrigger : public Trigger<> {
+ public:
+  ValveClosedTrigger(Valve *a_valve) {
+    a_valve->add_on_state_callback([this, a_valve]() {
+      if (a_valve->is_fully_closed()) {
+        this->trigger();
+      }
+    });
+  }
+};
+
+}  // namespace valve
+}  // namespace esphome
diff --git a/esphome/components/valve/valve.cpp b/esphome/components/valve/valve.cpp
new file mode 100644
index 0000000000..d1ec17945a
--- /dev/null
+++ b/esphome/components/valve/valve.cpp
@@ -0,0 +1,179 @@
+#include "valve.h"
+#include "esphome/core/log.h"
+
+namespace esphome {
+namespace valve {
+
+static const char *const TAG = "valve";
+
+const float VALVE_OPEN = 1.0f;
+const float VALVE_CLOSED = 0.0f;
+
+const char *valve_command_to_str(float pos) {
+  if (pos == VALVE_OPEN) {
+    return "OPEN";
+  } else if (pos == VALVE_CLOSED) {
+    return "CLOSE";
+  } else {
+    return "UNKNOWN";
+  }
+}
+const char *valve_operation_to_str(ValveOperation op) {
+  switch (op) {
+    case VALVE_OPERATION_IDLE:
+      return "IDLE";
+    case VALVE_OPERATION_OPENING:
+      return "OPENING";
+    case VALVE_OPERATION_CLOSING:
+      return "CLOSING";
+    default:
+      return "UNKNOWN";
+  }
+}
+
+Valve::Valve() : position{VALVE_OPEN} {}
+
+ValveCall::ValveCall(Valve *parent) : parent_(parent) {}
+ValveCall &ValveCall::set_command(const char *command) {
+  if (strcasecmp(command, "OPEN") == 0) {
+    this->set_command_open();
+  } else if (strcasecmp(command, "CLOSE") == 0) {
+    this->set_command_close();
+  } else if (strcasecmp(command, "STOP") == 0) {
+    this->set_command_stop();
+  } else if (strcasecmp(command, "TOGGLE") == 0) {
+    this->set_command_toggle();
+  } else {
+    ESP_LOGW(TAG, "'%s' - Unrecognized command %s", this->parent_->get_name().c_str(), command);
+  }
+  return *this;
+}
+ValveCall &ValveCall::set_command_open() {
+  this->position_ = VALVE_OPEN;
+  return *this;
+}
+ValveCall &ValveCall::set_command_close() {
+  this->position_ = VALVE_CLOSED;
+  return *this;
+}
+ValveCall &ValveCall::set_command_stop() {
+  this->stop_ = true;
+  return *this;
+}
+ValveCall &ValveCall::set_command_toggle() {
+  this->toggle_ = true;
+  return *this;
+}
+ValveCall &ValveCall::set_position(float position) {
+  this->position_ = position;
+  return *this;
+}
+void ValveCall::perform() {
+  ESP_LOGD(TAG, "'%s' - Setting", this->parent_->get_name().c_str());
+  auto traits = this->parent_->get_traits();
+  this->validate_();
+  if (this->stop_) {
+    ESP_LOGD(TAG, "  Command: STOP");
+  }
+  if (this->position_.has_value()) {
+    if (traits.get_supports_position()) {
+      ESP_LOGD(TAG, "  Position: %.0f%%", *this->position_ * 100.0f);
+    } else {
+      ESP_LOGD(TAG, "  Command: %s", valve_command_to_str(*this->position_));
+    }
+  }
+  if (this->toggle_.has_value()) {
+    ESP_LOGD(TAG, "  Command: TOGGLE");
+  }
+  this->parent_->control(*this);
+}
+const optional<float> &ValveCall::get_position() const { return this->position_; }
+const optional<bool> &ValveCall::get_toggle() const { return this->toggle_; }
+void ValveCall::validate_() {
+  auto traits = this->parent_->get_traits();
+  if (this->position_.has_value()) {
+    auto pos = *this->position_;
+    if (!traits.get_supports_position() && pos != VALVE_OPEN && pos != VALVE_CLOSED) {
+      ESP_LOGW(TAG, "'%s' - This valve device does not support setting position!", this->parent_->get_name().c_str());
+      this->position_.reset();
+    } else if (pos < 0.0f || pos > 1.0f) {
+      ESP_LOGW(TAG, "'%s' - Position %.2f is out of range [0.0 - 1.0]", this->parent_->get_name().c_str(), pos);
+      this->position_ = clamp(pos, 0.0f, 1.0f);
+    }
+  }
+  if (this->toggle_.has_value()) {
+    if (!traits.get_supports_toggle()) {
+      ESP_LOGW(TAG, "'%s' - This valve device does not support toggle!", this->parent_->get_name().c_str());
+      this->toggle_.reset();
+    }
+  }
+  if (this->stop_) {
+    if (this->position_.has_value()) {
+      ESP_LOGW(TAG, "Cannot set position when stopping a valve!");
+      this->position_.reset();
+    }
+    if (this->toggle_.has_value()) {
+      ESP_LOGW(TAG, "Cannot set toggle when stopping a valve!");
+      this->toggle_.reset();
+    }
+  }
+}
+ValveCall &ValveCall::set_stop(bool stop) {
+  this->stop_ = stop;
+  return *this;
+}
+bool ValveCall::get_stop() const { return this->stop_; }
+
+ValveCall Valve::make_call() { return {this}; }
+
+void Valve::add_on_state_callback(std::function<void()> &&f) { this->state_callback_.add(std::move(f)); }
+void Valve::publish_state(bool save) {
+  this->position = clamp(this->position, 0.0f, 1.0f);
+
+  ESP_LOGD(TAG, "'%s' - Publishing:", this->name_.c_str());
+  auto traits = this->get_traits();
+  if (traits.get_supports_position()) {
+    ESP_LOGD(TAG, "  Position: %.0f%%", this->position * 100.0f);
+  } else {
+    if (this->position == VALVE_OPEN) {
+      ESP_LOGD(TAG, "  State: OPEN");
+    } else if (this->position == VALVE_CLOSED) {
+      ESP_LOGD(TAG, "  State: CLOSED");
+    } else {
+      ESP_LOGD(TAG, "  State: UNKNOWN");
+    }
+  }
+  ESP_LOGD(TAG, "  Current Operation: %s", valve_operation_to_str(this->current_operation));
+
+  this->state_callback_.call();
+
+  if (save) {
+    ValveRestoreState restore{};
+    memset(&restore, 0, sizeof(restore));
+    restore.position = this->position;
+    this->rtc_.save(&restore);
+  }
+}
+optional<ValveRestoreState> Valve::restore_state_() {
+  this->rtc_ = global_preferences->make_preference<ValveRestoreState>(this->get_object_id_hash());
+  ValveRestoreState recovered{};
+  if (!this->rtc_.load(&recovered))
+    return {};
+  return recovered;
+}
+
+bool Valve::is_fully_open() const { return this->position == VALVE_OPEN; }
+bool Valve::is_fully_closed() const { return this->position == VALVE_CLOSED; }
+
+ValveCall ValveRestoreState::to_call(Valve *valve) {
+  auto call = valve->make_call();
+  call.set_position(this->position);
+  return call;
+}
+void ValveRestoreState::apply(Valve *valve) {
+  valve->position = this->position;
+  valve->publish_state();
+}
+
+}  // namespace valve
+}  // namespace esphome
diff --git a/esphome/components/valve/valve.h b/esphome/components/valve/valve.h
new file mode 100644
index 0000000000..0e14a8d8f0
--- /dev/null
+++ b/esphome/components/valve/valve.h
@@ -0,0 +1,152 @@
+#pragma once
+
+#include "esphome/core/component.h"
+#include "esphome/core/entity_base.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/preferences.h"
+#include "valve_traits.h"
+
+namespace esphome {
+namespace valve {
+
+const extern float VALVE_OPEN;
+const extern float VALVE_CLOSED;
+
+#define LOG_VALVE(prefix, type, obj) \
+  if ((obj) != nullptr) { \
+    ESP_LOGCONFIG(TAG, "%s%s '%s'", prefix, LOG_STR_LITERAL(type), (obj)->get_name().c_str()); \
+    auto traits_ = (obj)->get_traits(); \
+    if (traits_.get_is_assumed_state()) { \
+      ESP_LOGCONFIG(TAG, "%s  Assumed State: YES", prefix); \
+    } \
+    if (!(obj)->get_device_class().empty()) { \
+      ESP_LOGCONFIG(TAG, "%s  Device Class: '%s'", prefix, (obj)->get_device_class().c_str()); \
+    } \
+  }
+
+class Valve;
+
+class ValveCall {
+ public:
+  ValveCall(Valve *parent);
+
+  /// Set the command as a string, "STOP", "OPEN", "CLOSE", "TOGGLE".
+  ValveCall &set_command(const char *command);
+  /// Set the command to open the valve.
+  ValveCall &set_command_open();
+  /// Set the command to close the valve.
+  ValveCall &set_command_close();
+  /// Set the command to stop the valve.
+  ValveCall &set_command_stop();
+  /// Set the command to toggle the valve.
+  ValveCall &set_command_toggle();
+  /// Set the call to a certain target position.
+  ValveCall &set_position(float position);
+  /// Set whether this valve call should stop the valve.
+  ValveCall &set_stop(bool stop);
+
+  /// Perform the valve call.
+  void perform();
+
+  const optional<float> &get_position() const;
+  bool get_stop() const;
+  const optional<bool> &get_toggle() const;
+
+ protected:
+  void validate_();
+
+  Valve *parent_;
+  bool stop_{false};
+  optional<float> position_{};
+  optional<bool> toggle_{};
+};
+
+/// Struct used to store the restored state of a valve
+struct ValveRestoreState {
+  float position;
+
+  /// Convert this struct to a valve call that can be performed.
+  ValveCall to_call(Valve *valve);
+  /// Apply these settings to the valve
+  void apply(Valve *valve);
+} __attribute__((packed));
+
+/// Enum encoding the current operation of a valve.
+enum ValveOperation : uint8_t {
+  /// The valve is currently idle (not moving)
+  VALVE_OPERATION_IDLE = 0,
+  /// The valve is currently opening.
+  VALVE_OPERATION_OPENING,
+  /// The valve is currently closing.
+  VALVE_OPERATION_CLOSING,
+};
+
+const char *valve_operation_to_str(ValveOperation op);
+
+/** Base class for all valve devices.
+ *
+ * Valves currently have three properties:
+ *  - position - The current position of the valve from 0.0 (fully closed) to 1.0 (fully open).
+ *    For valves with only binary OPEN/CLOSED position this will always be either 0.0 or 1.0
+ *  - current_operation - The operation the valve is currently performing, this can
+ *    be one of IDLE, OPENING and CLOSING.
+ *
+ * For users: All valve operations must be performed over the .make_call() interface.
+ * To command a valve, use .make_call() to create a call object, set all properties
+ * you wish to set, and activate the command with .perform().
+ * For reading out the current values of the valve, use the public .position, etc.
+ * properties (these are read-only for users)
+ *
+ * For integrations: Integrations must implement two methods: control() and get_traits().
+ * Control will be called with the arguments supplied by the user and should be used
+ * to control all values of the valve. Also implement get_traits() to return what operations
+ * the valve supports.
+ */
+class Valve : public EntityBase, public EntityBase_DeviceClass {
+ public:
+  explicit Valve();
+
+  /// The current operation of the valve (idle, opening, closing).
+  ValveOperation current_operation{VALVE_OPERATION_IDLE};
+  /** The position of the valve from 0.0 (fully closed) to 1.0 (fully open).
+   *
+   * For binary valves this is always equals to 0.0 or 1.0 (see also VALVE_OPEN and
+   * VALVE_CLOSED constants).
+   */
+  float position;
+
+  /// Construct a new valve call used to control the valve.
+  ValveCall make_call();
+
+  void add_on_state_callback(std::function<void()> &&f);
+
+  /** Publish the current state of the valve.
+   *
+   * First set the .position, etc. values and then call this method
+   * to publish the state of the valve.
+   *
+   * @param save Whether to save the updated values in RTC area.
+   */
+  void publish_state(bool save = true);
+
+  virtual ValveTraits get_traits() = 0;
+
+  /// Helper method to check if the valve is fully open. Equivalent to comparing .position against 1.0
+  bool is_fully_open() const;
+  /// Helper method to check if the valve is fully closed. Equivalent to comparing .position against 0.0
+  bool is_fully_closed() const;
+
+ protected:
+  friend ValveCall;
+
+  virtual void control(const ValveCall &call) = 0;
+
+  optional<ValveRestoreState> restore_state_();
+
+  CallbackManager<void()> state_callback_{};
+
+  ESPPreferenceObject rtc_;
+};
+
+}  // namespace valve
+}  // namespace esphome
diff --git a/esphome/components/valve/valve_traits.h b/esphome/components/valve/valve_traits.h
new file mode 100644
index 0000000000..7e9aab2f26
--- /dev/null
+++ b/esphome/components/valve/valve_traits.h
@@ -0,0 +1,27 @@
+#pragma once
+
+namespace esphome {
+namespace valve {
+
+class ValveTraits {
+ public:
+  ValveTraits() = default;
+
+  bool get_is_assumed_state() const { return this->is_assumed_state_; }
+  void set_is_assumed_state(bool is_assumed_state) { this->is_assumed_state_ = is_assumed_state; }
+  bool get_supports_position() const { return this->supports_position_; }
+  void set_supports_position(bool supports_position) { this->supports_position_ = supports_position; }
+  bool get_supports_toggle() const { return this->supports_toggle_; }
+  void set_supports_toggle(bool supports_toggle) { this->supports_toggle_ = supports_toggle; }
+  bool get_supports_stop() const { return this->supports_stop_; }
+  void set_supports_stop(bool supports_stop) { this->supports_stop_ = supports_stop; }
+
+ protected:
+  bool is_assumed_state_{false};
+  bool supports_position_{false};
+  bool supports_toggle_{false};
+  bool supports_stop_{false};
+};
+
+}  // namespace valve
+}  // namespace esphome
diff --git a/esphome/components/web_server/list_entities.cpp b/esphome/components/web_server/list_entities.cpp
index dd9fd4afe4..13e396dc82 100644
--- a/esphome/components/web_server/list_entities.cpp
+++ b/esphome/components/web_server/list_entities.cpp
@@ -86,6 +86,15 @@ bool ListEntitiesIterator::on_lock(lock::Lock *a_lock) {
 }
 #endif
 
+#ifdef USE_VALVE
+bool ListEntitiesIterator::on_valve(valve::Valve *valve) {
+  if (this->web_server_->events_.count() == 0)
+    return true;
+  this->web_server_->events_.send(this->web_server_->valve_json(valve, DETAIL_ALL).c_str(), "state");
+  return true;
+}
+#endif
+
 #ifdef USE_CLIMATE
 bool ListEntitiesIterator::on_climate(climate::Climate *climate) {
   if (this->web_server_->events_.count() == 0)
diff --git a/esphome/components/web_server/list_entities.h b/esphome/components/web_server/list_entities.h
index fc48186b32..d0b8dda233 100644
--- a/esphome/components/web_server/list_entities.h
+++ b/esphome/components/web_server/list_entities.h
@@ -56,6 +56,9 @@ class ListEntitiesIterator : public ComponentIterator {
 #ifdef USE_LOCK
   bool on_lock(lock::Lock *a_lock) override;
 #endif
+#ifdef USE_VALVE
+  bool on_valve(valve::Valve *valve) override;
+#endif
 #ifdef USE_ALARM_CONTROL_PANEL
   bool on_alarm_control_panel(alarm_control_panel::AlarmControlPanel *a_alarm_control_panel) override;
 #endif
diff --git a/esphome/components/web_server/web_server.cpp b/esphome/components/web_server/web_server.cpp
index 2e6206b691..412f8816ee 100644
--- a/esphome/components/web_server/web_server.cpp
+++ b/esphome/components/web_server/web_server.cpp
@@ -1260,6 +1260,68 @@ void WebServer::handle_lock_request(AsyncWebServerRequest *request, const UrlMat
 }
 #endif
 
+#ifdef USE_VALVE
+void WebServer::on_valve_update(valve::Valve *obj) {
+  if (this->events_.count() == 0)
+    return;
+  this->events_.send(this->valve_json(obj, DETAIL_STATE).c_str(), "state");
+}
+void WebServer::handle_valve_request(AsyncWebServerRequest *request, const UrlMatch &match) {
+  for (valve::Valve *obj : App.get_valves()) {
+    if (obj->get_object_id() != match.id)
+      continue;
+
+    if (request->method() == HTTP_GET && match.method.empty()) {
+      std::string data = this->valve_json(obj, DETAIL_STATE);
+      request->send(200, "application/json", data.c_str());
+      continue;
+    }
+
+    auto call = obj->make_call();
+    if (match.method == "open") {
+      call.set_command_open();
+    } else if (match.method == "close") {
+      call.set_command_close();
+    } else if (match.method == "stop") {
+      call.set_command_stop();
+    } else if (match.method == "toggle") {
+      call.set_command_toggle();
+    } else if (match.method != "set") {
+      request->send(404);
+      return;
+    }
+
+    auto traits = obj->get_traits();
+    if (request->hasParam("position") && !traits.get_supports_position()) {
+      request->send(409);
+      return;
+    }
+
+    if (request->hasParam("position")) {
+      auto position = parse_number<float>(request->getParam("position")->value().c_str());
+      if (position.has_value()) {
+        call.set_position(*position);
+      }
+    }
+
+    this->schedule_([call]() mutable { call.perform(); });
+    request->send(200);
+    return;
+  }
+  request->send(404);
+}
+std::string WebServer::valve_json(valve::Valve *obj, JsonDetail start_config) {
+  return json::build_json([obj, start_config](JsonObject root) {
+    set_json_icon_state_value(root, obj, "valve-" + obj->get_object_id(), obj->is_fully_closed() ? "CLOSED" : "OPEN",
+                              obj->position, start_config);
+    root["current_operation"] = valve::valve_operation_to_str(obj->current_operation);
+
+    if (obj->get_traits().get_supports_position())
+      root["position"] = obj->position;
+  });
+}
+#endif
+
 #ifdef USE_ALARM_CONTROL_PANEL
 void WebServer::on_alarm_control_panel_update(alarm_control_panel::AlarmControlPanel *obj) {
   if (this->events_.count() == 0)
@@ -1394,6 +1456,11 @@ bool WebServer::canHandle(AsyncWebServerRequest *request) {
     return true;
 #endif
 
+#ifdef USE_VALVE
+  if ((request->method() == HTTP_POST || request->method() == HTTP_GET) && match.domain == "valve")
+    return true;
+#endif
+
 #ifdef USE_ALARM_CONTROL_PANEL
   if (request->method() == HTTP_GET && match.domain == "alarm_control_panel")
     return true;
@@ -1535,6 +1602,13 @@ void WebServer::handleRequest(AsyncWebServerRequest *request) {
   }
 #endif
 
+#ifdef USE_VALVE
+  if (match.domain == "valve") {
+    this->handle_valve_request(request, match);
+    return;
+  }
+#endif
+
 #ifdef USE_ALARM_CONTROL_PANEL
   if (match.domain == "alarm_control_panel") {
     this->handle_alarm_control_panel_request(request, match);
diff --git a/esphome/components/web_server/web_server.h b/esphome/components/web_server/web_server.h
index 1935f8d076..9f807efc43 100644
--- a/esphome/components/web_server/web_server.h
+++ b/esphome/components/web_server/web_server.h
@@ -276,6 +276,16 @@ class WebServer : public Controller, public Component, public AsyncWebHandler {
   std::string lock_json(lock::Lock *obj, lock::LockState value, JsonDetail start_config);
 #endif
 
+#ifdef USE_VALVE
+  void on_valve_update(valve::Valve *obj) override;
+
+  /// Handle a valve request under '/valve/<id>/<open/close/stop/set>'.
+  void handle_valve_request(AsyncWebServerRequest *request, const UrlMatch &match);
+
+  /// Dump the valve state as a JSON string.
+  std::string valve_json(valve::Valve *obj, JsonDetail start_config);
+#endif
+
 #ifdef USE_ALARM_CONTROL_PANEL
   void on_alarm_control_panel_update(alarm_control_panel::AlarmControlPanel *obj) override;
 
diff --git a/esphome/core/application.h b/esphome/core/application.h
index 73330d27e3..ee931282a6 100644
--- a/esphome/core/application.h
+++ b/esphome/core/application.h
@@ -54,6 +54,9 @@
 #ifdef USE_LOCK
 #include "esphome/components/lock/lock.h"
 #endif
+#ifdef USE_VALVE
+#include "esphome/components/valve/valve.h"
+#endif
 #ifdef USE_MEDIA_PLAYER
 #include "esphome/components/media_player/media_player.h"
 #endif
@@ -147,6 +150,10 @@ class Application {
   void register_lock(lock::Lock *a_lock) { this->locks_.push_back(a_lock); }
 #endif
 
+#ifdef USE_VALVE
+  void register_valve(valve::Valve *valve) { this->valves_.push_back(valve); }
+#endif
+
 #ifdef USE_MEDIA_PLAYER
   void register_media_player(media_player::MediaPlayer *media_player) { this->media_players_.push_back(media_player); }
 #endif
@@ -348,6 +355,15 @@ class Application {
     return nullptr;
   }
 #endif
+#ifdef USE_VALVE
+  const std::vector<valve::Valve *> &get_valves() { return this->valves_; }
+  valve::Valve *get_valve_by_key(uint32_t key, bool include_internal = false) {
+    for (auto *obj : this->valves_)
+      if (obj->get_object_id_hash() == key && (include_internal || !obj->is_internal()))
+        return obj;
+    return nullptr;
+  }
+#endif
 #ifdef USE_MEDIA_PLAYER
   const std::vector<media_player::MediaPlayer *> &get_media_players() { return this->media_players_; }
   media_player::MediaPlayer *get_media_player_by_key(uint32_t key, bool include_internal = false) {
@@ -429,6 +445,9 @@ class Application {
 #ifdef USE_LOCK
   std::vector<lock::Lock *> locks_{};
 #endif
+#ifdef USE_VALVE
+  std::vector<valve::Valve *> valves_{};
+#endif
 #ifdef USE_MEDIA_PLAYER
   std::vector<media_player::MediaPlayer *> media_players_{};
 #endif
diff --git a/esphome/core/component_iterator.cpp b/esphome/core/component_iterator.cpp
index 228cf64d54..b00154c685 100644
--- a/esphome/core/component_iterator.cpp
+++ b/esphome/core/component_iterator.cpp
@@ -277,6 +277,21 @@ void ComponentIterator::advance() {
       }
       break;
 #endif
+#ifdef USE_VALVE
+    case IteratorState::VALVE:
+      if (this->at_ >= App.get_valves().size()) {
+        advance_platform = true;
+      } else {
+        auto *valve = App.get_valves()[this->at_];
+        if (valve->is_internal() && !this->include_internal_) {
+          success = true;
+          break;
+        } else {
+          success = this->on_valve(valve);
+        }
+      }
+      break;
+#endif
 #ifdef USE_MEDIA_PLAYER
     case IteratorState::MEDIA_PLAYER:
       if (this->at_ >= App.get_media_players().size()) {
diff --git a/esphome/core/component_iterator.h b/esphome/core/component_iterator.h
index d7f19f2850..3fbbc0bc19 100644
--- a/esphome/core/component_iterator.h
+++ b/esphome/core/component_iterator.h
@@ -72,6 +72,9 @@ class ComponentIterator {
 #ifdef USE_LOCK
   virtual bool on_lock(lock::Lock *a_lock) = 0;
 #endif
+#ifdef USE_VALVE
+  virtual bool on_valve(valve::Valve *valve) = 0;
+#endif
 #ifdef USE_MEDIA_PLAYER
   virtual bool on_media_player(media_player::MediaPlayer *media_player);
 #endif
@@ -135,6 +138,9 @@ class ComponentIterator {
 #ifdef USE_LOCK
     LOCK,
 #endif
+#ifdef USE_VALVE
+    VALVE,
+#endif
 #ifdef USE_MEDIA_PLAYER
     MEDIA_PLAYER,
 #endif
diff --git a/esphome/core/controller.cpp b/esphome/core/controller.cpp
index db5818d455..eab818bdb4 100644
--- a/esphome/core/controller.cpp
+++ b/esphome/core/controller.cpp
@@ -91,6 +91,12 @@ void Controller::setup_controller(bool include_internal) {
       obj->add_on_state_callback([this, obj]() { this->on_lock_update(obj); });
   }
 #endif
+#ifdef USE_VALVE
+  for (auto *obj : App.get_valves()) {
+    if (include_internal || !obj->is_internal())
+      obj->add_on_state_callback([this, obj]() { this->on_valve_update(obj); });
+  }
+#endif
 #ifdef USE_MEDIA_PLAYER
   for (auto *obj : App.get_media_players()) {
     if (include_internal || !obj->is_internal())
diff --git a/esphome/core/controller.h b/esphome/core/controller.h
index 9b1cfd93c6..94a4acb7c7 100644
--- a/esphome/core/controller.h
+++ b/esphome/core/controller.h
@@ -46,6 +46,9 @@
 #ifdef USE_LOCK
 #include "esphome/components/lock/lock.h"
 #endif
+#ifdef USE_VALVE
+#include "esphome/components/valve/valve.h"
+#endif
 #ifdef USE_MEDIA_PLAYER
 #include "esphome/components/media_player/media_player.h"
 #endif
@@ -100,6 +103,9 @@ class Controller {
 #ifdef USE_LOCK
   virtual void on_lock_update(lock::Lock *obj){};
 #endif
+#ifdef USE_VALVE
+  virtual void on_valve_update(valve::Valve *obj){};
+#endif
 #ifdef USE_MEDIA_PLAYER
   virtual void on_media_player_update(media_player::MediaPlayer *obj){};
 #endif
diff --git a/esphome/core/defines.h b/esphome/core/defines.h
index f13ae968f0..2064ca1356 100644
--- a/esphome/core/defines.h
+++ b/esphome/core/defines.h
@@ -54,6 +54,7 @@
 #define USE_TIME
 #define USE_TOUCHSCREEN
 #define USE_UART_DEBUGGER
+#define USE_VALVE
 #define USE_WIFI
 #define USE_WIFI_AP
 #define USE_GRAPHICAL_DISPLAY_MENU
diff --git a/script/ci-custom.py b/script/ci-custom.py
index c3cfc7a331..c591bfe5c3 100755
--- a/script/ci-custom.py
+++ b/script/ci-custom.py
@@ -638,6 +638,7 @@ def lint_trailing_whitespace(fname, match):
         "esphome/components/stepper/stepper.h",
         "esphome/components/switch/switch.h",
         "esphome/components/text_sensor/text_sensor.h",
+        "esphome/components/valve/valve.h",
         "esphome/core/component.h",
         "esphome/core/gpio.h",
         "esphome/core/log.h",
diff --git a/tests/components/template/test.all.yaml b/tests/components/template/test.all.yaml
index 29dc83b649..64faec36c2 100644
--- a/tests/components/template/test.all.yaml
+++ b/tests/components/template/test.all.yaml
@@ -125,6 +125,23 @@ lock:
     open_action:
       - logger.log: open_action
 
+valve:
+  - platform: template
+    name: "Template Valve"
+    lambda: |-
+      if (id(some_binary_sensor).state) {
+        return VALVE_OPEN;
+      } else {
+        return VALVE_CLOSED;
+      }
+    open_action:
+      - logger.log: open_action
+    close_action:
+      - logger.log: close_action
+    stop_action:
+      - logger.log: stop_action
+    optimistic: true
+
 text:
   - platform: template
     name: "Template text"