import json
from datetime import date
from datetime import datetime
from dateutil.relativedelta import relativedelta
import pandas as pd
[docs]class Json_factory():
def __init__(self):
pass
## first approach to be overhauled
##TODO Decide upon way of json creation/storage for OS operations, the current idea is to set mandatory values to
## boolean true by default. than a random generator can create the values in fixed order and they can be set
## according to the boolean. Even better would be the restriction to use pandas data frames directly and create the
## json once data entry is done. Another important question is how to deal with custom value fields in various cases.
## Further, where possible, the csv import option should be used and json creation should be left to the numerous
## methods within Open Specimen
# creation jsons:
# participant
[docs] def create_participant_json(self, regdate = None, id_ = None, cpid = None, cptitle = None,
cpshorttitle =None, ppid = None, firstname = None, middlename = None,
lastname = None, uid = None, birthdate = None, vitalstatus = None,
deathdate = None, gender = None, race = None, ethnicities = None, cprid=None,
sexgenotype = None, pmis = None, mrn = None, sitename = None, empi =None):
"""Creates a JSON-formated string for participant creation
This function creates the json corresponding to the cpr_util function
Note
-----
Mandatory parameters are passed as positional arguments in the caliing function
Parameters
----------
regdate : string
Mandatory field with date of registration in the format which is defined in the systemsettings of openSpecimen.
cprid : int
Unique ID of the Participant's Registration.
id_ : int
Unique ID of the Participant.
cpid : int
Unique ID of the Collection Protocol, which is autogenerated from OpenSpecimen. cpid or cptitle or cpshorttile is mandatory.
cptitle : string
Unique title of the Collection Protocol. cpid or cptitle or cpshorttile is mandatory.
cpshorttitle : string
Unique Acronym of the Collection Protocol. cpid or cptitle or cpshorttile is mandatory.
ppid : string
Participant protocol ID, is mandatory if created manaully and have to be empty if it is autogenerated.
This is a protocol setting.
firstname : string
Participants first name.
middlename : string
Participants middle name.
lastname : string
Participants last name.
uid : string
Unique identifier e.g. social security number.
birthdate : string
Birthdate in the format which is defined in the systemsettings of OpenSpecimen.
vitalstatus : string
Vital status of the Participant.
deathdate : string
Deathdate in the format which is defined in the systemsettings of OpenSpecimen.
gender : string
Gender of the participant, permissable values are Male, Female, Unknown, Unspecified.
race : string
Participants racial origination, permissable values are {American Indian or Alaska Native, Asian, black or Afro American, Native Hawaiian
or other Pacific Islander, Not REported, Unknown, White}
ethnicities : string
Participants ethnicities, permissable values are: {Hispanic or Latino, Not Hispanic or Latino, Not Reported, Unknown}
sexgenotype : string
Participants sex Genotype, permissable values are {XX Genotype, XY Genotype, XXX, Klinefelter’s Syndrome, XXXY syndrome,
XXYY syndrome, Mosaic including XXXXY, Penta X syndrome}
pmis : string
Collection of the Participants medical record numner.
mrn : string
Participants medical record number.
sitename : string
Name of the site, where the participant is registrated.
empi : string
Enterprise master patient index number.
Returns
-------
JSON-dict
Details of the updated Participant or the OpenSpecimen error message as Dictornary.
"""
part = {
"id": id_,
"firstName": firstname,
"middleName": middlename,
"lastName": lastname,
"uid":uid,
"birthDate": birthdate,
"vitalStatus": vitalstatus,
"deathDate": deathdate,
"gender": gender,
"race": race,
"sexGenotype": sexgenotype,
"ethnicities": ethnicities,
"pmis": pmis,
"mrn": mrn,
"siteName": sitename,
"empi": empi
}
part = {k: v for k, v in part.items() if v is not None}
data = {
"cpId": str(cpid),
"registrationDate": regdate,
"ppid": str(ppid),
"cpShortTitle": cpshorttitle,
"cpTitle": cptitle
}
data = {k: v for k, v in data.items() if v is not None}
data['participant']=part
return json.dumps(data)
# Collection Protocoll
[docs] def create_CP_json(self, short_title = None, title=None, pi_mail=None, time_start=None, time_end=None, sites=None, man_id=False, coords=None,
consentsWaived=None,eth_cons_id=None, part_no=None, desc_url=None, visitNameFmt=None, specimenLabelFmt=None,
derivativeLabelFmt =None, man_visit_name=False, man_spec_label=True, aliquots_in_same=None, activity=None,
aliquotLabelFmt = None, ppidFmt= None, specimenCentric = None, cpid=None):
"""Creates the JSON-formated string corresponding to the collection_protocol_util funciton create_CP
Note
-----
Mandatory paramters are passed as positional args within the calling util class
Parameters
----------
short_title : string
Short title of the Collection Protocol.
title : string
Title of the Collection Protocol.
pi_mail : string
Email Address of the Principal Investigator.
time_start: string
String with the start_time of the collection Protocol in the timeformat specified in the System configuration.
time_end: string
String with the end_time of the collection Protocol in the timeformat specified in the System configuration.
sites: list
Sites which are assigned to the collection Protocol.
man_id : string
OpenSpecimen's boolean true/false if the manual PPID creation is enabled.
coords: dict
dict with Coordinators and coordinator ids in it.
consentsWaived : string
OpenSpecimen's boolean true/false if consent should be waived.
eth_cons_id : string
Ethical aproavel id.
part_no : string
String with number of anticipated Participant count.
desc_url = string
URL with the decription of the Collection Protocol.
visitNameFMT : string
String which contains the OpenSpecimen's token for creating Visit Names automatically.
man_visit_name : string
String with OpenSpecimen's boolean format if the Visits should be created manually.
man_spec_label : string
String with OpenSpecimen's boolean format if the Specimen Labels should be created manually.
man_spec_label : string
String with OpenSpecimen's boolean format if the Aliquotes are stored in the same Container.
activity : string
String with the acitivity status of the Specimen.
cpid : int
Integer with the Unique ID of the Collection Protocol.
Returns
-------
JSON-formated-string
Collection protocol information neccesary for creation
"""
site_arr = []
for item in sites:
site_arr.append(
{"siteName": item},
)
data = {
"id": cpid,
"shortTitle": short_title,
"title": title,
"code": None,
"principalInvestigator":
{
"loginName": pi_mail,
"domain": "openspecimen"
},
"startDate": time_start,
"endDate": time_end,
"ppidFmt": ppidFmt,
"manualPpidEnabled": man_id,
"cpSites": site_arr,
"consentsWaived": consentsWaived,
"irbId": eth_cons_id,
"anticipatedParticipantsCount": part_no,
"descriptionUrl": desc_url,
"specimenLabelFmt": specimenLabelFmt,
"derivativeLabelFmt": derivativeLabelFmt,
"aliquotLabelFmt": aliquotLabelFmt,
"visitNameFmt": visitNameFmt,
"manualVisitNameEnabled": man_visit_name,
"manualSpecLabelEnabled": man_spec_label,
"aliquotsInSameContainer": aliquots_in_same,
"activityStatus": activity,
"specimenCentric": specimenCentric
}
if coords != None:
data["coordinators"] = {"loginName" : coords,"domain" : "openspecimen"}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
# Collection Protocol event
[docs] def create_cp_event_json(self, label=None, point=None, cp=None, site=None, diagnosis=None, status=None, activity=None, unit=None, code=None):
"""Create JSON-formated string needed for event creation
Create an event for a given Collection Protocol. Details of the parameters can be found in
the parameters section.
Note
-----
Mandatory paramters are passed as positional args within the calling util class
Parameters
----------
label : string
Label of the Event, has to be unique.
point : string or int
Starting Point of the event, Value + unit (e.g. DAYS).
cp : string
title of the collection protocol.
site : string
The default Site of the event.
diagnosis : string
Defines the permissable values of the diagnosis.
status : string
Defines the permissable values of the clinical status.
acitivity : string
DEfines the activity status of the event.
unit : string
Defines which unit has the starting point.
code : string
the Event code, is optional. In order to define condionals in the workflow, one need the Event code.
Returns
-------
JSON-formated string
Returns a JSON-formated string with the given details for cp event creation
"""
params = {
"eventLabel": label,
"clinicalDiagnosis": diagnosis,
"clinicalStatus": status,
"collectionProtocol": cp,
"defaultSite": site,
"activityStatus": activity,
"eventPoint": point,
"eventPointUnit": unit
}
return json.dumps(params)
[docs] def create_specimen_json(self, label=None, specimenclass=None, specimentype=None, pathology=None, anatomic=None,
laterality=None, initqty=None, avaqty=None, visitid=None, userid=None, colltime=None,
rectime=None, recqlt=None, lineage='New', status='Collected', storloc=None, concentration=None,
biohazard=None, comments=None, collproc=None, conttype=None, extension =None):
"""Create a API Json String for a Specimen
Create the JSON String neccesary for creating a specimen
Note
-----
Mandatory paramters are passed as positional args within the calling util class
Parameters
----------
label : string
UUID of specimen generated automatically if not set to manual in corresponding collection protocol
specimenclass : string
Class of the specimen.
specimentype : string
Type of the specimen, belongs to the class.
pathology : string
Pathologystatus of the Specimen.
anatomic : string
The anatomic site of the specimen.
laterality : string
The laterality of the specimen.
initqty : int
The initial quantity of a specimen.
avaqty : int
The available quantity of a specimen.
visitid : int
The unique identifier of the visit.
recqlt : string
The received quality.
colltime : string
Date and Time of the collection event, the format is in the OpenSpecimen's System configuration.[optional]
rectime : string
Date and Time of the received event, the format is in the OpenSpecimen's System configuration.[optional]
lineage : string
Lineage of the specimen, default value is New.
status : string
Status of the Specimen, default is 'Collected'.
stor_name : string
Name of the container. [optional]
storlocx : int
Position of the specimen in the Container in x direction.[optional]
storlocy : int
Position of the specimen int the container in y direction.[optional]
concetration : int
Concentration of the specimen[optional].
biohazard : string
Biohazards of that specimen.[optional]
userid : int
ID of the user who creates the specimen. If not specified the API user is taken.
comments : string
Comments regarding to the specimen[optional].
collproc : string
The procedure of the collection[otpional].
conttype : string
Type of the storage conatiner.
extension : JSON-String
JSON-formated-string containing the dict of specimnen extensions created during call to the corresponding util class
Returns
-------
dict
JSON-formated-string with the complete specimen to be created
"""
data = {
"label": label,
"specimenClass": specimenclass,
"type": specimentype,
"pathology": pathology,
"anatomicSite": anatomic,
"laterality": laterality,
"initialQty": initqty,
"availableQty": avaqty,
"visitId": visitid,
"status": status,
"storageLocation": storloc,
"concetration": concentration,
"biohazard": biohazard,
"comments": comments,
"collectionEvent": {
"user": {"id": userid},
"time": colltime,
"container": conttype,
"procedure": collproc
},
"receivedEvent":{
"user":{"id": userid},
"time": rectime,
"receivedQuality": recqlt
},
"extensionDetail": extension
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def create_extension(self, attrsmap=None, extensiondict=None, useudn=None):
data = {
"useUdn": useudn,
"attryMap": attrsmap,
"value": extensiondict
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
# Create CSV export job
[docs] def create_csv_export_job(self, objecttype=None, recordids=None, cpid=None, ppids=None, entitytype=None, formname=None,
specimenlabels=None):
"""Create JSON formated string neccesary for exporting an collection protocol (should be implemented but is not see OpenSpecimen 7.2)
Parameters
----------
objecttype: string
Identifying the general object to be exported.
Permissible Values: "institute", "site", "user", "cpr", "specimen", "extensions", "storageContainer"
recordids: list or string
Comma seperated list of record ids for fetching selected entries by their identifier. (Sites, Institutes, Users and Containers)
cp_id: string
Collection protocol id of export target not neccesary for objects higher in the hierachy like institue or site.
For all others it can be specified or set to "-1" which means all data in the system.
ppids: list or string
List of comma seperated participant identifiers; String if its a singular participant to be exported
Used in combination with specimen object type as a paramter
entitytype: string
Paramter defining the entity for data extraction (e.g.: attached form at participant level)
used with the extension object type
formname: string
Defines the form to be downloaded in context of the extension object type together with the specified entity
specimenlabels: list or string
List of comma seperated specimen identifiers; str if its a singular specimen to be exported
Returns
-------
JSON-formated-string
Json data needed for creating an export job for the given entity
"""
object_types = ["institute", "site", "user", "cpr", "specimen", "extensions", "storageContainer", "distributionProtocol", "cp", "cpe"]
entity_types = ["Participant", "Visit", "Specimen", "SpecimenEvent", "SpecimenCollectionGroup", "CollectionProtocol"]
objecttype = objecttype.lower()
assert objecttype in object_types, "Object Type {} not allowed check documentation".format(objecttype)
if entitytype is not None:
assert entitytype in entity_types, "Entity Type {} not allowed check documentation".format(entitytype)
if objecttype == "cpr" or objecttype == "specimen":
assert cpid is not None , "cpid may not be none with objecttype {}".format(objecttype)
elif objecttype == "extensions":
assert formname is not None and entitytype is not None, "Please specify formname and the entitytype the given form is attached to"
assert cpid is not None , "cpid may not be none with objecttype {}".format(objecttype)
params = {
"ppids": ppids,
"specimenLabels": specimenlabels,
"entityType": entitytype,
"formName": formname,
"cpId": cpid
}
params = {k: v for k, v in params.items() if v is not None}
data = {
"objectType": objecttype,
"recordIds": recordids,
"params": params
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
# Create Any AQL Query
[docs] def execute_aql(self, cpid, aql, rowmode='OFF', columnexpr='true', isodate='true'):
"""Create JSON formated string to execute a specified query passed to the method
Parameters
----------
cp_id: string
Collection protocoll id of export target
aql: string
query command in the OpenSpecimen Syntax (see OpenSpecimen API)
rowmode: string
Specifies whether multi-valued attributes result in a single row or one row per value.
Default value is OFF. Other permitted values are SHALLOW and DEEP.
Try out to see what fits best for your use case.
columnexpr: string
Specifies whether the column labels or AQL expression needs to be included in the query response.
By default, user friendly column labels are included in query response.
isodate: string
Specifies how the date column values needs to be serialised in the query response.
If true, then date/time values are serialised using ISO format: yyyy-MM-dd'T'HH:mm:ss.
Otherwise, date/time values are serialised using the format specified in OS locale settings
Returns
-------
JSON-formated-string
Json data needed for executing a system query
"""
params = {
"cpId" : cpid,
"aql" : aql,
"wideRowMode" : rowmode,
"outputColumnExprs" : columnexpr,
"outputIsoDateTime" : isodate
}
params = {k: v for k, v in params.items() if v is not None}
return json.dumps(params)
# Execute Saved Query
[docs] def execute_query(self, start, results, drivingform="Participant", rowmode="OFF"):
"""Create JSON formated string neccesary for execution of a saved query
Parameters
----------
start: string
Used for paginating the results. If start=5, the output result will start from row number 5.
results: string
Used for paginating the results. If results=10, the output result will be maximum 10 rows
drivingform: string
Driving form determines the search perspective. When left empty, it defaults to Participant.
(For Examples when drivingForm is Participant, the root table is 'catissue_coll_prot_reg'
which is then used to join with the other tables.
Similarly when drivingForm is Specimen, the root table will be catissue_specimen,
which will be joined with the other tables.) rowmode: string
Returns
-------
JSON-formatted-string
Json data needed for executing a query saved within an Openspecimen appliance
"""
params= {
"drivingForm": drivingform,
"wideRowMode": rowmode,
"startAt": start,
"maxResults":results
}
params = {k: v for k, v in params.items() if v is not None}
return json.dumps(params)
[docs] def create_bulk_import_job(self, schemaname=None, operation=None, fileid=None,
dateformat=None, timeformat=None):
"""Create JSON formated string neccesary for creating a bulk import operation to be handled by OpenSpecimen
Parameters
----------
schemaname : string
The Openspecimen schematype, that defines the expected CSV file structure for the entity to be created
operation : string
Either CREATE or UPDATE duplicate entries will be ignored in case of creation.
fileid: string
The file id created internally by OpenSpecimen, which needs to be passed to the API
dateformat: string
Specification of date format used in the OpenSpecimen API. See link for details
https://openspecimen.atlassian.net/wiki/spaces/CAT/pages/68976690/Date+and+time+formats
timeformat: string
Specification of time format used in the OpenSpecimen API. See link for details
https://openspecimen.atlassian.net/wiki/spaces/CAT/pages/68976690/Date+and+time+formats
Returns
-------
JSON-formated-string
Json data needed for creating an CSV file import job for the given entity
"""
data = {"objectType": schemaname,
"importType": operation,
"inputFileId": fileid,
"dateFormat":dateformat,
"timeFormat":timeformat
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def get_registrations(self, cpid=None, registrationdate=None, ppid=None, name=None, birthdate=None, uid=None, specimen=None,
includestats=None, startat=None,maxresults=None, exactmatch=None):
"""Create JSON formated string neccesary for creating a bulk import operation to be handled by OpenSpecimen
Parameters
----------
cpid : string
Collection protol id of the participants to be queried
registrationdate: string
Date formatted string to query participants registered especially on this date
ppid: string
OpenSpecimen internal unique id or part of a unique id to be matched (exact matching is defind in exact match)
name: string
Name of the participant to be matched
birthdate: string
Date formatted string containing the birthdate of a particiapnt to be queried
uid: string
Social Security number or different national identifier
specimen: string
Participants whose specimen labels or barcodes contain this Parameters value as a substring will be matched
includestats: string
Include additional statitics within the response (visits specimens etc.)
startat: string
startat and maxResults are useful in implementing pagination of participants list. When not specified,
startAt defaults to 0. When startAt = n, the first element of the response is (n + 1)th participant satisfying the query criteria.
maxresults: string
This Parameters specifies how many participant records should be included in the API response.
exactmatch: string
Specifies whether the PPID should be exact match or sub-string match. Boolean true means exact match. Otherwise it is substring match.
Returns
-------
JSON-formated-string
CP id of target of merge action
Json data needed for creating an CSV file import job for the given entity
"""
params = {
"cpId": cpid,
"registrationDate": registrationdate,
"ppid": ppid,
"name": name,
"dob": birthdate,
"uid": uid,
"specimen": specimen,
"includeStats": includestats,
"startAt": startat,
"maxResults": maxresults,
"exactMatch": exactmatch
}
params = {k: v for k, v in params.items() if v is not None}
return json.dumps(params)
[docs] def add_visit_json(self, cprid, name, site, eventid=None,eventlabel=None,ppid=None, cptitle=None, cpshorttitle=None,
diagnosis=None, clinicalstatus=None, activity=None, visitstatus="Complete", missedreason=None,
missedby=None, comments=None,pathologynumber=None,cohort=None, visitdate=None, cpid=None):
"""Create JSON formated string neccesary for adding an visit
Parameters
----------
cprid : int
Identifier of the Collection Protocoll Registration to which the Visit belongs.
cprid or (cptitle and ppid) or (cpid and ppid) or (cpshorttitle and ppid) are mandatory.
name : string
Name of the Visit.
site : string
Site to which the Visit belongs.
eventid : int
ID of the event to which the visit belongs.[optional]
eventlabel : string
Label of the event to which the visit belongs.[optional]
ppid : string
Identifier of the Participant to whom the Visit belongs.
cprid or (cptitle and ppid) or (cpid and ppid) or (cpshorttitle and ppid) are mandatory.
cptitle : string
Name of the Collection Protocol.
cprid or (cptitle and ppid) or (cpid and ppid) or (cpshorttitle and ppid) are mandatory.
cpshorttitle : title
Acronym of the Collection Protocol.
cprid or (cptitle and ppid) or (cpid and ppid) or (cpshorttitle and ppid) are mandatory.
diagnosis : string
Name of the diagnoses of the visit.
clinicalstatus : string
Clinical Status of the visit.[optional]
activity : string
Activity Status of the Visit.[optional]
visitstatus : string
Status of the visit.[optional]
missedreason : string
Reason why the visit was missed.[optional]
missedby : string
Details of the person who missed the visit.[optional]
comments : string
Commets regarding the visit.[optional]
pathologynumber : string
Surgical Pathology number. [optional]
cohort : string
Cohorts to which the Visit belongs. [optional]
visitdate : string
Date when the visit will occur, if empty takes the current date.[optional]
cpid : int
Identifier of the Collection Protocol.
cprid or (cptitle and ppid) or (cpid and ppid) or (cpshorttitle and ppid) are mandatory.
Returns
-------
JSON-formatted string
Json data necessary to add an visit via API call
"""
params = {
"cprId": cprid,
"eventId": eventid,
"eventLabel": eventlabel,
"ppid": ppid,
"cpTitle": cptitle,
"cpShortTitle":cpshorttitle,
"cpId": cpid,
"name": name,
"clinicalDiagnoses": diagnosis,
"clinicalStatus": clinicalstatus,
"activityStatus": activity,
"site": site,
"status": visitstatus,
"missedReason": missedreason,
"missedBy": missedby,
"comments": comments,
"surgicalPathologyNumber": pathologynumber,
"cohort": cohort,
"visitDate": visitdate
}
params = {k: v for k, v in params.items() if v is not None}
return json.dumps(params)
[docs] def storage_location_json(self, id_=None, name=None, xpos=None, ypos=None):
"""Create JSON-formatted string to neccesary to retrieve storage location via API
Parameters
----------
id: string
Identifier of the Storage Container within OpenSpecimen.
name: string
Storage Container name
xpos: string
x-coordinate within the given storage conatiner grid (if applicable)
ypos: string
y-coordinate within the given storage container grid (if applicable)
Returns
-------
JSON-formated string
JSON foramted string that is needed to retrieve the storage location of a specimen
"""
data = {
"id": id_,
"name": name,
"positionX": xpos,
"positionY": ypos
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def merge_cps(self, src_cp, trg_cp):
"""Create JSON-formatted string to neccesary to retrieve storage location via API
Parameters
----------
src_cp: string
CP id of source collection protocoll for merge
trg_cp: string
CP id of target of merge action
Returns
-------
JSON-formated string
JSON-formated string neccesary for merging to CP's via API
"""
data = {
"srcCpShortTitle": src_cp,
"tgtCpShortTitle": trg_cp
}
return json.dumps(data)
[docs] def create_institute(self, institutename=None, inst_id=None, get_csv = False):
"""Create JSON-formatted string to neccesary to retrieve storage location via API
Parameters
----------
institutename: string
Name of the institute
inst_id:
institute_id needed for updating when creating an institute via csv
get_csv: bool
Return a CSV for creating an institute
Returns
-------
JSON-formated string
JSON-formated string neccesary for creating an institute
"""
data = {
"name": institutename
}
data = {k: v for k, v in data.items() if v is not None}
if get_csv:
data["identifier"] = inst_id
data = {k: v for k, v in data.items() if v is not None}
return pd.DataFrame(data, index=[0]).to_csv(index=False)
else:
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def get_participants(self, lastname = None, uid = None, birthdate = None, pmi = None, empi = None, reqreginfo = None):
"""Create JSON-formatted string to neccesary to retrieve storage location via API
Parameters
---------
lastname : string
Substring of the Lastname of a Paritcipant.
uid : string
Country specific unique social security number.
birthdate : string
The date of registration in the format, which is defined in the System settings.
pmi : dict
Dict with details of the Medical records number mrn and the assigned site with key siteName.
empi : string
Enterprise wide unique ID assigned to the participant.
reqreginfo : string
OpenSpecimen's boolean (true/false). If true it returns details of the participant
Returns
-------
JSON-formated string
JSON-formated string neccesary for creating an institute
"""
data ={
"lastName" : lastname,
"birthDate" : birthdate,
"uid" : uid,
"empi" : empi,
"reqRegInfo" : reqreginfo
}
if pmi==None:
data['pmi']=None
else:
data["pmi"]={
"mrn" : str(pmi[0]),
"siteName" : pmi[1]
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def create_site(self, name = None, institutename = None, type_ = None, coordinators = None, address = None):
"""Create a Site
Create the payload to create a site in OpenSpecimen via API.
Parameters
----------
name : string
String with the name of the site.
institutename : string
String with the name of the institute.
type_ : string
String with the type of the site.
coordinators : string
String with the login-name of the coordinators.
address : string
String with the address of the institute.
site_id: string
String with site id for upadting via csv
Returns
-------
string
JSON-formatted string to create a site such that OpenSpecimen can read it.
"""
data = {
"name" : name,
"instituteName" : institutename,
"coordinators" : coordinators,
"type" : type_,
"address" : address
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def create_user_json(self, first = None, last = None, email = None, phone = None, login = None, institute = None,
type_ = None, address = None, domain = None):
"""Create User
Create the payload for creating a user in OpenSpecimen.
Parameters
----------
first : string
First name of the user.
last : string
Last name of the user.
email : string
Email-address of the user.
phone : string
Phone number of the user.
login : string
Login name of the user.
institute : string
Institute to which the user belongs.
type_ : string
Type of the user.
address : string
Address of the user.
domain : string
Domain where the user belongs to.
Returns
-------
string
JSON formatted string to create an user such that Openspecimen can read it
"""
data = {
"firstName" : first,
"lastName" : last,
"emailAddress" : email,
"phoneNumber" : phone,
"domainName" : domain,
"loginName" : login,
"instituteName" : institute,
"type" : type_,
"address" : address
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def change_user_pw_json(self, userid, newpw, oldpw = None):
"""Change Password
Change the password of an user via API.
Note
-----
If an user changes its own password, the old password is mandatory.
If the user who changes the password is a superadmin, the new password is enough.
Parameters
----------
userid : int
Integer with the unique ID of the user, which password should be changed.
newpw : string
String with the new password, care about the password complexity.
oldpw : string
String with the old password.
Returns
-------
string
JSO-formatted string to change the password, such that OpenSpecimen can read it.
"""
data = {
"userId" : userid,
"oldPassword" : oldpw,
"newPassword" : newpw
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)
[docs] def assign_user_role_json(self, siteid, cpid, role):
"""Assign Role
Assign a Role to a Collection Protocol.
Parameters
----------
siteid : int
Integer with the Unique ID of the site, where the User belongs.
cpid : int
Integer with the unique ID of the Collection protocol.
role : string
String with the name of the Role.
Returns
-------
string
JSON-formatted string such that OpenSpecimen can read it.
"""
data = {
"site":{"id":siteid},
"collectionProtocol":{"id":cpid},
"role":{"name":role}
}
return json.dumps(data)
[docs] def register_to_cp(self, cprid, regdate, cpid, ppid):
"""Register a Registration to another Protocol
Register an existing Registration to another Collection Protocol.
Parameters
----------
cprid : int
Integer with the existing Registration Id of the Participant.
regdate: string
Dete of the Regisrtation to the new protocol.
cpid : int
Integer with the unique ID of the Collection Protocol.
ppid : string
String with the Participant protocol ID.
Returns
-------
string
Json-formatted string for register a Participant to another Protocol such that OpenSpecimen can read it.
"""
data = {
"participant":{"id":cprid},
"registrationDate":regdate,
"cpId":cpid,
"ppid":ppid
}
return json.dumps(data)
[docs] def create_container_json(self, name=None, barcode=None, typename=None, activitystatus=None, sitename=None,
storageloc=None, numcols=None, numrows=None, storespecimens=None, childcontainers=None,
temp=None, columnlabelscheme=None, rowlablescheme=None, comment=None, specimenclasses=None,
specimentypes=None, collectionprotocols=None):
"""Create a storage container in OpenSpecimen
Parameters
----------
name : string
Name of the sorage container
barcode: string
Barcode of the sotrage container
typename : string
Container type (e.g. Freezer see https://openspecimen.atlassian.net/wiki/spaces/CAT/overview
for available types and customization)
activitystatus : string
Whether the container is operational: Active or disabled; Default active
sitename : string
Site location of the container
storageloc : dict
used if conatiner is a child container gives the name, id and position of the child
within the parent container
numcols : string
Number of container columns (if type allows it)
numrows : string
Number of rows (if type allows it)
storespecimens: string
Boolean value (true, false) whether container can hold specimens or not
childcontainers : string
Filled if container sotres multiple other conatiners with different attributes
temp : string
Number literal giving the container temperature in celcius (e.g. -80, -20)
columnlabelscheme : string
Column labels - Numbers by default;
Permissible Values are {Numbers, Alphabets Upper Case, Alphabets Lower Case, Roman Upper Case, Roman Lower Case}
rowlabelscheme : string
Row labels - Numbers by default;
Permissible Values are {Numbers, Alphabets Upper Case, Alphabets Lower Case, Roman Upper Case, Roman Lower Case}
comment : string
Additional comments for the sorage container
specimenclasses : list
Specimen Classes allowed to be stored within the container
specimentypes : list
Specimen Types allowed to be stored within the container
collecitonprotocols : list
List of collection Protocols that can store samples within the given container
Returns
-------
string
Json-formatted string with details needed to create a container.
"""
data = {
"name":name,
"barcode":barcode,
"typeName":typename,
"activityStatus":activitystatus,
"siteName":sitename,
"storageLocation":storageloc,
"noOfColumns":numcols,
"noOfRows":numrows,
"storeSpecimensEnabled":storespecimens,
"childContainers":childcontainers,
"temperature":temp,
"columnLabelingScheme":columnlabelscheme,
"rowLabelingScheme":rowlablescheme,
"comments":comment,
"allowedSpecimenClasses":specimenclasses,
"allowedSpecimenTypes":specimentypes,
"allowedCollectionProtocols":collectionprotocols
}
data = {k: v for k, v in data.items() if v is not None}
return json.dumps(data)