services client in python

verso_spelling/services.py

@@ -0,0 +1,179 @@
+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])

verso_spelling/verso_spell_via_script.py

@@ -1,17 +1,16 @@
-#Cellario modules
+# Cellario modules
 from HRB.Cellario.Scripting import *
 from HRB.Cellario.Scripting.API import *
 import json
-from sys import argv
 from os import path, listdir
 from dataclasses import dataclass
 
 
-def Execute(api : PythonScriptingApi):
+def Execute(api: PythonScriptingApi):
     available_barcodes = "C:/temp/availabletubes.txt"
     letter_dir = "C:/temp/letters"
     input = "EH"
-    
+
     available_tubes = read_available(available_barcodes)
     letters = []
     for letter in input:
@@ -23,10 +22,11 @@ def Execute(api : PythonScriptingApi):
         letters += x
 
     output = letter_tubes_to_obj(letters)
-    
+
     api.Messaging.WriteDiagnostic(ScriptLogLevel.Normal, json.dumps(output))
-    list(api.CurrentRun.RunOrderParameters)[0].ParameterValue = json.dumps(output)
-    
+    list(api.CurrentRun.RunOrderParameters)[
+        0].ParameterValue = json.dumps(output)
+
 
 @dataclass(frozen=True)
 class Tube:
@@ -85,4 +85,4 @@ def validate_word(input: str, letter_dir: str) -> bool:
 def get_letter_positions(letter: str, letter_dir: str):
     filename = [f for f in listdir(letter_dir) if f[0] == letter][0]
     with open(path.join(letter_dir, filename), 'r') as file:
-        return [int(x.strip()) for x in file.readlines()]
+        return [int(x.strip()) for x in file.readlines()]