Blinken – per Logik

Die Idee für die Logik kam aus meinem Alarmmodul. Dort sollten im Alarm-/Feuer-/Panikfall die Lichter an der Hauswand und die Gartenbeleuchtung im Wechsel blinken. Aber auch für die Signalisieren auf Taster LEDs könnte die Logik ihren Einsatz finden. Beispielsweise um im Schlafzimmer einen Status per LED auszugeben, ohne dass die LED die ganze Nacht leuchtet.

Somit war der Grundstein für die Logik gelegt, die möglichst flexibel, für unterschiedliche Anwendungen, konfigurierbar sein soll.

Funktion:

Gesteuert wird die Logik über ein control-item, welches zugleich in der Logik Konfiguration als watch-item dient. Wir das control-item auf True gesetzt, so läuft die Logik. False stoppt die Blinkroutine.

blink_cycles setzt Initial den Countdown, der mit jedem Aufruf um 1 nach unten zählt. Bei 0 stoppt die Logik und setzt zudem das control-item auf False. 

blink_interval bestimmt die Blinkabstand in Sekunden. 

Mit items_cyclic und items_anticyclic werden die Items übergeben, die blinken sollen. Antizyklische Items werden immer dann aus/an wenn die zyklischen Items an/aus sind. 

Bevor die items zum blinken gebracht werden, sichert die Logik die ursprünglichen Werte, die auch am Ende wieder hergestellt werden.

Um die Routine ohne aktives schalten der Items testen zu können, wurde noch dry_run aufgenommen.

Zusammengefasst ergibt sich folgende Beispiel-Konfiguration für ../etc/logic.yaml:

visual_alarm:
    filename: blinker.py
    watch_item: HOME.alarm.action.blinkinglight
    set_items_cyclic:
        - AUSSEN.south.light.outer
        - AUSSEN.west.light.outer
    set_items_anticyclic:
        - AUSSEN.south.light.inner
        - AUSSEN.east.light.door
        - AUSSEN.west.light.nested
    set_blink_cycles: 4 # -1 for endless loop 
    set_blink_interval: 4
    dry_run: 1

Alle items müssen vom Typ boolean sein, da nur einfache Schaltoperationen durchgeführt werden.

Die Logik lässt sich relativ einfach nutzen, indem das im watch_item verbundene Item, also im Beispiel HOME.alarm.action.blinkinglight, auf True gesetzt wird. Die Logik ruft sich selbst solange auf, bis set_blink_cycles erreicht ist oder vorher das watch_item auf False gesetzt wurde. Am Ende wird das watch_item auf von der Logik auf False gesetzt.

Die Logik selbst ist in ../logics/blinker.py:

#!/usr/bin/env python3
# vim: tabstop=2 softtabstop=2 shiftwidth=2 expandtab

# Logic blinker.py

# LOGIC config
# -------------------------------------------------------
# visual_alarm:
#   filename: blinker.py
#   watch_item: HOME.alarm.action.blinkinglight
#   set_items_cyclic:
#     - AUSSEN.south.light.outer
#     - AUSSEN.west.light.outer
#   set_items_anticyclic: 
#     - AUSSEN.south.light.inner
#     - AUSSEN.east.light.door 
#     - AUSSEN.west.light.nested
#   set_blink_cycles: -1 " -1: endless 
#   set_blink_interval: 2
#   dry_run: 0 

control_item_name = trigger['source'] 
control_item = sh.return_item(control_item_name)
triggered_by = trigger['by'] 


