Fix watchman doing blocking I/O in the event loop

Author: andreasbrett

custom_components/watchman/__init__.py

@@ -1,4 +1,5 @@
 """https://github.com/andreasbrett/thewatchman§"""
+
 from datetime import timedelta
 import logging
 import os
@@ -133,18 +134,17 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
     await add_event_handlers(hass)
     if hass.is_running:
         # integration reloaded or options changed via UI
-        parse_config(hass, reason="changes in watchman configuration")
+        await parse_config(hass, reason="changes in watchman configuration")
         await coordinator.async_config_entry_first_refresh()
     else:
         # first run, home assistant is loading
         # parse_config will be scheduled once HA is fully loaded
         _LOGGER.info("Watchman started [%s]", VERSION)
 
-
-#    resources = hass.data["lovelace"]["resources"]
-#    await resources.async_get_info()
-#    for itm in resources.async_items():
-#        _LOGGER.debug(itm)
+    # resources = hass.data["lovelace"]["resources"]
+    # await resources.async_get_info()
+    # for itm in resources.async_items():
+    #     _LOGGER.debug(itm)
 
     return True
 
@@ -210,7 +210,7 @@ async def async_handle_report(call):
             await async_notification(hass, "Watchman error", message, error=True)
 
         if call.data.get(CONF_PARSE_CONFIG, False):
-            parse_config(hass, reason="service call")
+            await parse_config(hass, reason="service call")
 
         if send_notification:
             chunk_size = call.data.get(CONF_CHUNK_SIZE, config.get(CONF_CHUNK_SIZE))
@@ -270,7 +270,7 @@ async def async_delayed_refresh_states(timedate):  # pylint: disable=unused-argu
         await coordinator.async_refresh()
 
     async def async_on_home_assistant_started(event):  # pylint: disable=unused-argument
-        parse_config(hass, reason="HA restart")
+        await parse_config(hass, reason="HA restart")
         startup_delay = get_config(hass, CONF_STARTUP_DELAY, 0)
         await async_schedule_refresh_states(hass, startup_delay)
 
@@ -283,12 +283,12 @@ async def async_on_configuration_changed(event):
                 "reload_core_config",
                 "reload",
             ]:
-                parse_config(hass, reason="configuration changes")
+                await parse_config(hass, reason="configuration changes")
                 coordinator = hass.data[DOMAIN]["coordinator"]
                 await coordinator.async_refresh()
 
         elif typ in [EVENT_AUTOMATION_RELOADED, EVENT_SCENE_RELOADED]:
-            parse_config(hass, reason="configuration changes")
+            await parse_config(hass, reason="configuration changes")
             coordinator = hass.data[DOMAIN]["coordinator"]
             await coordinator.async_refresh()
 
@@ -340,14 +340,14 @@ def state_or_missing(state_id):
     hass.data[DOMAIN]["cancel_handlers"] = hdlr
 
 
-def parse_config(hass: HomeAssistant, reason=None):
+async def parse_config(hass: HomeAssistant, reason=None):
     """parse home assistant configuration files"""
     assert hass.data.get(DOMAIN_DATA)
     start_time = time.time()
     included_folders = get_included_folders(hass)
     ignored_files = hass.data[DOMAIN_DATA].get(CONF_IGNORED_FILES, None)
 
