Demo entry 3442333

sdf

   

Submitted by sdf on Dec 29, 2015 at 16:56
Language: Python. Code size: 4.8 kB.

import pickle
import gup_config
import thom_seq_db_con
import pdb
FAKE_GUP_DBNAME = gup_config.GUP_HOME + "fake_gup_db.p"

class GupDb():
    """ Cheating: Using dictionaries instead of database connections and 
    tables. But I'm going to pretend, using db type names for methods. So 
    nobody outside of this module will know I'm cheating. """
    def __init__(self):
        self.loadDb()

    def loadDb(self):
        try:
            self.tables = pickle.load(open(FAKE_GUP_DBNAME, "rb"))
        except:
            self.tables = {}
            self.tables['Samp'] = {}
            self.tables['Calc'] = {}
            self.tables['Flow'] = {}
            self.saveDb()

    def saveDb(self):
        pickle.dump(self.tables, open(FAKE_GUP_DBNAME, "wb")) 

    def getBarcodeFromNglusterDirName(self,nglusterDirName):
        for bc in self.tables['Flow'].keys():
            if self.tables['Flow'][bc]['nglusterDirName']== nglusterDirName:
                return bc
        return None

    def importFlowcell(self,nglusterDirName):
        try: 
            if self.getBarcodeFromNglusterDirName(nglusterDirName) != None:
                return 
        except:
            pass

        dbCon = thom_seq_db_con.DbCon()
        barcode = self.getBarcode(dbCon,nglusterDirName)

        # A row in the Flow table is keyed by barcode. Right now the 
        # only other columns are boolean isControl values for each lane.
        # the keys should be "Lane {0}.isControl" being lazy for now
        # note self.getAllLanes() would need to be updates as well
        self.tables['Flow'][barcode] = {} 
        self.tables['Flow'][barcode]['nglusterDirName'] = nglusterDirName

        for (i,isControl) in enumerate( dbCon.returnControlLanes(barcode)):
            self.tables['Flow'][barcode][i+1] = isControl #Lanes 1-8, not 0-7

        # A row in the Samp table is keyed by sampleID, the combination of
        # Accession number and flowcell_sample_id. Other values include barcode
        # as a foreign key, as well as 'flowcell_lane', 'genome', 
        # 'index_label', 'sample_name', 'recipe', 'seq_prep', 'project_name'
        for el in dbCon.returnSamples(barcode):
            samp = self.formatQueryResult(el)
            self.tables['Samp'][samp['sampleID']] = dict([ el for el in
                                                          samp.items() if el[0]
                                                          != 'sampleID'])
        self.saveDb()

    def getBarcode(self,dbCon,nglusterDirName):
        guessedBarcode = nglusterDirName.split('_')[3][1:]
        barcodesAvail = dbCon.returnAllBarcodes()
        try:
            bcIndex = [(guessedBarcode in el) for el in 
                       barcodesAvail].index(True)
            return barcodesAvail[bcIndex].upper()
        except: 
            raise Exception("Can't find {0} in {1}'s".format(guessedBarcode, 
                            nglusterDirName) + " seq_flowcell.barcode field")
 
    def formatQueryResult(self,row):
        colNames = [ 'barcode', 'flowcell_lane', 'sample_accession', 
                    'flowcell_sample_id', 'genome', 'index_label', 
                    'sample_name', 'recipe', 'seq_prep', 'project_name']

        samp = dict(zip(colNames,[str(el) for el in row]))
       
        samp["sampleID"] = (samp['sample_accession'].zfill(6) + '.' + 
                            samp['flowcell_sample_id'].zfill(7))
        del samp['sample_accession']
        del samp['flowcell_sample_id'] 

        colsToStrip = ['barcode','sample_name','project_name']
        for c in colsToStrip:
            samp[c]=samp[c].strip()

        colsSpaceToUnderscore = ['sample_name','project_name'] 
        for c in colsSpaceToUnderscore:
            samp[c]=samp[c].replace(" ","_") 

        samp["sample_name"] = ('\"' + samp["sample_name"] + '\"') 
        
        samp["is_control"] = "N"
        
        return samp 

    def getAllLanes(self,barcode):
        return [el for el in self.tables['Flow'][barcode].keys() if 
                el in range(1,9)]

    def getIsControl(self,barcode,lane):
        return self.tables['Flow'][barcode][lane]

    def getAllSamplesForFlowcell(self,barcode):
        return {key:value for key,value in 
                self.tables['Samp'].items() if
                self.tables['Samp'][key]['barcode']==barcode}

    def writeCalc(self, calcType, barcode, lnStr, runInfoForLane):
        pdb.set_trace()
        key = (calcType, barcode, lnStr) 
        self.tables['Calc'][key] = runInfoForLane
        self.saveDb()

    def getCalc(self,calcType,barcode,laneStr):
        try:
            return self.tables['Calc'][(calcType, barcode, laneStr)]
        except:
            return None

This snippet took 0.01 seconds to highlight.

Back to the Entry List or Home.

Delete this entry (admin only).