Exit loop with KeyboardInterrupt then begin another loop in python-CodePudding

I currently have a piece of code designed to communicate with two motors via a TCP Device Server which sends ASCII commands to them in order to trace sinusoidal paths. I wish to have the movement continue indefinitely and then immediately stop when KeyboardInterrupt is triggered, then have the motors move back to their defined home positions before the program ends.

This code can currently replicate sinusodial motion, but it currently does not stop immediately when KeyboardInterrupt is triggered, nor do the motors move back to their home positions. The sinusodial loop is designed such that when KeyboardInterrupt occurs, a global variable called move changes from True to False and this breaks the loop. The simplified code is given below:

import socket
import time
import numpy as np
import math

pi = math.pi

try:
    s1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM) #TCP Server Connection
    s2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
except:
    print("Failed to connect")
    exit()
print("Sockets Created")

s1.connect(("192.168.177.200", 4001)) # Y motor
s2.connect(("192.168.177.200", 4002)) # X motor

# Disengage the motor to allow manual movement by wheel
s1.send("DI\n".encode("ASCII")) 
message = s1.recv(1024).decode()
s2.send("DI\n".encode("ASCII"))
message = s1.recv(1024).decode()



homeposition = input("Are the motors centred? Press 'y' to confirm: ")
if homeposition == 'y':
    s1.send("EN\n".encode("ASCII"))
    s2.send("EN\n".encode("ASCII")) # energise the motors so they cannot be moved
    print("Motors Engaged")
    s1.send("HO\n".encode("ASCII")) # set current position as home position
    s2.send("HO\n".encode("ASCII")) # set current position as home position
else:
    print("Set home position and restart program")
    exit()


#----ADD DATA FOR SAMPLING SINUSODIAL FUNCTIONS----

radius = input("Desired radius of movement (mm):")
radius = float(radius)
print("radius of circular path (mm): ", radius)

# need slightly different ratios for each motor due to different thread sizes
gearpositionlim_x = 20000 #one rotation equals 2.5mm (  bearing ration on manipulator of 2:1)
gearpositionlim_y = 10000 #one rotation equals 2mm



# sample sine and cosine
step = 2*pi / 1000
time_range = np.arange(0,2*pi   step,step)
x_motorrange = gearpositionlim_x*np.cos(time_range)
y_motorrange = gearpositionlim_y*np.sin(time_range)
x_motorrange = ['la' str(int(i)) for i in x_motorrange]
y_motorrange = ['la' str(int(i)) for i in y_motorrange]
#print(x_motorrange)

x_motorrange_wcom = []
y_motorrange_wcom = []
{x_motorrange_wcom.extend([e, 'm', 'np']) for e in x_motorrange} # add movement prompts and wait for movement to complete
{y_motorrange_wcom.extend([e, 'm', 'np']) for e in y_motorrange} # add movement prompts and wait for movement to complete

# Set Acceleration and Deceleration of Motors
s1.send("AC10\n".encode("ASCII"))
message = s1.recv(1024).decode()
s2.send("AC10\n".encode("ASCII"))
message = s2.recv(1024).decode()
print("Acceleration set to 10 ") 

s1.send("DEC10\n".encode("ASCII"))
message = s1.recv(1024).decode()
s2.send("DEC10\n".encode("ASCII"))
message = s2.recv(1024).decode()
print("Deceleration set to 10")

def setup():  #move to initial position before starting movement
    s2.send(str(str(x_motorrange_wcom[0]) "\n").encode("ASCII"))
    s2.send("m\n".encode("ASCII"))
    s2.send("np\n".encode("ASCII"))
    s2.send("delay200\n".encode("ASCII"))
    
def end():
    print("Movement ended, return to home position")
    s1.send("la0\n".encode("ASCII"))
    s1.send("m\n".encode("ASCII"))
    s1.send("np\n".encode("ASCII"))
    s1.send("delay200\n".encode("ASCII"))
    s1.send("DI\n".encode("ASCII"))
    s1.send("delay200\n".encode("ASCII"))
    time.sleep(2)
    s2.send("la0\n".encode("ASCII"))
    s2.send("m\n".encode("ASCII"))
    s2.send("np\n".encode("ASCII"))
    s2.send("delay200\n".encode("ASCII"))
    s2.send("DI\n".encode("ASCII"))
    s2.send("delay200\n".encode("ASCII"))


