Merge 9e6deb4 into 56cea7e

utils/measure/.env.dist

@@ -4,6 +4,9 @@ POWER_METER=shelly
 # Set the light controller to use (hass, hue)
 LIGHT_CONTROLLER=hue
 
+# Set the media controller to use (for measuring smart speakers).
+MEDIA_CONTROLLER=hass
+
 # Set the log level (CRITICAL, ERROR, WARNING, INFO, DEBUG)
 LOG_LEVEL=INFO
 

utils/measure/config.py

@@ -0,0 +1,90 @@
+import logging
+from decouple import Choices, UndefinedValueError, config
+from light_controller.const import LightControllerType
+from media_controller.const import MediaControllerType
+from powermeter.const import PowerMeterType
+
+MIN_BRIGHTNESS = min(max(
+    config(
+        "MIN_BRIGHTNESS",
+        default=config("START_BRIGHTNESS", default=1, cast=int),
+        cast=int
+    ), 1), 255
+)
+MAX_BRIGHTNESS = 255
+MIN_SAT = min(max(config("MIN_SAT", default=1, cast=int), 1), 255)
+MAX_SAT = min(max(config("MAX_SAT", default=255, cast=int), 1), 255)
+MIN_HUE = min(max(config("MIN_HUE", default=1, cast=int), 1), 65535)
+MAX_HUE = min(max(config("MAX_HUE", default=65535, cast=int), 1), 65535)
+CT_BRI_STEPS = min(config("CT_BRI_STEPS", default=5, cast=int), 10)
+CT_MIRED_STEPS = min(config("CT_MIRED_STEPS", default=10, cast=int), 10)
+BRI_BRI_STEPS = 1
+
+HS_BRI_PRECISION = config("HS_BRI_PRECISION", default=1, cast=float)
+HS_BRI_PRECISION = min(HS_BRI_PRECISION, 4)
+HS_BRI_PRECISION = max(HS_BRI_PRECISION, 0.5)
+HS_BRI_STEPS = round(32 / HS_BRI_PRECISION)
+del HS_BRI_PRECISION
+
+HS_HUE_PRECISION = config("HS_HUE_PRECISION", default=1, cast=float)
+HS_HUE_PRECISION = min(HS_HUE_PRECISION, 4)
+HS_HUE_PRECISION = max(HS_HUE_PRECISION, 0.5)
+HS_HUE_STEPS = round(2731 / HS_HUE_PRECISION)
+del HS_HUE_PRECISION
+
+HS_SAT_PRECISION = config("HS_SAT_PRECISION", default=1, cast=float)
+HS_SAT_PRECISION = min(HS_SAT_PRECISION, 4)
+HS_SAT_PRECISION = max(HS_SAT_PRECISION, 0.5)
+HS_SAT_STEPS = round(32 / HS_SAT_PRECISION)
+del HS_SAT_PRECISION
+
+SELECTED_LIGHT_CONTROLLER = config("LIGHT_CONTROLLER", cast=Choices([t.value for t in LightControllerType]))
+SELECTED_MEDIA_CONTROLLER = config("MEDIA_CONTROLLER", cast=Choices([t.value for t in MediaControllerType]), default=MediaControllerType.HASS)
+SELECTED_POWER_METER = config("POWER_METER", cast=Choices([t.value for t in PowerMeterType]))
+
+LOG_LEVEL = config("LOG_LEVEL", default=logging.INFO)
+SLEEP_INITIAL = 10
+SLEEP_STANDBY = config("SLEEP_STANDBY", default=20, cast=int)
+SLEEP_TIME = config("SLEEP_TIME", default=2, cast=int)
+SLEEP_TIME_SAMPLE = config("SLEEP_TIME_SAMPLE", default=1, cast=int)
+SLEEP_TIME_HUE = config("SLEEP_TIME_HUE", default=5, cast=int)
+SLEEP_TIME_SAT = config("SLEEP_TIME_SAT", default=10, cast=int)
+SLEEP_TIME_CT = config("SLEEP_TIME_CT", default=10, cast=int)
+SLEEP_TIME_NUDGE = config("SLEEP_TIME_NUDGE", default=10, cast=float)
+
+PULSE_TIME_NUDGE = config("PULSE_TIME_NUDGE", default=2, cast=float)
+MAX_RETRIES = config("MAX_RETRIES", default=5, cast=int)
+MAX_NUDGES = config("MAX_NUDGES", default=0, cast=int)
+SAMPLE_COUNT = config("SAMPLE_COUNT", default=1, cast=int)
+
+SHELLY_IP = config("SHELLY_IP")
+SHELLY_TIMEOUT = config("SHELLY_TIMEOUT", default=5, cast=int)
+TUYA_DEVICE_ID = config("TUYA_DEVICE_ID")
+TUYA_DEVICE_IP = config("TUYA_DEVICE_IP")
+TUYA_DEVICE_KEY = config("TUYA_DEVICE_KEY")
+TUYA_DEVICE_VERSION = config("TUYA_DEVICE_VERSION", default="3.3")
+HUE_BRIDGE_IP = config("HUE_BRIDGE_IP")
+HASS_URL = config("HASS_URL")
+HASS_TOKEN = config("HASS_TOKEN")
+HASS_CALL_UPDATE_ENTITY_SERVICE = config("HASS_CALL_UPDATE_ENTITY_SERVICE", default=False, cast=bool)
+TASMOTA_DEVICE_IP = config("TASMOTA_DEVICE_IP")
+KASA_DEVICE_IP = config("KASA_DEVICE_IP")
+
+CSV_ADD_DATETIME_COLUMN = config("CSV_ADD_DATETIME_COLUMN", default=False, cast=bool)
+
+try:
+    SELECTED_DEVICE_TYPE = config("SELECTED_DEVICE_TYPE")
+except UndefinedValueError:
+    SELECTED_DEVICE_TYPE = None
+
+try:
+    RESUME = config("RESUME", cast=bool)
+except UndefinedValueError:
+    RESUME = None
+
+# Change some settings when selected power meter is manual
+if SELECTED_POWER_METER == PowerMeterType.MANUAL:
+    SAMPLE_COUNT = 1
+    BRI_BRI_STEPS = 3
+    CT_BRI_STEPS = 15
+    CT_MIRED_STEPS = 50

