NSPanel Logic for shNG

__schnell_mal.py

#!/usr/bin/env python3
# __schnell_mal.py

import time
import yaml
from lib.item import Items
items = Items.get_instance()

# Define Tasmota Topic
tasmota_topic = "NSPanel1"

# Define location of config file
page_config_file = "/usr/local/smarthome/logics/nspanel_pages.yaml"

# Generate Full Topics
nspanel_mqtt_topic_out_custom = f"cmnd/{tasmota_topic}/CustomSend"
nspanel_mqtt_topic_in_result = f"tele/{tasmota_topic}/RESULT"
nspanel_mqtt_topic_in_power = f"stat/{tasmota_topic}/POWER"
nspanel_mqtt_topic_in_result2 = f"stat/{tasmota_topic}/RESULT"


################################
# MQTT Stuff
################################


def logic_publish_topic(topic: str = nspanel_mqtt_topic_out_custom, payload: str = ""):
    # logger.warning(f"Function '{inspect.stack()[0][3]}()' - called by '{inspect.stack()[1][3]}()' in logic '{logic.name}'")
    if mqtt.publish_topic(logic.name, topic, payload):
        logger.warning(f"Function '{inspect.stack()[0][3]}()' - topic={topic} with payload={payload} was published.")
    else:
        logger.warning(f"Function '{inspect.stack()[0][3]}()' - topic={topic} with payload={payload} NOT published")


def _logic_subscribe_topic(topic, payload_type: str = 'str'):
    logger.warning(f"Function '{inspect.stack()[0][3]}()' - called by '{inspect.stack()[1][3]}()' with topic={topic}, payload_type={payload_type}")
    mqtt.subscribe_topic(logic.name, topic, callback=logic.name, payload_type=payload_type)


def _logic_unsubscribe_topic(topic):
    logger.warning(f"Function '{inspect.stack()[0][3]}()' - called by '{inspect.stack()[1][3]}()' with topic={topic}")
    mqtt.unsubscribe_topic(logic.name, topic)


def subscribe_topics():
    _logic_subscribe_topic(nspanel_mqtt_topic_in_result, 'dict')
    _logic_subscribe_topic(nspanel_mqtt_topic_in_power, 'str')
    _logic_subscribe_topic(nspanel_mqtt_topic_in_result2, 'dict')


def unsubscribe_topics():
    _logic_unsubscribe_topic(nspanel_mqtt_topic_in_result)
    _logic_unsubscribe_topic(nspanel_mqtt_topic_in_power)
    _logic_unsubscribe_topic(nspanel_mqtt_topic_in_result2)


################################
# Logic Support Stuff
################################


def parse_config_file():
    with open(page_config_file, 'r') as stream:
        try:
            config = yaml.safe_load(stream)
        except yaml.YAMLError as exc:
            logger.warning(f"Exception during parsing of  page config yaml file occurred: {exc}")
            raise LeaveLogic(f"Logic aborted: Exception '{exc}' during import of page config yaml file.")
        else:
            logger.warning(f"config={config} available!")
            return config


def next_page():
    logic.current_page += 1
    if logic.current_page >= len(logic.config['cards']):
        logic.current_page -= len(logic.config['cards'])
    logger.warning(f"next_page={logic.current_page}")
    return logic.current_page


def previous_page():
    logic.current_page -= 1
    if logic.current_page < 0:
        logic.current_page += len(logic.config['cards'])
    logger.warning(f"previous_page={logic.current_page}")
    return logic.current_page


def send_page(page):
    logger.warning(f"send_page called for page={page}")

    # Payload vorbereiten: Aufbau: entityUpd~heading~navigationLeft|navigationRight~[~*type*~*internalNameEntity*~*iconId*~*iconColor*~*displayNameEntity*~*optionalValue*]
    payload = f"entityUpd~{page['heading']}~{page['navigationLeft']}|{page['navigationRight']}"
    for entity in page['entities']:
        logger.warning(f"entity={entity}")

        item = _get_item(entity['internalNameEntity'])
        value = item() if item else entity['optionalValue']

        payload = f"{payload}~{entity['type']}~{entity['internalNameEntity']}~{entity['iconId']}~{entity['iconColor']}~{entity['displayNameEntity']}~{value}"
    
    # Bildschirm vorbereiten und anzeigen
    logic_publish_topic(payload=f"pageType~{page['pageType']}")
    logic_publish_topic(payload=payload)


def _get_item(entityName):
    item = items.return_item(entityName)
    if item:
        logger.warning(f"item={item} for entityName={entityName} identified")
    else:
        logger.warning(f"No corresponding item for entityName={entityName} identified")
    return item


def _set_current_time():
    logic_publish_topic(payload=f"time~{time.strftime('%H:%M', time.localtime())}")


def _set_current_date():
    logic_publish_topic(payload=f"date~{time.strftime('%A, %d.%B %Y', time.localtime())}")