def motormove():
    global move
    try:
        for i in np.arange(0,len(x_motorrange)):
            if (move == True):
                s1.send(str(str(x_motorrange[i]) "\n").encode("ASCII"))
                s2.send(str(str(y_motorrange[i]) "\n").encode("ASCII"))
            else:
                break
    except KeyboardInterrupt:
        move = False
        print(move)
        end()
            

#-------------------------------------------
setup()
name = input("Code Ready, press enter to proceed: ")
if name == "":
        print("Code Running: Press ctrl   c to end")
        while (move == True):
            motormove()

I believe my issue is with my function motormove(), but I am unsure of what I should do in order to achieve my desired operation. Does anyone know how this can be achieved?

Thanks in advance

CodePudding user response：

Using library signal should be sufficient for your usecase. See code bellow.

import socket
import signal
import time
import numpy as np
import math

pi = math.pi
try:
    s1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM) #TCP Server Connection
    s2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
except:
    print("Failed to connect")
    exit()
print("Sockets Created")

s1.send("HO\n".encode("ASCII")) # set current position as home position
s2.send("HO\n".encode("ASCII")) # set current position as home position

gearpositionlim = int(10000)
# sample sine and cosine
step = 2*pi / 2000
time_range = np.arange(0,2*pi   step,step)
x_motorrange = gearpositionlim*np.sin(time_range)
y_motorrange = gearpositionlim*np.cos(time_range)

def handler(signum, frame):
    res = input("Ctrl-c was pressed. Do you really want to exit? y/n ")
    if res == 'y':
        exit(1) 
    else:
        #Do STH


def setup():
    s2.send(y_motorrange[0]) "\n").encode("ASCII"))
    s2.send("m\n".encode("ASCII"))
    s2.send("np\n".encode("ASCII"))
    s2.send("delay200\n".encode("ASCII"))

def end():
    print("Movement ended, return to home position")
    s1.send("la0\n".encode("ASCII"))
    s1.send("m\n".encode("ASCII"))
    s1.send("np\n".encode("ASCII"))
    s1.send("delay200\n".encode("ASCII"))
    s1.send("DI\n".encode("ASCII"))
    s1.send("delay200\n".encode("ASCII"))
    time.sleep(2)
    s2.send("la0\n".encode("ASCII"))
    s2.send("m\n".encode("ASCII"))
    s2.send("np\n".encode("ASCII"))
    s2.send("delay200\n".encode("ASCII"))
    s2.send("DI\n".encode("ASCII"))
    s2.send("delay200\n".encode("ASCII"))


def motormove():
    global move
    try:
        for i in np.arange(0,len(x_motorrange)):
            if (move == True):
                s1.send(str(str(x_motorrange[i]) "\n").encode("ASCII"))
                s2.send(str(str(y_motorrange[i]) "\n").encode("ASCII"))
            else:
                break
    except KeyboardInterrupt:
        signal.signal(signal.SIGINT, handler)

            

#-------------------------------------------
setup()
name = input("Code Ready, press enter to proceed: ")
if name == "":
        print("Code Running: Press ctrl   c to end")
        while (move == True):
            motormove()

This should be working just fine, note the signal function being called on KeyboardInterrupt, this redirects to signal function where, you can either exit or do something else.

CodePudding user response：

I would re-arrange your code to remove the duplication (note that the following is untested, but you should get the idea):

from socket import socket, AF_INET, SOCK_STREAM
from time import sleep
import numpy as np
import math

class Motor:
    def __init__(self, ip, port, gearposition_limit=10000, step_count=2000):
        self.socket = socket(AF_INET, SOCK_STREAM)
        self.socket.connect((ip, port))
        self.step = 2 * math.pi / step_count
        self.time_range = np.arange(0, 2 * math.pi   self.step, self.step)
        self.move_range = gearposition_limit * np.sin(self.time_range)
        self.reset()

    def send(self, *args):
        try:
            for c in args:
                self.socket.send(f'{c}\n'.encode("ASCII"))
        except KeyboardInterrupt:
            self.end()

    def reset(self):
        self.send(self.move_range[0], "m", "np", "delay200", "HO")

    def end(self):
        self.send("la0", "m", "np", "delay200", "DI", "delay200")

motor_x = Motor(ip="192.168.177.200", port=4002)
motor_y = Motor(ip="192.168.177.200", port=4001)

input("Ready. Press enter to proceed")
print("Running. Press CTRL C to end")

motor_x.send(*motor_x.move_range)
sleep(2)
motor_y.send(*motor_y.move_range)

If you want independent parallel movement on both motors, you will need to look into multi-threading.