diff --git a/api/src/opentrons/protocol_api/labware.py b/api/src/opentrons/protocol_api/labware.py index 583196246e22..e7647c52877b 100644 --- a/api/src/opentrons/protocol_api/labware.py +++ b/api/src/opentrons/protocol_api/labware.py @@ -6,8 +6,7 @@ from typing import List, Dict from enum import Enum, auto from opentrons.types import Point -from opentrons.util.environment import PI_DATA_PATH -from opentrons.util import environment +from opentrons.util import environment as env from collections import defaultdict @@ -21,11 +20,7 @@ class WellShape(Enum): 'circular': WellShape.CIRCULAR } -if os.environ.get('RUNNING_ON_PI'): - presistentPath = os.path.join(PI_DATA_PATH, 'offsets') -else: - app_dir = environment.get_path('APP_DATA_DIR') - persistentPath = os.path.join(app_dir, 'offsets') +persistent_path = os.path.join(env.get_path('APP_DATA_DIR'), 'offsets') class Well: @@ -144,7 +139,7 @@ def __init__(self, definition: dict, parent: Point) -> None: y=offset['y'] + parent.y, z=offset['z'] + parent.z) # Applied properties - self.set_calibration(Point(0, 0, 0)) + self.set_calibration(self._calibrated_offset) self._pattern = re.compile(r'^([A-Z]+)([1-9][0-9]*)$', re.X) def _build_wells(self) -> List[Well]: @@ -261,28 +256,13 @@ def save_calibration(labware: Labware, delta: Point): using labware id as the filename. If the file does exist, load it and modify the delta and the lastModified field. """ - if not os.path.exists(persistentPath): - os.mkdir(persistentPath) + if not os.path.exists(persistent_path): + os.mkdir(persistent_path) labwareOffsetPath = os.path.join( - persistentPath, "{}.json".format(labware._id)) - if not os.path.exists(labwareOffsetPath): - schema = { - "default": { - "offset": [delta.x, delta.y, delta.z], - "lastModified": time.time() - } - } - - with open(labwareOffsetPath, 'w') as f: - json.dump(schema, f) - else: - with open(labwareOffsetPath, 'r') as f: - schema = json.load(f) - schema['default']['offset'] = [delta.x, delta.y, delta.z] - schema['default']['lastModified'] = time.time() - with open(labwareOffsetPath, 'w') as f: - json.dump(schema, f) - + persistent_path, "{}.json".format(labware._id)) + calibration_data = _helper_offset_data_format(labwareOffsetPath, delta) + with open(labwareOffsetPath, 'w') as f: + json.dump(calibration_data, f) labware.set_calibration(delta) @@ -292,15 +272,36 @@ def load_calibration(labware: Labware): """ offset = Point(0, 0, 0) labwareOffsetPath = os.path.join( - persistentPath, "{}.json".format(labware._id)) + persistent_path, "{}.json".format(labware._id)) if os.path.exists(labwareOffsetPath): - with open(labwareOffsetPath) as f: - schema = json.load(f) - offsetArray = schema['default']['offset'] + calibration_data = _read_file(labwareOffsetPath) + offsetArray = calibration_data['default']['offset'] offset = Point(x=offsetArray[0], y=offsetArray[1], z=offsetArray[2]) labware.set_calibration(offset) +def _helper_offset_data_format(filepath: str, delta: Point) -> dict: + if not os.path.exists(filepath): + calibration_data = { + "default": { + "offset": [delta.x, delta.y, delta.z], + "lastModified": time.time() + } + } + else: + calibration_data = _read_file(filepath) + calibration_data['default']['offset'] = [delta.x, delta.y, delta.z] + calibration_data['default']['lastModified'] = time.time() + return calibration_data + + +def _read_file(filepath: str) -> dict: + calibration_data: dict = {} + with open(filepath, 'r') as f: + calibration_data = json.load(f) + return calibration_data + + def _load_definition_by_name(name: str) -> dict: """ Look up and return a definition by name (name is expected to correspond to @@ -310,20 +311,20 @@ def _load_definition_by_name(name: str) -> dict: raise NotImplementedError -def load(name: str, slot: Point) -> Labware: +def load(name: str, cornerOffset: Point) -> Labware: """ Return a labware object constructed from a labware definition dict looked up by name (definition must have been previously stored locally on the robot) """ definition = _load_definition_by_name(name) - labware = load_from_definition(definition, slot) + labware = load_from_definition(definition, cornerOffset) load_calibration(labware) return labware -def load_from_definition(definition: dict, slot: Point) -> Labware: +def load_from_definition(definition: dict, cornerOffset: Point) -> Labware: """ Return a labware object constructed from a provided labware definition dict """ - return Labware(definition, slot) + return Labware(definition, cornerOffset) diff --git a/api/tests/opentrons/protocol_api/test_offsets.py b/api/tests/opentrons/protocol_api/test_offsets.py index f0f46578003e..f6b81ac0827d 100644 --- a/api/tests/opentrons/protocol_api/test_offsets.py +++ b/api/tests/opentrons/protocol_api/test_offsets.py @@ -1,8 +1,9 @@ +# pylama:ignore=W0612 import tempfile -import pytest import os import json import time +from functools import partial from opentrons.protocol_api import labware from opentrons.types import Point @@ -40,35 +41,37 @@ } tmpdir = tempfile.mkdtemp("offsets") -labware.persistentPath = tmpdir +labware.persistent_path = tmpdir testLabware = labware.Labware(minimalLabwareDef, Point(0, 0, 0)) -path = os.path.join(labware.persistentPath, "{}.json".format(testLabware._id)) -global testPoint +path = os.path.join(labware.persistent_path, "{}.json".format(testLabware._id)) -@pytest.fixture -def patch_calibration(monkeypatch): - def fake_set_calibration(delta: Point): - global testPoint - testPoint = delta - monkeypatch.setattr(testLabware, 'set_calibration', fake_set_calibration) +def mock_set_calibration(test_point, delta): + test_point = delta -def test_save_calibration(patch_calibration): +def test_save_calibration(monkeypatch): # Test the save calibration file assert not os.path.exists(path) + calibration_point = None + monkeypatch.setattr( + testLabware, + 'set_calibration', partial(mock_set_calibration, calibration_point)) labware.save_calibration(testLabware, Point(1, 1, 1)) assert os.path.exists(path) - global testPoint - testPoint == Point(1, 1, 1) + calibration_point == Point(1, 1, 1) -def test_schema_shape(patch_calibration, monkeypatch): +def test_schema_shape(monkeypatch): assert os.path.exists(path) def fake_time(): return 1 monkeypatch.setattr(time, 'time', fake_time) + calibration_point = None + monkeypatch.setattr( + testLabware, + 'set_calibration', partial(mock_set_calibration, calibration_point)) labware.save_calibration(testLabware, Point(1, 1, 1)) expected = {"default": {"offset": [1, 1, 1], "lastModified": 1}} with open(path) as f: @@ -76,18 +79,23 @@ def fake_time(): assert result == expected -def test_load_calibration(patch_calibration): +def test_load_calibration(monkeypatch): labware.load_calibration(testLabware) - global testPoint - testPoint == Point(1, 1, 1) + calibration_point = None + monkeypatch.setattr( + testLabware, + 'set_calibration', partial(mock_set_calibration, calibration_point)) + labware.save_calibration(testLabware, Point(1, 1, 1)) + calibration_point == Point(1, 1, 1) def test_wells_rebuilt_with_offset(): + testLabware2 = labware.Labware(minimalLabwareDef, Point(0, 0, 0)) old_wells = testLabware._wells - assert testLabware._offset == Point(10, 10, 5) - assert testLabware._calibrated_offset == Point(10, 10, 5) - labware.save_calibration(testLabware, Point(2, 2, 2)) - new_wells = testLabware._wells + assert testLabware2._offset == Point(10, 10, 5) + assert testLabware2._calibrated_offset == Point(10, 10, 5) + labware.save_calibration(testLabware2, Point(2, 2, 2)) + new_wells = testLabware2._wells assert old_wells[0] != new_wells[0] - assert testLabware._offset == Point(10, 10, 5) - assert testLabware._calibrated_offset == Point(12, 12, 7) + assert testLabware2._offset == Point(10, 10, 5) + assert testLabware2._calibrated_offset == Point(12, 12, 7)