def init_panel():
    logger.warning(f"init_panel called")

    # set time
    _set_current_time()

    # set date
    _set_current_date()

    # set screensavertimeout
    screensavertimeout = logic.config.get('config', {}).get('screensaver_timeout', 10)
    logic_publish_topic(payload=f"timeout~{screensavertimeout}")

    # set panel brightness
    brightness_screensaver = logic.config.get('config', {}).get('brightness_screensaver', 10)
    brightness_active = logic.config.get('config', {}).get('brightness_active', 100)
    logic_publish_topic(payload=f"dimmode~{brightness_screensaver}~{brightness_active}~6371")

    # Screensaver:
    logic_publish_topic(payload="pageType~screensaver")
    heading = logic.config.get('screensaver', {}).get('heading', '')
    text = logic.config.get('screensaver', {}).get('text', '')
    logic_publish_topic(payload=f"notify~{heading}~{text}")

    # logic_publish_topic(payload="weatherUpdate~tMainIcon~tMainText~tForecast1~tF1Icon~tForecast1Val~tForecast2~tF2Icon~tForecast2Val~tForecast3~tF3Icon~tForecast3Val~tForecast4~tF4Icon~tForecast4Val~optionalLayoutIcon~optionalLayoutText")
    # logic_publish_topic(payload="color~background~time~timeAMPM~date~tMainIcon~tMainText~tForecast1~tForecast2~tForecast3~tForecast4~tF1Icon~tF2Icon~tF3Icon~tF4Icon~tForecast1Val~tForecast2Val~tForecast3Val~tForecast4Val~bar~tMRIcon~tMR")


def callback(payload):
    """
    # general
    event,buttonPress2,pageName,bNext
    event,buttonPress2,pageName,bPrev
    event,buttonPress2,pageName,bExit,number_of_taps
    event,buttonPress2,pageName,sleepReached
    # startup page
    event,startup,version,model
    # screensaver page
    event,buttonPress2,screensaver,exit - Touch Event on Screensaver
    event,screensaverOpen - Screensaver has opened
    # cardEntities Page
    event,*eventName*,*entityName*,*actionName*,*optionalValue*
    event,buttonPress2,internalNameEntity,up
    event,buttonPress2,internalNameEntity,down
    event,buttonPress2,internalNameEntity,stop
    event,buttonPress2,internalNameEntity,OnOff,1
    event,buttonPress2,internalNameEntity,button
    # popupLight Page
    event,pageOpenDetail,popupLight,internalNameEntity
    event,buttonPress2,internalNameEntity,OnOff,1
    event,buttonPress2,internalNameEntity,brightnessSlider,50
    event,buttonPress2,internalNameEntity,colorTempSlider,50
    event,buttonPress2,internalNameEntity,colorWheel,x|y|wh
    # popupShutter Page
    event,pageOpenDetail,popupShutter,internalNameEntity
    event,buttonPress2,internalNameEntity,positionSlider,50
    # popupNotify Page
    event,buttonPress2,*internalName*,notifyAction,yes
    event,buttonPress2,*internalName*,notifyAction,no
    # cardThermo Page
    event,buttonPress2,*entityName*,tempUpd,*temperature*
    event,buttonPress2,*entityName*,hvac_action,*hvac_action*
    # cardMedia Page
    event,buttonPress2,internalNameEntity,media-back
    event,buttonPress2,internalNameEntity,media-pause
    event,buttonPress2,internalNameEntity,media-next
    event,buttonPress2,internalNameEntity,volumeSlider,75
    # cardAlarm Page
    event,buttonPress2,internalNameEntity,actionName,code
    """
    # payload = {'CustomRecv': 'event,buttonPress2,EG.Arbeiten.Licht,OnOff,1'}
    # mosquitto_pub -t 'tele/NSPanel1/RESULT' -m '{"CustomRecv": "event,buttonPress2,light.schreibtischlampe,OnOff,0"}'

    try:
        content_list = payload["CustomRecv"].split(',')
    except Exception as e:
        logger.warning(f"During handling of payload exception occurred: {e}")
        raise LeaveLogic(f"During handling of payload exception occurred: {e}")

    logger.warning(f"content_list={content_list}, length of content_list={len(content_list)}")
    event = content_list[0]
    eventName = content_list[1]

    if eventName == "startup":
        version = content_list[2]
        model = content_list[3]
        logger.warning(f"NSPanel started. version={version}, model={version}")

    elif eventName == "screensaverOpen":
        logger.warning(f"Screensaver has opened")

    elif eventName == "buttonPress2":
        entityName = content_list[2]
        actionName = content_list[3]
        optionalValue = content_list[4] if len(content_list) > 4 else None

        item = _get_item(entityName)

        if actionName == 'exit':
            if entityName == 'screensaver':
                logger.warning("Touch Event on Screensaver")

        elif actionName == "bExit":
            logger.warning(f"Display exited Standby from pageName={entityName}, Page will be reloaded.")
            send_page(logic.config['cards'][logic.current_page])

        elif actionName == "bNext":
            logger.warning(f"Show next Page from pageName={entityName}")
            send_page(logic.config['cards'][next_page()])

        elif actionName == "bPrev":
            logger.warning(f"Show previous Page from pageName={entityName}")
            send_page(logic.config['cards'][previous_page()])

        elif actionName == "sleepReached":
            logger.warning(f"Display went into Standby from pageName={entityName}")

        elif actionName == "up":
            logger.warning(f"ActionName=up for entityName={entityName} and item={item} called")

        elif actionName == "down":
            logger.warning(f"ActionName=down for entityName={entityName} and item={item} called")

        elif actionName == "stop":
            logger.warning(f"ActionName=stop for entityName={entityName} and item={item} called")

        elif actionName == "OnOff":
            if item and optionalValue:
                logger.warning(f"item={item.id()} will be set to new value={optionalValue}")
                item(optionalValue)

        elif actionName == "Button":
            logger.warning("Button pressed for entityName={entityName}")


