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 mailing list -- python-ideas@python.org
To unsubscribe send an email to python-ideas-le...@python.org
https://mail.python.org/mailman3/lists/python-ideas.python.org/
Message archived at
https://mail.python.org/archives/list/python-ideas@python.org/message/EP73EWIAGAXDLCMVFYVPPCNVUZMX7NAM/
Code of Conduct: http://python.org/psf/codeofconduct/
