TEST_CODE_DIR = os.path.split(__file__)[0]
TESTDATA_DIR = os.path.join(TEST_CODE_DIR, 'testdata')
-LANE_LIST = range(1,9)
-TILE_LIST = range(1,101)
+LANE_LIST = list(range(1,9))
+TILE_LIST = list(range(1,101))
HISEQ_TILE_LIST = [1101, 1102, 1103, 1104, 1105, 1106, 1107, 1108,
1201, 1202, 1203, 1204, 1205, 1206, 1207, 1208,
2101, 2102, 2103, 2104, 2105, 2106, 2107, 2108,
f.write(config)
f.close()
+def make_runinfo(runfolder_dir, flowcell_id):
+ """Simulate a RunInfo.xml file created by >= RTA 1.9
+ """
+ xml = '''<?xml version="1.0"?>
+<RunInfo xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" Version="2">
+ <Run Id="{runfolder}" Number="101">
+ <Flowcell>{flowcell}</Flowcell>
+ <Instrument>SN787</Instrument>
+ <Date>110815</Date>
+ <Reads>
+ <Read Number="1" NumCycles="50" IsIndexedRead="N" />
+ <Read Number="2" NumCycles="7" IsIndexedRead="Y" />
+ </Reads>
+ <FlowcellLayout LaneCount="8" SurfaceCount="2" SwathCount="3" TileCount="8" />
+ <AlignToPhiX />
+ </Run>
+</RunInfo>
+'''
+ path, runfolder = os.path.split(runfolder_dir)
+ runinfo = os.path.join(runfolder_dir, 'RunInfo.xml')
+ stream = open(runinfo, 'w')
+ stream.write(xml.format(runfolder=runfolder, flowcell=flowcell_id))
+ stream.close()
+ return runinfo
+
def make_bustard_config132(image_dir):
source = os.path.join(TESTDATA_DIR, 'bustard-config132.xml')
destination = os.path.join(image_dir, 'config.xml')
]
for src, dest in demultiplex_pairs:
shutil.copy(src, dest)
+
+def make_unaligned_status_1_12(unaligned_dir, flowcell_id):
+ basecall_status = ['All.htm', 'Demultiplex_Stats.htm', 'IVC.htm']
+ test_data_root = os.path.join(TESTDATA_DIR, '1_12', 'basecall_stats')
+ basecall_stats = os.path.join(unaligned_dir,
+ 'Basecall_Stats_{0}'.format(flowcell_id))
+ os.mkdir(basecall_stats)
+ for filename in basecall_status:
+ source = os.path.join(test_data_root, filename)
+ destination = os.path.join(basecall_stats, filename)
+ shutil.copy(source, destination)
def make_rta_intensities_1460(data_dir, version='1.4.6.0'):
"""
return pathname
def dump(self):
- print ('index seq: {0}'.format(self.index_seq))
+ print(('index seq: {0}'.format(self.index_seq)))
- print ('project dir: {0}'.format(self.project_dir))
- print ('sample dir: {0}'.format(self.sample_dir))
- print ('rootname: {0}'.format(self.rootname))
- print ('path: {0}'.format(
+ print(('project dir: {0}'.format(self.project_dir)))
+ print(('sample dir: {0}'.format(self.sample_dir)))
+ print(('rootname: {0}'.format(self.rootname)))
+ print(('path: {0}'.format(
os.path.join(self.project_dir,
self.sample_dir,
- self.rootname+'R1_001.fastq.gz')))
+ self.rootname+'R1_001.fastq.gz'))))
def get_unaligned_sample_fastq_data(flowcell_id, lane, index_seq):
""".format(lane=lane, index=index_seq)
return body
-def ls_tree(root):
+def print_ls_tree(root):
+ """List tree contents, useful for debugging.
+ """
for dirpath, dirnames, filenames in os.walk(root):
for filename in filenames:
- print os.path.join(dirpath, filename)
+ print(os.path.join(dirpath, filename))
class BaseCallInfo(object):