################################
# Logic Main Code
################################


logger.warning(f"Logic {logic.id()} triggered: trigger['source'] = {trigger['source']}, trigger[by] = {trigger['by']}, trigger[value] = {trigger['value']}")

# Prüfen, ob mqtt Objekt verfügbar
if mqtt is None:
    # no MQTT support available
    raise LeaveLogic("Logic aborted: MQTT module is not loaded or not yet initialized")
    
# Einlesen der PageDefinition aus yaml (wenn noch nicht vorhanden)
if not hasattr(logic, 'config'):
    logic.config = parse_config_file()

# Define current page (wenn noch nicht vorhanden)
if not hasattr(logic, 'current_page'):
    logic.current_page = 0

# Neues Abo auf eingehende Topics
subscribe_topics()

# Ausloeser der Logik ist der Scheduler oder das AdminIF // Initialisieren der NSPanels
if trigger['by'].lower() == 'scheduler' or trigger['by'].lower() == 'admin':
    init_panel()

# Ausloeserquelle der Logik ist das mqtt Objekt // Rückmeldung des NSPanels in Funktion verarbeiten
elif trigger['source'] == 'mqtt':
    # callback received
    topic = trigger['by']
    payload = trigger['value']
    (topic_type, tasmota_topic, info_topic) = topic.split('/')
    logger.warning(f"MQTT received topic '{topic}': payload = '{payload}' - type(payload) = {type(payload)})")

    if info_topic.startswith('POWER'):
        tasmota_relay = str(info_topic[5:])
        tasmota_relay = '1' if not tasmota_relay else None
        logger.warning(f"Power Message Received for {info_topic} with value {payload}")

    callback(payload)
    

nspanel_pages.yaml

config:
  screensaver_timeout: 11       # 0 disable, max. 65 sec
  brightness_screensaver: 11    # 0-100%
  brightness_active: 99         # 0-100%
  sleep_brightness:
    - time: "7:00:00"
      value: 10
    - time: "23:00:00"
      value: 0
  update_mode: auto-notify

cards:
  - pageType: cardEntities
    heading: Arbeitszimmer
    navigationLeft: 0
    navigationRight: 0
    entities:
      - entity: switch.example_item
        type: light
        internalNameEntity: EG.Arbeiten.Licht
        iconId: 1
        iconColor: 17299
        displayNameEntity: Arbeitszimmer Licht
        optionalValue: 0
      - entity: light.example_item
        type: shutter
        internalNameEntity: cover.entityName
        iconId: 0
        iconColor: 17299
        displayNameEntity: Shutter2
        optionalValue: iconUp|iconStop|iconDown
      - entity: cover.example_item
        type: text
        internalNameEntity: sensor.entityName
        iconId: 3
        iconColor: 17299
        displayNameEntity: Temperature
        optionalValue: 22
      - entity: input_boolean.example_item
        type: switch
        internalNameEntity: switch.entityName
        iconId: 4
        iconColor: 17299
        displayNameEntity: Switch1
        optionalValue: 0

  - pageType: cardGrid
    heading: Example Grid
    navigationLeft: 0
    navigationRight: 0
    entities:
      - entity: light.example_item
      - entity: switch.example_item
      - entity: delete
      - entity: button.example_item
      - entity: cover.example_item
      - entity: navigate.cardGrid_testKey

  - pageType: cardThermo
    entity: climate.example_item

    pageType: cardMedia
    entity: media_player.example_item

  - pageType: cardAlarm
    entity: alarm_control_panel.alarmo

screensaver:
    heading: Hello World!
    text: Hier könnte Ihre Werbung stehen
    # entity: weather.k3ll3r
    # weatherOverrideForecast4:
    #     entity: sensor.example_item
    #     name: name
    #     icon: lightbulb
    # alternativeLayout: True