utils/measure/light_controller/const.py

@@ -1,6 +1,16 @@
-MODE_HS = "hs"
-MODE_COLOR_TEMP = "color_temp"
-MODE_BRIGHTNESS = "brightness"
+from enum import Enum
 
 MIN_MIRED = 150
 MAX_MIRED = 500
+
+
+class ColorMode(str, Enum):
+    HS = "hs"
+    COLOR_TEMP = "color_temp"
+    BRIGHTNESS = "brightness"
+
+
+class LightControllerType(str, Enum):
+    DUMMY = "dummy"
+    HASS = "hass"
+    HUE = "hue"

utils/measure/light_controller/controller.py

@@ -2,6 +2,8 @@
 
 from typing import Any, Protocol
 
+import inquirer.questions
+
 from .const import MAX_MIRED, MIN_MIRED
 
 
@@ -42,7 +44,7 @@ def get_light_info(self) -> LightInfo:
         """Get device information about the light"""
         ...
 
-    def get_questions(self) -> list[dict]:
+    def get_questions(self) -> list[inquirer.questions.Question]:
         """Get questions to ask for the chosen light controller"""
         ...
 

utils/measure/light_controller/dummy.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+import inquirer.questions
+
 from .controller import LightInfo
 
 
@@ -10,9 +12,9 @@ def change_light_state(self, color_mode: str, on: bool = True, **kwargs):
         pass
 
     def get_light_info(self) -> LightInfo:
-        return LightInfo("")
+        return LightInfo("dummy")
 
-    def get_questions(self) -> list[dict]:
+    def get_questions(self) -> list[inquirer.questions.Question]:
         return []
 
     def process_answers(self, answers: dict[str, Any]):

utils/measure/light_controller/factory.py

@@ -0,0 +1,37 @@
+import logging
+
+from .hass import HassLightController
+from .hue import HueLightController
+from .dummy import DummyLightController
+from .const import LightControllerType
+from .controller import LightController
+import config
+
+_LOGGER = logging.getLogger("measure")
+
+
+class LightControllerFactory:
+    @staticmethod
+    def hass():
+        return HassLightController(config.HASS_URL, config.HASS_TOKEN)
+
+    @staticmethod
+    def hue():
+        return HueLightController(config.HUE_BRIDGE_IP)
+
+    @staticmethod
+    def dummy():
+        return DummyLightController()
+
+    def create(self) -> LightController:
+        """Create the light controller instance"""
+        factories = {
+            LightControllerType.DUMMY: self.dummy,
+            LightControllerType.HUE: self.hue,
+            LightControllerType.HASS: self.hass
+        }
+        factory = factories.get(config.SELECTED_LIGHT_CONTROLLER)
+        if factory is None:
+            raise Exception(f"Could not find a factory for {config.SELECTED_LIGHT_CONTROLLER}")
+
+        return factory()

utils/measure/light_controller/hass.py

@@ -5,7 +5,7 @@
 import inquirer
 from homeassistant_api import Client
 
-from .const import MAX_MIRED, MIN_MIRED, MODE_COLOR_TEMP, MODE_HS
+from .const import MAX_MIRED, MIN_MIRED, ColorMode
 from .controller import LightInfo
 from .errors import LightControllerError
 
@@ -25,9 +25,9 @@ def change_light_state(self, color_mode: str, on: bool = True, **kwargs):
             self.client.trigger_service('light', 'turn_off', entity_id=self._entity_id)
             return
 
-        if color_mode == MODE_HS:
+        if color_mode == ColorMode.HS:
             json = self.build_hs_json_body(**kwargs)
-        elif color_mode == MODE_COLOR_TEMP:
+        elif color_mode == ColorMode.COLOR_TEMP:
             json = self.build_ct_json_body(**kwargs)
         else:
             json = self.build_bri_json_body(**kwargs)
@@ -40,7 +40,7 @@ def get_light_info(self) -> LightInfo:
         max_mired = state.attributes.get("max_mireds") or MAX_MIRED
         return LightInfo(self._model_id, min_mired, max_mired)
 
-    def get_questions(self) -> list[dict]:
+    def get_questions(self) -> list[inquirer.questions.Question]:
         entities = self.client.get_entities()
         lights = entities["light"].entities.values()
         light_list = sorted([entity.entity_id for entity in lights])

utils/measure/light_controller/hue.py

@@ -62,7 +62,8 @@ def find_group_model(self, group_id: int) -> str:
 
         return model_ids.pop()
 
-    def initialize_hue_bridge(self, bridge_ip: str) -> Bridge:
+    @staticmethod
+    def initialize_hue_bridge(bridge_ip: str) -> Bridge:
         config_file_path = os.path.join(os.path.dirname(__file__), "../.persistent/.python_hue")
         try:
             bridge = Bridge(ip=bridge_ip, config_file_path=config_file_path)
@@ -73,7 +74,7 @@ def initialize_hue_bridge(self, bridge_ip: str) -> Bridge:
 
         return bridge
 
-    def get_questions(self) -> list[dict]:
+    def get_questions(self) -> list[inquirer.questions.Question]:
 
         def get_message(answers) -> str:
             if answers.get("multiple_lights"):