adeck/python-urx-getl-rt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use flake8 tool to clean up file to pep8 standards

Browse Source
This commit is contained in:
Chris Gilmer 2017-01-26 11:33:35 -08:00 committed by ORD
parent 606ecd5fa3
commit e711a1e929
5 changed files with 79 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
urx/__init__.py
View File

@ -1,11 +1,10 @@
""" """
Python library to control an UR robot through its TCP/IP interface Python library to control an UR robot through its TCP/IP interface
""" """
from urx.urrobot import RobotException, URRobot # noqa
__version__ = "0.9.0" __version__ = "0.9.0"
from urx.urrobot import RobotException, URRobot
try: try:
from urx.robot import Robot from urx.robot import Robot
except ImportError as ex: except ImportError as ex:

101
urx/robotiq_two_finger_gripper.py
View File

@ -26,61 +26,62 @@ Future Features
import logging import logging
class Robotiq_Two_Finger_Gripper(object): class Robotiq_Two_Finger_Gripper(object):
complete_program = "" complete_program = ""
header = "def myProg():" + b"\n" header = "def myProg():" + b"\n"
end = b"\n" + "end" end = b"\n" + "end"
logger = False logger = False
def __init__(self): def __init__(self):
self.logger = logging.getLogger("urx") self.logger = logging.getLogger("urx")
self.reset() self.reset()
def reset(self): def reset(self):
self.complete_program = "" self.complete_program = ""
self.add_line_to_program("set_analog_inputrange(0, 0)") self.add_line_to_program("set_analog_inputrange(0, 0)")
self.add_line_to_program("set_analog_inputrange(1, 0)") self.add_line_to_program("set_analog_inputrange(1, 0)")
self.add_line_to_program("set_analog_inputrange(2, 0)") self.add_line_to_program("set_analog_inputrange(2, 0)")
self.add_line_to_program("set_analog_inputrange(3, 0)") self.add_line_to_program("set_analog_inputrange(3, 0)")
self.add_line_to_program("set_analog_outputdomain(0, 0)") self.add_line_to_program("set_analog_outputdomain(0, 0)")
self.add_line_to_program("set_analog_outputdomain(1, 0)") self.add_line_to_program("set_analog_outputdomain(1, 0)")
self.add_line_to_program("set_tool_voltage(0)") self.add_line_to_program("set_tool_voltage(0)")
self.add_line_to_program("set_runstate_outputs([])") self.add_line_to_program("set_runstate_outputs([])")
self.add_line_to_program("set_payload(0.85)") #0.85 is the weight of the gripper in KG self.add_line_to_program("set_payload(0.85)") # 0.85 is the weight of the gripper in KG
self.add_line_to_program("socket_close(\"gripper_socket\")") self.add_line_to_program("socket_close(\"gripper_socket\")")
#self.add_line_to_program("sleep(1)") #in Robotiq's example they do a wait here... I haven't found it nec # self.add_line_to_program("sleep(1)") # in Robotiq's example they do a wait here... I haven't found it nec
self.add_line_to_program("socket_open(\"127.0.0.1\",63352,\"gripper_socket\")") self.add_line_to_program("socket_open(\"127.0.0.1\",63352,\"gripper_socket\")")
#self.add_line_to_program("sleep(1)") # self.add_line_to_program("sleep(1)")
self.add_line_to_program("socket_set_var(\"SEP\",255,\"gripper_socket\")") #Speed 0-255 is valid self.add_line_to_program("socket_set_var(\"SEP\",255,\"gripper_socket\")") # Speed 0-255 is valid
self.add_line_to_program("sync()") self.add_line_to_program("sync()")
self.add_line_to_program("socket_set_var(\"FOR\",50,\"gripper_socket\")") #Force 0-255 is valid self.add_line_to_program("socket_set_var(\"FOR\",50,\"gripper_socket\")") # Force 0-255 is valid
self.add_line_to_program("sync()") self.add_line_to_program("sync()")
self.add_line_to_program("socket_set_var(\"ACT\",1,\"gripper_socket\")") # Activate robot self.add_line_to_program("socket_set_var(\"ACT\",1,\"gripper_socket\")") # Activate robot
self.add_line_to_program("sync()") self.add_line_to_program("sync()")
self.add_line_to_program("socket_set_var(\"GTO\",1,\"gripper_socket\")") self.add_line_to_program("socket_set_var(\"GTO\",1,\"gripper_socket\")")
self.add_line_to_program("sync()") self.add_line_to_program("sync()")
def open_gripper(self): def open_gripper(self):
self.add_line_to_program("socket_set_var(\"POS\",0,\"gripper_socket\")") #0 is open; range is 0-255 self.add_line_to_program("socket_set_var(\"POS\",0,\"gripper_socket\")") # 0 is open; range is 0-255
self.add_line_to_program("sync()") self.add_line_to_program("sync()")
self.add_line_to_program("sleep(2)") self.add_line_to_program("sleep(2)")
def close_gripper(self): def close_gripper(self):
self.add_line_to_program("socket_set_var(\"POS\",255,\"gripper_socket\")") #255 is closed; range is 0-255 self.add_line_to_program("socket_set_var(\"POS\",255,\"gripper_socket\")") # 255 is closed; range is 0-255
self.add_line_to_program("sync()") self.add_line_to_program("sync()")
self.add_line_to_program("sleep(2)") self.add_line_to_program("sleep(2)")
def add_line_to_program(self,new_line): def add_line_to_program(self, new_line):
if(self.complete_program != ""): if(self.complete_program != ""):
self.complete_program += b"\n" self.complete_program += b"\n"
self.complete_program += new_line self.complete_program += new_line
def ret_program_to_run(self): def ret_program_to_run(self):
if(self.complete_program == ""): if(self.complete_program == ""):
self.logger.debug("robotiq_two_finger_gripper's program is empty") self.logger.debug("robotiq_two_finger_gripper's program is empty")
return "" return ""
prog = self.header prog = self.header
prog += self.complete_program prog += self.complete_program
prog += self.end prog += self.end
return prog return prog

