Skip to content

Commit

Permalink
feat(api): define & execute v3 json protocols (#3312)
Browse files Browse the repository at this point in the history 
* define v3 json protocol schema
* support v3 JSON protocol execution in APIv2 executor

Closes #3110
  • Loading branch information
@IanLondon
IanLondon authored Apr 9, 2019
1 parent 644ff35 commit 988407d
Show file tree
Hide file tree
Showing 11 changed files with 1,445 additions and 462 deletions.
272 changes: 42 additions & 230 deletions api/src/opentrons/protocol_api/execute.py
View file
Original file line number Diff line number Diff line change
@@ -1,14 +1,11 @@
import inspect
import itertools
import logging
import traceback
import sys
from typing import Any, Callable, Dict, Optional
from typing import Any, Callable, Dict

from .contexts import ProtocolContext, InstrumentContext
from .back_compat import BCLabware
from . import labware
from opentrons.types import Point, Location
from .contexts import ProtocolContext
from . import execute_v1, execute_v3
from opentrons import config

MODULE_LOG = logging.getLogger(__name__)
Expand Down Expand Up @@ -111,227 +108,27 @@ def _run_python(proto: Any, context: ProtocolContext):
raise ExceptionInProtocolError(e, tb, str(e), frame.lineno)


def load_pipettes_from_json(
ctx: ProtocolContext,
protocol: Dict[Any, Any]) -> Dict[str, InstrumentContext]:
pipette_data = protocol.get('pipettes', {})
pipettes_by_id = {}
for pipette_id, props in pipette_data.items():
model = props.get('model')
mount = props.get('mount')

# TODO: Ian 2018-11-06 remove this fallback to 'model' when
# backwards-compatability for JSON protocols with versioned
# pipettes is dropped (next JSON protocol schema major bump)
name = props.get('name')
if not name:
name = model.split('_v')[0]

instr = ctx.load_instrument(name, mount)

pipettes_by_id[pipette_id] = instr

return pipettes_by_id


def load_labware_from_json(
ctx: ProtocolContext,
protocol: Dict[Any, Any]) -> Dict[str, labware.Labware]:
data = protocol.get('labware', {})
loaded_labware = {}
bc = BCLabware(ctx)
for labware_id, props in data.items():
slot = props.get('slot')
model = props.get('model')
if slot == '12':
if model == 'fixed-trash':
# pass in the pre-existing fixed-trash
loaded_labware[labware_id] = ctx.fixed_trash
else:
raise RuntimeError(
"Nothing but the fixed trash may be loaded in slot 12; "
"this protocol attempts to load a {} there."
.format(model))
else:
loaded_labware[labware_id] = bc.load(
model, slot, label=props.get('display-name'))

return loaded_labware


def _get_well(loaded_labware: Dict[str, labware.Labware],
params: Dict[str, Any]):
labwareId = params['labware']
well = params['well']
plate = loaded_labware.get(labwareId)
if not plate:
raise ValueError(
'Command tried to use labware "{}", but that ID does not exist '
'in protocol\'s "labware" section'.format(labwareId))
return plate.wells_by_index()[well]


def _get_bottom_offset(command_type: str,
params: Dict[str, Any],
default_values: Dict[str, float]) -> Optional[float]:
# default offset from bottom for aspirate/dispense commands
offset_default = default_values.get(
'{}-mm-from-bottom'.format(command_type))

# optional command-specific value, fallback to default
offset_from_bottom = params.get(
'offsetFromBottomMm', offset_default)

return offset_from_bottom


def _get_location_with_offset(loaded_labware: Dict[str, labware.Labware],
command_type: str,
params: Dict[str, Any],
default_values: Dict[str, float]) -> Location:
well = _get_well(loaded_labware, params)

# Never move to the bottom of the fixed trash
if 'fixedTrash' in labware.quirks_from_any_parent(well):
return well.top()

offset_from_bottom = _get_bottom_offset(
command_type, params, default_values)

bot = well.bottom()
if offset_from_bottom:
with_offs = bot.move(Point(z=offset_from_bottom))
else:
with_offs = bot
MODULE_LOG.debug("offset from bottom for {}: {}->{}"
.format(command_type, bot, with_offs))
return with_offs


# TODO (Ian 2018-08-22) once Pipette has more sensible way of managing
# flow rate value (eg as an argument in aspirate/dispense fns), remove this
def _set_flow_rate(
pipette_name, pipette, command_type, params, default_values):
"""
Set flow rate in uL/mm, to value obtained from command's params,
or if unspecified in command params, then from protocol's "default-values".
"""
default_aspirate = default_values.get(
'aspirate-flow-rate', {}).get(pipette_name)

default_dispense = default_values.get(
'dispense-flow-rate', {}).get(pipette_name)

flow_rate_param = params.get('flow-rate')

if flow_rate_param is not None:
if command_type == 'aspirate':
pipette.flow_rate = {
'aspirate': flow_rate_param,
'dispense': default_dispense
}
return
if command_type == 'dispense':
pipette.flow_rate = {
'aspirate': default_aspirate,
'dispense': flow_rate_param
}
return

pipette.flow_rate = {
'aspirate': default_aspirate,
'dispense': default_dispense
}


def dispatch_json(context: ProtocolContext, # noqa(C901)
protocol_data: Dict[Any, Any],
instruments: Dict[str, InstrumentContext],
labware: Dict[str, labware.Labware]):
subprocedures = [
p.get('subprocedure', [])
for p in protocol_data.get('procedure', [])]

default_values = protocol_data.get('default-values', {})
flat_subs = itertools.chain.from_iterable(subprocedures)

for command_item in flat_subs:
command_type = command_item.get('command')
params = command_item.get('params', {})
pipette = instruments.get(params.get('pipette'))
protocol_pipette_data = protocol_data\
.get('pipettes', {})\
.get(params.get('pipette'), {})
pipette_name = protocol_pipette_data.get('name')

if (not pipette_name):
# TODO: Ian 2018-11-06 remove this fallback to 'model' when
# backwards-compatability for JSON protocols with versioned
# pipettes is dropped (next JSON protocol schema major bump)
pipette_name = protocol_pipette_data.get('model')

if command_type == 'delay':
wait = params.get('wait')
if wait is None:
raise ValueError('Delay cannot be null')
elif wait is True:
message = params.get('message', 'Pausing until user resumes')
context.pause(msg=message)
else:
context.delay(seconds=wait)

elif command_type == 'blowout':
well = _get_well(labware, params)
pipette.blow_out(well) # type: ignore

elif command_type == 'pick-up-tip':
well = _get_well(labware, params)
pipette.pick_up_tip(well) # type: ignore

elif command_type == 'drop-tip':
well = _get_well(labware, params)
pipette.drop_tip(well) # type: ignore

elif command_type == 'aspirate':
location = _get_location_with_offset(
labware, 'aspirate', params, default_values)
volume = params['volume']
_set_flow_rate(
pipette_name, pipette, command_type, params, default_values)
pipette.aspirate(volume, location) # type: ignore

elif command_type == 'dispense':
location = _get_location_with_offset(
labware, 'dispense', params, default_values)
volume = params['volume']
_set_flow_rate(
pipette_name, pipette, command_type, params, default_values)
pipette.dispense(volume, location) # type: ignore

elif command_type == 'touch-tip':
well = _get_well(labware, params)
offset = default_values.get('touch-tip-mm-from-top', -1)
pipette.touch_tip(location, v_offset=offset) # type: ignore

elif command_type == 'move-to-slot':
slot = params.get('slot')
if slot not in [str(s+1) for s in range(12)]:
raise ValueError('Invalid "slot" for "move-to-slot": {}'
.format(slot))
slot_obj = context.deck.position_for(slot)

offset = params.get('offset', {})
offsetPoint = Point(
offset.get('x', 0),
offset.get('y', 0),
offset.get('z', 0))

pipette.move_to( # type: ignore
slot_obj.move(offsetPoint),
force_direct=params.get('force-direct'),
minimum_z_height=params.get('minimum-z-height'))
else:
MODULE_LOG.warning("Bad command type {}".format(command_type))
def get_protocol_schema_version(protocol_json: Dict[Any, Any]) -> int:
# v3 and above uses `schemaVersion: integer`
version = protocol_json.get('schemaVersion')
if version:
return version
# v1 uses 1.x.x and v2 uses 2.x.x
legacyKebabVersion = protocol_json.get('protocol-schema')
# No minor/patch schemas ever were released,
# do not permit protocols with nonexistent schema versions to load
if legacyKebabVersion == '1.0.0':
return 1
elif legacyKebabVersion == '2.0.0':
return 2
elif legacyKebabVersion:
raise RuntimeError(
f'No such schema version: "{legacyKebabVersion}". Did you mean ' +
'"1.0.0" or "2.0.0"?')
# no truthy value for schemaVersion or protocol-schema
raise RuntimeError(
'Could not determine schema version for protocol. ' +
'Make sure there is a version number under "schemaVersion"')


def run_protocol(protocol_code: Any = None,
Expand Down Expand Up @@ -364,8 +161,23 @@ def run_protocol(protocol_code: Any = None,
if None is not protocol_code:
_run_python(protocol_code, true_context)
elif None is not protocol_json:
lw = load_labware_from_json(true_context, protocol_json)
ins = load_pipettes_from_json(true_context, protocol_json)
dispatch_json(true_context, protocol_json, ins, lw)
protocol_version = get_protocol_schema_version(protocol_json)
if protocol_version > 3:
raise RuntimeError(
f'JSON Protocol version {protocol_version} is not yet ' +
'supported in this version of the API')

if protocol_version >= 3:
ins = execute_v3.load_pipettes_from_json(
true_context, protocol_json)
lw = execute_v3.load_labware_from_json_defs(
true_context, protocol_json)
execute_v3.dispatch_json(true_context, protocol_json, ins, lw)
else:
ins = execute_v1.load_pipettes_from_json(
true_context, protocol_json)
lw = execute_v1.load_labware_from_json_loadnames(
true_context, protocol_json)
execute_v1.dispatch_json(true_context, protocol_json, ins, lw)
else:
raise RuntimeError("run_protocol must have either code or json")
Loading

0 comments on commit 988407d

Please sign in to comment.