<repo_name>dakl/screen-handler<file_sep>/run.py
import json

import paho.mqtt.client as mqtt
import structlog

from app import config
from app.accessory.screen import Screen

logger = structlog.getLogger(__name__)

# The callback for when the client receives a CONNACK response from the server.
logger.info("Starting screen-handler")
screen = Screen(name="raspi-touchscreen")

if not config.TOPIC_NAME:
    raise ValueError("The environment variable 'TOPIC_NAME' needs to be set.")


def on_connect(client, userdata, flags, rc):
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    topic = f"commands/{config.TOPIC_NAME}/#"
    logger.info("Subscribing", topic=topic)
    client.subscribe(topic)
    logger.info("Connected", rc=str(rc))


# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, message):
    payload = str(message.payload.decode("utf-8")).strip()
    logger.info("Handling message", topic=message.topic, payload=payload)

    screen.handle(payload=json.loads(payload))

    publish_topic = f"events/{config.TOPIC_NAME}"
    logger.info("Publishing state update message", topic=publish_topic, payload=payload)
    client.publish(topic=publish_topic, payload=payload, retain=True)


logger.info("Connecting to MQTT")
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

client.connect(config.BROKER, 1883, 60)

# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a
# manual interface.
client.loop_forever()
<file_sep>/.flake8
[flake8]
max-line-length = 100
import-order-style=smarkets
application-import-names=recommender,tests
ignore = W503,E203,E231,E501,S101
<file_sep>/app/accessory/screen.py
from app import config
import structlog

from .base import Accessory

logger = structlog.getLogger(__name__)

STATE_MAP = {
    "ON": 1,
    "OFF": 0,
}


class Screen(Accessory):
    def __init__(
        self,
        name: str,
        read_handle: str = None,
        write_handle: str = None,
    ):
        self.name = name
        self.read_handle = read_handle or config.READ_HANDLE
        self.write_handle = write_handle or config.WRITE_HANDLE

    def set_state(self, state: int) -> None:
        if state == 0:
            self.set_brightness(0)
        elif state == 1:
            self.set_brightness(255)

    def get_state(self) -> int:
        brightness = self.get_brightness()
        if brightness > 0:
            return 1
        else:
            return 0

    def set_brightness(self, value: int) -> None:
        """ Brightness is expected to come in as an int in the range 0-255."""
        with open(self.write_handle, "w") as f:
            f.write(str(value) + "\n")

    def get_brightness(self) -> int:
        with open(self.read_handle, "r") as f:
            return int(f.read().strip())

    def handle(self, payload):
        if "brightness" in payload:
            b = int(payload.get("brightness"))
            self.set_brightness(b)
        elif "state" in payload:
            if payload.get("state") == "OFF":
                self.set_brightness(0)
            elif payload.get("state") == "ON":
                self.set_brightness(255)
<file_sep>/requirements-dev.txt
-r requirements.txt

flake8
ipdb
mypy
pytest
watchdog
black
<file_sep>/app/accessory/mock.py
from .base import Accessory


class MockAccessory(Accessory):
    def set_state(self, state):
        return 1

    def get_status(self):
        return 0
<file_sep>/app/accessory/base.py
from abc import ABC, abstractmethod


class Accessory(ABC):
    name: str

    def __init__(self, name):
        self.name = name

    @abstractmethod
    def set_state(self, state: int) -> None:
        raise NotImplementedError()

    @abstractmethod
    def get_state(self) -> int:
        raise NotImplementedError()

    def handler(self, payload):
        self.set_state(payload)
<file_sep>/README.md
# Screen Handler

Sets the screen on a raspberry pi touchscreen based on HomeAssistant events on MQTT.

# Run

~~~bash
make run
~~~
<file_sep>/Makefile
run-dev:
	watchmedo auto-restart python run.py --patterns="*.py" --recursive

run:
	WRITE_HANDLE=/sys/class/backlight/rpi_backlight/brightness \
	READ_HANDLE=/sys/class/backlight/rpi_backlight/actual_brightness \
	python3 run.py

lint:
	flake8 app/
	mypy --ignore-missing-imports app/

test: lint
<file_sep>/Dockerfile
FROM dakl/arm32-python-alpine-qemu:3.7.1-slim

COPY requirements.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements.txt

WORKDIR /code
COPY . /code

CMD ["python", "run.py"]<file_sep>/app/config.py
import os


class Config:
    WRITE_HANDLE = os.getenv(
        "WRITE_HANDLE", "/tmp/brightness"
    )  # "/sys/class/backlight/rpi_backlight/brightness"
    READ_HANDLE = os.getenv(
        "READ_HANDLE", "/tmp/actual_brightness"
    )  # "/sys/class/backlight/rpi_backlight/actual_brightness"
    DEBUG = os.getenv("DEBUG", False)
    BROKER = os.getenv("BROKER", "worker0.local")
    TOPIC_NAME = os.getenv("TOPIC_NAME", "screen")
<file_sep>/app/__init__.py
import logging

from app.config import Config


def setup_logging():
    if config.DEBUG:
        level = logging.DEBUG
    else:
        level = logging.INFO
    logging.basicConfig(level=level)


config = Config()
setup_logging()
