Merge pull request #4 from gpongelli/partial_save

Partial save

README.md

@@ -127,8 +127,39 @@ listening state, until all IR code combination will being scan.
 * Optional fields (at least one must be present or nothing will be listened):
   * `operationModes`, `fanModes`,`swingModes`
 * Generated file can be used into SmartIR HomeAssistant component
+* It's possible to interrupt with CTRL-C at any moment, a temporary file will be saved
+* Temporary files are also saved at the end of each temperature range
+* In case of existing temporary file, the already learnt combination will be skipped
+
+
+## Example
+
+Example of cli command:
+```bash
+$ broadlink-listener generate-smart-ir ./real_data/1124.json <DEVICE_TYPE> <IP> <MAC_ADDR> -n dry -n fan_only -s eco_cool
+```
+
+`real_data/1124.json` file is [this one from SmartIR GitHub repo](https://github.com/smartHomeHub/SmartIR/blob/master/codes/climate/1124.json)
+in which I've added the missing "swingModes" array, supported by climate but not present on json:
+```json
+"swingModes": [
+  "auto",
+  "high",
+  "mid_high",
+  "middle",
+  "mid_low",
+  "low",
+  "swing"
+],
+```
+
+`<DEVICE_TYPE>`, `<IP>`, `<MAC_ADDR>` parameter can be obtained running:
+```bash
+$ broadlink-listener discover_ir
+```
 
 
 ## Credits
 
-This package was created with [Cookiecutter](https://github.com/audreyr/cookiecutter) and the [gpongelli/cookiecutter-pypackage](https://github.com/gpongelli/cookiecutter-pypackage) project template.
+This package was created with [Cookiecutter](https://github.com/audreyr/cookiecutter)
+and the [gpongelli/cookiecutter-pypackage](https://github.com/gpongelli/cookiecutter-pypackage) project template.

broadlink_listener/cli_tools/cli.py

@@ -111,7 +111,7 @@ def generate_smart_ir(
     smart_ir_mng = SmartIrManager(json_file, broadlink_mng, no_temp_on_mode, no_swing_on_mode)
 
     smart_ir_mng.learn_off()
-    smart_ir_mng.lear_all()
+    smart_ir_mng.learn_all()
     smart_ir_mng.save_dict()
 
 

broadlink_listener/cli_tools/smartir_manager.py

@@ -4,15 +4,21 @@
 
 """SmartIR json manager class."""
 
+import glob
 import json
-import time
+import os
+import platform
+import re
+import signal
+import sys
 from collections import namedtuple
 from copy import deepcopy
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
 from itertools import product
 from pathlib import Path
+from threading import Event
 from typing import Literal, Optional, Union
 
 import click
@@ -75,11 +81,11 @@ class SkipReason:
 _CombinationTupleNone = namedtuple('_CombinationTupleNone', ', '.join(_combination_arguments_none))  # type: ignore
 
 
-def _countdown(msg: str):
+def _countdown(msg: str, event: Event):
     click.echo(msg)
-    for i in range(5, 0, -1):
-        click.echo(i)
-        time.sleep(1)
+    if event.is_set():
+        input("-->> Press enter when ready <<--")
+    event.clear()
 
 
 class SmartIrManager:  # pylint: disable=too-many-instance-attributes
@@ -107,7 +113,10 @@ def __init__(  # pylint: disable=too-many-branches,too-many-statements
         """
         self.__broadlink_manager = broadlink_mng
 
-        self.__json_file_name = file_name
+        self.__partial_inc = 0
+        self.__prompt_event = Event()
+        self.__prompt_event.clear()
+        self.__json_file_name_path = Path(file_name)
         with open(str(file_name), "r", encoding='utf-8') as in_file:
             self.__smartir_dict = json.load(in_file)
 
@@ -137,6 +146,7 @@ def __init__(  # pylint: disable=too-many-branches,too-many-statements
             if not all(list(map(lambda x: x in self.__op_modes, no_temp_on_mode))):
                 raise click.exceptions.UsageError("no-temp-on-mode parameter is using a not-existent operating mode.")
 
+            # fill dict with all empty combination
             _temp_dict = {
                 f"{t}": deepcopy('') for t in range(self.__min_temp, self.__max_temp + 1, self.__precision_temp)
             }
@@ -161,6 +171,10 @@ def __init__(  # pylint: disable=too-many-branches,too-many-statements
                 else:
                     _operation_dict = {f"{o}": deepcopy(_temp_dict) for o in self.__op_modes}  # type: ignore
             self.__smartir_dict[_DictKeys.COMMANDS].update(_operation_dict)
+
+            # overwrite combination if tmp file exist
+            self._load_partial_dict()
+
         except KeyError as key_err:
             raise click.exceptions.UsageError(f"Missing mandatory field in json file: {key_err}") from None
         else:
@@ -169,6 +183,7 @@ def __init__(  # pylint: disable=too-many-branches,too-many-statements
             self.__operation_mode = ''
             self.__fan_mode = ''
             self.__swing_mode = ''
+        self._setup_signal_handler()
 
     @property
     def smartir_dict(self) -> dict:
@@ -179,6 +194,22 @@ def smartir_dict(self) -> dict:
         """
         return self.__smartir_dict
 
+    def _setup_signal_handler(self):
+        _system = platform.system().lower()
+        signal.signal(signal.SIGINT, self._signal_handler)
+        signal.signal(signal.SIGABRT, self._signal_handler)
+
+        if _system in ('linux', 'darwin'):
+            signal.signal(signal.SIGTERM, self._signal_handler)
+        if _system == 'windows':
+            signal.signal(signal.SIGBREAK, self._signal_handler)  # pylint: disable=no-member
+            signal.signal(signal.CTRL_C_EVENT, self._signal_handler)  # pylint: disable=no-member
+            signal.signal(signal.CTRL_BREAK_EVENT, self._signal_handler)  # pylint: disable=no-member
+
+    def _signal_handler(self, _signumber, _frame):
+        self._save_partial_dict()
+        sys.exit(2)
+
     def _setup_combinations(self):
         _variable_args = [self.__fan_modes, self.__swing_modes]
         if all(_variable_args):
@@ -307,6 +338,15 @@ def swing_mode(self, new_value: str) -> None:
         """
         self.__swing_mode = new_value
 
+    @property
+    def partial_inc(self) -> int:
+        """Partial increment value.
+
+        Returns:
+            int: last index of saved partial json files
+        """
+        return self.__partial_inc
+
     def _set_dict_value(self, value: str) -> None:
         if _DictKeys.FAN_MODES in self.__combination_arguments:
             if _DictKeys.SWING_MODES in self.__combination_arguments:
@@ -325,38 +365,100 @@ def _set_dict_value(self, value: str) -> None:
             else:
                 self.__smartir_dict[_DictKeys.COMMANDS.value][self.operation_mode][self.temperature] = value
 
+    def _get_dict_value(self) -> str:
+        _value = ''
+        if _DictKeys.FAN_MODES in self.__combination_arguments:
+            if _DictKeys.SWING_MODES in self.__combination_arguments:
+                _value = self.__smartir_dict[_DictKeys.COMMANDS.value][self.operation_mode][self.fan_mode][
+                    self.swing_mode
+                ][self.temperature]
+            else:
+                _value = self.__smartir_dict[_DictKeys.COMMANDS.value][self.operation_mode][self.fan_mode][
+                    self.temperature
+                ]
+        else:
+            if _DictKeys.SWING_MODES in self.__combination_arguments:
+                _value = self.__smartir_dict[_DictKeys.COMMANDS.value][self.operation_mode][self.swing_mode][
+                    self.temperature
+                ]
+            else:
+                _value = self.__smartir_dict[_DictKeys.COMMANDS.value][self.operation_mode][self.temperature]
+        return _value
+
     def save_dict(self):
         """Save modified dict to output json file."""
         now = datetime.now()
-        _modified_file_name = Path(self.__json_file_name.parent).joinpath(
-            f'{self.__json_file_name.stem}_' f'{now.strftime("%Y%m%d_%H%M%S")}.json'
+        _modified_file_name = self.__json_file_name_path.parent.joinpath(
+            f'{self.__json_file_name_path.stem}_{now.strftime("%Y%m%d_%H%M%S")}.json'
         )
         with open(_modified_file_name, 'w', encoding='utf-8') as out_file:
             json.dump(self.__smartir_dict, out_file)
         click.echo(f"Created new file {_modified_file_name}")
+        _previous = glob.glob(
+            os.path.join(self.__json_file_name_path.parent, f'{self.__json_file_name_path.stem}_tmp_*.json')
+        )
+        for _file in _previous:
+            os.remove(_file)
+
+    def _save_partial_dict(self):
+        # save with incremental 3 numbers to sort correctly when load
+        _modified_file_name = self.__json_file_name_path.parent.joinpath(
+            f'{self.__json_file_name_path.stem}_tmp_{self.__partial_inc:03}.json'
+        )
+
+        try:
+            _no_off = deepcopy(self.__smartir_dict)
+            del _no_off['commands']['off']
+        except KeyError:
+            return
+        else:
+            with open(_modified_file_name, 'w', encoding='utf-8') as out_file:
+                json.dump(_no_off[_DictKeys.COMMANDS.value], out_file, indent=2)
+            self.__partial_inc += 1
+
+    def _load_partial_dict(self):
+        _previous = glob.glob(
+            os.path.join(self.__json_file_name_path.parent, f'{self.__json_file_name_path.stem}_tmp_*.json')
+        )
+        _previous.sort()
+        try:
+            # load last file that's the most updated
+            _last_file = _previous[-1]
+            with open(str(_last_file), "r", encoding='utf-8') as partial_file:
+                self.__smartir_dict[_DictKeys.COMMANDS.value].update(json.load(partial_file))
+            _pattern = re.compile(fr'{self.__json_file_name_path.stem}_tmp_(\d+).json')
+            _res = _pattern.search(_last_file)
+            self.__partial_inc = int(_res.group(1))
+        except IndexError:
+            pass
 
     def learn_off(self):
         """Learn OFF command that's outside the combination.
 
         Raises:
             UsageError: if no IR signal is learnt within timeout
         """
+        self.__prompt_event.clear()
         _countdown(
-            "First of all, let's learn OFF command: turn ON the remote and then turn it OFF when "
-            "'Listening' message is on screen..."
+            "First of all, let's learn OFF command:\nturn ON the remote and then turn it OFF when "
+            "'Listening' message is on screen, or interrupt with CTRL-C...",
+            self.__prompt_event,
         )
+        # set event to wait for first code
+        self.__prompt_event.set()
         _off = self.__broadlink_manager.learn_single_code()
         if not _off:
             raise click.exceptions.UsageError("No IR signal learnt for OFF command within timeout.")
         self.__smartir_dict[_DictKeys.COMMANDS.value]["off"] = _off
 
-    def lear_all(self):  # pylint: disable=too-many-branches
+    def learn_all(self):  # pylint: disable=too-many-branches
         """Learn all the commands depending on calculated combination.
 
         Raises:
             UsageError: if no IR signal is learnt within timeout
         """
         _previous_code = None
+        _previous_combination: Optional[tuple] = None
         for comb in self.__all_combinations:  # pylint: disable=too-many-nested-blocks
             self.operation_mode = comb.operationModes
             if _DictKeys.FAN_MODES in comb._fields:
@@ -365,6 +467,10 @@ def lear_all(self):  # pylint: disable=too-many-branches
                 self.swing_mode = comb.swingModes
             self.temperature = str(comb.temperature)
 
+            if self._get_dict_value() != '':
+                self.__prompt_event.set()
+                continue
+
             _do_skip = self._skip_learning(comb)
             if _do_skip.skip:
                 # must read the first temperature and then reuse the same for next combination
@@ -388,15 +494,24 @@ def lear_all(self):  # pylint: disable=too-many-branches
                             self._set_dict_value(_previous_code)
                             continue
 
+            if _previous_combination:
+                for i in range(0, len(comb) - 1):
+                    if _previous_combination[i] != comb[i]:  # pylint: disable=unsubscriptable-object
+                        self.__prompt_event.set()
+                        self._save_partial_dict()
+            _previous_combination = comb
+
             _combination_str = self._get_combination(comb)
             _countdown(
-                f"Let's learn command of\n{_combination_str}\n"
-                "Prepare the remote to this combination, then turn it OFF. When 'Listening' message"
-                " is on screen, turn the remote ON to learn combination previously set..."
+                "-" * 30 + f"\nLet's learn IR command of\n{_combination_str}\n"
+                "Prepare the remote so Broadlink can listen the above combination when 'Listening' message"
+                " is on screen, or interrupt with CTRL-C...",
+                self.__prompt_event,
             )
             _code = self.__broadlink_manager.learn_single_code()
             _previous_code = _code
             if not _code:
+                self._save_partial_dict()
                 raise click.exceptions.UsageError(f"No IR signal learnt for {_combination_str} command within timeout.")
 
             # swing modes must be saved because all temperature need to be listened

broadlink_listener/py.typed

@@ -0,0 +1 @@
+

broadlink_listener/py.typed.license

@@ -0,0 +1,3 @@
+SPDX-FileCopyrightText: 2023 Gabriele Pongelli
+
+SPDX-License-Identifier: MIT

pyproject.toml

@@ -64,6 +64,7 @@
             types-termcolor = "^1.1.5"
             types-setuptools = "^65.3.0"
             reuse = "^1.1.0"
+            freezegun = "^1.2.2"
 
 
     [tool.poetry.scripts]