refactor: move call script back to core_utils

app/core_utils.py

@@ -1,9 +1,12 @@
 import array
+import asyncio
 import json
 import logging
 import os
 import random
+import re
 import time
+import warnings
 from typing import Dict, Optional
 
 from fastapi.encoders import jsonable_encoder
@@ -181,3 +184,76 @@ def config_get_hex_str(value: str, name: str = "") -> str:
             return m
 
     return value
+
+
+async def call_script(scriptPath) -> str:
+    warnings.warn("call_script is deprecated. Use call_script2 instead.")
+
+    cmd = f"bash {scriptPath}"
+    proc = await asyncio.create_subprocess_shell(
+        cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if stdout:
+        return stdout.decode()
+    if stderr:
+        logging.error(stderr.decode())
+    return ""
+
+
+async def call_script2(script_path) -> ProcessResult:
+    """
+    Call a local bash script and return the results
+
+    :param str script_path: full path with arguments
+    :return: The process result
+    :rtype: ProcessResult
+    """
+
+    cmd = f"bash {script_path}"
+    proc = await asyncio.create_subprocess_shell(
+        cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+
+    return ProcessResult(
+        proc.returncode,
+        stdout.decode() if stdout else "",
+        stderr.decode() if stderr else "",
+    )
+
+
+async def call_sudo_script(scriptPath) -> str:
+    cmd = f"sudo bash {scriptPath}"
+    proc = await asyncio.create_subprocess_shell(
+        cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if stdout:
+        return stdout.decode()
+    if stderr:
+        logging.error(stderr.decode())
+    return ""
+
+
+def parse_key_value_lines(lines: list) -> dict:
+    Dict = {}
+    for line in lines:
+        line = line.strip()
+        if len(line) == 0:
+            continue
+        if not re.match("^[a-zA-Z0-9]*=", line):
+            continue
+        key, value = line.strip().split("=", 1)
+        Dict[key] = value.strip('"').strip("'")
+    return Dict
+
+
+def parse_key_value_text(text: str) -> dict:
+    return parse_key_value_lines(text.splitlines())

app/repositories/apps_impl/raspiblitz.py

@@ -9,10 +9,14 @@ from decouple import config
 from fastapi import HTTPException, status
 from fastapi.encoders import jsonable_encoder
 
-from app.constants import available_app_ids
-from app.core_utils import SSE, broadcast_sse_msg
+from app.core_utils import (
+    SSE,
+    broadcast_sse_msg,
+    call_sudo_script,
+    parse_key_value_text,
+)
 from app.repositories.apps_impl.apps_base import AppsBase
-from app.repositories.utils.raspiblitz import call_sudo_script, parse_key_value_text
+from app.repositories.utils.raspiblitz import available_app_ids
 
 SHELL_SCRIPT_PATH = config("shell_script_path")
 

app/repositories/system.py

@@ -17,6 +17,7 @@ from app.models.system import (
 
 PLATFORM = config("platform", default=APIPlatform.RASPIBLITZ)
 if PLATFORM == APIPlatform.RASPIBLITZ:
+    from app.core_utils import call_script, call_sudo_script, parse_key_value_text
     from app.repositories.hardware_impl.raspiblitz import (
         HW_INFO_YIELD_TIME,
         get_hardware_info_impl,
@@ -27,7 +28,6 @@ if PLATFORM == APIPlatform.RASPIBLITZ:
         match_password,
         shutdown_impl,
     )
-    from app.repositories.utils.raspiblitz import call_script, parse_key_value_text
 elif PLATFORM == APIPlatform.NATIVE_PYTHON:
     from app.repositories.hardware_impl.native_python import (
         HW_INFO_YIELD_TIME,

app/repositories/utils/raspiblitz.py

@@ -1,12 +1,5 @@
-import asyncio
-import logging
-import re
-import warnings
 from decouple import config
 
-from app.core_utils import ProcessResult
-
-
 SHELL_SCRIPT_PATH = config("shell_script_path")
 
 available_app_ids = {
@@ -19,76 +12,3 @@ available_app_ids = {
     "mempool",
     "thunderhub",
 }
-
-
-async def call_script(scriptPath) -> str:
-    warnings.warn("call_script is deprecated. Use call_script2 instead.")
-
-    cmd = f"bash {scriptPath}"
-    proc = await asyncio.create_subprocess_shell(
-        cmd,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    stdout, stderr = await proc.communicate()
-    if stdout:
-        return stdout.decode()
-    if stderr:
-        logging.error(stderr.decode())
-    return ""
-
-
-async def call_script2(script_path) -> ProcessResult:
-    """
-    Call a local bash script and return the results
-
-    :param str script_path: full path with arguments
-    :return: The process result
-    :rtype: ProcessResult
-    """
-
-    cmd = f"bash {script_path}"
-    proc = await asyncio.create_subprocess_shell(
-        cmd,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    stdout, stderr = await proc.communicate()
-
-    return ProcessResult(
-        proc.returncode,
-        stdout.decode() if stdout else "",
-        stderr.decode() if stderr else "",
-    )
-
-
-async def call_sudo_script(scriptPath) -> str:
-    cmd = f"sudo bash {scriptPath}"
-    proc = await asyncio.create_subprocess_shell(
-        cmd,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    stdout, stderr = await proc.communicate()
-    if stdout:
-        return stdout.decode()
-    if stderr:
-        logging.error(stderr.decode())
-    return ""
-
-
-def parse_key_value_lines(lines: list) -> dict:
-    Dict = {}
-    for line in lines:
-        line = line.strip()
-        if len(line) == 0:
-            continue
-        if not re.match("^[a-zA-Z0-9]*=", line):
-            continue
-        key, value = line.strip().split("=", 1)
-        Dict[key] = value.strip('"').strip("'")
-    return Dict
-
-
-def parse_key_value_text(text: str) -> dict:
-    return parse_key_value_lines(text.splitlines())

app/routers/setup.py

@@ -6,9 +6,8 @@ from pydantic import BaseModel
 
 from app.auth.auth_bearer import JWTBearer
 from app.auth.auth_handler import sign_jwt
-from app.core_utils import redis_get
+from app.core_utils import call_script, parse_key_value_lines, redis_get
 from app.repositories.system import name_valid, password_valid, shutdown
-from app.repositories.utils.raspiblitz import call_script, parse_key_value_lines
 
 router = APIRouter(prefix="/setup", tags=["Setup"])