180 lines
4.8 KiB
Python
180 lines
4.8 KiB
Python
|
import requests
|
||
|
from dataclasses import dataclass, asdict, fields
|
||
|
from typing import Optional, Iterator
|
||
|
|
||
|
|
||
|
DEFAULT_URL = "http://localhost:8444"
|
||
|
|
||
|
|
||
|
@dataclass(frozen=True)
|
||
|
class TemplateSet:
|
||
|
TemplateId: int
|
||
|
BatchCount: int
|
||
|
|
||
|
|
||
|
@dataclass(frozen=True)
|
||
|
class OrderRecipe:
|
||
|
Description: str
|
||
|
User: str
|
||
|
Template: TemplateSet
|
||
|
|
||
|
|
||
|
@dataclass(frozen=True)
|
||
|
class OrderSample:
|
||
|
Id: int
|
||
|
OrderId: int
|
||
|
PlateProtocolId: int
|
||
|
Barcode: str
|
||
|
|
||
|
|
||
|
class DataclassFromDict:
|
||
|
@classmethod
|
||
|
def from_dict(cls, env):
|
||
|
return cls(**{
|
||
|
k: v for k, v in env.items()
|
||
|
if k in [x.name for x in fields(cls)]
|
||
|
})
|
||
|
|
||
|
|
||
|
@dataclass(frozen=True)
|
||
|
class Order(DataclassFromDict):
|
||
|
Id: int
|
||
|
State: str
|
||
|
ProtocolId: str
|
||
|
ProtocolName: str
|
||
|
StartTime: str
|
||
|
EndTime: str
|
||
|
OrderSamples: [OrderSample]
|
||
|
|
||
|
|
||
|
@dataclass(frozen=True)
|
||
|
class OrderParameter(DataclassFromDict):
|
||
|
ProtocolParameterId: int
|
||
|
RunOrderId: int
|
||
|
OrderSampleId: int
|
||
|
ParameterValue: str
|
||
|
ParameterName: str
|
||
|
Id: Optional[int] = None
|
||
|
ParameterLevel: Optional[str] = None
|
||
|
ParameterType: Optional[str] = None
|
||
|
|
||
|
|
||
|
@dataclass(frozen=True)
|
||
|
class ProtocolParameter(DataclassFromDict):
|
||
|
Id: int
|
||
|
ProtocolId: int
|
||
|
ParameterName: str
|
||
|
Name: str
|
||
|
DefaultValue: str
|
||
|
ParameterLevel: str
|
||
|
ParameterType: str
|
||
|
|
||
|
|
||
|
def serialize(obj) -> str:
|
||
|
return asdict(obj, dict_factory=lambda x:
|
||
|
{k: v for (k, v) in x if v is not None})
|
||
|
|
||
|
|
||
|
def deserialize(obj: DataclassFromDict, json):
|
||
|
return obj.from_dict(json)
|
||
|
|
||
|
|
||
|
def create_order(baseUrl: str,
|
||
|
templateSet: TemplateSet,
|
||
|
User: str = "EAllison",
|
||
|
Description: str = "Order generated by script") -> Order:
|
||
|
recipe = OrderRecipe(Description=Description,
|
||
|
User=User,
|
||
|
Template=templateSet)
|
||
|
recipe_json = serialize(recipe)
|
||
|
print(recipe_json)
|
||
|
req = requests.post(f"{baseUrl}/orders",
|
||
|
json=recipe_json)
|
||
|
|
||
|
if req.status_code == 200:
|
||
|
req_json = req.json()
|
||
|
order = deserialize(Order, req_json)
|
||
|
return order
|
||
|
else:
|
||
|
print(req)
|
||
|
return None
|
||
|
|
||
|
|
||
|
def get_order_parameters(baseUrl: str,
|
||
|
order: Order) -> [OrderParameter]:
|
||
|
url = f"{baseUrl}/orders/{order.Id}/parameters"
|
||
|
print(url)
|
||
|
req = requests.get(url)
|
||
|
|
||
|
if req.status_code == 200:
|
||
|
res_json = req.json()
|
||
|
return res_json
|
||
|
|
||
|
|
||
|
def get_protocol_parameters(baseUrl: str,
|
||
|
protocolId: int) -> [ProtocolParameter]:
|
||
|
url = f"{baseUrl}/protocols/{protocolId}/parameters"
|
||
|
req = requests.get(url)
|
||
|
|
||
|
if req.status_code == 200:
|
||
|
res_json = req.json()
|
||
|
if res_json != []:
|
||
|
return [deserialize(ProtocolParameter, x)
|
||
|
for x in res_json]
|
||
|
return None
|
||
|
|
||
|
|
||
|
def set_sample_parameter(baseUrl: str,
|
||
|
order: Order,
|
||
|
parameter: ProtocolParameter,
|
||
|
orderSampleId: int,
|
||
|
value):
|
||
|
newParameter = OrderParameter(RunOrderId=order.Id,
|
||
|
ProtocolParameterId=parameter.Id,
|
||
|
ParameterName=parameter.ParameterName,
|
||
|
ParameterValue=value,
|
||
|
OrderSampleId=orderSampleId
|
||
|
)
|
||
|
|
||
|
url = f"{baseUrl}/orders/{order.Id}/parameters"
|
||
|
req_json = [serialize(newParameter)]
|
||
|
print(url)
|
||
|
print(req_json)
|
||
|
req = requests.post(url, json=req_json)
|
||
|
|
||
|
if req.status_code == 200:
|
||
|
return req.json()
|
||
|
if req.status_code == 400:
|
||
|
if len(res_json := req.json()) == 1 and 'Message' in res_json:
|
||
|
if 'already been created' in res_json['Message']:
|
||
|
return True
|
||
|
|
||
|
|
||
|
def get_order_samples(order: Order) -> Iterator[int]:
|
||
|
return (sample['Id'] for sample in order.OrderSamples)
|
||
|
|
||
|
|
||
|
def create_order_with_parameters(baseUrl: str,
|
||
|
templateId: int,
|
||
|
parameterName: str,
|
||
|
parameterValues: [str],
|
||
|
User: str = "EAllison",
|
||
|
Description: str = "") -> Order:
|
||
|
templateSet = TemplateSet(templateId, len(parameterValues))
|
||
|
order = create_order(baseUrl, templateSet,
|
||
|
User, Description)
|
||
|
print(order)
|
||
|
|
||
|
parameter = [x for x in
|
||
|
get_protocol_parameters(baseUrl, order.ProtocolId)
|
||
|
if x.ParameterName == parameterName][0]
|
||
|
|
||
|
order_samples = get_order_samples(order)
|
||
|
|
||
|
for (i, sampleId) in enumerate(order_samples):
|
||
|
set_sample_parameter(baseUrl,
|
||
|
order,
|
||
|
parameter,
|
||
|
sampleId,
|
||
|
parameterValues[i])
|