I’m trying to communicate to some lab equipment that streams data to my PC through TCP. The equipment provider has given a .exe that is run with arguments to configure the set-up. Once the user wants to end the stream, a keyboard ctrl+c command must be sent in the terminal. This then tells the program to stop the stream, convert the data, save the data and then close. This works perfecty within the windows terminal as it does not immediately terminate the .exe. When the .exe is running and then a keyboard command sent:
I want to automate this process in Python. I want to load the .exe and then send a cancel command after a specific time, say 10 seconds. I can start the process in Python with the required arguments, but I cannot seem to end the process. All I can do is kill the process in Task Manager which stops the conversion of the data stream and I end up with no data:
import subprocess
import time
import signal
# Path to the executable and its arguments
exe_path = r"C:UserschrisDownloadsredpitayatestrpsa_client.exe"
arguments = [
"-s",
"-h", "169.254.219.196",
"-p", "8900",
"-f", "csv",
"-d", r"C:UserschrisDownloadsredpitayatesttest",
"-m", "volt"
]
#try:
result = subprocess.run(
[exe_path] + arguments,
cwd=r"C:UserschrisDownloadsredpitayatest", # Set the correct directory
shell=True,
capture_output=True,
text=True,
check=True
)
print("timer sstart:")
time.sleep(10)
print("Process ended")
subprocess.send_signal(signal.SIGINT)
What can I do to send a ctrl+c command to the process once the required time has elapsed?
3
You would need to call your rpsa_client.exe as part of a Process so that it can, effectively, run asynchronously.
With your current code, subprocess.run() will not return until the underlying executable terminates.
Here’s a simplified example that demonstrates a pattern that you could adapt:
from subprocess import run
import signal
from multiprocessing import Process
from time import sleep
from os import kill
def ping():
try:
run(["/sbin/ping", "www.google.com"])
except KeyboardInterrupt:
print("Interrupted")
if __name__ == "__main__":
# create the Process and start it
p = Process(target=ping)
p.start()
# wait a while
sleep(2)
# send SIGINT to the Process
kill(p.pid, signal.SIGINT) #type: ignore
# wait for the Process to end
p.join()
2