Newer
Older
kohleman
committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
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()