Synchronous and asynchronous motor movement

Question

I am working on an API that is mostly intended to be used interactively/little scripts. There are some classes with methods that could act slightly differently depending on the intention of the user, I have explored several options and would like to get some feedback on caveats, and alternative options.

I will use a simple example:

class Motor(object):
    def __init__(self, name):
        self.name = name
        self._blocking = True

    def move_to(self, position):
        self.block_until_idle() # protects from conflict in any case
        #internal move call here
        if self._blocking:
            self.block_until_idle()
        else:
            self._blocking = True

    def block_until_idle(self):
        #internal status query loop

I want to use an instance of Motor in two ways (dictated by the value of _blocking attribute)

• The movement call blocks until idle, because I want to take a picture at that position.
• The movement call returns ASAP, because I want to do other things meanwhile (for example move a completely different motor).

There are many different calls (not only move_to) that would follow this pattern.

The wish: Usage like:

motor = Motor("A")
motor2 = Motor("B")

motor.move_to(20)
# > wait until done

noblock(motor.move_to(50))  
# > returns immediately so I can move another
motor2.move_to(30)

So, in summary, the desire is that it is quick and fast to switch between behaviors, preferably within the same line of code and without lasting side effects.

Explored but discarded options:

Manually handling the flag variable: potentially dangerous (easy to forget), three liner every time.

Duplicate the methods into blocking and not blocking: API duplicates in size.

Context manager: Taking care of the flag and cleaning up the logic in the move_to method. Still a bit too verbose for my taste, specially for command line (I skip the context manager code). Note that the advantage of the context manager of being able to unblock many calls is not useful here, since the pre-blocking safety would kick in and only the last call would be affected. Also heavy to combine several unblocked motors.

with noblock(motor) as nbmotor:
    nbmotor.move_to(25)

Extra parameter: requires changes on each method, polluting call signature, still the best of the trivial options

def move_to(self, position, blocking = True):

Decoration / wrapper: I did not make it work so far in a way that would not scare users:

# syntax error, decorators do not work this way, and besides two liner
@noblock        
motor.move_to(20) 

# hi is evaluated before polite is applied
noblock(motor.move_to(15)) 

# almost, but problems to extract self inside the wrapper, the syntax starts to get tricky
noblock(motor.move_to) (20) 

# this works, but awful!
noblock(Motor.move_to) (motor2, 15) 

My best solution so far:

Some twisty implementation, but can be invoked in a single line.

class Motor(object):
    def __init__(self, name):
        self.name = name
        self._blocking = True
        self.noblock = Noblock(self)

    def move_to(self, position):
       self.block_until_idle() # protects from conflict in any case
       #internal move call here
       if self._blocking:
           self.block_until_idle()
       else:
           self._blocking = True

class Noblock(object):
    def __init__(self, other):
        self.other = other

    def __getattribute__(self, name):
        if name in ["move_to"]: #list of methods we intercept
            self.other._blocking = False
            return self.other.__getattribute__(name)
        else:
            return object.__getattribute__(self, name) 

motor = Motor("A")

motor.move_to(20)
# > wait
motor.noblock.move_to(20)
# > returns ASAP

Thoughts? Any wrapper/decoration strategy that I missed and does the same?

Answer 1

#almost, but problems to extract self inside the wrapper, the syntax starts to get tricky noblock(motor.move_to) (20)

What's tricky?

def noblock(motor_method,arg):
    motor_method.im_self._blocking = False
    motor_method(arg)

Or

def noblock(motor_method):
    motor_method.im_self._blocking = False
    return motor_method

Or catch non-existing attribute access in the Motor class.

def __getattr__(self,attr):
    if attr == 'unblock':
        self._blocking = False
        return self
    raise AttributeError('{} object has no attribute {}'.format(self.__class__.__name__,attr))  

Now,

motor.unblock.move_to(20)  

should work.

Answer 2

My suggestion would be to create a special result value (similar to Future) that would represent an asynchronous computation:

class Future(object):
  def result(self):
    raise NotImplementedError()

where result waits until the underlying computation finishes and returns the reuslt.

Then let all your methods return instances of Future asynchronously. If you want a synchronous invocation, just call

motor.move_to(position).result()

The actual implementation in Motor would depend on your needs, in particular:

• When calling two subsequent asynchronous computations, should the second block or should it be queued?
• Should result() block only until its corresponding call has finished, or until the motor is idle? The former makes more sense, but would require something different than block_until_idle.
• Should the class be thread-safe? (Your example code isn't.)