original_docstring: callback to check any zwave status updates
docstring: callback to check any zwave status updates
docstring_tokens: callback to check any zwave status updates
code: def zwave_callback(self, entity, attribute, old, new, kwargs):
      """callback to check any zwave status updates"""
      # make sure we care about htis device
      manufacturer = old["attributes"]["manufacturer_name"]
      product = old["attributes"]["product_name"]
      if (manufacturer, product) not in self.device_types:
        return
  
      # look up relevant light info
      light = f"light.{self.split_entity(entity)[1]}"
      new_state = self.state(light)
      old_state = self.states.get(light, {})
  
      # abort if there isn't a corresponding light
      if not new_state:
        return
  
      # only log the first and last time
      if not old_state:
        self.tracking[light] = 0
        self.log(f"tracking {light}")
  
      # track how many times we've matched
      count = self.counts.pop(light, 0)
      if new_state == old_state:
        count += 1
        self.counts[light] = count
  
      # we've stopped moving/polled long enough, clean up and stop polling
      if count >= self.COUNT or self.tracking[light] > self.MAX_POLLS:
        if count >= self.COUNT:
          self.log(f"{light} now stable after {self.tracking[light]} polls")
        else:
          self.log(f"tracking of {light} aborted after {self.MAX_POLLS} polls")
        self.states.pop(light, None)
        self.counts.pop(light, None)
        self.tracking.pop(light, None)
        return
  
      # still moving, save the current state and schedule another probe
      if light not in self.scheduled:
        #self.log(f"scheduling refresh for {light}")
        self.states[light] = new_state
        self.tracking[light] += 1
        self.scheduled.add(light)
        self.run_in(self.refresh_callback, self.DELAY, entity_id=light)
code_tokens: def zwave_callback ( self , entity , attribute , old , new , kwargs ) : manufacturer = old [ "attributes" ] [ "manufacturer_name" ] product = old [ "attributes" ] [ "product_name" ] if ( manufacturer , product ) not in self . device_types : return light = f"light.{self.split_entity(entity)[1]}" new_state = self . state ( light ) old_state = self . states . get ( light , { } ) if not new_state : return if not old_state : self . tracking [ light ] = 0 self . log ( f"tracking {light}" ) count = self . counts . pop ( light , 0 ) if new_state == old_state : count += 1 self . counts [ light ] = count if count >= self . COUNT or self . tracking [ light ] > self . MAX_POLLS : if count >= self . COUNT : self . log ( f"{light} now stable after {self.tracking[light]} polls" ) else : self . log ( f"tracking of {light} aborted after {self.MAX_POLLS} polls" ) self . states . pop ( light , None ) self . counts . pop ( light , None ) self . tracking . pop ( light , None ) return if light not in self . scheduled : self . states [ light ] = new_state self . tracking [ light ] += 1 self . scheduled . add ( light ) self . run_in ( self . refresh_callback , self . DELAY , entity_id = light )
short_docstring: callback to check any zwave status updates
short_docstring_tokens: callback to check any zwave status updates
comment: """callback to check any zwave status updates"""$SEP$# make sure we care about htis device$SEP$# look up relevant light info$SEP$# abort if there isn't a corresponding light$SEP$# only log the first and last time$SEP$# track how many times we've matched$SEP$# we've stopped moving/polled long enough, clean up and stop polling$SEP$# still moving, save the current state and schedule another probe$SEP$#self.log(f"scheduling refresh for {light}")