Coverage for crunch/client/utils.py: 100.00%

1import stat 

2import subprocess 

3from pathlib import Path 

4import hashlib 

5 

6from .enums import WorkflowType 

7 

8 

9def has_mamba()->bool: 

10 """ 

11 Checks to see if mamba is available. 

12 

13 Returns: 

14 bool: Whether or not mamba is in the path. 

15 """ 

16 try: 

17 subprocess.run(["mamba", "--version"], capture_output=True, check=True) 

18 except (subprocess.CalledProcessError, FileNotFoundError): 

19 return False 

20 

21 return True 

22 

23 

24def conda_frontend() -> str: 

25 return "mamba" if has_mamba() else "conda" 

26 

27 

28def write_workflow(data:str, working_directory:Path, workflow_type:WorkflowType) -> Path: 

29 assert data 

30 working_directory = Path(working_directory) 

31 working_directory.mkdir(exist_ok=True, parents=True) 

32 

33 if workflow_type == WorkflowType.snakemake: 

34 workflow_path = working_directory / "Snakefile" 

35 elif workflow_type == WorkflowType.script: 

36 workflow_path = working_directory / "script.sh" 

37 

38 with open(workflow_path, "w", encoding="utf-8") as f: 

39 f.write(data.replace('\r\n', '\n')) 

40 

41 if workflow_type == WorkflowType.script: 

42 workflow_path.chmod(workflow_path.stat().st_mode | stat.S_IEXEC) 

43 

44 return workflow_path 

45 

46 

47def md5_checksums(directory): 

48 directory = Path(directory) 

49 result = dict() 

50 for path in directory.rglob("*"): 

51 if path.is_dir(): 

52 continue 

53 

54 relative_path = path.relative_to(directory) 

55 md5 = hashlib.md5(path.read_bytes()).hexdigest() 

56 result[str(relative_path)] = md5 

57 

58 return result