[api] Emit every encode call through one generator helper

_encode_call() owns the cursor assignment and the _force suffix, so
the convention lives in one place instead of at every emission site;
the fixed32 fast path is an arm of the generic encode_content keyed by
a per type value template. write_fixed32_le uses convert_little_endian
instead of its own byte order switch. The integration test shares a
StateWaiter from state_utils and leaves the disconnect to the fixture.

tests/integration/test_api_encode_boundaries.py

@@ -1,12 +1,11 @@
 """Encode paths at their branch boundaries: zero skipped float, fixed32 state, negative int32,
-length prefixes of two varint bytes, two byte field tags and a clean disconnect."""
+length prefixes of two varint bytes and two byte field tags."""
 
 from __future__ import annotations
 
 import asyncio
 
 from aioesphomeapi import (
-    EntityState,
     NumberState,
     SelectInfo,
     SensorInfo,
@@ -15,7 +14,7 @@ from aioesphomeapi import (
 )
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import InitialStateHelper, StateWaiter, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 LONG_OPTION = (
@@ -30,12 +29,12 @@ async def test_api_encode_boundaries(
     run_compiled: RunCompiledFunction,
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        device_info = await client.device_info()
+        device_info, (entities, _) = await asyncio.gather(
+            client.device_info(), client.list_entities_services()
+        )
         assert device_info.suggested_area == "Kitchen"
 
-        entities, _ = await client.list_entities_services()
         sensor = require_entity(entities, "zero_then_value", SensorInfo)
         assert sensor.accuracy_decimals == -2
         select = require_entity(entities, "long_option_select", SelectInfo)
@@ -45,25 +44,9 @@ async def test_api_encode_boundaries(
         number = require_entity(entities, "negative_number")
         button = require_entity(entities, "publish_values")
 
-        sensor_value: asyncio.Future[float] = loop.create_future()
-        text_value: asyncio.Future[str] = loop.create_future()
         initial = InitialStateHelper(entities)
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, SensorState)
-                and state.key == sensor.key
-                and not sensor_value.done()
-            ):
-                sensor_value.set_result(state.state)
-            elif (
-                isinstance(state, TextSensorState)
-                and state.key == text.key
-                and not text_value.done()
-            ):
-                text_value.set_result(state.state)
-
-        client.subscribe_states(initial.on_state_wrapper(on_state))
+        waiter = StateWaiter()
+        client.subscribe_states(initial.on_state_wrapper(waiter.on_state))
         await initial.wait_for_initial_states()
 
         # A float of exactly zero is skipped on the wire and must still read as 0.0, not missing
@@ -75,8 +58,19 @@ async def test_api_encode_boundaries(
         assert first_number.state == -123.5
 
         client.button_command(button.key)
-        assert await asyncio.wait_for(sensor_value, 5.0) == 12.5
-        assert await asyncio.wait_for(text_value, 5.0) == "y" * 200
-
-        # DisconnectRequest and DisconnectResponse carry no fields
-        await client.disconnect()
+        await asyncio.gather(
+            waiter.expect(
+                lambda s: (
+                    isinstance(s, SensorState)
+                    and s.key == sensor.key
+                    and s.state == 12.5
+                )
+            ),
+            waiter.expect(
+                lambda s: (
+                    isinstance(s, TextSensorState)
+                    and s.key == text.key
+                    and s.state == "y" * 200
+                )
+            ),
+        )