iipyper: @osc.map auto-mapping to Python Class functions via introspection

[jarmitage]

Automatically create OSC handlers for all functions in a given class, e.g.:

class MyClass:
 def func(arg):
 pass
my_instance = MyClass()

@osc.map('my_instance_name')
my_instance

Would create a handler at the address /myclass/func.

Issues:

• Can we get the class name, function names, and arguments via introspection?
• kwargs vs args: the rule could be "a function with any default args or a **kwargs in the signature uses osc.kwargs, otherwise use osc.args"

[jarmitage]

Relatedly, a utility class for decoupling updates from OSC handlers:

class Updater:
 '''
 Decouples OSC handling from updating
 Updating is rate-limited by a counter
 '''
 def __init__(self, state=None, count=10, update=False):
 self.count = count
 self.counter = 0
 self.update = update
 self.state = state
 def set(self, state):
 '''
 Set the Updater's state
 '''
 self.state = state
 self.update = True
 def __call__(self, f):
 '''
 Update the target function with internal state
 '''
 self.counter += 1
 if not (self.update and 
 self.counter>self.count and 
 self.state is not None):
 return
 f(self.state)
 self.counter = 0
 self.update = False

E.g.:

def main(host="127.0.0.1", port=4000):
 osc = OSC(host, port, verbose=False, concurrent=True)
 ti.init(arch=ti.vulkan)
 x = 1920
 y = 1080
 n = 8
 obs = tol.vera.Obstacles(x, y, n)
 window = ti.ui.Window("Obstacles", (x, y))
 canvas = window.get_canvas()

 obs_updater = Updater()
 @osc.args("/tolvera/obstacles/set")
 def _(address, *args):
 obs_updater.set(args)

 while window.running:
 with _lock:
 obs.render_clear()
 obs_updater(obs.set)
 obs.move()
 obs.render()
 canvas.set_image(obs.px_rgb)
 window.show()

[jarmitage]

Using inspect

import inspect

class MyClass:
 def function1(self, arg1, arg2):
 pass
 def function2(self, arg1):
 pass
methods = inspect.getmembers(MyClass, predicate=inspect.isfunction)
for method_name, method in methods:
 method_args = inspect.getfullargspec(method).args
 print(f"Method name: {method_name}, Method Arguments: {method_args}")

Method name: function1, Method Arguments: ['self', 'arg1', 'arg2']
Method name: function2, Method Arguments: ['self', 'arg1']

This will only display methods defined in the class itself and not inherited methods. If you want to include inherited methods, you can use the following function:

def get_inherited_members(cls, predicate):
 return [x for x in inspect.getmembers(cls, predicate) if not x[0].startswith('__')]

methods = get_inherited_members(MyClass, predicate=inspect.isfunction)

Using typing

import typing
def foo(name: str) -> int:
 # ...
typing.get_type_hints(foo)
> {'name': <class 'str'>, 'return': <class 'int'>}

[jarmitage]

WIP:

class ReceiveUpdater:
 '''
 Decouples event handling from updating
 Updating is rate-limited by a counter
 '''
 def __init__(self, f, state=None, count=5, update=False):
 self.f = f
 self.count = count
 self.counter = 0
 self.update = update
 self.state = state
 def set(self, state):
 '''
 Set the Updater's state
 '''
 self.state = state
 self.update = True
 def __call__(self):
 '''
 Update the target function with internal state
 '''
 self.counter += 1
 if not (self.update and 
 self.counter>self.count and 
 self.state is not None):
 return
 self.f(self.state)
 self.counter = 0
 self.update = False

class OSCReceiveUpdater(ReceiveUpdater):
 def __init__(self, osc, address: str, f, state=None, count=10, update=False):
 super().__init__(f, state, count, update)
 self.osc = osc
 self.address = address
 osc.add_handler(self.address, self.receive)
 def receive(self, address, *args):
 # FIXME: ip:port/args
 self.set(args[1:])

