WIP: Allow to append default mapping file

.github/workflows/python-tests.yaml

@@ -0,0 +1,27 @@
+name: Python unittest
+
+on: [push]
+
+jobs:
+  test:
+
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python
+        # This is the version of the action for setting up Python, not the Python version.
+        uses: actions/setup-python@v4
+        with:
+          # Semantic version range syntax or exact version of a Python version
+          python-version: '3.11'
+          # Optional - x64 or x86 architecture, defaults to x64
+          architecture: 'x64'
+          cache: 'pip'
+      # You can test your matrix by printing the current Python version
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+      - name: Execute unit tests
+        run: python -m unittest tests/*.py

enoceanmqtt/overlays/homeassistant/ha_communicator.py

@@ -6,6 +6,8 @@
 import os
 import time
 import json
+from typing import Dict, Any
+
 import yaml
 
 import enocean.utils
@@ -27,6 +29,12 @@ def __init__(self, config, sensors):
             mapping_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'mapping.yaml')
         with open(mapping_file, 'r', encoding="utf-8") as file:
             self._ha_mapping = yaml.safe_load(file)
+        extra_mapping_file = config.get('extra_mapping_file')
+        if extra_mapping_file:
+            with open(extra_mapping_file, 'r', encoding="utf-8") as file:
+                extra_mapping = yaml.safe_load(file)
+                self._ha_mapping = custom_merge(self._ha_mapping, extra_mapping)
+
         logging.info("Mapping file correctly read: %s", mapping_file)
 
         # Overwrite some of the user-defined device configuration
@@ -376,3 +384,41 @@ def _publish_mqtt(self, sensor, mqtt_json):
 
         # Publish the packet
         super()._publish_mqtt(sensor, mqtt_json)
+
+
+def custom_merge(mapping_dict: Dict[int, Any], extra_mapping_dict: Dict[int, Any]) -> Dict[int, Any]:
+    """
+    Add extra mapping information to existing mapping.
+
+    :param mapping_dict: existing mapping
+    :param extra_mapping_dict: extra mapping
+    :return: merged mapping
+    """
+    mapping_copy = copy.deepcopy(mapping_dict)
+    for rorg, val_rorg in extra_mapping_dict.items():
+        for func, val_func in val_rorg.items():
+            for type_, val_type in val_func.items():
+                if 'entities' in val_type:
+                    for element in val_type['entities']:
+                        if 'action' in element and element['action'] == 'remove':
+                            mapping_copy[rorg][func][type_]['entities'] = list(
+                                filter(
+                                    lambda entity: (entity['component'],
+                                                    entity['name']) != (element['component'],
+                                                                        element['name']),
+                                    mapping_copy[rorg][func][type_]['entities'])
+                            )
+                        if 'action' not in element or element['action'] == 'add':
+                            element.pop('action', None)
+                            try:
+                                identical_entity = next(entity
+                                                        for entity in mapping_copy[rorg][func][type_]['entities']
+                                                        if (entity['component'],
+                                                            entity['name']) == (element['component'],
+                                                                                element['name'])
+                                                        )
+                                identical_entity.update(element)
+                            except StopIteration:
+                                mapping_copy[rorg][func][type_]['entities'].append(element)
+
+    return mapping_copy

requirements.txt

@@ -0,0 +1,4 @@
+pyyaml
+tinydb
+git+https://github.com/mak-gitdev/enocean.git
+git+https://github.com/embyt/enocean-mqtt.git

tests/resources/extra_mapping.yaml

@@ -0,0 +1,20 @@
+0xD2:
+  0x01:
+    0x01:
+      entities:
+        - component: "switch"
+          name: "switch2"
+          config:
+            state_topic: "CMD4"
+            state_on: "100"
+            state_off: "0"
+            value_template: "{{ value_json.OV }}"
+            command_topic: "req"
+            payload_on: >
+              {"CMD":"1","DV":"0","IO":"0","OV":"100","send":"clear"}
+            payload_off: >
+              {"CMD":"1","DV":"0","IO":"0","OV":"0","send":"clear"}
+            device_class: outlet
+        - component: "button"
+          name: "query_status"
+          action: "remove"

tests/test_mapping.py

@@ -0,0 +1,130 @@
+import unittest
+from pathlib import Path
+
+import yaml
+
+from enoceanmqtt.overlays.homeassistant.ha_communicator import custom_merge
+
+
+class MappingTestCase(unittest.TestCase):
+
+    def test_add_mapping(self):
+        mapping_path = Path(__file__).parent.parent / 'enoceanmqtt' / 'overlays' / 'homeassistant' / 'mapping.yaml'
+        with open(mapping_path, 'r', encoding='utf-8') as mapping_file:
+            mapping = yaml.safe_load(mapping_file)
+        entity_to_add = {'component': "cover",
+                         'name': "cover2",
+                         'config': {}}
+        extra_mapping = {
+            0xD2: {
+                0x05: {
+                    0x00: {
+                        'entities':
+                            [
+                                entity_to_add,
+                            ]
+                    }
+                }
+            }
+        }
+        merged_mapping = custom_merge(mapping, extra_mapping)
+        print(mapping[0xD2][0x05][0x00]['entities'])
+        print(merged_mapping[0xD2][0x05][0x00]['entities'])
+        self.assertTrue(entity_to_add in merged_mapping[0xD2][0x05][0x00]['entities'])  # add assertion here
+        self.assertTrue(entity_to_add not in mapping[0xD2][0x05][0x00]['entities'])
+
+    def test_add_mapping_2(self):
+        mapping_path = Path(__file__).parent.parent / 'enoceanmqtt' / 'overlays' / 'homeassistant' / 'mapping.yaml'
+        with open(mapping_path, 'r', encoding='utf-8') as mapping_file:
+            mapping = yaml.safe_load(mapping_file)
+        entity_to_add = {'component': "cover",
+                         'name': "cover2",
+                         'config': {},
+                         'action': 'add'}
+        extra_mapping = {
+            0xD2: {
+                0x05: {
+                    0x00: {
+                        'entities':
+                            [
+                                entity_to_add,
+                            ]
+                    }
+                }
+            }
+        }
+        merged_mapping = custom_merge(mapping, extra_mapping)
+        print(mapping[0xD2][0x05][0x00]['entities'])
+        print(merged_mapping[0xD2][0x05][0x00]['entities'])
+        self.assertTrue(entity_to_add in merged_mapping[0xD2][0x05][0x00]['entities'])  # add assertion here
+        self.assertTrue(entity_to_add not in mapping[0xD2][0x05][0x00]['entities'])
+        self.assertTrue('state' not in merged_mapping[0xD2][0x05][0x00]['entities'][-1])
+
+    def test_add_identical_mapping(self):
+        mapping_path = Path(__file__).parent.parent / 'enoceanmqtt' / 'overlays' / 'homeassistant' / 'mapping.yaml'
+        with open(mapping_path, 'r', encoding='utf-8') as mapping_file:
+            mapping = yaml.safe_load(mapping_file)
+        entity_to_add = {'component': "cover",
+                         'name': "cover",
+                         'added_key': 'added_value',
+                         'action': 'add'}
+        extra_mapping = {
+            0xD2: {
+                0x05: {
+                    0x00: {
+                        'entities':
+                            [
+                                entity_to_add,
+                            ]
+                    }
+                }
+            }
+        }
+        merged_mapping = custom_merge(mapping, extra_mapping)
+        print(mapping[0xD2][0x05][0x00]['entities'])
+        print(merged_mapping[0xD2][0x05][0x00]['entities'])
+        self.assertTrue((entity_to_add | mapping[0xD2][0x05][0x00]['entities'][0]) in merged_mapping[0xD2][0x05][0x00]['entities'])  # add assertion here
+        self.assertTrue(entity_to_add not in mapping[0xD2][0x05][0x00]['entities'])
+        self.assertTrue('action' not in merged_mapping[0xD2][0x05][0x00]['entities'][-1])
+
+    def test_remove_mapping(self):
+        mapping_path = Path(__file__).parent.parent / 'enoceanmqtt' / 'overlays' / 'homeassistant' / 'mapping.yaml'
+        with open(mapping_path, 'r', encoding='utf-8') as mapping_file:
+            mapping = yaml.safe_load(mapping_file)
+        entity_to_remove = {'component': "cover",
+                            'name': "cover",
+                            'action': 'remove'}
+        extra_mapping = {
+            0xD2: {
+                0x05: {
+                    0x00: {
+                        'entities':
+                            [
+                                entity_to_remove,
+                            ]
+                    }
+                }
+            }
+        }
+        merged_mapping = custom_merge(mapping, extra_mapping)
+        print(mapping[0xD2][0x05][0x00]['entities'])
+        print(merged_mapping[0xD2][0x05][0x00]['entities'])
+        self.assertTrue(entity_to_remove not in merged_mapping[0xD2][0x05][0x00]['entities'])  # add assertion here
+
+    def test_file_mapping(self):
+        mapping_path = Path(__file__).parent.parent / 'enoceanmqtt' / 'overlays' / 'homeassistant' / 'mapping.yaml'
+        extra_mapping_path = Path(__file__).parent / 'resources' / 'extra_mapping.yaml'
+        with open(mapping_path, 'r', encoding='utf-8') as mapping_file:
+            mapping = yaml.safe_load(mapping_file)
+        with open(extra_mapping_path, 'r', encoding='utf-8') as mapping_file:
+            extra_mapping = yaml.safe_load(mapping_file)
+        merged_mapping = custom_merge(mapping, extra_mapping)
+        print(mapping[0xD2][0x01][0x01]['entities'])
+        print(merged_mapping[0xD2][0x01][0x01]['entities'])
+        names = [entity['name'] for entity in merged_mapping[0xD2][0x01][0x01]['entities']]
+        self.assertTrue("switch2" in names)
+        self.assertTrue("query_status" not in names)
+
+
+if __name__ == '__main__':
+    unittest.main()