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

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

2"""Python wrapper for command line 

3""" 

4import os 

5import subprocess 

6from biobb_common.tools import file_utils as fu 

7import typing 

8from typing import Optional 

9import logging 

10from pathlib import Path 

11 

12 

13class CmdWrapper: 

14 """Command line wrapper using subprocess library 

15 """ 

16 

17 def __init__(self, cmd: typing.Iterable[str], shell_path: typing.Union[str, Path] = os.getenv('SHELL', '/bin/sh'), out_log: Optional[logging.Logger] = None, err_log: Optional[logging.Logger] = None, 

18 global_log: Optional[logging.Logger] = None, env: Optional[typing.Mapping] = None) -> None: 

19 

20 self.cmd = cmd 

21 self.shell_path = shell_path 

22 self.out_log = out_log 

23 self.err_log = err_log 

24 self.global_log = global_log 

25 self.env = env 

26 

27 def launch(self) -> int: 

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

29 if self.out_log is None: 

30 print('') 

31 print("cmd_wrapper commnand print: " + cmd) 

32 else: 

33 self.out_log.info(cmd + '\n') 

34 

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

36 process = subprocess.Popen(cmd, 

37 stdout=subprocess.PIPE, 

38 stderr=subprocess.PIPE, 

39 shell=True, 

40 executable=self.shell_path, 

41 env=new_env) 

42 

43 out, err = process.communicate() 

44 if self.out_log is None: 

45 print("Exit, code {}".format(process.returncode)) 

46 process.wait() 

47 

48 # Write output to log 

49 if self.out_log is not None: 

50 self.out_log.info("Exit code {}".format(process.returncode)+'\n') 

51 if out: 

52 self.out_log.info(out.decode("utf-8")) 

53 

54 if self.global_log is not None: 

55 self.global_log.info(fu.get_logs_prefix()+'Executing: '+cmd[0:80]+'...') 

56 self.global_log.info(fu.get_logs_prefix()+"Exit code {}".format(process.returncode)) 

57 

58 if self.err_log is not None: 

59 if err: 

60 self.err_log.info(err.decode("utf-8")) 

61 

62 return process.returncode