class OSCSendUpdater():
 def __init__(self, osc, address: str, f, count=30, client=None):
 self.osc = osc
 self.address = address
 self.f = f
 self.count = count
 self.counter = 0
 self.client = client
 def __call__(self):
 self.counter += 1
 if self.counter >= self.count:
 self.osc.send(self.address, *self.f(), client=self.client)
 self.counter = 0

class OSCReceiveUpdaters:
 '''
 o = OSCReceiveUpdaters(osc,
 {"/tolvera/particles/pos": s.osc_set_pos, 
 "/tolvera/particles/vel": s.osc_set_vel})
 '''
 def __init__(self, osc, receives=None, count=10):
 self.osc = osc
 self.receives = []
 self.count = count
 if receives is not None:
 self.add_dict(receives, count=self.count)
 def add_dict(self, receives, count=None):
 if count is None:
 count = self.count
 {a: self.add(a, f, count=count) for a, f in receives.items()}
 def add(self, address, function, state=None, count=None, update=False):
 if count is None:
 count = self.count
 self.receives.append(
 OSCReceiveUpdater(self.osc, address, function, \
 state, count, update))
 def __call__(self):
 [r() for r in self.receives]

class OSCSendUpdaters:
 '''
 o = OSCSendUpdaters(osc, client="particles", count=10,
 sends={
 "/tolvera/particles/get/pos/all": s.osc_get_pos_all
 })
 '''
 def __init__(self, osc, sends=None, count=10, client=None):
 self.osc = osc
 self.sends = []
 self.count = count
 self.client = client
 if sends is not None:
 self.add_dict(sends, self.count, self.client)
 def add_dict(self, sends, count=None, client=None):
 if count is None:
 count = self.count
 if client is None:
 client = self.client
 {a: self.add(a, f, count=count, client=client) for a, f in sends.items()}
 def add(self, address, function, state=None, count=None, update=False, client=None):
 if count is None:
 count = self.count
 if client is None:
 client = self.client
 self.sends.append(
 OSCSendUpdater(self.osc, address, function, count, client))
 def __call__(self):
 [s() for s in self.sends]

class OSCUpdaters:
 '''
 o = OSCUpdaters(osc, client="boids", count=10,
 receives={
 "/tolvera/boids/pos": b.osc_set_pos,
 "/tolvera/boids/vel": b.osc_set_vel
 },
 sends={
 "/tolvera/boids/pos/all": b.osc_get_all_pos
 }
 )
 '''
 def __init__(self, osc, sends=None, receives=None, count=10, client=None):
 self.osc = osc
 self.client = client
 self.count = count
 self.sends = OSCSendUpdaters(self.osc, count=self.count, client=self.client)
 self.receives = OSCReceiveUpdaters(self.osc, count=self.count)
 if sends is not None:
 self.add_sends(sends)
 if receives is not None:
 self.add_receives(receives)
 def add_sends(self, sends, count=None, client=None):
 if count is None:
 count = self.count
 if client is None:
 client = self.client
 self.sends.add_dict(sends, count, client)
 def add_send(self, send, count=None, client=None):
 if count is None:
 count = self.count
 if client is None:
 client = self.client
 self.sends.add(send, client=client, count=count)
 def add_receives(self, receives, count=None):
 if count is None:
 count = self.count
 self.receives.add_dict(receives, count=count)
 def add_receive(self, receive, count=None):
 if count is None:
 count = self.count
 self.receives.add(receive, count=count)
 def __call__(self):
 self.sends()
 self.receives()

[jarmitage]

Related: generating Max/MSP patches in Python https://github.com/shakfu/py2max

[jarmitage]

Another issue here is parameter ranges. Even if we can get the method names, arguments and their types, the parameter ranges are not by default in the source code. Probably normalisation of floats is required, which might make the Python code less iipyper agnostic

[jarmitage]

For parameter ranges, maybe something like this?

# pn=(default,min,max,step)
@osc.map(p1=(0.5,0,1,0.1))
def func(p1):
 # ...

• params are automatically clamped
• data used to generate OSC API & clients (e.g. Max)
• does having func & decorator arg with same name cause clash? how to associate them otherwise, arg ordering?

[jarmitage]

Worth considering https://github.com/ossia/libossia