From 44e614c5094768252b785a18bbd7775fec777d6a Mon Sep 17 00:00:00 2001
From: iggy <iggy@muc.ccc.de>
Date: Wed, 24 May 2023 02:49:09 +0200
Subject: [PATCH] barebone mpy event handler with callbacks and conditions and
timers and shenanigans
---
python_payload/event.py | 129 ++++++++++++++++++++++++++++++++++++++++
1 file changed, 129 insertions(+)
create mode 100644 python_payload/event.py
diff --git a/python_payload/event.py b/python_payload/event.py
new file mode 100644
index 0000000000..9cf6342896
--- /dev/null
+++ b/python_payload/event.py
@@ -0,0 +1,129 @@
+import hardware
+import time
+import math
+
+
+EVENTTYPE_TIMED = 1
+EVENTTYPE_INPUT = 2
+#EVENTTYPE_BUTTON = 2
+#EVENTTYPE_CAPTOUCH = 3
+#EVENTTYPE_CAPCROSS = 4
+
+class Engine():
+ def __init__(self):
+ self.events_timed = []
+ self.events_input = []
+ self.next_timed = None
+ self.last_input_state = None
+
+ def add_timed(self,event):
+ self.events_timed.append(event)
+ self.events_timed = sorted(self.events_timed, key = lambda event: event.deadline)
+
+
+ def add_input(self,event):
+ self.events_input.append(event)
+
+ def _handle_timed(self):
+ if not self.next_timed and self.events_timed:
+ self.next_timed = self.events_timed.pop(0)
+
+ now = time.ticks_ms()
+
+ if self.next_timed:
+ diff = time.ticks_diff(self.next_timed.deadline, now)
+ if diff <= 0:
+ self.next_timed.trigger({"ticks_ms":now, "ticks_delay": -diff})
+ self.next_timed = None
+
+ def _handle_input(self):
+ input_state={
+ "b0":(hardware.get_button(0),"button",0),
+ "b1":(hardware.get_button(1),"button",1)
+ }
+
+ for i in range(0,10):
+ input_state["c"+str(i)]=(hardware.get_captouch(i),"captouch",i)
+
+
+ if not self.last_input_state:
+ self.last_input_state=input_state
+ return
+
+ diff=[]
+ for key in input_state:
+ if input_state[key][0] != self.last_input_state[key][0]:
+ diff.append({
+ "type" : input_state[key][1],
+ "index" : input_state[key][2],
+ "to" : input_state[key][0],
+ "from" : self.last_input_state[key][0],
+ "ticks_ms": time.ticks_ms()
+ })
+
+
+ if diff:
+ #print(diff)
+ for d in diff:
+ triggered_events = list(filter(lambda e: e.condition(d),self.events_input))
+ #print (triggered_events)
+ #map(lambda e: e.trigger(d), triggered_events)
+ for e in triggered_events:
+ e.trigger(d)
+
+
+ self.last_input_state=input_state
+
+
+
+ def _eventloop_single(self):
+ self._handle_timed()
+ self._handle_input()
+
+ def eventloop(self):
+ while True:
+ self._eventloop_single()
+ time.sleep(0.005)
+
+class Event():
+ def __init__(self,name="unknown",data={},action=None,condition=None):
+ #print (action)
+ self.name = name
+ self.eventtype = None
+ self.data = data
+ self.action = action
+ self.condition = condition
+ if not condition:
+ self.condition = lambda x: True
+
+ #print (data)
+
+ def trigger(self,triggerdata={}):
+ print ("triggered {} (with {})".format(self.name,triggerdata))
+ if not self.action is None:
+ triggerdata.update(self.data)
+ #print("trigger")
+ self.action(triggerdata)
+
+
+class EventTimed(Event):
+ def __init__(self,ms,name="timer", *args, **kwargs):
+ #super().__init__(name,data,action)
+ super().__init__(*args, **kwargs)
+ self.name=name
+ self.deadline = time.ticks_add(time.ticks_ms(),ms)
+ self.type=EVENTTYPE_TIMED
+
+ def __repr__(self):
+ return ("event on tick {} ({})".format(self.deadline,self.name))
+
+e = Engine()
+e.add_timed(EventTimed(200,name="bar",action=lambda data: print("GNANGNAGNA")))
+e.add_timed(EventTimed(100,name="foo"))
+e.add_input(Event(name="baz",
+ action=lambda data: print(data),
+ condition=lambda data: data.get('type')=="captouch")
+)
+
+print (e.events_timed)
+#e.eventloop()
--
GitLab