Coverage for biobb_dna/curvesplus/biobb_canal.py: 81%
78 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-05-28 06:38 +0000
1#!/usr/bin/env python3
3"""Module containing the Canal class and the command line interface."""
4import os
5import zipfile
6from typing import Optional
7from pathlib import Path
9from biobb_common.generic.biobb_object import BiobbObject
10from biobb_common.tools import file_utils as fu
11from biobb_common.tools.file_utils import launchlogger
14class Canal(BiobbObject):
15 """
16 | biobb_dna Canal
17 | Wrapper for the Canal executable that is part of the Curves+ software suite.
18 | The Canal program is used to analyze the curvature of DNA structures.
20 Args:
21 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).
22 input_lis_file (str) (Optional): Input lis file, from Cur+ output. File type: input. Accepted formats: lis (edam:format_2330).
23 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).
24 properties (dic):
25 * **bases** (*str*) - (None) sequence of bases to be searched for in the I/P data (default is blank, meaning no specified sequence).
26 * **itst** (*int*) - (0) Iteration start index.
27 * **itnd** (*int*) - (0) Iteration end index.
28 * **itdel** (*int*) - (1) Iteration delimiter.
29 * **lev1** (*int*) - (0) Lower base level limit (i.e. base pairs) used for analysis.
30 * **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).
31 * **nastr** (*str*) - ('NA') character string used to indicate missing data in .ser files.
32 * **cormin** (*float*) - (0.6) minimal absolute value for printing linear correlation coefficients between pairs of analyzed variables.
33 * **series** (*bool*) - (False) if True then output spatial or time series data. Only possible for the analysis of single structures or single trajectories.
34 * **histo** (*bool*) - (False) if True then output histogram data.
35 * **corr** (*bool*) - (False) if True than output linear correlation coefficients between all variables.
36 * **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.
37 * **binary_path** (*str*) - ('Canal') Path to Canal executable, otherwise the program wil look for Canal executable in the binaries folder.
38 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files.
39 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist.
40 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory.
41 * **container_path** (*str*) - (None) Path to the binary executable of your container.
42 * **container_image** (*str*) - ("cmip/cmip:latest") Container Image identifier.
43 * **container_volume_path** (*str*) - ("/data") Path to an internal directory in the container.
44 * **container_working_dir** (*str*) - (None) Path to the internal CWD in the container.
45 * **container_user_id** (*str*) - (None) User number id to be mapped inside the container.
46 * **container_shell_path** (*str*) - ("/bin/bash") Path to the binary executable of the container shell.
47 Examples:
48 This is a use example of how to use the building block from Python::
50 from biobb_dna.curvesplus.biobb_canal import biobb_canal
51 prop = {
52 'series': 'True',
53 'histo': 'True',
54 'sequence': 'CGCGAATTCGCG'
55 }
56 biobb_canal(
57 input_cda_file='/path/to/curves/output.cda',
58 output_zip_path='/path/to/output.zip',
59 properties=prop)
60 Info:
61 * wrapped_software:
62 * name: Canal
63 * version: >=2.6
64 * license: BSD 3-Clause
65 * ontology:
66 * name: EDAM
67 * schema: http://edamontology.org/EDAM.owl
68 """
70 def __init__(self, input_cda_file, input_lis_file=None,
71 output_zip_path=None, properties=None, **kwargs) -> None:
72 properties = properties or {}
74 # Call parent class constructor
75 super().__init__(properties)
76 self.locals_var_dict = locals().copy()
78 # Input/Output files
79 self.io_dict = {
80 'in': {
81 'input_cda_file': input_cda_file,
82 'input_lis_file': input_lis_file,
83 },
84 'out': {
85 'output_zip_path': output_zip_path
86 }
87 }
89 # Properties specific for BB
90 self.bases = properties.get('bases', None)
91 self.nastr = properties.get('nastr', None)
92 self.cormin = properties.get('cormin', 0.6)
93 self.lev1 = properties.get('lev1', 0)
94 self.lev2 = properties.get('lev2', 0)
95 self.itst = properties.get('itst', 0)
96 self.itnd = properties.get('itnd', 0)
97 self.itdel = properties.get('itdel', 1)
98 self.series = ".t." if properties.get('series', False) else ".f."
99 self.histo = ".t." if properties.get('histo', False) else ".f."
100 self.corr = ".t." if properties.get('corr', False) else ".f."
101 self.sequence = properties.get('sequence', None)
102 self.binary_path = properties.get('binary_path', 'Canal')
103 self.properties = properties
105 # Check the properties
106 self.check_properties(properties)
107 self.check_arguments()
109 @launchlogger
110 def launch(self) -> int:
111 """Execute the :class:`Canal <biobb_dna.curvesplus.biobb_canal.Canal>` object."""
113 # Setup Biobb
114 if self.check_restart():
115 return 0
116 self.stage_files()
118 if self.sequence is None:
119 if self.stage_io_dict['in']['input_lis_file'] is None:
120 raise RuntimeError(
121 "if no sequence is passed in the configuration, "
122 "you must at least specify `input_lis_file` "
123 "so sequence can be parsed from there")
124 lis_lines = Path(
125 self.stage_io_dict['in']['input_lis_file']).read_text().splitlines()
126 for line in lis_lines:
127 if line.strip().startswith("Strand 1"):
128 self.sequence = line.split(" ")[-1]
129 fu.log(
130 f"using sequence {self.sequence} "
131 f"from {self.stage_io_dict['in']['input_lis_file']}",
132 self.out_log)
134 # define temporary file name
135 if self.container_path:
136 tmp_cda_path = Path(self.container_working_dir).joinpath(Path(self.stage_io_dict['in']['input_cda_file']).name)
137 else:
138 tmp_cda_path = Path(self.stage_io_dict['in']['input_cda_file']).name
140 # change directory to temporary folder
141 original_directory = os.getcwd()
143 if self.container_path:
144 os.chdir(self.container_working_dir)
145 else:
146 os.chdir(self.stage_io_dict.get("unique_dir", ""))
148 # create intructions
149 instructions = [
150 f"{self.binary_path} <<! ",
151 "&inp",
152 " lis=canal_output,"]
153 if self.bases is not None:
154 # add topology file if needed
155 fu.log('Appending sequence of bases to be searched to command',
156 self.out_log, self.global_log)
157 instructions.append(f" seq={self.bases},")
158 if self.nastr is not None:
159 # add topology file if needed
160 fu.log('Adding null values string specification to command',
161 self.out_log, self.global_log)
162 instructions.append(f" nastr={self.nastr},")
164 instructions = instructions + [
165 f" cormin={self.cormin},",
166 f" lev1={self.lev1},lev2={self.lev2},",
167 f" itst={self.itst},itnd={self.itnd},itdel={self.itdel},",
168 f" histo={self.histo},",
169 f" series={self.series},",
170 f" corr={self.corr},",
171 "&end",
172 f"{tmp_cda_path} {self.sequence}",
173 "!"]
175 self.cmd = ["\n".join(instructions)]
176 fu.log('Creating command line with instructions and required arguments',
177 self.out_log, self.global_log)
179 # Run Biobb block
180 self.run_biobb()
182 # change back to original directory
183 os.chdir(original_directory)
185 workdir = self.stage_io_dict.get("unique_dir", "")
186 zip_host_path = Path(workdir) / Path(self.io_dict["out"]["output_zip_path"]).name
188 # create zipfile and write output inside
189 with zipfile.ZipFile(zip_host_path, "w") as zf:
190 for canal_outfile in Path(workdir).glob("canal_output*"):
191 fu.log(f"Adding {canal_outfile} to zip file", self.out_log, self.global_log)
192 if canal_outfile.suffix != ".zip":
193 zf.write(
194 canal_outfile,
195 arcname=canal_outfile.name)
197 # Copy files to host
198 self.copy_to_host()
200 # Remove temporary file(s)
201 self.remove_tmp_files()
203 self.check_arguments(output_files_created=True, raise_exception=False)
205 return self.return_code
208def biobb_canal(
209 input_cda_file: str,
210 output_zip_path: str,
211 input_lis_file: Optional[str] = None,
212 properties: Optional[dict] = None,
213 **kwargs) -> int:
214 """Create :class:`Canal <biobb_dna.curvesplus.biobb_canal.Canal>` class and
215 execute the :meth:`launch() <biobb_dna.curvesplus.biobb_canal.Canal.launch>` method."""
216 return Canal(**dict(locals())).launch()
219biobb_canal.__doc__ = Canal.__doc__
220main = Canal.get_main(biobb_canal, "Execute Canal from the Curves+ software suite.")
223if __name__ == '__main__':
224 main()