6
urx/urrobot.py
View File

@ -208,7 +208,7 @@ class URRobot(object):
start_dist = self._get_dist(target, joints) start_dist = self._get_dist(target, joints)
if threshold is None: if threshold is None:
threshold = start_dist * 0.8 threshold = start_dist * 0.8
if threshold < 0.001: # roboten precision is limited if threshold < 0.001: # roboten precision is limited
threshold = 0.001 threshold = 0.001
self.logger.debug("No threshold set, setting it to %s", threshold) self.logger.debug("No threshold set, setting it to %s", threshold)
count = 0 count = 0
@ -234,13 +234,13 @@ class URRobot(object):
return self._get_lin_dist(target) return self._get_lin_dist(target)
def _get_lin_dist(self, target): def _get_lin_dist(self, target):
#FIXME: we have an issue here, it seems sometimes the axis angle received from robot # FIXME: we have an issue here, it seems sometimes the axis angle received from robot
pose = URRobot.getl(self, wait=True) pose = URRobot.getl(self, wait=True)
dist = 0 dist = 0
for i in range(3): for i in range(3):
dist += (target[i] - pose[i]) ** 2 dist += (target[i] - pose[i]) ** 2
for i in range(3, 6): for i in range(3, 6):
dist += ((target[i] - pose[i]) / 5) ** 2 # arbitraty length like dist += ((target[i] - pose[i]) / 5) ** 2 # arbitraty length like
return dist ** 0.5 return dist ** 0.5
def _get_joints_dist(self, target): def _get_joints_dist(self, target):

20
urx/urrtmon.py
View File

