Device Guides
Inheco ODTC
Basic usage of the Inheco ODTC.
This guide is based on the Jupyter Notebook guide which is located in your tenants Jupyter environment under ~/Guides.
Getting started with Inheco ODTC
This jupyter notebook introduces how the Inheco ODTC (prototype) SiLA 2 server is used with the UniteLabs SDK. The ODTC server is provided by Inheco and is in a prototype stage. The server is connected to the UniteLabs platform with by the UniteLabs Edge Gateway. The UniteLabs SDK (extension) provides classes and methods for better usability.
The general concept of the ODTC is stateful and requests must be sent in a defined order (See SiLA Device Control & Data Interface Specification). The ODTC is a PCR thermocycler. It is operated by running defined methods on the device itself. These methods define the temperature profile to be run. These methods are specified in an INheco specific XML-format. Traditionally, these are build using the Inheco Script Editor tool. However, this is not necessary when using the UniteLabs SDK. Respective classes for Methods, Pre-Methods, Steps, and PID controller settings are provided. Additional information regarding these can be found in the SiLA FWCommandSet manual.
Responses from the server are returned in an XML-format. Utility classes and methods in the SDK take care of the conversion.
General Imports
import asyncio
from datetime import datetime, timezone
from unitelabs.sdk import core, client, connect
await connect.init()
This guide is based on the SDK version 0.2.5.
Get an instance of the Inheco connector
connect.__all__
The default name of the connector will include InhecoODTCPrototypeServer.
(
...
'InhecoODTCPrototypeServerLocal',
...
)
There are two ways to get the instance. By importing it directly or by getting it using a specific method.
# Option 1
from unitelabs.sdk.connect import InhecoODTC as InhecoODTC
odtc = InhecoODTC()
# Option 2
odtc = await unitelabs.sdk.connect(name='InhecoODTC')
Explore the connector
A connector consists of modules which consist of actions. Modules group thematically related actions. The ODTC only has a single module, the odtc_sila1 module. The silas_ervice module is a standard module that all SiLA connectors have.
odtc.modules
{'sila_service': SiLAService(client=<unitelabs.sdk.client.client.Client object at 0x7f7d28d9a110>, id='e7d181bb-ee3b-4602-8c49-35cbf0200b4b', name='SiLA Service'),
'temperature_controller': TemperatureController(client=<unitelabs.sdk.client.client.Client object at 0x7f7d28d9a110>, id='b7a458fc-9cd1-4df1-b376-1154269f1354', name='Temperature Controller'),
'door_controller': DoorController(client=<unitelabs.sdk.client.client.Client object at 0x7f7d28d9a110>, id='57956c73-9d15-4f15-bc80-8b5b24b7eecc', name='Door Controller')}
The actions of the module can be viewed with:
odtc.sila_service.actions
{'get_target_temperature': TargetTemperature(client=<unitelabs.sdk.client.client.Client object at 0x7f7d28d9a110>, id='f687108a-edb8-451a-ae59-10b170625132', name='Target Temperature', type='PROPERTY'),
'subscribe_current_temperature': CurrentTemperature(client=<unitelabs.sdk.client.client.Client object at 0x7f7d28d9a110>, id='43ab866f-c63c-457d-a6bb-2fe7356ea3fe', name='Current Temperature', type='SENSOR'),
'set_target_temperature': SetTargetTemperature(client=<unitelabs.sdk.client.client.Client object at 0x7f7d28d9a110>, id='d4c9e751-2c7f-45ac-85ff-301ba20f054a', name='Set Target Temperature', type='CONTROL')}
Action types and parameters
Each action is either a CONTROL, PROPERTY, or SENSOR.
- A PROPERTY just returns a response and doesn't require any parameters.
- A SENSOR returns a subscription of responses and doesn't require parameters.
- A CONTROL requires parameters and can return a single response or a subscription of responses.
Parameters, and response types and constraints can be viewed using the respective attributes:
print(odtc.sila_service.set_server_name.type)
print(odtc.sila_service.set_server_name.parameters)
print(odtc.sila_service.set_server_name.responses)
CONTROL
{'server_name': Parameter(id='ServerName', schema='')}
{}
print(odtc.sila_service.get_server_name.type)
print(odtc.sila_service.get_server_name.parameters)
print(odtc.sila_service.get_server_name.responses)
PROPERTY
{}
{'ServerName': Response(name='server_name', schema={'dataType': {'name': 'Constrained', 'dataType': {'name': 'String'}, 'constraints': [{'name': 'MaximalLength', 'value': '255'}]}, 'identifier': 'ServerName', 'description': "Human readable name of the SiLA Server. The name can be set using the 'Set Server Name' command.", 'displayName': 'Server Name'})}
General connector info
General info about the connector can be retrieved from the SiLA Service feature:
print(await odtc.sila_service.get_server_name())
print(await odtc.sila_service.get_server_vendor_url())
print(await odtc.sila_service.get_server_version())
print(await odtc.sila_service.get_server_uuid())
Inheco ODTC Prototype Server Local
https://www.inheco.com
7.5.3
cb2cdf15-ab7d-4ec1-b7c8-cd0bce40136d
ODTC SiLA 1 module
This is the main module that contains all the functionality provided by the SiLA Server of the vendor.
odtc.odtc_sila1.actions
{'subscribe_last_data_event': LastDataEvent(id='f573563a-fa58-4163-831e-74cca6f796af', name='Last Data Event', type='SENSOR'),
'get_status': GetStatus(id='5b438774-171a-46a3-b1d5-341f3697a040', name='Get Status', type='CONTROL'),
'get_device_identification': GetDeviceIdentification(id='5e262690-c58a-4fb1-b43e-8182b6031e2d', name='Get Device Identification', type='CONTROL'),
'subscribe_last_status_event': LastStatusEvent(id='33a1dc7b-bd0d-4c3d-946b-ea784c7cb100', name='Last Status Event', type='SENSOR'),
'reset': Reset(id='b149b1e1-37c6-4573-897e-9a13c6a5f5b2', name='Reset', type='CONTROL'),
'pause': Pause(id='ffe5f1d1-2126-4c77-b7be-a056a4f350ce', name='Pause', type='CONTROL'),
'initialize': Initialize(id='0320aa35-6bed-476f-909f-15fb97b25772', name='Initialize', type='CONTROL'),
'abort': Abort(id='da4f5b05-3785-4b34-b105-1020c763ddfe', name='Abort', type='CONTROL'),
'docontinue': DoContinue(id='6c981469-dfb7-4f98-b56f-7b8dc6fee67a', name='DoContinue', type='CONTROL'),
'lockdevice': LockDevice(id='c68799ca-81ac-48d4-ab21-65f92e6b7b00', name='LockDevice', type='CONTROL'),
'unlockdevice': UnlockDevice(id='a007c168-f101-469d-9835-c4e42132131f', name='UnlockDevice', type='CONTROL'),
'execute_method': ExecuteMethod(id='0fe4b10d-aca6-4acd-b95b-8f89f7ff0055', name='Execute Method', type='CONTROL'),
'stop_method': StopMethod(id='e9959426-ebed-47ea-8a9a-9ce05e4bd305', name='Stop Method', type='CONTROL'),
'read_actual_temperature': ReadActualTemperature(id='7a48b663-e587-4459-a425-6fe5dc74e8c1', name='Read Actual Temperature', type='CONTROL'),
'get_last_data': GetLastData(id='0e3aeb40-58b3-47f0-9013-a0fbf7163e27', name='Get Last Data', type='CONTROL'),
'open_door': OpenDoor(id='91166b68-f87f-402e-8e4f-f1990dad6341', name='Open Door', type='CONTROL'),
'close_door': CloseDoor(id='82fecaa5-1750-414c-b95b-2975c84e470e', name='Close Door', type='CONTROL'),
'get_data_events': GetDataEvents(id='22c23e31-403f-4a38-8cf6-1b8448a55920', name='Get Data Events', type='CONTROL'),
'get_parameters': GetParameters(id='4a811894-63bb-4879-9318-419482f0d044', name='Get Parameters', type='CONTROL'),
'set_parameters': SetParameters(id='5e417e53-c3a0-4484-a6e4-debce442211d', name='Set Parameters', type='CONTROL'),
'get_configuration': GetConfiguration(id='55330b85-cf8b-4a0c-b25e-db3586517405', name='Get Configuration', type='CONTROL'),
'set_configuration': SetConfiguration(id='ce8eb3fb-3dcf-4c57-905a-d90887f3f37e', name='Set Configuration', type='CONTROL')}
Get the current status and device information
Using the get_status and the get_device_identification controls information on the connected device can be accessed.
The "required" parameters "request_id" and "lock_id" are not needed, but random integers (request_id) and strings
(lock_id) must be passed. They are a relict from SiLA1 and will be removed in subsequent updates.
print(odtc.odtc_sila1.get_status.parameters)
status = await odtc.odtc_sila1.get_status(request_id=1)
print(status[0]['ReturnValue']['Message']['value'])
print(status[1]['Status']['State']['value'])
{'request_id': Parameter(id='RequestId', schema='')}
Success.
Idle
This command returns specific device info including the serial number and node name. This information should be the same as on the labels on the device and its power unit.
device_info = await odtc.odtc_sila1.get_device_identification(request_id=1, lock_id='a')
print(device_info[0]['ReturnValue']['Message']['value'])
print(device_info[1]['DeviceDescriptionValue']['DeviceName']['value'])
device_info
Success.
ODTC_1A3C00
({'ReturnValue': {'ReturnCode': {'value': '1'},
'Message': {'value': 'Success.'},
'Duration': {'value': 'PT1S'},
'DeviceClass': {'value': '30'}}},
{'DeviceDescriptionValue': {'Wsdl': {'value': 'http://10.97.22.128/odtc.wsdl'},
'SiLAInterfaceVersion': {'value': '1.2.01'},
'SiLADeviceClass': {'value': '30'},
'SiLADeviceClassVersion': {'value': '1.0'},
'DeviceManufacturer': {'value': 'http://www.inheco.com'},
'DeviceName': {'value': 'ODTC_1A3C00'},
'DeviceSerialNumber': {'value': 'ArticleNo PCU: 8900035, SerialNo PCU: 6748, SerialNo PCU PCB: 605, ArticleNo ODTC: 8100001, Type ODTC: 960000, SerialNo ODTC: 6707, SerialNo ODTC PCB: 6676, SerialNo ODTC Lid: 7319, SerialNo VCM: 01301341'},
'DeviceFirmwareVersion': {'value': 'BUILD:7783.22980,DEVICE:ODTC_277_BOOT_001_HW_00003'}}})
Using the device
Most functions of the ODTC can only be used when the device is in a certain state. Before using it, it needs to be reset using the "reset" method and initialized using the "initialize" method.
Risks and safety
The methods "open_door" and "close_door" result in a movement of the ODTC lid. Make sure the lid is not obstructed in
any way to avoid damage to the device or injury to a person! When interacting with the ODTC with another device, i.e.
placing a plate into the ODTC, make sure that the transfer runs smoothly and positions used are accurate. Use error
handling to avoid potential collisions and read out the state of the device to ensure operations have executed as
intended!
When running methods using the "execute_method" method that alter the temperature of the device and its various parts,
extremely high temperatures can be reached. Make sure that you check the device temeprature before touching the device
or before placing heat-sensitive objects into the device.
Always ensure the device is not left heating, when unused. Read the hardware manual provided by the manufacturer. All
risk and safety warnings in this manual apply and must be read carefully.
When sending commands to the device connector, always make sure that the transmission was successful by having adequate
error handling in place. Make sure that if your automation workflow crashes, a safe state is entered!
All units are metric and Celsius.
Resetting the device to get started (Step 1).
assert await odtc.odtc_sila1.reset(simulation_mode=False)
Initializing the device to get started (Step 2).
assert await odtc.odtc_sila1.initialize()
Getting the current temperature data
raw_data = await odtc.odtc_sila1.read_actual_temperature()
raw_data
{'ResponseValue': {'RequestId': {'value': '7'},
'CommandId': {'value': 'ReadActualTemperature'},
'SiLAReturnValue': {'ReturnValue': {'ReturnCode': {'value': '3'},
'Message': {'value': 'Asynchronous command has finished.'},
'Duration': {'value': 'PT1S'},
'DeviceClass': {'value': '30'}}},
'ResponseData': {'value': '<?xml version="1.0" encoding="utf-8"?><ResponseData xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="ResponseType_1.2.xsd"><Parameter><String><SensorValues timestamp="2024-02-02T14:29:31Z"><Mount>2207</Mount><Mount_Monitor>2207</Mount_Monitor><Lid>2261</Lid><Lid_Monitor>2243</Lid_Monitor><Ambient>2200</Ambient><PCB>2619</PCB><Heatsink>2212</Heatsink><Heatsink_TEC>2241</Heatsink_TEC></SensorValues></String></Parameter></ResponseData>'},
'Success': {'value': True},
'ErrorMessage': {},
'Aborted': {},
'ExpectedDuration': {'value': '1'},
'SynchronousDuration': {'value': '18'},
'ResponseDuration': {}}}
The vendor interfaces returns the temperature data in an XML-format. The extract_current_temperature_data utility
method converts that into a dictionary.
from unitelabs.incubation.odtc import extract_current_temperature_data
temperature_data = extract_current_temperature_data(raw_data=raw_data['ResponseValue']['ResponseData']['value'])
temperature_data
{'timestamp': '2024-02-02T14:29:44Z',
'Mount': 22.07,
'Mount_Monitor': 22.07,
'Lid': 22.61,
'Lid_Monitor': 22.42,
'Ambient': 22.0,
'PCB': 26.19,
'Heatsink': 22.12,
'Heatsink_TEC': 22.41}
Subscribing the last experiment data
Similar to the current temperature data, the extract_last_experiment_data method extracts the experiment data from the
XML-format.
from unitelabs.incubation.odtc import extract_last_experiment_data
temp_data = await odtc.odtc_sila1.get_last_data(request_id=4, lock_id='f')
experiment_data = extract_last_experiment_data(temp_data['ResponseValue']['ResponseData']['value']);
experiment_data;
Get last data event as a stream
This data can also be subscribed to. In this example we justpick first element and convert the XML format it into a
DataEvent object.
from unitelabs.sdk.incubation.inheco import DataEvent
await client.Client().aclose()
async with client.Client() as c:
dev = await connect(name='Inheco ODTC Prototype Server')
action = odtc.modules[f"odtc_sila1"].subscribe_last_data_event
parameters = {}
subscription = await c.create_subscription(action_id=action.id, parameters=action._parse_parameters(parameters))
async with subscription:
for i in range(0, 2, 1):
event, value = await subscription.__anext__()
data = DataEvent(value['LastDataEvent']['DataEvent']['DataValue']['value'])
odtc = await unitelabs.connect(name='Inheco ODTC Prototype Server')
print(data.container_data.experiment_steps[0].data_series_list)
Opening and closing the door
To transfer plates into and out of the thermocycler with a robotic arm or grippers, the door can be opened and closed.
This control is observable and will block for the duration of the execution if the await statement is used. Using
asyncio.run() can be used to circumvent this behaviour.
# response = await odtc.odtc_sila1.open_door()
asyncio.run(odtc.odtc_sila1.open_door())
# print(response)
The response contains information on the door opening process, like execution time in seconds, and the final state of the door.
{'ResponseValue': {'RequestId': {'value': '1'},
'CommandId': {'value': 'OpenDoor'},
'SiLAReturnValue': {'ReturnValue': {'ReturnCode': {'value': '3'},
'Message': {'value': 'Asynchronous command has finished.'},
'Duration': {'value': 'PT9S'},
'DeviceClass': {'value': '30'}}},
'ResponseData': {'value': '<?xml version="1.0" encoding="utf-8"?><ResponseData xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="ResponseType_1.2.xsd"/>'},
'Success': {'value': True},
'ErrorMessage': {},
'Aborted': {},
'ExpectedDuration': {'value': '14'},
'SynchronousDuration': {'value': '30'},
'ResponseDuration': {'value': '8'}}}
The execution behaviour and response information is provided by the close_door control respectively.
asyncio.run(odtc.odtc_sila1.close_door())
{'ResponseValue': {'RequestId': {'value': '1'},
'CommandId': {'value': 'CloseDoor'},
'SiLAReturnValue': {'ReturnValue': {'ReturnCode': {'value': '3'},
'Message': {'value': 'Asynchronous command has finished.'},
'Duration': {'value': 'PT8S'},
'DeviceClass': {'value': '30'}}},
'ResponseData': {'value': '<?xml version="1.0" encoding="utf-8"?><ResponseData xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="ResponseType_1.2.xsd"/>'},
'Success': {'value': True},
'ErrorMessage': {},
'Aborted': {},
'ExpectedDuration': {'value': '14'},
'SynchronousDuration': {'value': '28'},
'ResponseDuration': {'value': '7'}}}
Creating an object for an ODTC method
Either create an object by loading an XML from file or build it using the provided classes.
# Loading a mehtod XML-file
odtc_method = ParameterSet()
odtc_method.load_parameter_set_from_xml_file(path='./example_methods/sample_parameterset.xml')
print(odtc_method.method_set[0])
print(odtc_method.method_set[0].steps[0])
print(odtc_method.method_set[0].pid_set.pids[0].p_cooling)
Method: test, Creator: mbw, Date Time: 2022-12-13T13:57:03.8452497+08:00, Variant: 960000, Plate type: 0, Fluid quantity: 1, Post heating: True, Start block temperature: 4, Start lid temperature: 110, Steps: 9, PID set: PIDs: [PID(number=1, p_heating=60, p_cooling=80, i_heating=250, i_cooling=100, d_heating=10, d_cooling=10, p_lid=100, i_lid=70)]
Step 1 - Slope: 3.0, Plateau Temperature: 95
80
Creating a new method from scratch
New methods can be created with the classes provided by the SDK.
from unitelabs.sdk.incubation.inheco import ParameterSet, Method, PreMethod, Step, PIDSet, PID
print(datetime.now().astimezone(timezone.utc).isoformat())
# Creating a new one
from datetime import datetime
odtc_method = ParameterSet(delete_all_methods=False)
pre_method = PreMethod(
method_name="pre-method",
creator="LB",
description="A pre-method for testing",
date_time=datetime.now().astimezone(timezone.utc).isoformat(),
target_block_temperature=60,
target_lid_temp=80,
)
pid_controller = PID(
number=1,
p_heating=60,
p_cooling=80,
i_heating=250,
i_cooling=100,
d_heating=10,
d_cooling=10,
p_lid=100,
i_lid=70
)
step_1 = Step(
number=1,
slope=3,
plateau_temperature=95,
plateau_time=180,
overshoot_slope1=3,
overshoot_temperature=7,
overshoot_time=5.6,
overshoot_slope2=2.2,
goto_number=0,
loop_number=0,
pid_number=1,
lid_temp=110
)
test_method = Method(
method_name='my_first_method',
creator='unitelabs',
date_time='2024-01-29T07:46:01.1', #"2022-12-13T13:57:03.8452497+08:00", # datetime.now().astimezone(timezone.utc).isoformat(),
variant=960001,
plate_type=0,
fluid_quantity=50,
post_heating=True,
start_block_temperature=60,
start_lid_temperature=80,
steps=[step_1],
pid_set=PIDSet(pid_controller)
)
odtc_method.delete_all_methods = True
odtc_method.pre_method = pre_method
odtc_method.method_set=[test_method]
2024-02-02T13:32:46.955066+00:00
method_xml = odtc_method.export_parameter_set_xml()
method_xml
'<ParameterSet><Parameter name="MethodsXML"><String><?xml version="1.0" encoding="utf-8"?><MethodSet><DeleteAllMethods>true</DeleteAllMethods><PreMethod methodName="pre-method" creator="LB" description="A pre-method for testing" dateTime="2024-02-02T13:32:46.955673+00:00"><TargetBlockTemperature>60</TargetBlockTemperature><TargetLidTemp>80</TargetLidTemp></PreMethod><Method methodName="my_first_method" creator="unitelabs" dateTime="2024-01-29T07:46:01.1"><Variant>960001</Variant><PlateType>0</PlateType><FluidQuantity>50</FluidQuantity><PostHeating>true</PostHeating><StartBlockTemperature>60</StartBlockTemperature><StartLidTemperature>80</StartLidTemperature><Step><Number>1</Number><Slope>3</Slope><PlateauTemperature>95</PlateauTemperature><PlateauTime>180</PlateauTime><OverShootSlope1>3</OverShootSlope1><OverShootTemperature>7</OverShootTemperature><OverShootTime>5.6</OverShootTime><OverShootSlope2>2.2</OverShootSlope2><GotoNumber>0</GotoNumber><LoopNumber>0</LoopNumber><PIDNumber>1</PIDNumber><LidTemp>110</LidTemp></Step><PIDSet><PID number="1"><PHeating>60</PHeating><PCooling>80</PCooling><IHeating>250</IHeating><ICooling>100</ICooling><DHeating>10</DHeating><DCooling>10</DCooling><PLid>100</PLid><ILid>70</ILid></PID></PIDSet></Method></MethodSet></String></Parameter></ParameterSet>'
Storing the method on the ODTC
Using the SetParameter method
# Save a new method
# print(odtc.odtc_sila1.set_parameters.parameters)
device_parameters = await odtc.odtc_sila1.set_parameters(paramsxml=method_xml)
print(device_parameters)
{'ResponseValue': {'RequestId': {'value': '1'}, 'CommandId': {'value': 'SetParameters'}, 'SiLAReturnValue': {'ReturnValue': {'ReturnCode': {'value': '3'}, 'Message': {'value': 'Asynchronous command has finished.'}, 'Duration': {'value': 'PT1S'}, 'DeviceClass': {'value': '30'}}}, 'ResponseData': {'value': '<?xml version="1.0" encoding="utf-8"?><ResponseData xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="ResponseType_1.2.xsd"/>'}, 'Success': {'value': True}, 'ErrorMessage': {}, 'Aborted': {}, 'ExpectedDuration': {'value': '180'}, 'SynchronousDuration': {'value': '294'}, 'ResponseDuration': {}}}
Executing a method
This is a blocking call. If you want to keep going, use asyncio.run() or another way of parallelization.
# Execute a method: Pre-Method must be executed first!
# print(odtc.odtc_sila1.execute_method.parameters)
response = await odtc.odtc_sila1.execute_method(method_name='pre-method', priority=1)
print(response)
{'ResponseValue': {'RequestId': {'value': '1'}, 'CommandId': {'value': 'ExecuteMethod'}, 'ResponseData': {}, 'Success': {'value': True}, 'ErrorMessage': {}, 'Aborted': {'value': True}, 'ExpectedDuration': {'value': '251'}, 'SynchronousDuration': {'value': '61'}, 'ResponseDuration': {'value': '31'}}}
A method requires the initialization step, i.e. the pre-method. Executing a method before the pre-method has finished is possible. The block and lid temperature of the pre-method must be equal to the start temperature of the lid and block of the method.
# Execute a method
# print(odtc.odtc_sila1.execute_method.parameters)
response = await odtc.odtc_sila1.execute_method(method_name='my_first_method', priority=1)
print(response)
Automate data acquisition and method execution in one experimental workflow
import time, traceback
async def record_current_temperature(interval, duration: float = None):
t_ls = []
temp_block_ls = []
temp_lid_ls = []
i = 0
async def get_latest_current_temperature(i, interval):
nonlocal t_ls, temp_block_ls, temp_lid_ls
try:
temp_data = await odtc.odtc_sila1.read_actual_temperature()
raw_data = extract_current_temperature_data(raw_data=temp_data['ResponseValue']['ResponseData']['value'])
t_ls.append(i+interval)
temp_block_ls.append(raw_data['Mount'])
temp_lid_ls.append(raw_data['Lid'])
print(f'Block temperature: {temp_block_ls}')
print(f'Lid temperature: {temp_lid_ls}')
except Exception as e:
print('Error')
await asyncio.sleep(interval)
while True:
if duration is not None and i*interval > duration:
break
await get_latest_current_temperature(i, interval)
i=i+1
async def long_poll_status(interval, duration: float = None):
i = 0
while True:
if duration is not None and i*interval > duration:
break
status = await odtc.odtc_sila1.get_status()
# print(status)
print(f"Status: {status[1]['Status']['State']['value']}")
await asyncio.sleep(interval)
i=i+1
async def main():
status = await odtc.odtc_sila1.get_status()
print(status)
try:
# Start acquiring status updates
task_1 = asyncio.create_task(
long_poll_status(duration = None, interval=5)
)
# Start acquiring temperature updates
task_2 = asyncio.create_task(
record_current_temperature(duration = None, interval=5)
)
# Random wait
await asyncio.sleep(20)
# Start pre-method
print('Starting pre-method')
resp_exec_method = await odtc.odtc_sila1.execute_method(request_id=1, lock_id='a', method_name='pre-method', priority=1)
print(resp_exec_method)
except (asyncio.CancelledError, KeyboardInterrupt) as e:
raise e
except Exception as e:
raise e
finally:
# Stop the data and status acquisition tasks
task_1.cancel()
task_2.cancel()
# Abort the current method
await odtc.odtc_sila1.abort()
await asyncio.sleep(1)
# Reset the device
await odtc.odtc_sila1.reset()
await asyncio.sleep(1)
await odtc.odtc_sila1.initialize()
await asyncio.sleep(1)
# Potentially add checks that state is ok
status = await odtc.odtc_sila1.get_status()
print(status)
try:
asyncio.run(main())
except (asyncio.CancelledError, KeyboardInterrupt):
print("Run interupped by user.")
except Exception as e:
print(e)
print(type(error))
print(traceback.format_exc())
({'ReturnValue': {'ReturnCode': {'value': '1'}, 'Message': {'value': 'Success.'}, 'Duration': {'value': 'PT1S'}, 'DeviceClass': {'value': '30'}}}, {'Status': {'DeviceId': {}, 'State': {'value': 'Idle'}, 'Locked': {}, 'PMSId': {}, 'CurrentTime': {'second': 50, 'minute': 53, 'hour': 9, 'day': 2, 'month': 2, 'year': 2024}}})
Status: Idle
Block temperature: [22.96]
Lid temperature: [35.14]
Status: Idle
...
Block temperature: [22.96, 22.82, 22.71, 22.62, 24.0, 25.88, 27.71, 29.52]
Lid temperature: [35.14, 34.94, 34.76, 34.57, 38.16, 41.27, 44.02, 46.6]
Status: Busy
{'ResponseValue': {'RequestId': {'value': '1'}, 'CommandId': {'value': 'ExecuteMethod'}, 'ResponseData': {}, 'Success': {'value': True}, 'ErrorMessage': {}, 'Aborted': {'value': True}, 'ExpectedDuration': {'value': '251'}, 'SynchronousDuration': {'value': '22'}, 'ResponseDuration': {'value': '27'}}}
({'ReturnValue': {'ReturnCode': {'value': '1'}, 'Message': {'value': 'Success.'}, 'Duration': {'value': 'PT1S'}, 'DeviceClass': {'value': '30'}}}, {'Status': {'DeviceId': {}, 'State': {'value': 'Idle'}, 'Locked': {}, 'PMSId': {}, 'CurrentTime': {'second': 42, 'minute': 54, 'hour': 9, 'day': 2, 'month': 2, 'year': 2024}}})
async def subscribe_data():
#async with client.Client() as c:
# dev = await connect(name='Inheco ODTC Prototype Server')
action = odtc.modules[f"odtc_sila1"].subscribe_last_data_event
parameters = {}
subscription = await c.create_subscription(action_id=action.id, parameters=action._parse_parameters(parameters))
async with subscription:
for i in range(0, 20, 1):
event, value = await subscription.__anext__()
data = DataEvent(value['LastDataEvent']['DataEvent']['DataValue']['value'])
print(data.container_data.experiment_steps[0].name)
print(data.container_data.experiment_steps[0].data_series_list[1])