mqtt_pi_led-display/src/controller.py

import paho.mqtt.client as mqtt

import time
import board
import busio
from adafruit_ht16k33 import segments

__TOPIC_BRIGHTNESS__ = "brightness"
__TOPIC_SCROLL_INTERVAL__ = "scroll_interval"
__TOPIC_TEXT__ = "text"

class Controller:
	connected = False
	display = None

	def __init__(self, base_topic, mqtt_host, mqtt_port = 1883):

		self.scroll_interval = 0.25
		self.text = ""
		self.poscount = 0
		
		# Create the I2C interface.
		self.i2c = busio.I2C(board.SCL, board.SDA)
		
		# Create the LED segment class.
		# This creates a 14 segment 4 character display:
		self.display = segments.Seg14x4(self.i2c)
		
		# Clear the display.
		self.display.fill(0)
		
		# set brightness, range 0-1.0, 1.0 max brightness
		self.display.brightness = 1.0
		
		self.mqtt = mqtt.Client()
		self.mqtt.on_connect = self.on_connect
		self.mqtt.on_message = self.on_message
		self.mqtt.on_disconnect = self.on_disconnect
		self.mqtt.connect(mqtt_host, mqtt_port)
		self.topic = base_topic
		
		self.display.print("BOOT")


	def on_disconnect(self, client, userdata, rc):
		print("MQTT disconnected")
		self.connected = False

	def on_message(self, client, userdata, message):
		msg = str(message.payload.decode("utf-8"))
		print("msg = " +  msg)
		
		if message.topic.endswith(__TOPIC_BRIGHTNESS__ + "/set"):
			val = float(msg)
			if val >= 0.0 and val <=1.0:
				self.display.brightness = val
				print("changed brightness to ", self.display.brightness)
				self.mqtt.publish(self.topic + "/" + __TOPIC_BRIGHTNESS__, self.display.brightness, 1 )
			else:
				print("invalid brightness ", val)
		
		if message.topic.endswith(__TOPIC_SCROLL_INTERVAL__ + "/set"):
			val = float(msg)
			if val > 0.0:
				self.scroll_interval = val
				print("changed scroll_interval to ", self.scroll_interval)
				self.mqtt.publish(self.topic + "/" + __TOPIC_SCROLL_INTERVAL__, self.scroll_interval, 1 )
			else:
				print("error not >0.0: ", val)
		
		
		if message.topic.endswith(__TOPIC_TEXT__ + "/set"):
			self.text = msg
			print("changed text to ", self.text)
			self.mqtt.publish(self.topic + "/" + __TOPIC_TEXT__, self.text, 1 )


	def on_connect(self, client, userdata, flags, rc):
		print("Connected to MQTT Broker")
		self.mqtt.subscribe(self.topic + "/#")
		self.connected = True


	def loop_forever(self):
		run = True

		while run:
			self.mqtt.loop()

			if self.display is not None:
				self.display.print(self.text[self.poscount])
				self.poscount += 1
				self.poscount %= len(self.text)
				time.sleep(self.scroll_interval)