@ -4,13 +4,6 @@ Confer http://support.universal-robots.com/Technical/RealTimeClientInterface
Note: The packet lenght given in the web-page is 740. What is actually received from the controller is 692. It is assumed that the motor currents, the last group of 48 bytes, are not send. Note: The packet lenght given in the web-page is 740. What is actually received from the controller is 692. It is assumed that the motor currents, the last group of 48 bytes, are not send.
Originally Written by Morten Lind Originally Written by Morten Lind
''' '''
__author__ = "Morten Lind, Olivier Roulet-Dubonnet"
__copyright__ = "Copyright 2011, NTNU/SINTEF Raufoss Manufacturing AS"
__credits__ = ["Morten Lind, Olivier Roulet-Dubonnet"]
__license__ = "LGPLv3"
import logging import logging
import socket import socket
import struct import struct
@ -22,6 +15,11 @@ import numpy as np
import math3d as m3d import math3d as m3d
__author__ = "Morten Lind, Olivier Roulet-Dubonnet"
__copyright__ = "Copyright 2011, NTNU/SINTEF Raufoss Manufacturing AS"
__credits__ = ["Morten Lind, Olivier Roulet-Dubonnet"]
__license__ = "LGPLv3"
class URRTMonitor(threading.Thread): class URRTMonitor(threading.Thread):
@ -51,7 +49,7 @@ class URRTMonitor(threading.Thread):
self._tcp_force = None self._tcp_force = None
self.__recvTime = 0 self.__recvTime = 0
self._last_ctrl_ts = 0 self._last_ctrl_ts = 0
#self._last_ts = 0 # self._last_ts = 0
self._buffering = False self._buffering = False
self._buffer_lock = threading.Lock() self._buffer_lock = threading.Lock()
self._buffer = [] self._buffer = []
@ -155,8 +153,8 @@ class URRTMonitor(threading.Thread):
self._timestamp = timestamp self._timestamp = timestamp
# it seems that packet often arrives packed as two... maybe TCP_NODELAY is not set on UR controller?? # it seems that packet often arrives packed as two... maybe TCP_NODELAY is not set on UR controller??
# if (self._timestamp - self._last_ts) > 0.010: # if (self._timestamp - self._last_ts) > 0.010:
#self.logger.warning("Error the we did not receive a packet for {}s ".format( self._timestamp - self._last_ts)) # self.logger.warning("Error the we did not receive a packet for {}s ".format( self._timestamp - self._last_ts))
#self._last_ts = self._timestamp # self._last_ts = self._timestamp
self._ctrlTimestamp = np.array(unp[0]) self._ctrlTimestamp = np.array(unp[0])
if self._last_ctrl_ts != 0 and ( if self._last_ctrl_ts != 0 and (
self._ctrlTimestamp - self._ctrlTimestamp -
@ -239,7 +237,7 @@ class URRTMonitor(threading.Thread):
tcp_force=self._tcp_force) tcp_force=self._tcp_force)
def stop(self): def stop(self):
#print(self.__class__.__name__+': Stopping') # print(self.__class__.__name__+': Stopping')
self._stop_event = True self._stop_event = True
def close(self): def close(self):

18
urx/ursecmon.py
View File

