A method to continuously target an angle and RuntimeError: can't start a new thread

[iwanimsand]

Hi, I have a car with an ackermann steering which I would like to control with a ps4 controller. I searched for a way to continuously target the angle delivered by the controller. But I can't find a satisfactory solution with the library at the moment. I hope that someone here can put me on the right track.

I found also a related question #175 (comment) asked by natikgadzhi:

On a similar note, let's say I'm using a motor for steering and accepting and queueing remote commands. If I queue a command run_for_degrees and it's not yet started executing, but another command run_for_degrees in different direction comes in, I should be able to go to hat.motorqueue[target_port].pop or delete another command from queue, right?

That's unless that command is currently executing. If it's already in progress, any way to see what command is in progress on each port, and interrupt it if needed?

Here is a little program which should simulate the behavior:

from buildhat import Motor, Hat
import time

h = Hat(debug=True)

motor_a = Motor('A')

start_time = time.time()

motor_a.run_to_position(degrees=0, speed=20, blocking=True, direction="shortest")

sleep_time_seconds = 0.05

angles_from_controller = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, -20, -40, -40, -39, -40, -35, -38, -35, -30, 0, 20, 40, 35, 35,
                          35, 35, 35, 35, 35, 35, 30, 0, 0, 0]

try:
    while True:
        for angle in angles_from_controller:
            motor_a.run_to_position(degrees=angle, speed=50, blocking=False, direction="shortest")
            time.sleep(sleep_time_seconds)

except RuntimeError as e:
    duration = time.time() - start_time
    print(f"-> Duration: {duration}")
    print(e)

Depending on how the sleep time was chosen, the sooner no more threads can be started.

I know that pybricks has a method https://docs.pybricks.com/en/stable/pupdevices/motor.html#pybricks.pupdevices.Motor.track_target or https://docs.pybricks.com/en/stable/pupdevices/motor.html#pybricks.pupdevices.Motor.run_target which does exactly what I would need.

Questions:

1. Is there already a way in the library to continuously target an angle of a motor as fast as possible for the given use case and I don't see it actually?
2. What would be the best way to realize this use case with the functions implemented by the library at the moment?
3. @chrisruk if there is a concept or existing idea how to implement something like a track_target method for motors in the library, maybe I could try to implement a draft of it - but I would have to know a little more first so that the whole thing would fit into the existing concepts and future development. Maybe there are also changes in a future firmware that would help with it.
   • A quick idea was to add a method in the library that doesn't start a thread and just sends the command to the build HAT. Because I don't expect any feedback, this would be enough. But I haven't checked this idea further and if this fits with the existing concepts at all.

Currently I'm trying to implement my project directly with the serial protocol of the Build HAT without using the Python library - but of course it would be great to be able to use the library.

Thank you for inputs and ideas

[chrisruk]

Hi,

Do you mind trying running your example on the 'main' version of buildhat in this repo (The change isn't released
yet to PyPi).

It implements a queuing system, so that you shouldn't get an error.

Does that help re. the questions?

Thanks

You can install from git doing

pip uninstall buildhat
pip install git+https://github.com/RaspberryPiFoundation/python-build-hat.git

[iwanimsand]

Hi,

I ran my example with the 'main' version of buildhat and my first observations are:

• The error is gone. I let it run for more than 5 minutes without a problem. Cool! 🥳 Thanks for the great work!
• The memory usage is growing and growing... slowly, but it's growing more and more. 🤔 I think this is the result of the queuing system and resulting growing queue because the commands can not be finished fast enough. But the methods I use are not designed for my use case and everything works as expected.

I tried a little other example:

from buildhat import Motor, Hat
import time

h = Hat(debug=True)

motor_a = Motor('A')

start_time = time.time()

motor_a.run_to_position(degrees=0, speed=20, blocking=True, direction="shortest")

sleep_time_seconds = 0.001

angles_from_controller = [-90, 0, 0, 0, 90, 0, 0, 0]

try:
    while True:
        for angle in angles_from_controller:
            motor_a.run_to_position(degrees=angle, speed=20, blocking=False, direction="shortest")
            time.sleep(sleep_time_seconds)

except RuntimeError as e:
    duration = time.time() - start_time
    print(f"-> Duration: {duration}")
    print(e)

You can see that the angles are [-90, 0, 0, 0, 90, 0, 0, 0]. If I have a controller which should control the steering of the car, I would expect that the -90 and 90 degrees are not reached, the motor should mostly bouncing around 0 degrees in this example. But the actual behavior is that the motor finishes every command in the queue and goes to -90 degrees, then 0, 0, 0 and then to fully 90 degrees and so on... This works as designed. (And fills the queue and the memory continuously. 😄 )

But for my use case I should have a way to delete (or overwrite) an actual command. I think the comment in #175 (comment) meant the same use case.

If I send a command run_to_position(degrees=-90, blocking=False, ...) and then a millisecond later the command run_to_position(degrees=90, blocking=False, ...) the first command should be overwritten somehow by the second. It means if the motor has reached -30 degrees and the second command comes in, it should not go to -90 degrees first and then to 90 degrees, it should stop on -30 and then directly execute the next command to 90 degrees. I know that the method run_to_position is not designed for this behavior.

There comes my third question in:
Is there a possible way or an idea how to implement a track_angle(degrees=x) method where I can throw in angles as fast as possible and the motor tries to reach this angle, but if a new angle comes in and the motor hasn't reached the actual angle, then ignore the actual angle and try to reach directly the new angle. IMHO for this use case, the queuing system will not work and it needs a way to bypass it.

Maybe I should try to implement a draft of such a method and we could discuss about it when I created a pull request.

[chrisruk]

"If I send a command run_to_position(degrees=-90, blocking=False, ...) and then a millisecond later the command run_to_position(degrees=90, blocking=False, ...) the first command should be overwritten somehow by the second"

That's a good question, I assume provided a command to move the motor hasn't been issued to the hat itself, there should be a way to do that.

It does look to me there aren't ways to alter the synchronised queue though, I can't see any public methods to do this - https://docs.python.org/3/library/queue.html.

I wonder if something like https://docs.python.org/3/library/collections.html#collections.deque may be useful here.

If you've got any suggestions/ideas/pull reqs. though, that'd be cool.

I'm just testing a new firmware currently, but can get back to thinking about this soon.