Skip to content

GitLab

Commit dfb9cffd authored by Jerome Mariette's avatar Jerome Mariette
Browse files

delete iotypes

parent 69b6172a
Hide whitespace changes
Inline Side-by-side
src/jflow/iotypes.py deleted 100644 → 0
View file @ 69b6172a
#
# Copyright (C) 2012 INRA
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os, re
import abc
from _pyio import __metaclass__
class Formats(object):
ANY = "any"
BAM = "bam"
FASTQ = "fastq"
FASTA = "fasta"
SFF = "sff"
QUAL = "qual"
FLOW = "flow"
HTML = "html"
#
# Inputs classes
#
class InputFile(str):
def __new__(self, input, format=Formats.ANY):
self.format = format
if input.__class__.__name__ == "str":
return str.__new__(self, input)
elif input.__class__.__name__ == "OutputFile" and (input.format == self.format or self.format == Formats.ANY):
return str.__new__(self, input)
elif input.__class__.__name__ == "OutputFile" and not (input.format == self.format or self.format == Formats.ANY):
raise IOError(input.format+" format is incompatible with allowed format "+self.format)
else:
raise IOError(input.__class__.__name__+" cannot be used as InputFile")
def check(self):
return "TODO: check input file"
class InputFileList(list):
def __init__(self, inputs, format=Formats.ANY):
self.format = format
if inputs.__class__.__name__ == "str":
return list.__init__(self, [inputs])
elif inputs.__class__.__name__ == "OutputFile" and (inputs.format == self.format or self.format == Formats.ANY):
return list.__init__(self, [inputs])
elif inputs.__class__.__name__ == "list":
return list.__init__(self, inputs)
elif inputs.__class__.__name__ == "OutputFileList" and (inputs.format == self.format or self.format == Formats.ANY):
return list.__init__(self, inputs)
elif inputs.__class__.__name__ == "OutputFileEndsWith" and (inputs.format == self.format or self.format == Formats.ANY):
return list.__init__(self, inputs)
elif inputs.__class__.__name__ == "OutputFilePattern" and (inputs.format == self.format or self.format == Formats.ANY):
return list.__init__(self, inputs)
elif (inputs.__class__.__name__ == "OutputFile" or inputs.__class__.__name__ == "OutputFileList") and \
not (inputs.format == self.format or self.format == Formats.ANY):
raise IOError(inputs.format+" format is incompatible with allowed format "+self.format)
else:
raise IOError(inputs.__class__.__name__+" cannot be used as InputFileList")
def check(self):
return "TODO: check input file list"
class InputDirectory(str):
def __new__(self, input):
if input.__class__.__name__ == "str":
return str.__new__(self, input)
elif input.__class__.__name__ == "OutputDirectory":
return str.__new__(self, input)
else:
raise IOError(input.__class__.__name__+" cannot be used as InputDirectory")
def check(self):
return "TODO: check input directory"
class InputDirectoryList(list):
def __init__(self, inputs):
if inputs.__class__.__name__ == "str":
return list.__init__(self, [inputs])
elif inputs.__class__.__name__ == "OutputDirectory":
return list.__init__(self, [inputs])
elif inputs.__class__.__name__ == "list":
return list.__init__(self, inputs)
elif inputs.__class__.__name__ == "OutputDirectoryList":
return list.__init__(self, inputs)
else:
raise IOError(inputs.__class__.__name__+" cannot be used as InputDirectoryList")
#
# Outputs classes
#
class DynamicOutput(list):
"""
@warning : with this class of output, the component become dynamic.
"""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def update(self):
"""
This method is used at the end of component execution to update output list.
"""
raise NotImplementedError
class OutputFile(str):
def __new__(self, output, format=Formats.ANY):
self.format = format
if output.__class__.__name__ == "str":
return str.__new__(self, output)
else:
raise IOError(output.__class__.__name__+" cannot be used as OutputFile")
def check(self):
return "TODO: check output file"
class OutputFileList(list):
def __init__(self, outputs, format=Formats.ANY):
self.format = format
if outputs.__class__.__name__ == "str":
return list.__init__(self, [outputs])
elif outputs.__class__.__name__ == "list":
return list.__init__(self, outputs)
else:
raise IOError(outputs.__class__.__name__+" cannot be used as OutputFileList")
def check(self):
return "TODO: check output file list"
class OutputFilePattern(DynamicOutput):
def __init__(self, output_directory, pattern, format=Formats.ANY, include=True):
"""
@warning : with this class of output, the component become dynamic.
@param output_directory : path to the directory where outputs will be created.
@param pattern : the pattern of (a part) the file names.
@param format : the files format.
@param include : if true, the files with the pattern in file name are added into output files.
If false, the files with the pattern in file name are added into output files.
"""
self.output_directory = output_directory
self.pattern = pattern
self.format = format
self.include = include
return list.__init__(self, [])
def update(self):
output_files = list()
for file in os.listdir( self.output_directory ):
if self.include and re.search( self.pattern, file ) is not None:
output_files.append( os.path.join(self.output_directory, file) )
elif not self.include and re.search( self.pattern, file ) is None:
output_files.append( os.path.join(self.output_directory, file) )
return list.__init__(self, output_files)
def check(self):
return "TODO: check output file list"
class OutputFileEndsWith(DynamicOutput):
def __init__(self, output_directory, end_str, format=Formats.ANY, include=True):
"""
@warning : with this class of output, the component become dynamic.
@param output_directory : path to the directory where outputs will be created.
@param end_str : the end of the files names.
@param format : the files format.
@param include : if true, the files with name terminated by end_str are added into output files.
If false, the files with name not terminated by end_str are added into output files.
"""
self.output_directory = output_directory
self.end_str = end_str
self.format = format
self.include = include
return list.__init__(self, [])
def update(self):
output_files = list()
for file in os.listdir( self.output_directory ):
if file.endswith( self.end_str ) and self.include :
output_files.append( os.path.join(self.output_directory, file) )
elif not file.endswith( self.end_str ) and not self.include:
output_files.append( os.path.join(self.output_directory, file) )
return list.__init__(self, output_files)
def check(self):
return "TODO: check output file list"
class OutputFileList(list):
def __init__(self, outputs, format=Formats.ANY):
self.format = format
if outputs.__class__.__name__ == "str":
return list.__init__(self, [outputs])
elif outputs.__class__.__name__ == "list":
return list.__init__(self, outputs)
else:
raise IOError(outputs.__class__.__name__+" cannot be used as OutputFileList")
def check(self):
return "TODO: check output file list"
class OutputDirectory(str):
def __new__(self, output):
if output.__class__.__name__ == "str":
return str.__new__(self, output)
else:
raise IOError(output.__class__.__name__+" cannot be used as OutputDirectory")
def check(self):
return "TODO: check output directory"
class OutputDirectoryList(list):
def __init__(self, outputs):
if outputs.__class__.__name__ == "str":
return list.__init__(self, [outputs])
elif outputs.__class__.__name__ == "list":
return list.__init__(self, outputs)
else:
raise IOError(outputs.__class__.__name__+" cannot be used as OutputDirectoryList")
def check(self):
return "TODO: check output directory list"
\ No newline at end of file
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment