Coverage for biobb_common / biobb_common / command_wrapper / cmd_wrapper.py: 67%

1# -*- coding: utf-8 -*- 

2"""Python wrapper for command line 

3""" 

4import os 

5import subprocess 

6from biobb_common.tools import file_utils as fu 

7from typing import Optional, Union 

8import logging 

9from pathlib import Path 

10 

11 

12class CmdWrapper: 

13 """Command line wrapper using subprocess library 

14 """ 

15 

16 def __init__(self, 

17 cmd: list[str], 

18 shell_path: Union[str, Path] = os.getenv('SHELL', '/bin/sh'), 

19 out_log: Optional[logging.Logger] = None, 

20 err_log: Optional[logging.Logger] = None, 

21 global_log: Optional[logging.Logger] = None, 

22 env: Optional[dict] = None, 

23 timeout: Optional[int] = None, 

24 disable_logs: Optional[bool] = None) -> None: 

25 

26 self.cmd = cmd 

27 self.shell_path = shell_path 

28 self.out_log = out_log 

29 self.err_log = err_log 

30 self.global_log = global_log 

31 self.env = env 

32 self.timeout = timeout 

33 self.disable_logs = disable_logs 

34 

35 def log_output(self, exit_code: str, command: str, out: Optional[bytes] = None, err: Optional[bytes] = None, timeout: Optional[str] = None, 

36 out_log: Optional[logging.Logger] = None, err_log: Optional[logging.Logger] = None, global_log: Optional[logging.Logger] = None) -> None: 

37 

38 timeout_str = '' 

39 if timeout: 

40 timeout_str = f"Timeout: {timeout} seconds expired, killing process\n" 

41 command_str = f"Command '{command[0:80]}...' finalized with exit code {exit_code}" 

42 if out_log: 

43 out_log.info(command_str) 

44 if timeout_str: 

45 out_log.info(timeout_str) 

46 if out: 

47 out_log.info(out.decode("utf-8")) 

48 elif not self.disable_logs: 

49 print(command_str) 

50 if timeout_str: 

51 print(timeout_str) 

52 print("") 

53 if err_log and err: 

54 err_log.info(err.decode("utf-8")) 

55 

56 if global_log: 

57 global_log.info(f"{fu.get_logs_prefix()}{command_str}") 

58 if timeout_str: 

59 global_log.info(f"{fu.get_logs_prefix()}{timeout_str}") 

60 

61 def launch(self) -> int: 

62 cmd = " ".join(self.cmd) 

63 if self.out_log: 

64 self.out_log.info(f'Launching command (it may take a while): {cmd}') 

65 elif not self.disable_logs: 

66 print(f"\ncmd_wrapper command print: {cmd}") 

67 

68 new_env = {**os.environ.copy(), **self.env} if self.env else os.environ.copy() 

69 process = subprocess.Popen(cmd, 

70 stdout=subprocess.PIPE, 

71 stderr=subprocess.PIPE, 

72 shell=True, 

73 executable=self.shell_path, 

74 env=new_env) 

75 try: 

76 out, err = process.communicate(timeout=self.timeout) 

77 except subprocess.TimeoutExpired: 

78 process.kill() 

79 out, err = process.communicate() 

80 process.returncode = 1 

81 self.log_output(exit_code=str(process.returncode), command=" ".join(self.cmd), out=out, err=err, timeout=str(self.timeout), out_log=self.out_log, err_log=self.err_log, global_log=self.global_log) 

82 return process.returncode 

83 

84 process.wait() 

85 self.log_output(exit_code=str(process.returncode), command=" ".join(self.cmd), out=out, err=err, out_log=self.out_log, err_log=self.err_log, global_log=self.global_log) 

86 return process.returncode