René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"QAPage","mainEntity":{"@type":"Question","name":"Access Rsync event","text":"
Hello,
\n
This may be fairly trivial and hopefully possible. I'm trying to listen to when a rsync event occurs while a task is being run, have a conditional catch it, and in my case have an LED turn on every time a pulse goes out.
\n
The issue is that when I do have a conditional listening to rsync events (which are successfully going out and logging), nothing occurs. I'm assuming I'm referencing it wrong.
\n
Below is a short example.
\n
  from pyControl.utility import *\n  from devices import *\n  import hardware_definition as hw\n\n  sync_output = Rsync(pin=board.BNC_1, event_name='rsync', mean_IPI=1000)\n\n  states = [\"trial\"]\n\n  events = [\"session_timer\", \"rsync\"]\n\n  initial_state = \"trial\"\n\n\n  def trial(event):\n\n      if event == \"rsync\":\n          #--------> Do something <--------------\n
","upvoteCount":1,"answerCount":3,"acceptedAnswer":{"@type":"Answer","text":"
The rsync event does not get processed by the state machine, it just goes straight to the data log. The line of interest is 
\n  
\n    
\n      code/source/pyControl/hardware.py\n    
\n    
\n         Line 509\n      in\n      e6fe329\n    
\n  
\n  
\n    \n\n        \n          \n          \n        \n    
 fw.data_output_queue.put(fw.Datatuple(fw.current_time, fw.EVENT_TYP, \"s\", self.event_ID)) 
\n  
\n
\n
\n
I believe this is intentional so the sync signal is as low overhead as possible? @ThomasAkam, should we add an option to the Rsync initialization for sending the event to the event queue? Something like...
\n
class Rsync(IO_object):\n    # Class for generating sync pulses with random inter-pulse interval.\n\n    def __init__(self, pin, event_name=\"rsync\", mean_IPI=5000, pulse_dur=50, process_event=False):\n        assert 0.1 * mean_IPI > pulse_dur, \"0.1*mean_IPI must be greater than pulse_dur\"\n        self.sync_pin = pyb.Pin(pin, pyb.Pin.OUT)\n        self.event_name = event_name\n        self.pulse_dur = pulse_dur  # Sync pulse duration (ms)\n        self.process_event = process_event\n        self.min_IPI = int(0.1 * mean_IPI)\n        self.max_IPI = int(1.9 * mean_IPI)\n        assign_ID(self)\n\n    def _initialise(self):\n        self.event_ID = sm.events[self.event_name] if self.event_name in sm.events else False\n\n    def _run_start(self):\n        if self.event_ID:\n            self.state = False  # Whether output is high or low.\n            self._timer_callback()\n\n    def _run_stop(self):\n        self.sync_pin.value(False)\n\n    def _timer_callback(self):\n        if self.state:  # Pin high -> low, set timer for next pulse.\n            timer.set(randint(self.min_IPI, self.max_IPI), fw.HARDW_TYP, \"\", self.ID)\n        else:  # Pin low -> high, set timer for pulse duration.\n            timer.set(self.pulse_dur, fw.HARDW_TYP, \"\", self.ID)\n            if self.process_event:\n                fw.event_queue.put(fw.Datatuple(fw.current_time, fw.EVENT_TYP, \"s\", self.event_ID))\n            else:\n                fw.data_output_queue.put(fw.Datatuple(fw.current_time, fw.EVENT_TYP, \"s\", self.event_ID))\n        self.state = not self.state\n        self.sync_pin.value(self.state)
","upvoteCount":1,"url":"https://github.com/orgs/pyControl/discussions/162#discussioncomment-14821569"}}}