if control_item is not None:
  logger.info("LOGIC blinker.py: {0} triggered from source {1} by {2}".format(logic.name, control_item_name, triggered_by))

  # control_item_active = trigger['value']
  control_item_active = control_item()

  # set parameters from logic config
  if hasattr(logic, 'set_items_cyclic'):
    blink_items_cyclic = logic.set_items_cyclic

  if hasattr(logic, 'set_items_anticyclic'):
    blink_items_anticyclic = logic.set_items_anticyclic

  if hasattr(logic, 'set_blink_cycles'):
    blink_cycles = int(logic.set_blink_cycles)
  else:
    blink_cycles = 2
    logger.info("LOGIC blinker.py: no blink cycles set, setting to default: {2}".format(blink_cycles))

  if hasattr(logic, 'set_blink_interval'):
    blink_interval = int(logic.set_blink_interval)
  else:
    blink_interval = 2
    logger.info("LOGIC blinker.py: no blink interval set, setting to default: {2}".format(blink_interval))

  if hasattr(logic, 'dry_run'):
    dry_run = (int(logic.dry_run) > 0)
  else:
    dry_run = False

  if control_item_active == True:

    # first call?
    if not hasattr(logic, 'blink_init'):
      logger.info("LOGIC blinker.py: initiating {0} blinks for {1} cyclic and {2} anti-cyclic items".format(blink_cycles, len(blink_items_cyclic), len(blink_items_anticyclic)))
      logic.blink_init = True

      blink_items = blink_items_cyclic + blink_items_anticyclic
      if len(blink_items) > 0:
        # SAVE item states
        logic.saved_items = {}
        for item_name in blink_items:
          item = sh.return_item(item_name)
          logic.saved_items[item_name] = item()
          logger.info("LOGIC blinker.py: save {0} = {1}".format(item_name, item()))

      if dry_run == True:
        logger.warn("LOGIC blinker.py: test mode, dry run active")

      else:
        logger.warn("LOGIC blinker.py: no blink items set")
 

    # prepare blink
    if hasattr(logic, 'blink_value'):
      logic.blink_value = not logic.blink_value
    else:
      logic.blink_value = True

    if hasattr(logic, 'blink_countdown'):
      logic.blink_countdown -= 1
    else:
      logic.blink_countdown = blink_cycles
     
    # any blinks left?
    if logic.blink_countdown > 0 or blink_cycles == -1:
      if blink_cycles == -1:
        logger.info("LOGIC blinker.py BLINK (endless)")
      else:
        logger.info("LOGIC blinker.py BLINK ({0} cycles remaining)".format(logic.blink_countdown))

      # set all cyclic items
      for item_name in blink_items_cyclic:
        item = sh.return_item(item_name)
        logger.info("LOGIC blinker.py SET {item} to {value})".format(item=item_name, value=logic.blink_value))
        if dry_run == False:
          item(logic.blink_value)

      # set all anti-cyclic items
      for item_name in blink_items_anticyclic:
        item = sh.return_item(item_name)
        logger.info("LOGIC blinker.py SET {item} to {value})".format(item=item_name, value=(not logic.blink_value)))
        if dry_run == False:
          item(not logic.blink_value)

      sh.trigger("logics.{0}".format(logic.name), by=trigger['by'], source=trigger['source'], value=trigger['value'], dt=sh.now() + datetime.timedelta(seconds=blink_interval))

    else:
      logger.info("LOGIC blinker.py countdown finished, set control item to False")
      control_item(False) # this triggers logic again!

  else:
    logger.info("LOGIC blinker.py control item set to false: stop blink logic, restore saved item values")

    # RESTORE item states
    if hasattr(logic, 'saved_items'):
      for item_name in logic.saved_items:
        item = sh.return_item(item_name)
        value = logic.saved_items[item_name]
        logger.info("LOGIC blinker.py: restore {0} = {1}".format(item_name, item()))
        if dry_run == False:
          item(value)
      
    # reset persistent values
    if hasattr(logic, 'saved_items'):
      delattr(logic, 'saved_items')  
    if hasattr(logic, 'blink_countdown'):
      delattr(logic, 'blink_countdown')  
    if hasattr(logic, 'blink_value'):
      delattr(logic, 'blink_value')  
    if hasattr(logic, 'blink_init'):
      delattr(logic, 'blink_init')  
    
else:
  logger.info("LOGIC blinker.py triggered without source... ignoring...")

Überarbeitung im Januar 2021:

Diese Logik wurde aufgrund weitere Anforderungen im Januar 2021 ergänzt. (siehe Thread im Forum) Die folgende Version vom 18. Januar 2021 blinkt mit einem oder zwei Items:

#!/usr/bin/env python3
# vim: tabstop=4 softtabstop=4 shiftwidth=4 expandtab

# Logic blinker2.py

# LOGIC config
# -------------------------------------------------------
# visual_alarm:
#     filename: blinker2.py
#     watch_item: HOME.alarm.action.blinkinglight
#     set_items_cyclic:
#         - AUSSEN.south.light.outer
#         - AUSSEN.west.light.outer
#     set_items_anticyclic:
#         - AUSSEN.south.light.inner
#         - AUSSEN.east.light.door
#         - AUSSEN.west.light.nested
#     set_blink_cycles: -1 ' -1: endless
#     set_blink_interval: 2
#     dry_run: 0

control_item_name = trigger['source']
control_item = sh.return_item(control_item_name)
triggered_by = trigger['by']

