Merge pull request #11 from protolab-rosenheim/dev

Dev

artnet/artnet_configurator.py

@@ -6,6 +6,7 @@
 from .artnet_node import ArtNetNode, LEDStrip
 from .artnet_node_planboard import ArtNetNodePlanboard
 from .artnet_server import ArtNetServer
+from .models import Slot
 
 
 class ArtNetConfigurator:
@@ -43,17 +44,27 @@ def get_artnet_server(config_artnet=None, config_led_mapping=None):
                         strip_length = int(config_artnet[node_entry][node_option_key])
                         artnet_node.universe[universe_id] = LEDStrip(strip_length)
                     if node_option_key == 'color_history':
-                            artnet_node.color_history = config_artnet[node_entry][node_option_key]
+                        artnet_node.color_history = config_artnet[node_entry][node_option_key]
 
                 for mapping_entry in config_led_mapping:
                     if mapping_entry.startswith(artnet_node.name):
-                        universe = config_led_mapping[mapping_entry]['universe']
+                        universe = str(config_led_mapping[mapping_entry]['universe'])
 
                         for slot_entry in config_led_mapping[mapping_entry]:
-                            if slot_entry.startswith('slot_'):
-                                artnet_node.slots.update({int(slot_entry.split('_')[1]): {'universe': universe,
-                                                                       'led': config_led_mapping[mapping_entry][slot_entry]}})
-
+                            if not slot_entry.startswith('slot_'):
+                                continue
+                                # artnet_node.slots.update({int(slot_entry.split('_')[1]): {'universe': universe,
+                                #                                        'led': config_led_mapping[mapping_entry][slot_entry]}})
+                            for sub_slot in config_led_mapping[mapping_entry][slot_entry]:
+                                start, end = config_led_mapping[mapping_entry][slot_entry][sub_slot].split('-')
+                                slot_name = slot_entry.split('_')[1] + '.' + str(sub_slot)
+                                if slot_name in artnet_node.slots:
+                                    artnet_node.slots[slot_name].universe.append(str(universe))
+                                    artnet_node.slots[slot_name].start.append(int(start))
+                                    artnet_node.slots[slot_name].end.append(int(end))
+                                else:
+                                    artnet_node.slots.update(
+                                        {slot_name: Slot([str(universe)], [int(start)], [int(end)], slot_name)})
                 artnet_server.art_net_nodes.append(artnet_node)
 
         return artnet_server

artnet/artnet_node.py

@@ -3,6 +3,7 @@
 from collections import deque
 
 from .color import Color
+from .models import Slot
 from .packets import PacketType, ArtNetDMXPacket
 
 
@@ -35,13 +36,29 @@ def __init__(self, name, ip_address, port=6454, max_history_size=5):
         self._max_history_size = max_history_size
         self.color_history = []
 
-    def illuminate_multiple_slots(self, slots, color):
+    def flush_slot_history(self):
+        self.slot_history = []
+
+    def flush_slot_history_opcua_call(self, parent):
+        """Only use this method for calls from python-opcua"""
+        from asyncua import ua
+        self.flush_slot_history()
+        return [ua.Variant(True, ua.VariantType.Boolean)]
+
+    def illuminate_multiple_slots(self, slots: str, color: str):
         """Illuminate multiple slots, in one color, don't add them to illuminated history. Slots delimiter is ';'"""
-        slot_color = {}
+        slot_color = []
+        if not Color.colors.get(color):
+            return False
+
         for slot_name in slots.split(';'):
-            slot_area, slot_num = map(int, str(slot_name).split('.'))
-            if slot_num in self.slots[slot_area]['led'].keys() and Color.colors.get(color):
-                slot_color[slot_name] = color
+            # slot_area, slot_num = map(int, str(slot_name).split('.'))
+            # if slot_num in self.slots[slot_area]['led'].keys() and Color.colors.get(color):
+            #     slot_color[slot_name] = color
+            slot = self.slots.get(slot_name)
+            if slot:
+                # slot_color = (slot, color)
+                slot_color.append((slot, color))
 
         led_strip_dict = self._history_led_strip_builder(slot_color)
 
@@ -58,11 +75,13 @@ def illuminate_multiple_slots_opcua_call(self, parent, slot_name, color):
         from asyncua import ua
         return [ua.Variant(self.illuminate_multiple_slots(slot_name.Value, color.Value), ua.VariantType.Boolean)]
 
-    def illuminate_slot(self, slot_name, color, history_to_illu, coll_history):
+    def illuminate_slot(self, slot_name: str, color: str, history_to_illu: int, coll_history: bool) -> bool:
         """Illuminates a slot and adds it to the illuminated slots history"""
-        slot_area, slot_num = map(int, str(slot_name).split('.'))
+        # slot_area, slot_num = map(int, str(slot_name).split('.'))
 
-        if slot_num in self.slots[slot_area]['led'].keys() and Color.colors.get(color):
+        # if slot_num in self.slots[slot_area]['led'].keys() and Color.colors.get(color):
+        slot = self.slots.get(slot_name)
+        if slot and Color.colors.get(color):
             slot_history = self._history_builder(slot_name, color, history_to_illu)
             led_strip_dict = self._history_led_strip_builder(slot_history)
 
@@ -101,8 +120,8 @@ def illuminate_slot_with_history(self, slot_name, color, history_to_illu):
     def illuminate_slot_with_history_opcua_call(self, parent, slot_name, color, history_to_illu):
         from asyncua import ua
         return [ua.Variant(self.illuminate_slot_with_history(slot_name.Value,
-                                                color.Value,
-                                                history_to_illu.Value), ua.VariantType.Boolean)]
+                                                             color.Value,
+                                                             history_to_illu.Value), ua.VariantType.Boolean)]
 
     def illuminate_universe(self, universe, color_str):
         if Color.colors.get(color_str):
@@ -114,7 +133,7 @@ def illuminate_universe(self, universe, color_str):
 
     def illuminate_universe_opcua_call(self, parent, universe, color_str):
         from asyncua import ua
-        return [ua.Variant(self.illuminate_universe(str(universe.Value), color_str.Value),ua.VariantType.Boolean)]
+        return [ua.Variant(self.illuminate_universe(str(universe.Value), color_str.Value), ua.VariantType.Boolean)]
 
     def illuminate_universe_rgb(self, universe, red, green, blue):
         if self.universe.get(universe):
@@ -211,10 +230,10 @@ def _illuminate_from_to(self, start_led, end_led, led_strip, color):
             led_strip.led_strip[i].set_color(color)
         return led_strip
 
-    def _history_builder(self, slot_name, color, history_to_illu):
+    def _history_builder(self, slot_name: str, color: str, history_to_illu: int) -> [(Slot, str)]:
         """Build history of slots that have to be illuminated"""
-        history_to_build = {}
-        slot_area, slot_num = map(int, str(slot_name).split('.'))
+        history_to_build = []
+        # slot_area, slot_num = map(int, str(slot_name).split('.'))
         tmp_slot_history = copy.deepcopy(self.slot_history)
 
         if slot_name in tmp_slot_history:
@@ -226,30 +245,35 @@ def _history_builder(self, slot_name, color, history_to_illu):
         if len(tmp_slot_history) < history_to_illu:
             history_to_illu = len(tmp_slot_history)
         for i in range(history_to_illu):
-            history_to_build[tmp_slot_history[i]] = self.color_history[i]
-
-        if self.slots[slot_area]['led'][slot_num] and Color.colors.get(color):
-            history_to_build[slot_name] = color
+            # history_to_build[tmp_slot_history[i]] = self.color_history[i]
+            slot = self.slots.get(tmp_slot_history[i])
+            if slot:
+                history_to_build.append((slot, self.color_history[i]))
+
+        slot = self.slots.get(slot_name)
+        if slot:
+            history_to_build.append((slot, color))
         return history_to_build
 
-    def _history_led_strip_builder(self, slot_history):
+    def _history_led_strip_builder(self, slot_color: [(Slot, str)]) -> {}:
         """Needs a slot_history dict from _history_builder to build LEDStrips that can be send"""
         led_strip_dict = {}
 
         for universe in self.universe:
             led_strip_dict[universe] = copy.deepcopy(self.universe.get(universe))
 
-        for slot in slot_history:
-            slot_area, slot_num = map(int, str(slot).split('.'))
-            universe = self.slots.get(slot_area).get('universe')
-            if universe not in led_strip_dict.keys():
-                led_strip_dict[universe] = copy.deepcopy(self.universe[universe])
-
-            start_led, end_led = map(int, self.slots[slot_area]['led'][slot_num].split('-'))
-            led_strip_dict[universe] = self._illuminate_from_to(start_led,
-                                                                end_led,
-                                                                led_strip_dict[universe],
-                                                                slot_history[slot])
+        for slot, color in slot_color:
+            # slot_area, slot_num = map(int, str(slot).split('.'))
+            # universe = self.slots.get(slot_area).get('universe')
+            # if universe not in led_strip_dict.keys():
+            #     led_strip_dict[universe] = copy.deepcopy(self.universe[universe])
+
+            # start_led, end_led = map(int, self.slots[slot_area]['led'][slot_num].split('-'))
+            for i in range(len(slot.start)):
+                led_strip_dict[slot.universe[i]] = self._illuminate_from_to(slot.start[i],
+                                                                            slot.end[i],
+                                                                            led_strip_dict[slot.universe[i]],
+                                                                            color)
 
         return led_strip_dict
 

artnet/artnet_server.py

@@ -16,6 +16,7 @@ def __init__(self, ip, broadcast_addr, port=6454):
         self.ip = ip
         self.broadcast_addr = broadcast_addr
         self.art_net_nodes = []
+        self.max_send_packets_in_a_row = 15
 
         if port not in range(65536):
             raise ValueError('Only values between 0-65535 are valid for ports')
@@ -48,10 +49,14 @@ def server(self):
                         logging.error('ArtNet Server error: {}').format(e)
 
                 for node in self.art_net_nodes:
-                    if node.send_queue:
+                    max_packet_row_counter = 0
+                    while node.send_queue:
+                        max_packet_row_counter += 1
                         packet = node.send_queue.popleft()
                         server_socket.sendto(packet, (node.ip_address, node.port))
                         server_socket.sendto(packet, (node.ip_address, node.port))
+                        if max_packet_row_counter >= self.max_send_packets_in_a_row:
+                            break
 
             server_socket.close()
 

artnet/models.py

@@ -0,0 +1,7 @@
+
+class Slot:
+    def __init__(self, universe, start, end, slot_name):
+        self.universe = universe
+        self.start = start
+        self.end = end
+        self.slot_name = slot_name

requirements.txt

@@ -1 +1 @@
-PyYAML==5.3
+PyYAML==5.3.1

samples/sample_1/__main__.py

@@ -15,6 +15,10 @@
 # Run till thread has stopped
 while artnet_server.thread.isAlive():
     try:
+        artnet_server.art_net_nodes[0].flush_slot_history()
+        print('flush history')
+        sleep(sleep_time)
+
         artnet_server.art_net_nodes[0].illuminate_all('green')
         print('green')
         sleep(sleep_time)
@@ -23,9 +27,31 @@
         print('red')
         sleep(sleep_time)
 
-        artnet_server.art_net_nodes[0].illuminate_slot('3.1', 'red', 0, False)
-        print('partly red')
+        artnet_server.art_net_nodes[0].illuminate_slot('3.1', 'red', 3, True)
+        print('slot 3.1')
+        sleep(sleep_time)
+
+        artnet_server.art_net_nodes[0].illuminate_slot('9.2', 'red', 3, True)
+        print('slot 9.2')
+        sleep(sleep_time)
+
+        artnet_server.art_net_nodes[0].illuminate_slot('1.1', 'red', 3, True)
+        print('slot 1.1')
+        sleep(sleep_time)
+
+        artnet_server.art_net_nodes[0].illuminate_multiple_slots('6.1;6.2;6.3;6.4;7.1;7.2;7.3;7.4', 'magenta')
+        print('slot 6.1;6.2;6.3;6.4;7.1;7.2;7.3;7.4')
+        sleep(sleep_time)
+
+        artnet_server.art_net_nodes[0].flush_slot_history()
+        print('flush history')
+        sleep(sleep_time)
+
+        artnet_server.art_net_nodes[0].illuminate_slot('2.1', 'magenta', 3, True)
+        print('slot 2.1')
+        sleep(sleep_time)
         sleep(sleep_time)
 
     except KeyboardInterrupt:
+        artnet_server.art_net_nodes[0].all_off()
         artnet_server.stop_server()