srenblad
/
artiq
forked from M-Labs/artiq
1
0
Fork 0
Code Pull Requests Activity

devices/thorlabs_tcube: fix more style issues

This commit is contained in:
Sebastien Bourdeauducq 2015-11-24 17:10:31 +08:00
parent 76eadc0a61
commit bda11149df
1 changed files with 17 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
artiq/devices/thorlabs_tcube/driver.py
View File

@ -186,7 +186,7 @@ class Message:
raise ValueError
class Tcube:
class _Tcube:
def __init__(self, serial_dev):
self.port = serial.serial_for_url(serial_dev, baudrate=115200,
rtscts=True)
@ -202,16 +202,15 @@ class Tcube:
msg.send(self.port)
def handle_message(self, msg):
pass
# derived classes must implement this
raise NotImplementedError
def send_request(self, msgreq_id, wait_for_msgs, param1=0, param2=0, data=None):
Message(msgreq_id, param1, param2, data=data).send(self.port)
msg = msg_id = None
while msg is None or msg_id not in wait_for_msgs:
msg = None
while msg is None or msg.id not in wait_for_msgs:
msg = Message.recv(self.port)
self.handle_message(msg)
msg_id = msg.id
return msg
def set_channel_enable_state(self, activated):
@ -238,7 +237,7 @@ class Tcube:
self.chan_enabled = False
else:
raise MsgError("Channel state response is invalid: neither "
+ "1 nor 2: {}".format(self.chan_enabled))
"1 nor 2: {}".format(self.chan_enabled))
return self.chan_enabled
def module_identify(self):
@ -247,7 +246,6 @@ class Tcube:
Instruct hardware unit to identify itself by flashing its front panel
led.
"""
self.send(Message(MGMSG.MOD_IDENTIFY))
def hardware_start_update_messages(self, update_rate):
@ -258,13 +256,10 @@ class Tcube:
:param update_rate: Rate at which you will receive status updates
"""
self.send(Message(MGMSG.HW_START_UPDATEMSGS, param1=update_rate))
def hardware_stop_update_messages(self):
"""Stop status updates from the controller.
"""
"""Stop status updates from the controller."""
self.send(Message(MGMSG.HW_STOP_UPDATEMSGS))
def hardware_request_information(self):
@ -282,9 +277,9 @@ class Tcube:
return True
class Tpz(Tcube):
class Tpz(_Tcube):
def __init__(self, serial_dev):
Tcube.__init__(self, serial_dev)
_Tcube.__init__(self, serial_dev)
self.voltage_limit = self.get_tpz_io_settings()[0]
def handle_message(self, msg):
@ -294,8 +289,8 @@ class Tpz(Tcube):
if msg_id == MGMSG.HW_DISCONNECT:
raise MsgError("Error: Please disconnect the TPZ001")
elif msg_id == MGMSG.HW_RESPONSE:
raise MsgError("Hardware error, please disconnect"
+ "and reconnect the TPZ001")
raise MsgError("Hardware error, please disconnect "
"and reconnect the TPZ001")
elif msg_id == MGMSG.HW_RICHRESPONSE:
(code, ) = st.unpack("<H", data[2:4])
raise MsgError("Hardware error {}: {}"
@ -317,7 +312,6 @@ class Tpz(Tcube):
0x04 for Closed Loop Smooth.
"""
self.send(Message(MGMSG.PZ_SET_POSCONTROLMODE, param1=1,
param2=control_mode))
@ -336,7 +330,6 @@ class Tpz(Tcube):
:rtype: int
"""
get_msg = self.send_request(MGMSG.PZ_REQ_POSCONTROLMODE,
[MGMSG.PZ_GET_POSCONTROLMODE], 1)
return get_msg.param2
@ -354,7 +347,6 @@ class Tpz(Tcube):
<artiq.devices.thorlabs.driver.Tpz.set_tpz_io_settings>` method
between the three values 75 V, 100 V and 150 V.
"""
if voltage < 0 or voltage > self.voltage_limit:
raise ValueError("Voltage must be in range [0;{}]"
.format(self.voltage_limit))
@ -368,7 +360,6 @@ class Tpz(Tcube):
:return: The output voltage.
:rtype: float
"""
get_msg = self.send_request(MGMSG.PZ_REQ_OUTPUTVOLTS,
[MGMSG.PZ_GET_OUTPUTVOLTS], 1)
return st.unpack("<H", get_msg.data[2:])[0]*self.voltage_limit/32767
@ -385,7 +376,6 @@ class Tpz(Tcube):
[0; 65535] depending on the unit. This corresponds to 0 to 100% of
the maximum piezo extension.
"""
payload = st.pack("<HH", 1, position_sw)
self.send(Message(MGMSG.PZ_SET_OUTPUTPOS, data=payload))
@ -396,7 +386,6 @@ class Tpz(Tcube):
position.
:rtype: int
"""
get_msg = self.send_request(MGMSG.PZ_REQ_OUTPUTPOS,
[MGMSG.PZ_GET_OUTPUTPOS], 1)
return st.unpack("<H", get_msg.data[2:])[0]
@ -424,7 +413,6 @@ class Tpz(Tcube):
The values can be bitwise or'ed to sum the software source with
either or both of the other source options.
"""
payload = st.pack("<HH", 1, volt_src)
self.send(Message(MGMSG.PZ_SET_INPUTVOLTSSRC, data=payload))
@ -438,7 +426,6 @@ class Tpz(Tcube):
docstring for meaning of bits.
:rtype: int
"""
get_msg = self.send_request(MGMSG.PZ_REQ_INPUTVOLTSSRC,
[MGMSG.PZ_GET_INPUTVOLTSSRC], 1)
return st.unpack("<H", get_msg.data[2:])[0]
@ -456,7 +443,6 @@ class Tpz(Tcube):
:param prop_const: Value of the proportional term in range [0; 255].
:param int_const: Value of the integral term in range [0; 255].
"""
payload = st.pack("<HHH", 1, prop_const, int_const)
self.send(Message(MGMSG.PZ_SET_PICONSTS, data=payload))
@ -467,7 +453,6 @@ class Tpz(Tcube):
term and the second element is the integral term.
:rtype: a 2 int elements tuple : (int, int)
"""
get_msg = self.send_request(MGMSG.PZ_REQ_PICONSTS,
[MGMSG.PZ_GET_PICONSTS], 1)
return st.unpack("<HH", get_msg.data[2:])
@ -512,7 +497,6 @@ class Tpz(Tcube):
:py:meth:`set_tpz_io_settings
<artiq.devices.thorlabs.driver.Tpz.set_tpz_io_settings>` method.
"""
volt = round(output*32767/self.voltage_limit)
payload = st.pack("<HHH", 1, lut_index, volt)
self.send(Message(MGMSG.PZ_SET_OUTPUTLUT, data=payload))
@ -524,10 +508,9 @@ class Tpz(Tcube):
the voltage output value.
:rtype: a 2 elements tuple (int, float)
"""
get_msg = self.send_request(MGMSG.PZ_REQ_OUTPUTLUT,
[MGMSG.PZ_GET_OUTPUTLUT], 1)
(index, output) = st.unpack("<Hh", get_msg.data[2:])
index, output = st.unpack("<Hh", get_msg.data[2:])
return index, output*self.voltage_limit/32767
def set_output_lut_parameters(self, mode, cycle_length, num_cycles,
@ -590,7 +573,6 @@ class Tpz(Tcube):
and 2147483648 sample intervals. The system then outputs the last
value in the cycle until the postcycle_rest time has expired.
"""
# triggering is not supported by the TPZ device
payload = st.pack("<HHHLLLLHLH", 1, mode, cycle_length, num_cycles,
delay_time, precycle_rest, postcycle_rest,
@ -604,21 +586,16 @@ class Tpz(Tcube):
num_cycles, delay_time, precycle_rest, postcycle_rest).
:rtype: 6 int elements tuple
"""
get_msg = self.send_request(MGMSG.PZ_REQ_OUTPUTLUTPARAMS,
[MGMSG.PZ_GET_OUTPUTLUTPARAMS], 1)
return st.unpack("<HHLLLL", get_msg.data[2:22])
def start_lut_output(self):
"""Start the voltage waveform (LUT) outputs.
"""
"""Start the voltage waveform (LUT) outputs."""
self.send(Message(MGMSG.PZ_START_LUTOUTPUT, param1=1))
def stop_lut_output(self):
"""Stop the voltage waveform (LUT) outputs.
"""
"""Stop the voltage waveform (LUT) outputs."""
self.send(Message(MGMSG.PZ_STOP_LUTOUTPUT, param1=1))
def set_eeprom_parameters(self, msg_id):
@ -627,7 +604,6 @@ class Tpz(Tcube):
:param msg_id: The message ID of the message containing the parameters
to be saved.
"""
payload = st.pack("<HH", 1, msg_id)
self.send(Message(MGMSG.PZ_SET_EEPROMPARAMS, data=payload))
@ -637,7 +613,6 @@ class Tpz(Tcube):
:param intensity: The intensity is set as a value from 0 (Off) to 255
(brightest).
"""
payload = st.pack("<H", intensity)
self.send(Message(MGMSG.PZ_SET_TPZ_DISPSETTINGS, data=payload))
@ -647,7 +622,6 @@ class Tpz(Tcube):
:return: The intensity as a value from 0 (Off) to 255 (brightest).
:rtype: int
"""
get_msg = self.send_request(MGMSG.PZ_REQ_TPZ_DISPSETTINGS,
[MGMSG.PZ_GET_TPZ_DISPSETTINGS], 1)
return st.unpack("<H", get_msg.data)[0]
@ -682,7 +656,6 @@ class Tpz(Tcube):
0x03: the feedback signals run through the read panel SMA
connectors.
"""
self.voltage_limit = voltage_limit
if self.voltage_limit == 75:
@ -706,7 +679,6 @@ class Tpz(Tcube):
meaning of those parameters.
:rtype: a 2 elements tuple (int, int)
"""
get_msg = self.send_request(MGMSG.PZ_REQ_TPZ_IOSETTINGS,
[MGMSG.PZ_GET_TPZ_IOSETTINGS], 1)
voltage_limit, hub_analog_input = st.unpack("<HH", get_msg.data[2:6])
@ -722,7 +694,7 @@ class Tpz(Tcube):
return voltage_limit, hub_analog_input
class Tdc(Tcube):
class Tdc(_Tcube):
def send_status_ack(self):
self.send(Message(MGMSG.MOT_ACK_DCSTATUSUPDATE))
@ -749,7 +721,7 @@ class Tdc(Tcube):
else:
self.status_report_counter += 1
# 'r' is a currently unused and reserved field
(self.position, self.velocity, r, self.status) = st.unpack(
self.position, self.velocity, r, self.status = st.unpack(
"<LHHL", data[2:])
def is_moving(self):
@ -773,7 +745,6 @@ class Tdc(Tcube):
wnd2 to 127) to apply vel3.
:param vel4: The velocity to move when beyond wnd3.
"""
payload = st.pack("<HHLHLHLHL", 1, zero_wnd, vel1, wnd1, vel2, wnd2,
vel3, wnd3, vel4)
self.send(Message(MGMSG.MOT_SET_POTPARAMS, data=payload))
@ -788,7 +759,6 @@ class Tdc(Tcube):
description of each tuple element meaning.
:rtype: An 8 int tuple
"""
get_msg = self.send_request(MGMSG.MOT_REQ_POTPARAMS,
[MGMSG.MOT_GET_POTPARAMS], 1)
return st.unpack("<HLHLHLHL", get_msg.data[2:])
@ -808,7 +778,6 @@ class Tdc(Tcube):
:param position: The new value of the position counter.
"""
payload = st.pack("<Hl", 1, position)
self.send(Message(MGMSG.MOT_SET_POSCOUNTER, data=payload))
@ -818,7 +787,6 @@ class Tdc(Tcube):
:return: The value of the position counter.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_POSCOUNTER,
[MGMSG.MOT_GET_POSCOUNTER], 1)
return st.unpack("<l", get_msg.data[2:])[0]
@ -832,7 +800,6 @@ class Tdc(Tcube):
:param encoder_count: The new value of the encoder counter.
"""
payload = st.pack("<Hl", 1, encoder_count)
self.send(Message(MGMSG.MOT_SET_ENCCOUNTER, data=payload))
@ -842,7 +809,6 @@ class Tdc(Tcube):
:return: The value of the encoder counter.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_ENCCOUNTER,
[MGMSG.MOT_GET_ENCCOUNTER], 1)
return st.unpack("<l", get_msg.data[2:])[0]
@ -853,7 +819,6 @@ class Tdc(Tcube):
:param acceleration: The acceleration in encoder counts/sec/sec.
:param max_velocity: The maximum (final) velocity in counts/sec.
"""
payload = st.pack("<HLLL", 1, 0, acceleration, max_velocity)
self.send(Message(MGMSG.MOT_SET_VELPARAMS, data=payload))
@ -863,7 +828,6 @@ class Tdc(Tcube):
:return: A 2 int tuple: (acceleration, max_velocity).
:rtype: A 2 int tuple (int, int)
"""
get_msg = self.send_request(MGMSG.MOT_REQ_VELPARAMS,
[MGMSG.MOT_GET_VELPARAMS], 1)
return st.unpack("<LL", get_msg.data[6:])
@ -880,7 +844,6 @@ class Tdc(Tcube):
:param stop_mode: 1 for immediate (abrupt) stop, 2 for profiled stop
(with controlled deceleration).
"""
payload = st.pack("<HHLLLLH", 1, mode, step_size, 0, acceleration,
max_velocity, stop_mode)
self.send(Message(MGMSG.MOT_SET_JOGPARAMS, data=payload))
@ -892,7 +855,6 @@ class Tdc(Tcube):
step_size, acceleration, max_velocity, stop_mode
:rtype: A 5 int tuple.
"""
get_msg = self.send_request(MGMSG.MOT_REQ_JOGPARAMS,
[MGMSG.MOT_GET_JOGPARAMS], 1)
(jog_mode, step_size, _, acceleration, max_velocity,
@ -905,7 +867,6 @@ class Tdc(Tcube):
:param backlash_distance: The value of the backlash distance,
which specifies the relative distance in position counts.
"""
payload = st.pack("<Hl", 1, backlash_distance)
self.send(Message(MGMSG.MOT_SET_GENMOVEPARAMS, data=payload))
@ -915,7 +876,6 @@ class Tdc(Tcube):
:return: The value of the backlash distance.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_GENMOVEPARAMS,
[MGMSG.MOT_GET_GENMOVEPARAMS], 1)
return st.unpack("<l", get_msg.data[2:])[0]
@ -927,7 +887,6 @@ class Tdc(Tcube):
integer that specifies the relative distance in position encoder
counts.
"""
payload = st.pack("<Hl", 1, relative_distance)
self.send(Message(MGMSG.MOT_SET_MOVERELPARAMS, data=payload))
@ -937,7 +896,6 @@ class Tdc(Tcube):
:return: The relative distance move parameter.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_MOVERELPARAMS,
[MGMSG.MOT_GET_MOVERELPARAMS], 1)
return st.unpack("<l", get_msg.data[2:])[0]
@ -949,7 +907,6 @@ class Tdc(Tcube):
signed integer that specifies the absolute move position in encoder
counts.
"""
payload = st.pack("<Hl", 1, absolute_position)
self.send(Message(MGMSG.MOT_SET_MOVEABSPARAMS, data=payload))
@ -959,7 +916,6 @@ class Tdc(Tcube):
:return: The absolute position to move.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_MOVEABSPARAMS,
[MGMSG.MOT_GET_MOVEABSPARAMS], 1)
return st.unpack("<l", get_msg.data[2:])[0]
@ -969,7 +925,6 @@ class Tdc(Tcube):
:param home_velocity: Homing velocity.
"""
payload = st.pack("<HHHLL", 1, 0, 0, home_velocity, 0)
self.send(Message(MGMSG.MOT_SET_HOMEPARAMS, data=payload))
@ -979,7 +934,6 @@ class Tdc(Tcube):
:return: The homing velocity.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_HOMEPARAMS,
[MGMSG.MOT_GET_HOMEPARAMS], 1)
return st.unpack("<L", get_msg.data[6:10])[0]
@ -989,7 +943,6 @@ class Tdc(Tcube):
This call is blocking until device is homed or move is stopped.
"""
self.send_request(MGMSG.MOT_MOVE_HOME,
[MGMSG.MOT_MOVE_HOMED, MGMSG.MOT_MOVE_STOPPED], 1)
@ -1020,7 +973,6 @@ class Tdc(Tcube):
:param ccw_hw_limit: The operation of counter clockwise hardware limit
switch when contact is made.
"""
payload = st.pack("<HHHLLH", 1, cw_hw_limit, ccw_hw_limit, 0, 0, 0)
self.send(Message(MGMSG.MOT_SET_LIMSWITCHPARAMS, data=payload))
@ -1034,7 +986,6 @@ class Tdc(Tcube):
method.
:rtype: A 2 int tuple.
"""
get_msg = self.send_request(MGMSG.MOT_REQ_LIMSWITCHPARAMS,
[MGMSG.MOT_GET_LIMSWITCHPARAMS], 1)
return st.unpack("<HH", get_msg.data[2:6])
@ -1047,7 +998,6 @@ class Tdc(Tcube):
<artiq.devices.thorlabs.driver.Tdc.set_move_relative_parameters>`
command.
"""
self.send_request(MGMSG.MOT_MOVE_RELATIVE,
[MGMSG.MOT_MOVE_COMPLETED, MGMSG.MOT_MOVE_STOPPED], 1)
@ -1057,7 +1007,6 @@ class Tdc(Tcube):
:param relative_distance: The distance to move in position encoder
counts.
"""
payload = st.pack("<Hl", 1, relative_distance)
self.send_request(MGMSG.MOT_MOVE_RELATIVE,
[MGMSG.MOT_MOVE_COMPLETED, MGMSG.MOT_MOVE_STOPPED],
@ -1071,7 +1020,6 @@ class Tdc(Tcube):
<artiq.devices.thorlabs.driver.Tdc.set_move_absolute_parameters>`
command.
"""
self.send_request(MGMSG.MOT_MOVE_ABSOLUTE,
[MGMSG.MOT_MOVE_COMPLETED, MGMSG.MOT_MOVE_STOPPED],
param1=1)
@ -1083,18 +1031,16 @@ class Tdc(Tcube):
integer that specifies the absolute distance in position encoder
counts.
"""
payload = st.pack("<Hl", 1, absolute_distance)
self.send_request(MGMSG.MOT_MOVE_ABSOLUTE,
[MGMSG.MOT_MOVE_COMPLETED, MGMSG.MOT_MOVE_STOPPED],
data=payload)
def move_jog(self, direction):
"""Start a job move.
"""Start a jog move.
:param direction: The direction to jog. 1 is forward, 2 is backward.
"""
self.send_request(MGMSG.MOT_MOVE_JOG,
[MGMSG.MOT_MOVE_COMPLETED, MGMSG.MOT_MOVE_STOPPED],
param1=1, param2=direction)
@ -1113,7 +1059,6 @@ class Tdc(Tcube):
:param direction: The direction to jog: 1 to move forward, 2 to move
backward.
"""
self.send(Message(MGMSG.MOT_MOVE_VELOCITY, param1=1, param2=direction))
def move_stop(self, stop_mode):
@ -1126,7 +1071,6 @@ class Tdc(Tcube):
or profiled stop. Set this byte to 1 to stop immediately, or to 2
to stop in a controlled (profiled) manner.
"""
if self.is_moving():
self.send_request(MGMSG.MOT_MOVE_STOP,
[MGMSG.MOT_MOVE_STOPPED,
@ -1148,7 +1092,6 @@ class Tdc(Tcube):
setting the corresponding bit to 1. By default, all parameters are
applied, and this parameter is set to 0x0F (1111).
"""
payload = st.pack("<HLLLLH", 1, proportional, integral,
differential, integral_limit, filter_control)
self.send(Message(MGMSG.MOT_SET_DCPIDPARAMS, data=payload))
@ -1163,7 +1106,6 @@ class Tdc(Tcube):
precise description.
:rtype: A 5 int tuple.
"""
get_msg = self.send_request(MGMSG.MOT_REQ_DCPIDPARAMS,
[MGMSG.MOT_GET_DCPIDPARAMS], 1)
return st.unpack("<LLLLH", get_msg.data[2:])
@ -1180,7 +1122,6 @@ class Tdc(Tcube):
forward or reverse limit switch.
Set the bit 3 (value 8) will make the LED lit when motor is moving.
"""
payload = st.pack("<HH", 1, mode_bits)
self.send(Message(MGMSG.MOT_SET_AVMODES, data=payload))
@ -1190,7 +1131,6 @@ class Tdc(Tcube):
:return: The LED indicator mode bits.
:rtype: int
"""
get_msg = self.send_request(MGMSG.MOT_REQ_AVMODES,
[MGMSG.MOT_GET_AVMODES], 1)
return st.unpack("<H", get_msg.data[2:])[0]
@ -1214,7 +1154,6 @@ class Tdc(Tcube):
:param position2: The position (in encoder counts) to which the motor
will move when the bottom button is pressed.
"""
payload = st.pack("<HHllHH", 1, mode, position1, position2,
0, 0)
self.send(Message(MGMSG.MOT_SET_BUTTONPARAMS, data=payload))
@ -1228,7 +1167,6 @@ class Tdc(Tcube):
description.
:rtype: A 3 int tuple
"""
get_msg = self.send_request(MGMSG.MOT_REQ_BUTTONPARAMS,
[MGMSG.MOT_GET_BUTTONPARAMS], 1)
return st.unpack("<Hll", get_msg.data[2:12])
@ -1239,7 +1177,6 @@ class Tdc(Tcube):
:param msg_id: The message ID of the message containing the parameters
to be saved.
"""
payload = st.pack("<HH", 1, msg_id)
self.send(Message(MGMSG.MOT_SET_EEPROMPARAMS, data=payload))
@ -1252,7 +1189,6 @@ class Tdc(Tcube):
velocity, status bits.
:rtype: A 3 int tuple
"""
get_msg = self.send_request(MGMSG.MOT_REQ_DCSTATUSUPDATE,
[MGMSG.MOT_GET_DCSTATUSUPDATE], 1)
pos, vel, _, stat = st.unpack("<LHHL", get_msg.data[2:])
@ -1264,7 +1200,6 @@ class Tdc(Tcube):
:return: The motor status.
:rtype:
"""
get_msg = self.send_request(MGMSG.MOT_REQ_STATUSBITS,
[MGMSG.MOT_GET_STATUSBITS], 1)
return st.unpack("<L", get_msg.data[2:])[0]
@ -1276,7 +1211,6 @@ class Tdc(Tcube):
i.e., MGMSG.MOT_MOVE_STOPPED, MGMSG.MOT_MOVE_COMPLETED,
MGMSGS_MOT_MOVE_HOMED
"""
self.send(Message(MGMSG.MOT_SUSPEND_ENDOFMOVEMSGS))
def resume_end_of_move_messages(self):
@ -1291,7 +1225,6 @@ class Tdc(Tcube):
MGMSG.HW_RESPONSE,
MGMSG.HW_RICHRESPONSE
"""
self.send(Message(MGMSG.MOT_RESUME_ENDOFMOVEMSGS))