From 5f94925cae9bc2354d1d8274882e9a9c8cb3aebe Mon Sep 17 00:00:00 2001 From: kohleman <kohleman> Date: Mon, 6 Aug 2012 11:26:08 +0000 Subject: [PATCH] openBIS script for checking uniqueness of indices in a single- or dual-indexed sample pool SVN: 26304 --- .../source/Jython/duplicateBarcodeChecker.py | 104 ++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 deep_sequencing_unit/source/Jython/duplicateBarcodeChecker.py diff --git a/deep_sequencing_unit/source/Jython/duplicateBarcodeChecker.py b/deep_sequencing_unit/source/Jython/duplicateBarcodeChecker.py new file mode 100644 index 00000000000..e39e569845f --- /dev/null +++ b/deep_sequencing_unit/source/Jython/duplicateBarcodeChecker.py @@ -0,0 +1,104 @@ +import smtplib +import string +import copy + +def sendMail(registratorEmail, indexStrings='', sampleIdentifier='', additionalText=''): + ''' + Send out an email to the specified recipients + ''' + COMMASPACE = ", " + + loe = ["manuel.kohler@bsse.ethz.ch"] + if registratorEmail and registratorEmail not in loe: + loe.append(registratorEmail) + + TO = COMMASPACE.join(loe) + SUBJECT = "WARNING! For pooled sample " + sampleIdentifier + FROM = "qgf@openbis-dsu.ethz.ch" + HOST = "smtp0.ethz.ch" + text = "Sample: "+ sampleIdentifier + "\nBarcode: " + indexStrings + "\n" + additionalText + BODY = string.join(( + "From: %s" % FROM, + "To: %s" % TO, + "Subject: %s" % SUBJECT , + "", + text + ), "\r\n") + server = smtplib.SMTP(HOST) + server.sendmail(FROM, [TO], BODY) + server.quit() + +def checkDuplicateBarcodes(): + ''' + 'parents' are a HashSet of SamplePropertyPE + ''' + VOCABULARY_INDEX1 = 'BARCODE' + VOCABULARY_INDEX2 = 'INDEX2' + + SUCCESS_MESSAGE="OK" + registratorEmail = entity.entityPE().registrator.email + poolName = entity.entityPE().sampleIdentifier + + listofIndices = [] + listofIndices2 = [] + returnString = " " + + for e in entity.entityPE().parents: + for s in e.properties: + + if s.entityTypePropertyType.propertyType.simpleCode == VOCABULARY_INDEX1: + index = s.getVocabularyTerm().code + listofIndices.append(index) + + if s.entityTypePropertyType.propertyType.simpleCode == VOCABULARY_INDEX2: + index = s.getVocabularyTerm().code + listofIndices2.append(index) + + sampleIdentifier = e.sampleIdentifier + + # if dual-indexed + if listofIndices2: + dualIndexList = [] + if len(listofIndices) != len(listofIndices2): + indices = "\nIndex1 :" + str(listofIndices) + "\nIndex2: " + str(listofIndices2) + returnString = "Dual indexing assignment is not complete for pool sample " + str(poolName) + sendMail(registratorEmail, indices , str(poolName), returnString) + return returnString + + # build up a list of dual indices + for l in range(len(listofIndices)): + try: + dualIndexList.insert(l, [listofIndices[l], listofIndices2[l]]) + except: + pass + + # copy, needed by the for loop + dualIndexListOrig = copy.deepcopy(dualIndexList) + + for dualIndex in dualIndexListOrig: + dualIndexList.remove(dualIndex) + if dualIndex in dualIndexList: + returnString = "WARNING! You assigned duplicate indices \n " + str(dualIndex) + \ + " in the dual-indexed pooled sample!" + sendMail(registratorEmail, str(dualIndex), str(poolName), returnString) + return returnString + else: + returnString = SUCCESS_MESSAGE + return returnString + + # else single index + else: + uniqueList=list(set(listofIndices)) + if len(listofIndices) != len(uniqueList): + for index in uniqueList: + listofIndices.remove(index) + returnString = "WARNING! You assigned duplicate indices \n " + str(listofIndices) + \ + " in the single-indexed pooled sample!" + sendMail(registratorEmail, str(listofIndices), str(poolName), returnString) + else: + returnString = SUCCESS_MESSAGE + return returnString + +def calculate(): + """Main script function. The result will be used as the value of appropriate dynamic property.""" + return checkDuplicateBarcodes() -- GitLab