if control_item is not None:
    logger.info(f'LOGIC blinker.py: {logic.name} triggered from source {control_item_name} by {triggered_by}')

    # control_item_active = trigger['value']
    control_item_active = control_item()

    # set parameters from logic config
    blink_items_cyclic = []
    if hasattr(logic, 'set_items_cyclic'):
        blink_items_cyclic = logic.set_items_cyclic
    if blink_items_cyclic == 'None':
        blink_items_cyclic = []
    if isinstance(blink_items_cyclic, str):
        blink_items_cyclic = [blink_items_cyclic, ]

    # do we have blink items?
    if not isinstance(blink_items_cyclic, list) or len(blink_items_cyclic) == 0:
        logger.info('LOGIC blinker.py: error, no cyclic blink items set. Aborting')
    else:

        blink_items_anticyclic = []
        if hasattr(logic, 'set_items_anticyclic'):
            blink_items_anticyclic = logic.set_items_anticyclic
        if blink_items_anticyclic == 'None':
            blink_items_anticyclic = []
        elif isinstance(blink_items_anticyclic, str):
            blink_items_anticyclic = [blink_items_anticyclic, ]

        # did we get strange data?
        if not isinstance(blink_items_anticyclic, list):
            blink_items_anticyclic = []

        if hasattr(logic, 'set_blink_cycles'):
            blink_cycles = int(logic.set_blink_cycles)
        else:
            blink_cycles = 2
            logger.info(f'LOGIC blinker.py: no blink cycles set, setting to default: {blink_cycles}')

        if hasattr(logic, 'set_blink_interval'):
            blink_interval = int(logic.set_blink_interval)
        else:
            blink_interval = 2
            logger.info(f'LOGIC blinker.py: no blink interval set, setting to default: {blink_interval}')

        if hasattr(logic, 'dry_run'):
            dry_run = (int(logic.dry_run) > 0)
        else:
            dry_run = False

        if control_item_active:

            # first call?
            if not hasattr(logic, 'blink_init'):
                logger.info(f'LOGIC blinker.py: initiating {blink_cycles} blinks for {len(blink_items_cyclic)} cyclic and {len(blink_items_anticyclic)} anti-cyclic items')
                logic.blink_init = True

                blink_items = blink_items_cyclic + blink_items_anticyclic
                if len(blink_items) > 0:
                    # SAVE item states
                    logic.saved_items = {}
                    for item_name in blink_items:
                        item = sh.return_item(item_name)
                        if item:
                            logic.saved_items[item_name] = item()
                            logger.info(f'LOGIC blinker.py: save {item_name} = {item()}')

                    if dry_run:
                        logger.warn('LOGIC blinker.py: test mode, dry run active')

                else:
                    logger.warn('LOGIC blinker.py: no blink items set')

            # prepare blink
            if hasattr(logic, 'blink_value'):
                logic.blink_value = not logic.blink_value
            else:
                logic.blink_value = True

            if hasattr(logic, 'blink_countdown'):
                logic.blink_countdown -= 1
            else:
                logic.blink_countdown = blink_cycles

            # any blinks left?
            if logic.blink_countdown > 0 or blink_cycles == -1:
                if blink_cycles == -1:
                    logger.info('LOGIC blinker.py BLINK (endless)')
                else:
                    logger.info(f'LOGIC blinker.py BLINK ({logic.blink_countdown} cycles remaining)')

                # set all cyclic items
                for item_name in blink_items_cyclic:
                    item = sh.return_item(item_name)
                    logger.info(f'LOGIC blinker.py SET {item_name} to {logic.blink_value}')
                    if not dry_run:
                        item(logic.blink_value)
                        print(f'    - {item.id()}:{item()}')

                # set all anti-cyclic items
                for item_name in blink_items_anticyclic:
                    item = sh.return_item(item_name)
                    logger.info(f'LOGIC blinker.py SET {item_name} to {not logic.blink_value}')
                    if not dry_run:
                        item(not logic.blink_value)
                        print(f'    - {item.id()}:{item()}')

                sh.trigger(f'logics.{logic.name}', by=trigger['by'], source=trigger['source'], value=trigger['value'], dt=sh.now() + datetime.timedelta(seconds=blink_interval))

            else:
                logger.info('LOGIC blinker.py countdown finished, set control item to False')
                control_item(False)  # this triggers logic again!

        else:
            logger.info('LOGIC blinker.py control item set to false: stop blink logic, restore saved item values')

            # RESTORE item states
            if hasattr(logic, 'saved_items'):
                for item_name in logic.saved_items:
                    item = sh.return_item(item_name)
                    value = logic.saved_items[item_name]
                    logger.info(f'LOGIC blinker.py: restore {item_name} = {item()}')
                    if not dry_run:
                        item(value)

            # reset persistent values
            if hasattr(logic, 'saved_items'):
                delattr(logic, 'saved_items')
            if hasattr(logic, 'blink_countdown'):
                delattr(logic, 'blink_countdown')
            if hasattr(logic, 'blink_value'):
                delattr(logic, 'blink_value')
            if hasattr(logic, 'blink_init'):
                delattr(logic, 'blink_init')

else:
    logger.info('LOGIC blinker.py triggered without source... ignoring...')