Coverage for biobb_flexserv / pcasuite / pcz_info.py: 97%

1#!/usr/bin/env python3 

2 

3"""Module containing the PCZinfo class and the command line interface.""" 

4from typing import Optional 

5import shutil 

6import json 

7from pathlib import PurePath 

8from biobb_common.tools import file_utils as fu 

9from biobb_common.generic.biobb_object import BiobbObject 

10from biobb_common.tools.file_utils import launchlogger 

11 

12 

13class PCZinfo(BiobbObject): 

14 """ 

15 | biobb_flexserv PCZinfo 

16 | Extract PCA info (variance, Dimensionality) from a compressed PCZ file. 

17 | Wrapper of the pczdump tool from the PCAsuite FlexServ module. 

18 

19 Args: 

20 input_pcz_path (str): Input compressed trajectory file. File type: input. `Sample file <https://github.com/bioexcel/biobb_flexserv/raw/master/biobb_flexserv/test/data/pcasuite/pcazip.pcz>`_. Accepted formats: pcz (edam:format_3874). 

21 output_json_path (str): Output json file with PCA info such as number of components, variance and dimensionality. File type: output. `Sample file <https://github.com/bioexcel/biobb_flexserv/raw/master/biobb_flexserv/test/reference/pcasuite/pcz_info.json>`_. Accepted formats: json (edam:format_3464). 

22 properties (dict - Python dictionary object containing the tool parameters, not input/output files): 

23 * **binary_path** (*str*) - ("pczdump") pczdump binary path to be used. 

24 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files. 

25 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist. 

26 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory. 

27 

28 Examples: 

29 This is a use example of how to use the building block from Python:: 

30 

31 from biobb_flexserv.pcasuite.pcz_info import pcz_info 

32 

33 pcz_info( input_pcz_path='/path/to/pcazip_input.pcz', 

34 output_json_path='/path/to/pcz_info.json') 

35 

36 Info: 

37 * wrapped_software: 

38 * name: FlexServ PCAsuite 

39 * version: >=1.0 

40 * license: Apache-2.0 

41 * ontology: 

42 * name: EDAM 

43 * schema: http://edamontology.org/EDAM.owl 

44 

45 """ 

46 

47 def __init__(self, input_pcz_path: str, 

48 output_json_path: str, properties: Optional[dict] = None, **kwargs) -> None: 

49 

50 properties = properties or {} 

51 

52 # Call parent class constructor 

53 super().__init__(properties) 

54 self.locals_var_dict = locals().copy() 

55 

56 # Input/Output files 

57 self.io_dict = { 

58 'in': {'input_pcz_path': input_pcz_path}, 

59 'out': {'output_json_path': output_json_path} 

60 } 

61 

62 # Properties specific for BB 

63 self.properties = properties 

64 self.binary_path = properties.get('binary_path', 'pczdump') 

65 

66 # Check the properties 

67 self.check_properties(properties) 

68 self.check_arguments() 

69 

70 @launchlogger 

71 def launch(self): 

72 """Launches the execution of the FlexServ pcz_info module.""" 

73 

74 # Setup Biobb 

75 if self.check_restart(): 

76 return 0 

77 # self.stage_files() 

78 

79 # Internal file paths 

80 # try: 

81 # # Using rel paths to shorten the amount of characters due to fortran path length limitations 

82 # input_pcz = str(Path(self.stage_io_dict["in"]["input_pcz_path"]).relative_to(Path.cwd())) 

83 # output_json = str(Path(self.stage_io_dict["out"]["output_json_path"]).relative_to(Path.cwd())) 

84 # except ValueError: 

85 # # Container or remote case 

86 # input_pcz = self.stage_io_dict["in"]["input_pcz_path"] 

87 # output_json = self.stage_io_dict["out"]["output_json_path"] 

88 

89 # Manually creating a Sandbox to avoid issues with input parameters buffer overflow: 

90 # Long strings defining a file path makes Fortran or C compiled programs crash if the string 

91 # declared is shorter than the input parameter path (string) length. 

92 # Generating a temporary folder and working inside this folder (sandbox) fixes this problem. 

93 # The problem was found in Galaxy executions, launching Singularity containers (May 2023). 

94 

95 # Creating temporary folder 

96 tmp_folder = fu.create_unique_dir() 

97 fu.log('Creating %s temporary folder' % tmp_folder, self.out_log) 

98 

99 shutil.copy2(self.io_dict["in"]["input_pcz_path"], tmp_folder) 