@ -69,7 +69,7 @@ class ParserUtils(object):
self.version = (3, 0) self.version = (3, 0)
allData['RobotModeData'] = self._get_data(pdata, "!IBQ???????BBdd", ("size", "type", "timestamp", "isRobotConnected", "isRealRobotEnabled", "isPowerOnRobot", "isEmergencyStopped", "isSecurityStopped", "isProgramRunning", "isProgramPaused", "robotMode", "controlMode", "speedFraction", "speedScaling")) allData['RobotModeData'] = self._get_data(pdata, "!IBQ???????BBdd", ("size", "type", "timestamp", "isRobotConnected", "isRealRobotEnabled", "isPowerOnRobot", "isEmergencyStopped", "isSecurityStopped", "isProgramRunning", "isProgramPaused", "robotMode", "controlMode", "speedFraction", "speedScaling"))
elif psize == 46: # It's 46 bytes in 3.2 elif psize == 46: # It's 46 bytes in 3.2
self.version = (3,2) self.version = (3, 2)
allData['RobotModeData'] = self._get_data(pdata, "!IBQ???????BBdd", ("size", "type", "timestamp", "isRobotConnected", "isRealRobotEnabled", "isPowerOnRobot", "isEmergencyStopped", "isSecurityStopped", "isProgramRunning", "isProgramPaused", "robotMode", "controlMode", "speedFraction", "speedScaling", "speedFractionLimit")) allData['RobotModeData'] = self._get_data(pdata, "!IBQ???????BBdd", ("size", "type", "timestamp", "isRobotConnected", "isRealRobotEnabled", "isPowerOnRobot", "isEmergencyStopped", "isSecurityStopped", "isProgramRunning", "isProgramPaused", "robotMode", "controlMode", "speedFraction", "speedScaling", "speedFractionLimit"))
else: else:
allData["RobotModeData"] = self._get_data(pdata, "!iBQ???????Bd", ("size", "type", "timestamp", "isRobotConnected", "isRealRobotEnabled", "isPowerOnRobot", "isEmergencyStopped", "isSecurityStopped", "isProgramRunning", "isProgramPaused", "robotMode", "speedFraction")) allData["RobotModeData"] = self._get_data(pdata, "!iBQ???????Bd", ("size", "type", "timestamp", "isRobotConnected", "isRealRobotEnabled", "isPowerOnRobot", "isEmergencyStopped", "isSecurityStopped", "isProgramRunning", "isProgramPaused", "robotMode", "speedFraction"))
@ -104,9 +104,9 @@ class ParserUtils(object):
elif ptype == 7 and self.version >= (3, 2): elif ptype == 7 and self.version >= (3, 2):
allData["ForceModeData"] = self._get_data(pdata, "iBddddddd", ("size", "type", "x", "y", "z", "rx", "ry", "rz", "robotDexterity")) allData["ForceModeData"] = self._get_data(pdata, "iBddddddd", ("size", "type", "x", "y", "z", "rx", "ry", "rz", "robotDexterity"))
# elif ptype == 8: # elif ptype == 8:
#allData["varMessage"] = self._get_data(pdata, "!iBQbb iiBAcAc", ("size", "type", "timestamp", "source", "robotMessageType", "code", "argument", "titleSize", "messageTitle", "messageText")) # allData["varMessage"] = self._get_data(pdata, "!iBQbb iiBAcAc", ("size", "type", "timestamp", "source", "robotMessageType", "code", "argument", "titleSize", "messageTitle", "messageText"))
# elif ptype == 7: # elif ptype == 7:
#allData["keyMessage"] = self._get_data(pdata, "!iBQbb iiBAcAc", ("size", "type", "timestamp", "source", "robotMessageType", "code", "argument", "titleSize", "messageTitle", "messageText")) # allData["keyMessage"] = self._get_data(pdata, "!iBQbb iiBAcAc", ("size", "type", "timestamp", "source", "robotMessageType", "code", "argument", "titleSize", "messageTitle", "messageText"))
elif ptype == 20: elif ptype == 20:
tmp = self._get_data(pdata, "!iB Qbb", ("size", "type", "timestamp", "source", "robotMessageType")) tmp = self._get_data(pdata, "!iB Qbb", ("size", "type", "timestamp", "source", "robotMessageType"))
@ -217,11 +217,11 @@ class ParserUtils(object):
# ok we we have somehting which looks like a packet" # ok we we have somehting which looks like a packet"
return (data[:psize], data[psize:]) return (data[:psize], data[psize:])
else: else:
#packet is not complete # packet is not complete
self.logger.debug("Packet is not complete, advertised size is %s, received size is %s, type is %s", psize, len(data), ptype) self.logger.debug("Packet is not complete, advertised size is %s, received size is %s, type is %s", psize, len(data), ptype)
return None return None
else: else:
#self.logger.debug("data smaller than 5 bytes") # self.logger.debug("data smaller than 5 bytes")
return None return None
@ -314,7 +314,7 @@ class SecondaryMonitor(Thread):
self.logger.error("Robot not running: " + str(self._dict["RobotModeData"])) self.logger.error("Robot not running: " + str(self._dict["RobotModeData"]))
self.running = False self.running = False
with self._dataEvent: with self._dataEvent:
#print("X: new data") # print("X: new data")
self._dataEvent.notifyAll() self._dataEvent.notifyAll()
def _get_data(self): def _get_data(self):
@ -322,14 +322,14 @@ class SecondaryMonitor(Thread):
returns something that looks like a packet, nothing is guaranted returns something that looks like a packet, nothing is guaranted
""" """
while True: while True:
#self.logger.debug("data queue size is: {}".format(len(self._dataqueue))) # self.logger.debug("data queue size is: {}".format(len(self._dataqueue)))
ans = self._parser.find_first_packet(self._dataqueue[:]) ans = self._parser.find_first_packet(self._dataqueue[:])
if ans: if ans:
self._dataqueue = ans[1] self._dataqueue = ans[1]
#self.logger.debug("found packet of size {}".format(len(ans[0]))) # self.logger.debug("found packet of size {}".format(len(ans[0])))
return ans[0] return ans[0]
else: else:
#self.logger.debug("Could not find packet in received data") # self.logger.debug("Could not find packet in received data")
tmp = self._s_secondary.recv(1024) tmp = self._s_secondary.recv(1024)
self._dataqueue += tmp self._dataqueue += tmp