update os.popen to subprocess.Popen

autosubmit_api/components/jobs/utils.py

@@ -5,6 +5,7 @@
 from autosubmit_api.common.utils import Status
 from typing import List, Dict, Tuple
 from bscearth.utils.date import parse_date
+import subprocess
 
 wrapped_title_format = " <span class='badge' style='background-color:#94b8b8'>Wrapped {0} </span>"
 source_tag = " <span class='badge' style='background-color:#80d4ff'>SOURCE</span>"
@@ -144,8 +145,9 @@ def get_job_total_stats(status_code: int, name: str, tmp_path: str) -> Tuple[dat
     current_status = status_from_job
     path = os.path.join(tmp_path, name + '_TOTAL_STATS')
     if os.path.exists(path):
-        request = 'tail -1 ' + path
-        last_line = os.popen(request).readline()
+        process = subprocess.Popen(['tail', '-1', path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        stdout, _ = process.communicate()
+        last_line = stdout.decode()
         # print(last_line)
 
         values = last_line.split()

autosubmit_api/experiment/common_requests.py

@@ -39,6 +39,7 @@
 from autosubmit_api.components.jobs import utils as JUtils
 
 from autosubmit_api.autosubmit_legacy.job.job_list import JobList
+from autosubmit_api.experiment.utils import get_files_from_dir_with_pattern, read_tail
 from autosubmit_api.logger import logger
 
 from autosubmit_api.performance.utils import calculate_SYPD_perjob
@@ -489,29 +490,33 @@ def get_experiment_log_last_lines(expid):
     error = False
     error_message = ""
     logcontent = []
-    reading = ""
 
     try:
         APIBasicConfig.read()
         exp_paths = ExperimentPaths(expid)
-        path = exp_paths.tmp_as_logs_dir
-        reading = os.popen('ls -t ' + path + ' | grep "run.log"').read() if (os.path.exists(path)) else ""
 
-        # Finding log files
-        if len(reading) == 0:
-            path = APIBasicConfig.LOCAL_ROOT_DIR + '/' + expid + '/' + APIBasicConfig.LOCAL_TMP_DIR
-            reading = os.popen('ls -t ' + path + ' | grep "run.log"').read() if (os.path.exists(path)) else ""
+        dir_files = []
+        path = None
 
-        if len(reading) > 0:
-            log_file_name = reading.split()[0]
-            current_stat = os.stat(path + '/' + log_file_name)
+        # Try to read from the tmp_as_logs folder
+        if os.path.exists(exp_paths.tmp_as_logs_dir):
+            dir_files = get_files_from_dir_with_pattern(exp_paths.tmp_as_logs_dir, 'run.log')
+            path = exp_paths.tmp_as_logs_dir
+
+        # Try to read from the tmp folder
+        if len(dir_files) == 0 and os.path.exists(exp_paths.tmp_dir):
+            dir_files = get_files_from_dir_with_pattern(exp_paths.tmp_dir, 'run.log')
+            path = exp_paths.tmp_dir
+
+        if len(dir_files) > 0:
+            log_file_name = dir_files[0]
+            log_file_path = str(os.path.join(path, log_file_name))
+            current_stat = os.stat(log_file_path)
             timest = int(current_stat.st_mtime)
             log_file_lastmodified = common_utils.timestamp_to_datetime_format(timest)
             found = True
-            request = 'tail -150 ' + path + '/' + log_file_name
-            last_lines = os.popen(request)
-            for i, item in enumerate(last_lines.readlines()):
-                logcontent.append({'index': i, 'content': item[0:-1]})
+
+            logcontent = read_tail(log_file_path, 150)
     except Exception as e:
         error = True
         error_message = str(e)
@@ -550,12 +555,8 @@ def get_job_log(expid, logfile, nlines=150):
             timest = int(current_stat.st_mtime)
             log_file_lastmodified = common_utils.timestamp_to_datetime_format(timest)
             found = True
-            request = "tail -{0} {1}".format(nlines, logfilepath)
-            last50 = os.popen(request)
-            i = 0
-            for item in last50.readlines():
-                logcontent.append({'index': i, 'content': item[0:-1]})
-                i += 1
+
+            logcontent = read_tail(logfilepath, nlines)
     except Exception as e:
         error = True
         error_message = str(e)

autosubmit_api/experiment/utils.py

@@ -1,4 +1,6 @@
 
+import subprocess
+from typing import List
 import xml.etree.ElementTree as ET
 import traceback
 
@@ -20,4 +22,58 @@ def get_cas_user_from_xml(xmlstring):
     except Exception as exp:
       print(exp)
       print((traceback.format_exc))
-    return user
+    return user
+
+def read_tail(file_path: str, num_lines: int =150) -> List[dict]:
+    """
+    Reads the last num_lines of a file and returns them as a dictionary.
+   
+    :param file_path: path to the file
+    :param num_lines: number of lines to read
+    """
+    tail_content = []
+
+    process = subprocess.Popen(
+        ["tail",  "-{0}".format(num_lines), file_path],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+    stdout, stderr = process.communicate()
+
+    if process.returncode != 0:
+        raise ValueError(stderr.decode())
+
+    lines = stdout.decode().splitlines()
+    for i, item in enumerate(lines):
+        tail_content.append({"index": i, "content": item})
+
+    return tail_content
+
+def get_files_from_dir_with_pattern(dir_path: str, pattern: str) -> List[str]:
+  """
+  Returns a list of files ordered by creation date in a directory that match a pattern.
+  """
+  # Submits the commands
+  ls_process = subprocess.Popen(
+    ['ls', '-t', dir_path],
+    stdout=subprocess.PIPE
+  )
+  grep_process = subprocess.Popen(
+    ['grep', pattern],
+    stdin=ls_process.stdout,
+    stdout=subprocess.PIPE
+  )
+
+  # Allow ls_process to receive a SIGPIPE if grep_process exits
+  ls_process.stdout.close()
+
+  # Read the output of grep
+  stdout, stderr = grep_process.communicate()
+
+  if grep_process.returncode != 0:
+    raise ValueError(stderr.decode())
+
+  stdout = stdout.decode()
+  files = stdout.split()
+  return files