Coverage for biobb_dna/curvesplus/biobb_canal.py: 71%
84 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-28 10:36 +0000
1#!/usr/bin/env python3
3"""Module containing the Canal class and the command line interface."""
4import os
5import zipfile
6import argparse
7from typing import Optional
8from pathlib import Path
10from biobb_common.generic.biobb_object import BiobbObject
11from biobb_common.configuration import settings
12from biobb_common.tools import file_utils as fu
13from biobb_common.tools.file_utils import launchlogger
16class Canal(BiobbObject):
17 """
18 | biobb_dna Canal
19 | Wrapper for the Canal executable that is part of the Curves+ software suite.
20 | The Canal program is used to analyze the curvature of DNA structures.
22 Args:
23 input_cda_file (str): Input cda file, from Cur+ output. File type: input. `Sample file <https://raw.githubusercontent.com/bioexcel/biobb_dna/master/biobb_dna/test/data/curvesplus/curves_output.cda>`_. Accepted formats: cda (edam:format_2330).
24 input_lis_file (str) (Optional): Input lis file, from Cur+ output. File type: input. Accepted formats: lis (edam:format_2330).
25 output_zip_path (str): zip filename for output files. File type: output. `Sample file <https://raw.githubusercontent.com/bioexcel/biobb_dna/master/biobb_dna/test/reference/curvesplus/canal_output.zip>`_. Accepted formats: zip (edam:format_3987).
26 properties (dic):
27 * **bases** (*str*) - (None) sequence of bases to be searched for in the I/P data (default is blank, meaning no specified sequence).
28 * **itst** (*int*) - (0) Iteration start index.
29 * **itnd** (*int*) - (0) Iteration end index.
30 * **itdel** (*int*) - (1) Iteration delimiter.
31 * **lev1** (*int*) - (0) Lower base level limit (i.e. base pairs) used for analysis.
32 * **lev2** (*int*) - (0) Upper base level limit used for analysis. If lev1 > 0 and lev2 = 0, lev2 is set to lev1 (i.e. analyze lev1 only). If lev1=lev2=0, lev1 is set to 1 and lev2 is set to the length of the oligmer (i.e. analyze all levels).
33 * **nastr** (*str*) - ('NA') character string used to indicate missing data in .ser files.
34 * **cormin** (*float*) - (0.6) minimal absolute value for printing linear correlation coefficients between pairs of analyzed variables.
35 * **series** (*bool*) - (False) if True then output spatial or time series data. Only possible for the analysis of single structures or single trajectories.
36 * **histo** (*bool*) - (False) if True then output histogram data.
37 * **corr** (*bool*) - (False) if True than output linear correlation coefficients between all variables.
38 * **sequence** (*str*) - (Optional) sequence of the first strand of the corresponding DNA fragment, for each .cda file. If not given it will be parsed from .lis file.
39 * **binary_path** (*str*) - ('Canal') Path to Canal executable, otherwise the program wil look for Canal executable in the binaries folder.
40 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files.
41 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist.
42 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory.
43 Examples:
44 This is a use example of how to use the building block from Python::
46 from biobb_dna.curvesplus.biobb_canal import biobb_canal
47 prop = {
48 'series': 'True',
49 'histo': 'True',
50 'sequence': 'CGCGAATTCGCG'
51 }
52 biobb_canal(
53 input_cda_file='/path/to/curves/output.cda',
54 output_zip_path='/path/to/output.zip',
55 properties=prop)
56 Info:
57 * wrapped_software:
58 * name: Canal
59 * version: >=2.6
60 * license: BSD 3-Clause
61 * ontology:
62 * name: EDAM
63 * schema: http://edamontology.org/EDAM.owl
64 """
66 def __init__(self, input_cda_file, input_lis_file=None,
67 output_zip_path=None, properties=None, **kwargs) -> None:
68 properties = properties or {}
70 # Call parent class constructor
71 super().__init__(properties)
72 self.locals_var_dict = locals().copy()
74 # Input/Output files
75 self.io_dict = {
76 'in': {
77 'input_cda_file': input_cda_file,
78 'input_lis_file': input_lis_file,
79 },
80 'out': {
81 'output_zip_path': output_zip_path
82 }
83 }
85 # Properties specific for BB
86 self.bases = properties.get('bases', None)
87 self.nastr = properties.get('nastr', None)
88 self.cormin = properties.get('cormin', 0.6)
89 self.lev1 = properties.get('lev1', 0)
90 self.lev2 = properties.get('lev2', 0)
91 self.itst = properties.get('itst', 0)
92 self.itnd = properties.get('itnd', 0)
93 self.itdel = properties.get('itdel', 1)
94 self.series = ".t." if properties.get('series', False) else ".f."
95 self.histo = ".t." if properties.get('histo', False) else ".f."
96 self.corr = ".t." if properties.get('corr', False) else ".f."
97 self.sequence = properties.get('sequence', None)
98 self.binary_path = properties.get('binary_path', 'Canal')
99 self.properties = properties
101 # Check the properties
102 self.check_properties(properties)
103 self.check_arguments()
105 @launchlogger
106 def launch(self) -> int:
107 """Execute the :class:`Canal <biobb_dna.curvesplus.biobb_canal.Canal>` object."""
109 # Setup Biobb
110 if self.check_restart():
111 return 0
112 self.stage_files()
114 if self.sequence is None:
115 if self.stage_io_dict['in']['input_lis_file'] is None:
116 raise RuntimeError(
117 "if no sequence is passed in the configuration, "
118 "you must at least specify `input_lis_file` "
119 "so sequence can be parsed from there")
120 lis_lines = Path(
121 self.stage_io_dict['in']['input_lis_file']).read_text().splitlines()
122 for line in lis_lines:
123 if line.strip().startswith("Strand 1"):
124 self.sequence = line.split(" ")[-1]
125 fu.log(
126 f"using sequence {self.sequence} "
127 f"from {self.stage_io_dict['in']['input_lis_file']}",
128 self.out_log)
130 # define temporary file name
131 tmp_cda_path = Path(self.stage_io_dict['in']['input_cda_file']).name
133 # change directory to temporary folder
134 original_directory = os.getcwd()
135 os.chdir(self.stage_io_dict.get("unique_dir", ""))
137 # create intructions
138 instructions = [
139 f"{self.binary_path} <<! ",
140 "&inp",
141 " lis=canal_output,"]
142 if self.bases is not None:
143 # add topology file if needed
144 fu.log('Appending sequence of bases to be searched to command',
145 self.out_log, self.global_log)
146 instructions.append(f" seq={self.bases},")
147 if self.nastr is not None:
148 # add topology file if needed
149 fu.log('Adding null values string specification to command',
150 self.out_log, self.global_log)
151 instructions.append(f" nastr={self.nastr},")
153 instructions = instructions + [
154 f" cormin={self.cormin},",
155 f" lev1={self.lev1},lev2={self.lev2},",
156 f" itst={self.itst},itnd={self.itnd},itdel={self.itdel},",
157 f" histo={self.histo},",
158 f" series={self.series},",
159 f" corr={self.corr},",
160 "&end",
161 f"{tmp_cda_path} {self.sequence}",
162 "!"]
164 self.cmd = ["\n".join(instructions)]
165 fu.log('Creating command line with instructions and required arguments',
166 self.out_log, self.global_log)
168 # Run Biobb block
169 self.run_biobb()
171 # change back to original directory
172 os.chdir(original_directory)
174 # create zipfile and write output inside
175 zf = zipfile.ZipFile(
176 Path(self.stage_io_dict["out"]["output_zip_path"]), "w")
177 for canal_outfile in Path(self.stage_io_dict.get("unique_dir", "")).glob("canal_output*"):
178 if canal_outfile.suffix not in (".zip"):
179 zf.write(
180 canal_outfile,
181 arcname=canal_outfile.name)
182 zf.close()
184 # Copy files to host
185 self.copy_to_host()
187 # Remove temporary file(s)
188 # self.tmp_files.extend([
189 # self.stage_io_dict.get("unique_dir", "")
190 # ])
191 self.remove_tmp_files()
193 self.check_arguments(output_files_created=True, raise_exception=False)
195 return self.return_code
198def biobb_canal(
199 input_cda_file: str,
200 output_zip_path: str,
201 input_lis_file: Optional[str] = None,
202 properties: Optional[dict] = None,
203 **kwargs) -> int:
204 """Create :class:`Canal <biobb_dna.curvesplus.biobb_canal.Canal>` class and
205 execute the :meth:`launch() <biobb_dna.curvesplus.biobb_canal.Canal.launch>` method."""
207 return Canal(
208 input_cda_file=input_cda_file,
209 input_lis_file=input_lis_file,
210 output_zip_path=output_zip_path,
211 properties=properties, **kwargs).launch()
213 biobb_canal.__doc__ = Canal.__doc__
216def main():
217 """Command line execution of this building block. Please check the command line documentation."""
218 parser = argparse.ArgumentParser(description='Execute Canal from the Curves+ software suite.',
219 formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999))
220 parser.add_argument('--config', required=False, help='Configuration file')
222 required_args = parser.add_argument_group('required arguments')
223 required_args.add_argument('--input_cda_file', required=True,
224 help='cda input file from Curves+ output. Accepted formats: cda.')
225 required_args.add_argument('--output_zip_path', required=True,
226 help='Filename for .zip file with Canal output. Accepted formats: zip.')
227 parser.add_argument('--input_lis_file', required=False,
228 help='lis input file from Curves+ output. Accepted formats: lis.')
230 args = parser.parse_args()
231 args.config = args.config or "{}"
232 properties = settings.ConfReader(config=args.config).get_prop_dic()
234 biobb_canal(
235 input_cda_file=args.input_cda_file,
236 input_lis_file=args.input_lis_file,
237 output_zip_path=args.output_zip_path,
238 properties=properties)
241if __name__ == '__main__':
242 main()