J. Nick Koston
214b419db2
* Use mdns or freshen cache when device connection is OTA Since we already have a service browser running, we likely already know the IP of the deivce we want to connect to so we can replace OTA with the address to avoid the esphome app having to look it up again * isort * Fix zeroconf name resolution refactoring error HostResolver should get the type as the first arg instead of the name * no i/o * tornado support native coros * lint * use new tornado start methods * use new tornado start methods * use new tornado start methods * break * lint * lint * typing, missing awaits * io in executor * missed one * fix: missing if * stale comment * rename run_command to build_device_command since it does not actually run anything
57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
|
|
|
|
class ThreadedAsyncEvent:
|
|
"""This is a shim to allow the asyncio event to be used in a threaded context.
|
|
|
|
When more of the code is moved to asyncio, this can be removed.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the ThreadedAsyncEvent."""
|
|
self.event = threading.Event()
|
|
self.async_event: asyncio.Event | None = None
|
|
self.loop: asyncio.AbstractEventLoop | None = None
|
|
|
|
def async_setup(
|
|
self, loop: asyncio.AbstractEventLoop, async_event: asyncio.Event
|
|
) -> None:
|
|
"""Set the asyncio.Event instance."""
|
|
self.loop = loop
|
|
self.async_event = async_event
|
|
|
|
def async_set(self) -> None:
|
|
"""Set the asyncio.Event instance."""
|
|
self.async_event.set()
|
|
self.event.set()
|
|
|
|
def set(self) -> None:
|
|
"""Set the event."""
|
|
self.loop.call_soon_threadsafe(self.async_event.set)
|
|
self.event.set()
|
|
|
|
def wait(self) -> None:
|
|
"""Wait for the event."""
|
|
self.event.wait()
|
|
|
|
async def async_wait(self) -> None:
|
|
"""Wait the event async."""
|
|
await self.async_event.wait()
|
|
|
|
def clear(self) -> None:
|
|
"""Clear the event."""
|
|
self.loop.call_soon_threadsafe(self.async_event.clear)
|
|
self.event.clear()
|
|
|
|
def async_clear(self) -> None:
|
|
"""Clear the event async."""
|
|
self.async_event.clear()
|
|
self.event.clear()
|
|
|
|
def is_set(self) -> bool:
|
|
"""Return if the event is set."""
|
|
return self.event.is_set()
|