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

dashboard: remove usage of codecs module (#5741)

Browse Source
This commit is contained in:
J. Nick Koston 2023-11-13 18:44:49 -06:00 committed by Jesse Hills
parent 3e2b83acb0
commit bd568eecf5
No known key found for this signature in database
GPG Key ID: BEAAE804EFD8E83A
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

@ -2,7 +2,6 @@ from __future__ import annotations
import base64 import base64
import binascii import binascii
import codecs
import collections import collections
import datetime import datetime
import functools import functools
@ -339,8 +338,8 @@ class EsphomeCommandWebSocket(tornado.websocket.WebSocketHandler):
def handle_stdin(self, json_message): def handle_stdin(self, json_message):
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)
@ -351,18 +350,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): def _stdout_thread(self):
if not self._use_popen: if not self._use_popen: