0.2 Big Buck Beta (Release Candidate 2)

.gitignore

@@ -5,11 +5,11 @@ Thumbs.db
 .idea/
 
 # any specific configuration
-config.py
+config.yml
 
 # setuptools
 server/dist
 *.egg-info
 
-# tests
-tests/
+# temporary stuff
+playground/

logo

@@ -0,0 +1,6 @@
+\033[1;31m
+ _______ ___ __
+ < / __ \__ \ _____/ /_ ____ _ _______ \033[0;32m
+ / / / / /_/ // ___/ __ \/ __ \ | /| / / ___/
+ / / /_/ / __/(__ ) / / / /_/ / |/ |/ (__ ) \033[1;34m
+/_/\____/____/____/_/ /_/\____/|__/|__/____/ \033[0m

server/config.example.yml

@@ -0,0 +1,22 @@
+sys_name: null
+log_level: INFO
+
+Strip:
+ driver: APA102
+ num_leds: null
+ max_clock_speed_hz: 4000000 # [Hz] 4 MHz is the maximum for "large" strips of more than 500 LEDs.
+ initial_brightness: 50 # integer from 0 to 100
+ max_brightness: 75 # maximum brightness
+ gamma: 2.8 # for gamma correction, put 1 to turn it off
+
+MQTT:
+ prefix: led
+ general_path: "{prefix}/{sys_name}/show/{show_name}/{command}"
+ notification_path: "{prefix}/{sys_name}/notification"
+ username: None
+ password: None
+
+ Broker:
+ host: localhost
+ port: 1883
+ keepalive: 60 # [seconds]

server/drivers/__init__.py

@@ -52,6 +52,20 @@ def __init__(self, num_leds: int, max_clock_speed_hz: int = 4000000, gamma: floa
  self.synced_blue_buffer = SyncedArray('f', [0.0] * self.num_leds)
  self.synced_brightness_buffer = SyncedArray('i', self.brightness_buffer)
 
+ def __del__(self):
+ """ invokes self.close() and deletes all the buffers"""
+ self.close()
+
+ del self.color_buffer, self.synced_red_buffer, self.synced_green_buffer, self.synced_blue_buffer
+ del self.brightness_buffer, self.synced_brightness_buffer
+
+ logger.info("Driver successfully closed")
+
+ @abstractmethod
+ def close(self):
+ """ close the bus connection and clean up remains"""
+ pass
+
  def freeze(self):
  """
  freezes the strip. All state-changing methods (on_color_change() and on_brightness_change())

server/drivers/apa102.py

@@ -103,6 +103,9 @@ def led_prefix(cls, brightness: int) -> int:
 
  return prefix_byte
 
+ def close(self):
+ self.spi.close()
+
  @staticmethod
  def spi_start_frame() -> list:
  """ To start a transmission, one must send 32 empty bits """

server/drivers/dummy.py

@@ -34,3 +34,6 @@ def show(self) -> None:
  g=green,
  b=blue,
  brightness=brightness))
+
+ def close(self):
+ pass

server/helpers/__init__.py

@@ -7,3 +7,15 @@
 """
 
 __all__ = ['color', 'exceptions', 'mqtt', 'preprocessors', 'verify']
+
+
+def get_logo(filename: str ='../logo') -> str:
+ with open(filename, encoding='unicode_escape') as file:
+ contents = file.read()
+ return contents.rstrip().rstrip() # return without newline at the end
+
+
+def get_version(filename: str ='../version') -> str:
+ with open(filename) as file:
+ contents = file.read()
+ return contents.rstrip().rstrip() # return without newline at the end

server/helpers/test_verify.py

@@ -0,0 +1,134 @@
+import unittest
+
+from helpers import verify
+from helpers.exceptions import *
+
+
+class TestVerifyNumeric(unittest.TestCase):
+ def test_string_fails(self):
+ self.assertRaises(InvalidParameters, verify.numeric, candidate="6.7")
+
+ def test_out_of_bounds_fails(self):
+ self.assertRaises(InvalidParameters, verify.numeric, candidate=3.98, minimum=5.0, maximum=10.0)
+
+ def test_max_bounds_passes(self):
+ self.assertIsNone(verify.numeric(candidate=5.123, minimum=1.0, maximum=10.0))
+
+ def test_is_min_bound_passes(self):
+ self.assertIsNone(verify.numeric(candidate=1.0, minimum=1.0, maximum=10.0))
+
+ def test_is_max_bound_passes(self):
+ self.assertIsNone(verify.numeric(candidate=10.0, minimum=1.0, maximum=10.0))
+
+
+class TestVerifyNotNegativeNumeric(unittest.TestCase):
+ def test_null_int_passes(self):
+ self.assertIsNone(verify.not_negative_numeric(candidate=0))
+
+ def test_null_float_passes(self):
+ self.assertIsNone(verify.not_negative_numeric(candidate=0.0))
+
+ def test_positive_passes(self):
+ self.assertIsNone(verify.not_negative_numeric(candidate=1.3))
+
+ def test_negative_fails(self):
+ self.assertRaises(InvalidParameters, verify.not_negative_numeric, candidate=-2.4)
+
+
+class TestVerifyPositiveNumeric(unittest.TestCase):
+ def test_null_int_fails(self):
+ self.assertRaises(InvalidParameters, verify.positive_numeric, candidate=0)
+
+ def test_null_float_fails(self):
+ self.assertRaises(InvalidParameters, verify.positive_numeric, candidate=0.0)
+
+ def test_positive_passes(self):
+ self.assertIsNone(verify.positive_numeric(candidate=1.3))
+
+ def test_negative_fails(self):
+ self.assertRaises(InvalidParameters, verify.positive_numeric, candidate=-2.4)
+
+
+class TestVerifyInteger(unittest.TestCase):
+ def test_string_fails(self):
+ self.assertRaises(InvalidParameters, verify.integer, candidate="6")
+
+ def test_out_of_bounds_fails(self):
+ self.assertRaises(InvalidParameters, verify.integer, candidate=3, minimum=5, maximum=10)
+
+ def test_max_bounds_passes(self):
+ self.assertIsNone(verify.integer(candidate=5, minimum=1, maximum=10))
+
+ def test_is_min_bound_passes(self):
+ self.assertIsNone(verify.integer(candidate=1, minimum=1, maximum=10))
+
+ def test_is_max_bound_passes(self):
+ self.assertIsNone(verify.integer(candidate=10, minimum=1, maximum=10))
+
+
+class TestVerifyNotNegativeInteger(unittest.TestCase):
+ def test_null_int_passes(self):
+ self.assertIsNone(verify.not_negative_integer(candidate=0))
+
+ def test_null_float_fails(self):
+ self.assertRaises(InvalidParameters, verify.not_negative_integer, candidate=0.0)
+
+ def test_positive_passes(self):
+ self.assertIsNone(verify.not_negative_integer(candidate=1))
+
+ def test_negative_fails(self):
+ self.assertRaises(InvalidParameters, verify.not_negative_integer, candidate=-2)
+
+
+class TestVerifyPositiveInteger(unittest.TestCase):
+ def test_null_int_fails(self):
+ self.assertRaises(InvalidParameters, verify.positive_integer, candidate=0)
+
+ def test_null_float_fails(self):
+ self.assertRaises(InvalidParameters, verify.positive_integer, candidate=0.0)
+
+ def test_positive_passes(self):
+ self.assertIsNone(verify.positive_integer(candidate=1))
+
+ def test_negative_fails(self):
+ self.assertRaises(InvalidParameters, verify.positive_integer, candidate=-2)
+
+
+class TestVerifyBoolean(unittest.TestCase):
+ def test_str_fails(self):
+ self.assertRaises(InvalidParameters, verify.boolean, candidate="True")
+
+ def test_none_fails(self):
+ self.assertRaises(InvalidParameters, verify.boolean, candidate=None)
+
+ def test_zero_fails(self):
+ self.assertRaises(InvalidParameters, verify.boolean, candidate=0)
+
+ def test_one_fails(self):
+ self.assertRaises(InvalidParameters, verify.boolean, candidate=1)
+
+ def test_True_passes(self):
+ self.assertIsNone(verify.boolean(candidate=True))
+
+ def test_False_passes(self):
+ self.assertIsNone(verify.boolean(candidate=False))
+
+
+class TestVerifyRGBColorTuple(unittest.TestCase):
+ def test_list_fails(self):
+ self.assertRaises(InvalidParameters, verify.rgb_color_tuple, candidate=[1, 2, 3])
+
+ def test_false_dimensional_tuple_fails(self):
+ self.assertRaises(InvalidParameters, verify.rgb_color_tuple, candidate=(0, 1, 2, 3))
+
+ def test_smaller_than_0_fails(self):
+ self.assertRaises(InvalidParameters, verify.rgb_color_tuple, candidate=(-0.5, 1, 2))
+
+ def test_bigger_than_255_fails(self):
+ self.assertRaises(InvalidParameters, verify.rgb_color_tuple, candidate=(1, 0.2, 275.8))
+
+ def test_float_passes(self):
+ self.assertIsNone(verify.rgb_color_tuple(candidate=(100.1, 101.2, 102.3)))
+
+ def test_int_passes(self):
+ self.assertIsNone(verify.rgb_color_tuple(candidate=(123, 234, 12)))

server/lightshows/templates/base.py

@@ -53,7 +53,7 @@ def name(self) -> str:
  def start(self):
  """ invokes the run() method and after that synchronizes the shared buffer """
  # before
- signal.signal(signal.SIGINT, self.stop) # attach stop() to SIGSTOP
+ signal.signal(signal.SIGINT, self.stop) # attach stop() to SIGINT
  self.strip.sync_down()
  self.mqtt.start_listening()
 

server/mqttcontrol.py

@@ -178,5 +178,14 @@ def run(self) -> None:
  # start a show show to listen for brightness changes and refresh the strip regularly
  self.start_show("clear", {})
 
- client.loop_forever()
- logger.critical("MQTTControl.py has exited")
+ try:
+ signal.signal(signal.SIGTERM, self.stop_controller) # attach stop_controller() to SIGTERM
+ client.loop_forever()
+ except KeyboardInterrupt:
+ self.stop_controller()
+ finally:
+ logger.critical("MQTTControl.py has exited")
+
+ def stop_controller(self, signum=None, frame=None):
+ """ what happens if the controller exits """
+ del self.strip # close driver connection

server/server.py

@@ -23,17 +23,18 @@
 
 import coloredlogs
 
+from helpers import get_logo, get_version
 from helpers.configparser import get_configuration
 from mqttcontrol import MQTTControl
 
-__version__ = '0.2rc1'
-
-logo = """
- _______ ___ __
- < / __ \__ \ _____/ /_ ____ _ _______
- / / / / /_/ // ___/ __ \/ __ \ | /| / / ___/
- / / /_/ / __/(__ ) / / / /_/ / |/ |/ (__ )
-/_/\____/____/____/_/ /_/\____/|__/|__/____/ v{}"""
+# constants
+logo = get_logo()
+version = get_version()
+license_hint = """
+This is free software. You are welcome to redistribute
+it under the conditions of the GNU Public License v2.
+For details, see https://www.gnu.org/licenses/gpl-2.0.html
+"""
 
 # configuration
 user_config = get_configuration()
@@ -42,12 +43,11 @@
 logger = logging.getLogger('102shows.server')
 coloredlogs.install(level=user_config.log_level)
 
-# friendly greetings
-print(logo.format(__version__))
+# friendly greeting
+print(logo + " version: {}".format(version))
+print()
+print(license_hint)
 print()
-print("This is free software. You are welcome to redistribute")
-print("it under the conditions of the GNU Public License v2.")
-print("For details, see https://www.gnu.org/licenses/gpl-2.0.html")
 print()
 print()