-    entity_list, service_list, files_parsed, files_ignored = parse(
+    entity_list, service_list, files_parsed, files_ignored = await parse(
         hass, included_folders, ignored_files, hass.config.config_dir
     )
     hass.data[DOMAIN]["entity_list"] = entity_list
@@ -387,11 +387,16 @@ async def async_report_to_file(hass, path, test_mode):
     """save report to a file"""
     coordinator = hass.data[DOMAIN]["coordinator"]
     await coordinator.async_refresh()
-    report_chunks = report(hass, table_renderer, chunk_size=0, test_mode=test_mode)
-    # OSError exception is handled in async_handle_report
-    with open(path, "w", encoding="utf-8") as report_file:
-        for chunk in report_chunks:
-            report_file.write(chunk)
+    report_chunks = await report(
+        hass, table_renderer, chunk_size=0, test_mode=test_mode
+    )
+
+    def write(path):
+        with open(path, "w", encoding="utf-8") as report_file:
+            for chunk in report_chunks:
+                report_file.write(chunk)
+
+    await hass.async_add_executor_job(write, path)
 
 
 async def async_report_to_notification(hass, service_str, service_data, chunk_size):
@@ -422,7 +427,7 @@ async def async_report_to_notification(hass, service_str, service_data, chunk_si
 
     coordinator = hass.data[DOMAIN]["coordinator"]
     await coordinator.async_refresh()
-    report_chunks = report(hass, text_renderer, chunk_size)
+    report_chunks = await report(hass, text_renderer, chunk_size)
     for chunk in report_chunks:
         data["message"] = chunk
         # blocking=True ensures execution order

custom_components/watchman/utils.py

@@ -1,4 +1,6 @@
 """Miscellaneous support functions for watchman"""
+
+import aiofiles
 import glob
 import re
 import fnmatch
@@ -231,7 +233,7 @@ def check_entitites(hass):
     return entities_missing
 
 
-def parse(hass, folders, ignored_files, root=None):
+async def parse(hass, folders, ignored_files, root=None):
     """Parse a yaml or json file for entities/services"""
     files_parsed = 0
     entity_pattern = re.compile(
@@ -255,19 +257,22 @@ def parse(hass, folders, ignored_files, root=None):
             continue
 
         try:
-            for i, line in enumerate(open(yaml_file, encoding="utf-8")):
-                line = re.sub(comment_pattern, "", line)
-                for match in re.finditer(entity_pattern, line):
-                    typ, val = match.group(1), match.group(2)
-                    if (
-                        typ != "service:"
-                        and "*" not in val
-                        and not val.endswith(".yaml")
-                    ):
-                        add_entry(entity_list, val, short_path, i + 1)
-                for match in re.finditer(service_pattern, line):
-                    val = match.group(1)
-                    add_entry(service_list, val, short_path, i + 1)
+            lineno = 1
+            async with aiofiles.open(yaml_file, mode="r", encoding="utf-8") as f:
+                async for line in f:
+                    line = re.sub(comment_pattern, "", line)
+                    for match in re.finditer(entity_pattern, line):
+                        typ, val = match.group(1), match.group(2)
+                        if (
+                            typ != "service:"
+                            and "*" not in val
+                            and not val.endswith(".yaml")
+                        ):
+                            add_entry(entity_list, val, short_path, lineno)
+                    for match in re.finditer(service_pattern, line):
+                        val = match.group(1)
+                        add_entry(service_list, val, short_path, lineno)
+                    lineno += 1
             files_parsed += 1
             _LOGGER.debug("%s parsed", yaml_file)
         except OSError as exception:
@@ -312,7 +317,7 @@ def fill(data, width, extra=None):
     )
 
 
-def report(hass, render, chunk_size, test_mode=False):
+async def report(hass, render, chunk_size, test_mode=False):
     """generates watchman report either as a table or as a list"""
     if not DOMAIN in hass.data:
         raise HomeAssistantError("No data for report, refresh required.")
@@ -354,7 +359,11 @@ def report(hass, render, chunk_size, test_mode=False):
         rep += "your config are available!\n"
     else:
         rep += "\n-== No entities found in configuration files!\n"
-    timezone = pytz.timezone(hass.config.time_zone)
+
+    def get_timezone(hass):
+        return pytz.timezone(hass.config.time_zone)
+
+    timezone = await hass.async_add_executor_job(get_timezone, hass)
 
     if not test_mode:
         report_datetime = datetime.now(timezone).strftime("%d %b %Y %H:%M:%S")