sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-02-12 16:08:19 +00:00
Code Issues Releases Wiki Activity

Merge branch 'remove_codecs' into integration

Browse Source
This commit is contained in:
J. Nick Koston 2023-11-12 18:40:42 -06:00
commit ae31d75cbd
No known key found for this signature in database
1 changed files with 7 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
esphome/dashboard/dashboard.py
View File

@ -3,7 +3,6 @@ from __future__ import annotations
import asyncio import asyncio
import base64 import base64
import binascii import binascii
import codecs
import collections import collections
import datetime import datetime
import functools import functools
@ -345,8 +344,8 @@ class EsphomeCommandWebSocket(tornado.websocket.WebSocketHandler):
async def handle_stdin(self, json_message: dict[str, Any]) -> None: async def handle_stdin(self, json_message: dict[str, Any]) -> None:
if not self.is_process_active: if not self.is_process_active:
return return
data = json_message["data"] text: str = json_message["data"]
data = codecs.encode(data, "utf8", "replace") data = text.encode("utf-8", "replace")
_LOGGER.debug("< stdin: %s", data) _LOGGER.debug("< stdin: %s", data)
self._proc.stdin.write(data) self._proc.stdin.write(data)
@ -357,18 +356,18 @@ class EsphomeCommandWebSocket(tornado.websocket.WebSocketHandler):
while True: while True:
try: try:
if self._use_popen: if self._use_popen:
data = yield self._queue.get() data: bytes = yield self._queue.get()
if data is None: if data is None:
self._proc_on_exit(self._proc.poll()) self._proc_on_exit(self._proc.poll())
break break
else: else:
data = yield self._proc.stdout.read_until_regex(reg) data: bytes = yield self._proc.stdout.read_until_regex(reg)
except tornado.iostream.StreamClosedError: except tornado.iostream.StreamClosedError:
break break
data = codecs.decode(data, "utf8", "replace")
_LOGGER.debug("> stdout: %s", data) text = data.decode("utf-8", "replace")
self.write_message({"event": "line", "data": data}) _LOGGER.debug("> stdout: %s", text)
self.write_message({"event": "line", "data": text})
def _stdout_thread(self) -> None: def _stdout_thread(self) -> None:
if not self._use_popen: if not self._use_popen: