[Python-ideas] Re: Awaiting until a condition is met

May 31, 2022

      On 2022-05-14 10:10 p.m., Aaron Fink wrote:
...
start_moving_to(dest)
await lambda: distance_to(dest) <= tolerance
Instead of evaluating an expression you could build a notification chain 
that you await upon.

x = Variable(value=2)
y = Variable(value=2)

# we create a Future that watches x and y
condition = sqrt(x * x + y * y) <= 1

# some other task is launched to do work

await condition
print("done")

Here is an example implementation,

import asyncio
import math
import operator
import random
from asyncio import Future

class Variable(object):
     def __init__(self, value=None):
         self._value = value
         self.listeners = []

     @property
     def value(self):
         return self._value

     @value.setter
     def value(self, value):
         self._value = value
         for l in self.listeners:
             l.update()

     def __mul__(self, other):
         return Function(operator.mul, self, other)

     def __add__(self, other):
         return Function(operator.add, self, other)

     def __le__(self, other):
         return Boolean(operator.le, self, other)

     def __lt__(self, other):
         return Boolean(operator.lt, self, other)

class Function(Variable):
     def __init__(self, func, *params):
         Variable.__init__(self)
         self.func = func
         self.params = []
         for p in params:
             if not isinstance(p, Variable):
                 p = Variable(p)
             self.params.append(p)
             p.listeners.append(self)
         self.update()

     def update(self):
         self.value = self.func(*[p.value for p in self.params])

class Boolean(Future, Function):
     def __init__(self, func, *params):
         Function.__init__(self, func, *params)
         Future.__init__(self)

     def update(self):
         Function.update(self)
         if self.value and not self.done():
             self.set_result(self.value)

     def __bool__(self):
         return self._value

def sqrt(value):
     return Function(math.sqrt, value)

x = Variable(2)
y = Variable(2)

condition = sqrt(x * x + y * y) < 1

async def worker():
     while not condition:
         await asyncio.sleep(1)
         if random.Random().randrange(2):
             x.value = max(x.value-1, 0)
         else:
             y.value = max(y.value-1, 0)
         print(f"{x.value}, {y.value}")

async def watch():
     await condition
     print("watch done")

async def main():
     loop = asyncio.get_event_loop()
     t1 = loop.create_task(watch())
     t2 = loop.create_task(worker())
     await asyncio.wait([t1, t2])

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

[Python-ideas] Re: Awaiting until a condition is met

Kyle Lahnakoski