100 

101 # Temporary output 

102 # temp_out_1 = str(Path(self.stage_io_dict.get("unique_dir", "")).joinpath("output1.dat")) 

103 # temp_out_2 = str(Path(self.stage_io_dict.get("unique_dir", "")).joinpath("output2.dat")) 

104 temp_out_1 = "output1.dat" 

105 temp_out_2 = "output2.dat" 

106 temp_json = "output.json" 

107 

108 # Command line 

109 # pczdump -i structure.ca.std.pcz --info -o pcz.info 

110 # self.cmd = [self.binary_path, 

111 # "-i", input_pcz, 

112 # "-o", temp_out_1, 

113 # "--info", ';', 

114 # self.binary_path, 

115 # "-i", input_pcz, 

116 # "-o", temp_out_2, 

117 # "--evals" 

118 # ] 

119 

120 self.cmd = ['cd', tmp_folder, ';', 

121 self.binary_path, 

122 "-i", PurePath(self.io_dict["in"]["input_pcz_path"]).name, 

123 "-o", temp_out_1, 

124 "--info", ';', 

125 self.binary_path, 

126 "-i", PurePath(self.io_dict["in"]["input_pcz_path"]).name, 

127 "-o", temp_out_2, 

128 "--evals" 

129 ] 

130 

131 # Run Biobb block 

132 self.run_biobb() 

133 

134 # Parse output info 

135 # Title : MC generated trajectory 

136 # Atoms : 85 

137 # Vectors : 4 

138 # Frames : 1000 

139 # Total variance : 1137.20 

140 # Explained variance: 1043.32 

141 # Quality : 91.74% 

142 # Dimensionality : 21 

143 # RMSd type : Standard RMSd 

144 # Have atom names : True 

145 info_dict = {} 

146 with open(PurePath(tmp_folder).joinpath(temp_out_1), 'r') as file: 

147 for line in file: 

148 info = line.split(':') 

149 info_dict[info[0].strip().replace(' ', '_')] = info[1].strip() 

150 

151 # Parse output evals 

152 # 744.201782 

153 # 170.061981 

154 # 89.214905 

155 # 39.836308 

156 info_dict['Eigen_Values'] = [] 

157 info_dict['Eigen_Values_dimensionality_vs_total'] = [] 

158 info_dict['Eigen_Values_dimensionality_vs_explained'] = [] 

159 accum_tot = 0 

160 accum_exp = 0 

161 with open(PurePath(tmp_folder).joinpath(temp_out_2), 'r') as file: 

162 for line in file: 

163 eval = float(line.strip()) 

164 eval_var = (eval / float(info_dict['Total_variance']))*100 

165 accum_tot = accum_tot + eval_var 

166 eval_dim = (eval / float(info_dict['Explained_variance']))*100 

167 accum_exp = accum_exp + eval_dim 

168 info_dict['Eigen_Values'].append(eval) 

169 info_dict['Eigen_Values_dimensionality_vs_total'].append(accum_tot) 

170 info_dict['Eigen_Values_dimensionality_vs_explained'].append(accum_exp) 

171 

172 with open(PurePath(tmp_folder).joinpath(temp_json), 'w') as out_file: 

173 out_file.write(json.dumps(info_dict, indent=4)) 

174 

175 # Copy outputs from temporary folder to output path 

176 shutil.copy2(PurePath(tmp_folder).joinpath(temp_json), PurePath(self.io_dict["out"]["output_json_path"])) 

177 

178 # Copy files to host 

179 # self.copy_to_host() 

180 

181 # Remove temporary folder(s) 

182 self.tmp_files.append(tmp_folder) 

183 self.remove_tmp_files() 

184 

185 self.check_arguments(output_files_created=True, raise_exception=False) 

186 

187 return self.return_code 

188 

189 

190def pcz_info(input_pcz_path: str, output_json_path: str, 

191 properties: Optional[dict] = None, **kwargs) -> int: 

192 """Create :class:`PCZinfo <flexserv.pcasuite.pcz_info>`flexserv.pcasuite.PCZinfo class and 

193 execute :meth:`launch() <flexserv.pcasuite.pcz_info.launch>` method""" 

194 return PCZinfo(**dict(locals())).launch() 

195 

196 

197pcz_info.__doc__ = PCZinfo.__doc__ 

198main = PCZinfo.get_main(pcz_info, "Extract PCA info from a compressed PCZ file.") 

199 

200if __name__ == '__main__': 

201 main()