Skip to content

Commit 6b1dda0

Browse files
wweiyewweiye
committed
add lifetime move 0-74
1 parent 857e469 commit 6b1dda0

File tree

1 file changed

+251
-0
lines changed

1 file changed

+251
-0
lines changed

‎hardware-testing/hardware_testing/scripts/faster_plunger_lifetime_test_0_74.py

Lines changed: 251 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,251 @@
1+
"""Test Plunger."""
2+
from typing import Tuple, Dict, Literal, Optional
3+
from typing_extensions import TypedDict
4+
5+
from opentrons.hardware_control.ot3api import OT3API
6+
from opentrons.types import Mount
7+
from dataclasses import dataclass
8+
9+
import time
10+
from hardware_testing.opentrons_api import helpers_ot3
11+
from hardware_testing.opentrons_api.types import Axis, OT3Mount
12+
from hardware_testing.gravimetric.helpers import get_pipette_unique_name
13+
import enum
14+
import argparse
15+
import csv
16+
import asyncio
17+
from datetime import datetime
18+
19+
20+
class TestSection(enum.Enum):
21+
"""Test Section."""
22+
23+
PLUNGER = "PLUNGER"
24+
25+
26+
@dataclass
27+
class TestConfig:
28+
"""Test Config."""
29+
30+
simulate: bool
31+
pipette: Literal[200, 1000]
32+
blowout:int
33+
blowoutnum:int
34+
35+
36+
class TestData(TypedDict):
37+
"""Test Data entry."""
38+
39+
time_sec: Optional[float]
40+
cycle: Optional[int]
41+
stall: Optional[str]
42+
position: Optional[str]
43+
position_check: Optional[bool]
44+
error: Optional[str]
45+
46+
47+
PLUNGER_MAX_SKIP_MM = 0.1
48+
SPEEDS_TO_TEST: float = 25
49+
CURRENTS_SPEEDS: Dict[float, float] = {
50+
0.7: SPEEDS_TO_TEST,
51+
}
52+
53+
54+
async def _is_plunger_still_aligned_with_encoder(
55+
api: OT3API,
56+
) -> Tuple[float, float, bool]:
57+
enc_pos = await api.encoder_current_position_ot3(OT3Mount.LEFT)
58+
motor_pos = await api.current_position_ot3(OT3Mount.LEFT)
59+
p_enc = enc_pos[Axis.P_L]
60+
p_est = motor_pos[Axis.P_L]
61+
is_aligned = abs(p_est - p_enc) < PLUNGER_MAX_SKIP_MM
62+
return p_enc, p_est, is_aligned
63+
64+
65+
async def main(args: argparse.Namespace, cfg: TestConfig) -> None:
66+
"""Run."""
67+
pipette_string = "p1000_96_v3.4" if cfg.pipette == 1000 else "p200_96_v3.1"
68+
69+
api = await helpers_ot3.build_async_ot3_hardware_api(
70+
is_simulating=cfg.simulate,
71+
pipette_left=pipette_string,
72+
stall_detection_enable=False,
73+
)
74+
ax = Axis.P_L
75+
xxx = Axis.X
76+
yyy = Axis.Y
77+
Z1 = Axis.Z_L
78+
Z2 = Axis.Z_R
79+
mount = OT3Mount.LEFT
80+
settings = helpers_ot3.get_gantry_load_per_axis_motion_settings_ot3(api, ax)
81+
settings.max_speed = 25
82+
settings.acceleration = 100
83+
settings.run_current = 0.7
84+
default_current = settings.run_current
85+
default_speed = settings.max_speed
86+
default_acceleration = 100
87+
top, bottom, blow_out, _ = helpers_ot3.get_plunger_positions_ot3(api, mount)
88+
print(f"Settings: {settings}")
89+
print(f"top: {top} bottom: {bottom} blow_out: {blow_out}")
90+
input(1)
91+
92+
async def position_check() -> bool:
93+
est, enc, aligned = await _is_plunger_still_aligned_with_encoder(api)
94+
print(f"Estimate: {est}, Encoder: {enc}, Aligned: {aligned}")
95+
return aligned
96+
97+
await api.cache_instruments()
98+
#await api.home_z(OT3Mount.LEFT)
99+
# LOOP THROUGH CURRENTS + SPEEDS
100+
await api.home([xxx,yyy,Z1,Z2])
101+
today = datetime.now().strftime("%m-%d-%y_%H-%M")
102+
pip_id = api.attached_pipettes[Mount.LEFT]["pipette_id"]
103+
with open(
104+
f"/data/testing_data/P200H_LT_{pip_id}_{today}.csv", "w", newline=""
105+
) as csvfile:
106+
test_data: TestData = {
107+
"time_sec": None,
108+
"cycle": None,
109+
"stall": None,
110+
"position": None,
111+
"position_check": None,
112+
"error": None,
113+
}
114+
writer = csv.DictWriter(csvfile, test_data)
115+
writer.writeheader()
116+
start_time = time.time()
117+
try:
118+
currents = list(CURRENTS_SPEEDS.keys())
119+
for cycle in range(1, args.cycles + 1):
120+
print(f"Cycle: {cycle}")
121+
for current in sorted(currents, reverse=True):
122+
speed = CURRENTS_SPEEDS[current]
123+
# HOME
124+
print("homing...")
125+
await api.home([ax])
126+
127+
print(f"run-current set to {current} amps")
128+
# await helpers_ot3.set_gantry_load_per_axis_current_settings_ot3(
129+
# api,
130+
# ax,
131+
# run_current=current,
132+
# )
133+
# await helpers_ot3.set_gantry_load_per_axis_motion_settings_ot3(
134+
# api,
135+
# ax,
136+
# default_max_speed=speed,
137+
# acceleration=default_acceleration,
138+
# )
139+
# MOVE DOWN
140+
print(f"moving down {74} mm at {speed} mm/sec")
141+
position_checked = await position_check()
142+
print(f"position checked: {position_checked}")
143+
try:
144+
await helpers_ot3.move_plunger_absolute_ot3(
145+
api, mount, 74
146+
)
147+
down_passed = await position_check()
148+
test_data["time_sec"] = time.time() - start_time
149+
test_data["cycle"] = cycle
150+
test_data["position"] = "bottom"
151+
test_data["position_check"] = down_passed
152+
test_data["stall"] = "NONE"
153+
print(test_data)
154+
writer.writerow(test_data)
155+
csvfile.flush()
156+
except Exception as e:
157+
print("STALL DETECTION")
158+
down_passed = await position_check()
159+
test_data["position_check"] = down_passed
160+
test_data["stall"] = str("Failed to move plunger down")
161+
test_data["error"] = str(e)
162+
print(test_data)
163+
writer.writerow(test_data)
164+
csvfile.flush()
165+
print("homing...")
166+
# await helpers_ot3.set_gantry_load_per_axis_current_settings_ot3(
167+
# api, ax, run_current=default_current
168+
# )
169+
# await helpers_ot3.set_gantry_load_per_axis_motion_settings_ot3(
170+
# api,
171+
# ax,
172+
# default_max_speed=default_speed,
173+
# acceleration=default_acceleration,
174+
# )
175+
# await api._backend.set_active_current(
176+
# {Axis.P_L: default_current}
177+
# )
178+
await api.home([ax])
179+
await helpers_ot3.move_plunger_absolute_ot3(
180+
api,
181+
mount,
182+
74
183+
)
184+
# MOVE UP
185+
print(f"moving up {top} mm at {speed} mm/sec")
186+
position_checked = await position_check()
187+
print(f"position checked: {position_checked}")
188+
# await helpers_ot3.set_gantry_load_per_axis_current_settings_ot3(
189+
# api,
190+
# ax,
191+
# run_current=current,
192+
# )
193+
# await helpers_ot3.set_gantry_load_per_axis_motion_settings_ot3(
194+
# api,
195+
# ax,
196+
# default_max_speed=speed,
197+
# acceleration=default_acceleration,
198+
# )
199+
try:
200+
await helpers_ot3.move_plunger_absolute_ot3(
201+
api, mount, 0
202+
)
203+
up_passed = await position_check()
204+
test_data["time_sec"] = time.time() - start_time
205+
test_data["cycle"] = cycle
206+
test_data["position"] = "top"
207+
test_data["position_check"] = up_passed
208+
print(test_data)
209+
writer.writerow(test_data)
210+
test_data["stall"] = "NONE"
211+
csvfile.flush()
212+
except Exception as e:
213+
print("STALL DETECTION")
214+
up_passed = await position_check()
215+
test_data["stall"] = str("Failed to move plunger down")
216+
test_data["position_check"] = up_passed
217+
test_data["error"] = str(e)
218+
print(test_data)
219+
writer.writerow(test_data)
220+
csvfile.flush()
221+
# RESET CURRENTS AND HOME
222+
print("homing...")
223+
# await helpers_ot3.set_gantry_load_per_axis_current_settings_ot3(
224+
# api, ax, run_current=default_current
225+
# )
226+
# await helpers_ot3.set_gantry_load_per_axis_motion_settings_ot3(
227+
# api,
228+
# ax,
229+
# default_max_speed=default_speed,
230+
# acceleration=default_acceleration,
231+
# )
232+
# await api._backend.set_active_current({Axis.P_L: default_current})
233+
234+
except Exception as e:
235+
test_data["error"] = str(e)
236+
writer.writerow(test_data)
237+
csvfile.flush()
238+
raise Exception(f"Error StallDetection: {e}")
239+
240+
241+
if __name__ == "__main__":
242+
parser = argparse.ArgumentParser()
243+
parser.add_argument("--cycles", type=int, default=100000)
244+
parser.add_argument("--simulate", action="store_true")
245+
parser.add_argument("--pipette", type=int, choices=[200, 1000], default=200)
246+
parser.add_argument("--blowout", type=int, default=1)
247+
parser.add_argument("--blowoutnum",type=int,default=100)
248+
args = parser.parse_args()
249+
_config = TestConfig(simulate=args.simulate, pipette=args.pipette,blowout=args.blowout,blowoutnum=args.blowoutnum)
250+
251+
asyncio.run(main(args, _config))

0 commit comments

Comments
 (0)