Shared system utils

DragonBuild · Jan 13, 2024 · c431d6c · c431d6c
1 parent 28cba47
commit c431d6c
Show file tree

Hide file tree

Showing 4 changed files with 49 additions and 69 deletions.
diff --git a/src/dragon/device.py b/src/dragon/device.py
@@ -12,50 +12,9 @@
 
 '''
 
-import os, sys, yaml, subprocess, socket
+import os, sys, yaml, socket
 from shared.util import dbstate, dbwarn, dberror
-
-
-def system(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
-    proc = subprocess.Popen("" + cmd,
-                            stdout=stdout,
-                            stderr=stderr,
-                            shell=True,
-                            universal_newlines=True)
-    std_out, std_err = proc.communicate()
-    # print(proc.returncode)
-    return proc.returncode  # , std_out, std_err
-
-
-def system_with_output(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
-    proc = subprocess.Popen("" + cmd,
-                            stdout=stdout,
-                            stderr=stderr,
-                            shell=True,
-                            universal_newlines=True)
-    std_out, std_err = proc.communicate()
-    return proc.returncode, std_out, std_err
-
-
-def system_pipe_output(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
-    process = subprocess.Popen(cmd,
-                          stdout=stdout,
-                          stderr=stderr,
-                          shell=True,
-                          universal_newlines=True)
-
-    while True:
-        realtime_output = process.stdout.readline()
-        realtime_err = process.stderr.readline()
-
-        if realtime_output == '' and realtime_err == '' and process.poll() is not None:
-            break
-
-        if realtime_output:
-            print(realtime_output.strip(), flush=True)
-        if realtime_err:
-            print(realtime_err.strip(), flush=True, file=sys.stderr)
-
+from shared.util import system, system_with_output, system_pipe_output
 
 class DeviceShell:
     @staticmethod

diff --git a/src/dragon/test.py b/src/dragon/test.py
@@ -1,7 +1,8 @@
 #!/usr/bin/env python3
 
-import shutil, subprocess, sys, os, timeit, yaml
+import shutil, sys, os, timeit, yaml
 from math import sin, cos, radians
+from shared.util import system
 
 TestDict = yaml.safe_load(open(os.environ['DRAGON_ROOT_DIR'] + '/internal/tests.yml'))
 projects = TestDict['ProjectTests']
@@ -16,18 +17,6 @@ def bench():
     return product
 
 
-
-def system(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
-    proc = subprocess.Popen("" + cmd,
-                            stdout=stdout,
-                            stderr=stderr,
-                            shell=True,
-                            universal_newlines=True)
-    std_out, std_err = proc.communicate()
-    # print(proc.returncode)
-    return proc.returncode  # , std_out, std_err
-
-
 def main():
     tests = {
 

diff --git a/src/dragongen/toolchain.py b/src/dragongen/toolchain.py
@@ -1,16 +1,5 @@
-import os, os.path
-import subprocess
-
-
-def system_with_output(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
-    proc = subprocess.Popen("" + cmd,
-                            stdout=stdout,
-                            stderr=stderr,
-                            shell=True,
-                            universal_newlines=True)
-    std_out, std_err = proc.communicate()
-    return proc.returncode, std_out, std_err
-
+import os
+from shared.util import system_with_output
 
 class Toolchain:
     def __init__(self):

diff --git a/src/shared/util.py b/src/shared/util.py
@@ -1,5 +1,7 @@
 from enum import IntEnum
 import sys
+import subprocess
+
 
 class OutputColors(IntEnum):
     Red = 31
@@ -28,3 +30,44 @@ def dbwarn(tool_name, msg):
 
 def dberror(tool_name, msg):
     dprintline(OutputColors.Red, tool_name, OutputColors.White, OutputWeight.Bold, False, msg)
+
+
+def system(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
+    proc = subprocess.Popen("" + cmd,
+                            stdout=stdout,
+                            stderr=stderr,
+                            shell=True,
+                            universal_newlines=True)
+    std_out, std_err = proc.communicate()
+    # print(proc.returncode)
+    return proc.returncode  # , std_out, std_err
+
+
+def system_with_output(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
+    proc = subprocess.Popen("" + cmd,
+                            stdout=stdout,
+                            stderr=stderr,
+                            shell=True,
+                            universal_newlines=True)
+    std_out, std_err = proc.communicate()
+    return proc.returncode, std_out, std_err
+
+
+def system_pipe_output(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
+    process = subprocess.Popen(cmd,
+                          stdout=stdout,
+                          stderr=stderr,
+                          shell=True,
+                          universal_newlines=True)
+
+    while True:
+        realtime_output = process.stdout.readline()
+        realtime_err = process.stderr.readline()
+
+        if realtime_output == '' and realtime_err == '' and process.poll() is not None:
+            break
+
+        if realtime_output:
+            print(realtime_output.strip(), flush=True)
+        if realtime_err:
+            print(realtime_err.strip(), flush=True, file=sys.stderr)