kobim/SmartIR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
Mirko Milovanovic
2026-05-29 11:46:44 +02:00
commit 2dd72d367c
465 changed files with 155400 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/smartir/__init__.py
+171
View File
@@ -0,0 +1,171 @@
import aiofiles
import aiohttp
import asyncio
import binascii
from distutils.version import StrictVersion
import json
import logging
import os.path
import requests
import struct
import voluptuous as vol
from aiohttp import ClientSession
from homeassistant.const import (
ATTR_FRIENDLY_NAME, __version__ as current_ha_version)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'smartir'
VERSION = '1.18.1'
MANIFEST_URL = (
"https://raw.githubusercontent.com/"
"smartHomeHub/SmartIR/{}/"
"custom_components/smartir/manifest.json")
REMOTE_BASE_URL = (
"https://raw.githubusercontent.com/"
"smartHomeHub/SmartIR/{}/"
"custom_components/smartir/")
COMPONENT_ABS_DIR = os.path.dirname(
os.path.abspath(__file__))
CONF_CHECK_UPDATES = 'check_updates'
CONF_UPDATE_BRANCH = 'update_branch'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_CHECK_UPDATES, default=True): cv.boolean,
vol.Optional(CONF_UPDATE_BRANCH, default='master'): vol.In(
['master', 'rc'])
})
}, extra=vol.ALLOW_EXTRA)
async def async_setup(hass, config):
"""Set up the SmartIR component."""
conf = config.get(DOMAIN)
if conf is None:
return True
check_updates = conf[CONF_CHECK_UPDATES]
update_branch = conf[CONF_UPDATE_BRANCH]
async def _check_updates(service):
await _update(hass, update_branch)
async def _update_component(service):
await _update(hass, update_branch, True)
hass.services.async_register(DOMAIN, 'check_updates', _check_updates)
hass.services.async_register(DOMAIN, 'update_component', _update_component)
if check_updates:
await _update(hass, update_branch, False, False)
return True
async def _update(hass, branch, do_update=False, notify_if_latest=True):
try:
async with aiohttp.ClientSession() as session:
async with session.get(MANIFEST_URL.format(branch)) as response:
if response.status == 200:
data = await response.json(content_type='text/plain')
min_ha_version = data['homeassistant']
last_version = data['updater']['version']
release_notes = data['updater']['releaseNotes']
if StrictVersion(last_version) <= StrictVersion(VERSION):
if notify_if_latest:
hass.components.persistent_notification.async_create(
"You're already using the latest version!",
title='SmartIR')
return
if StrictVersion(current_ha_version) < StrictVersion(min_ha_version):
hass.components.persistent_notification.async_create(
"There is a new version of SmartIR integration, but it is **incompatible** "
"with your system. Please first update Home Assistant.", title='SmartIR')
return
if do_update is False:
hass.components.persistent_notification.async_create(
"A new version of SmartIR integration is available ({}). "
"Call the ``smartir.update_component`` service to update "
"the integration. \n\n **Release notes:** \n{}"
.format(last_version, release_notes), title='SmartIR')
return
# Begin update
files = data['updater']['files']
has_errors = False
for file in files:
try:
source = REMOTE_BASE_URL.format(branch) + file
dest = os.path.join(COMPONENT_ABS_DIR, file)
os.makedirs(os.path.dirname(dest), exist_ok=True)
await Helper.downloader(source, dest)
except Exception:
has_errors = True
_LOGGER.error("Error updating %s. Please update the file manually.", file)
if has_errors:
hass.components.persistent_notification.async_create(
"There was an error updating one or more files of SmartIR. "
"Please check the logs for more information.", title='SmartIR')
else:
hass.components.persistent_notification.async_create(
"Successfully updated to {}. Please restart Home Assistant."
.format(last_version), title='SmartIR')
except Exception:
_LOGGER.error("An error occurred while checking for updates.")
class Helper():
@staticmethod
async def downloader(source, dest):
async with aiohttp.ClientSession() as session:
async with session.get(source) as response:
if response.status == 200:
async with aiofiles.open(dest, mode='wb') as f:
await f.write(await response.read())
else:
raise Exception("File not found")
@staticmethod
def pronto2lirc(pronto):
codes = [int(binascii.hexlify(pronto[i:i+2]), 16) for i in range(0, len(pronto), 2)]
if codes[0]:
raise ValueError("Pronto code should start with 0000")
if len(codes) != 4 + 2 * (codes[2] + codes[3]):
raise ValueError("Number of pulse widths does not match the preamble")
frequency = 1 / (codes[1] * 0.241246)
return [int(round(code / frequency)) for code in codes[4:]]
@staticmethod
def lirc2broadlink(pulses):
array = bytearray()
for pulse in pulses:
pulse = int(pulse * 269 / 8192)
if pulse < 256:
array += bytearray(struct.pack('>B', pulse))
else:
array += bytearray([0x00])
array += bytearray(struct.pack('>H', pulse))
packet = bytearray([0x26, 0x00])
packet += bytearray(struct.pack('<H', len(array)))
packet += array
packet += bytearray([0x0d, 0x05])
# Add 0s to make ultimate packet size a multiple of 16 for 128-bit AES encryption.
remainder = (len(packet) + 4) % 16
if remainder:
packet += bytearray(16 - remainder)
return packet
custom_components/smartir/climate.py
+454
View File
@@ -0,0 +1,454 @@
import asyncio
import aiofiles
import json
import logging
import os.path
import voluptuous as vol
from homeassistant.components.climate import ClimateEntity, PLATFORM_SCHEMA
from homeassistant.components.climate.const import (
ClimateEntityFeature, HVACMode, HVAC_MODES, ATTR_HVAC_MODE)
from homeassistant.const import (
CONF_NAME, STATE_ON, STATE_OFF, STATE_UNKNOWN, STATE_UNAVAILABLE, ATTR_TEMPERATURE,
PRECISION_TENTHS, PRECISION_HALVES, PRECISION_WHOLE)
from homeassistant.core import Event, EventStateChangedData, callback
from homeassistant.helpers.event import async_track_state_change, async_track_state_change_event
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from . import COMPONENT_ABS_DIR, Helper
from .controller import get_controller
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "SmartIR Climate"
DEFAULT_DELAY = 0.5
CONF_UNIQUE_ID = 'unique_id'
CONF_DEVICE_CODE = 'device_code'
CONF_CONTROLLER_DATA = "controller_data"
CONF_DELAY = "delay"
CONF_TEMPERATURE_SENSOR = 'temperature_sensor'
CONF_HUMIDITY_SENSOR = 'humidity_sensor'
CONF_POWER_SENSOR = 'power_sensor'
CONF_POWER_SENSOR_RESTORE_STATE = 'power_sensor_restore_state'
SUPPORT_FLAGS = (
ClimateEntityFeature.TURN_OFF |
ClimateEntityFeature.TURN_ON |
ClimateEntityFeature.TARGET_TEMPERATURE |
ClimateEntityFeature.FAN_MODE
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_DEVICE_CODE): cv.positive_int,
vol.Required(CONF_CONTROLLER_DATA): cv.string,
vol.Optional(CONF_DELAY, default=DEFAULT_DELAY): cv.positive_float,
vol.Optional(CONF_TEMPERATURE_SENSOR): cv.entity_id,
vol.Optional(CONF_HUMIDITY_SENSOR): cv.entity_id,
vol.Optional(CONF_POWER_SENSOR): cv.entity_id,
vol.Optional(CONF_POWER_SENSOR_RESTORE_STATE, default=False): cv.boolean
})
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the IR Climate platform."""
_LOGGER.debug("Setting up the smartir platform")
device_code = config.get(CONF_DEVICE_CODE)
device_files_subdir = os.path.join('codes', 'climate')
device_files_absdir = os.path.join(COMPONENT_ABS_DIR, device_files_subdir)
if not os.path.isdir(device_files_absdir):
os.makedirs(device_files_absdir)
device_json_filename = str(device_code) + '.json'
device_json_path = os.path.join(device_files_absdir, device_json_filename)
if not os.path.exists(device_json_path):
_LOGGER.warning("Couldn't find the device Json file. The component will " \
"try to download it from the GitHub repo.")
try:
codes_source = ("https://raw.githubusercontent.com/"
"smartHomeHub/SmartIR/master/"
"codes/climate/{}.json")
await Helper.downloader(codes_source.format(device_code), device_json_path)
except Exception:
_LOGGER.error("There was an error while downloading the device Json file. " \
"Please check your internet connection or if the device code " \
"exists on GitHub. If the problem still exists please " \
"place the file manually in the proper directory.")
return
try:
async with aiofiles.open(device_json_path, mode='r') as j:
_LOGGER.debug(f"loading json file {device_json_path}")
content = await j.read()
device_data = json.loads(content)
_LOGGER.debug(f"{device_json_path} file loaded")
except Exception:
_LOGGER.error("The device JSON file is invalid")
return
async_add_entities([SmartIRClimate(
hass, config, device_data
)])
class SmartIRClimate(ClimateEntity, RestoreEntity):
def __init__(self, hass, config, device_data):
_LOGGER.debug(f"SmartIRClimate init started for device {config.get(CONF_NAME)} supported models {device_data['supportedModels']}")
self.hass = hass
self._unique_id = config.get(CONF_UNIQUE_ID)
self._name = config.get(CONF_NAME)
self._device_code = config.get(CONF_DEVICE_CODE)
self._controller_data = config.get(CONF_CONTROLLER_DATA)
self._delay = config.get(CONF_DELAY)
self._temperature_sensor = config.get(CONF_TEMPERATURE_SENSOR)
self._humidity_sensor = config.get(CONF_HUMIDITY_SENSOR)
self._power_sensor = config.get(CONF_POWER_SENSOR)
self._power_sensor_restore_state = config.get(CONF_POWER_SENSOR_RESTORE_STATE)
self._manufacturer = device_data['manufacturer']
self._supported_models = device_data['supportedModels']
self._supported_controller = device_data['supportedController']
self._commands_encoding = device_data['commandsEncoding']
self._min_temperature = device_data['minTemperature']
self._max_temperature = device_data['maxTemperature']
self._precision = device_data['precision']
valid_hvac_modes = [x for x in device_data['operationModes'] if x in HVAC_MODES]
self._operation_modes = [HVACMode.OFF] + valid_hvac_modes
self._fan_modes = device_data['fanModes']
self._swing_modes = device_data.get('swingModes')
self._commands = device_data['commands']
self._target_temperature = self._min_temperature
self._hvac_mode = HVACMode.OFF
self._current_fan_mode = self._fan_modes[0]
self._current_swing_mode = None
self._last_on_operation = None
self._current_temperature = None
self._current_humidity = None
self._unit = hass.config.units.temperature_unit
#Supported features
self._support_flags = SUPPORT_FLAGS
self._support_swing = False
if self._swing_modes:
self._support_flags = self._support_flags | ClimateEntityFeature.SWING_MODE
self._current_swing_mode = self._swing_modes[0]
self._support_swing = True
self._temp_lock = asyncio.Lock()
self._on_by_remote = False
#Init the IR/RF controller
self._controller = get_controller(
self.hass,
self._supported_controller,
self._commands_encoding,
self._controller_data,
self._delay)
async def async_added_to_hass(self):
"""Run when entity about to be added."""
await super().async_added_to_hass()
_LOGGER.debug(f"async_added_to_hass {self} {self.name} {self.supported_features}")
last_state = await self.async_get_last_state()
if last_state is not None:
self._hvac_mode = last_state.state
self._current_fan_mode = last_state.attributes['fan_mode']
self._current_swing_mode = last_state.attributes.get('swing_mode')
self._target_temperature = last_state.attributes['temperature']
if 'last_on_operation' in last_state.attributes:
self._last_on_operation = last_state.attributes['last_on_operation']
if self._temperature_sensor:
async_track_state_change_event(self.hass, self._temperature_sensor,
self._async_temp_sensor_changed)
temp_sensor_state = self.hass.states.get(self._temperature_sensor)
if temp_sensor_state and temp_sensor_state.state != STATE_UNKNOWN:
self._async_update_temp(temp_sensor_state)
if self._humidity_sensor:
async_track_state_change_event(self.hass, self._humidity_sensor,
self._async_humidity_sensor_changed)
humidity_sensor_state = self.hass.states.get(self._humidity_sensor)
if humidity_sensor_state and humidity_sensor_state.state != STATE_UNKNOWN:
self._async_update_humidity(humidity_sensor_state)
if self._power_sensor:
async_track_state_change_event(self.hass, self._power_sensor,
self._async_power_sensor_changed)
@property
def unique_id(self):
"""Return a unique ID."""
return self._unique_id
@property
def name(self):
"""Return the name of the climate device."""
return self._name
@property
def state(self):
"""Return the current state."""
if self.hvac_mode != HVACMode.OFF:
return self.hvac_mode
return HVACMode.OFF
@property
def temperature_unit(self):
"""Return the unit of measurement."""
return self._unit
@property
def min_temp(self):
"""Return the polling state."""
return self._min_temperature
@property
def max_temp(self):
"""Return the polling state."""
return self._max_temperature
@property
def target_temperature(self):
"""Return the temperature we try to reach."""
return self._target_temperature
@property
def target_temperature_step(self):
"""Return the supported step of target temperature."""
return self._precision
@property
def hvac_modes(self):
"""Return the list of available operation modes."""
return self._operation_modes
@property
def hvac_mode(self):
"""Return hvac mode ie. heat, cool."""
return self._hvac_mode
@property
def last_on_operation(self):
"""Return the last non-idle operation ie. heat, cool."""
return self._last_on_operation
@property
def fan_modes(self):
"""Return the list of available fan modes."""
return self._fan_modes
@property
def fan_mode(self):
"""Return the fan setting."""
return self._current_fan_mode
@property
def swing_modes(self):
"""Return the swing modes currently supported for this device."""
return self._swing_modes
@property
def swing_mode(self):
"""Return the current swing mode."""
return self._current_swing_mode
@property
def current_temperature(self):
"""Return the current temperature."""
return self._current_temperature
@property
def current_humidity(self):
"""Return the current humidity."""
return self._current_humidity
@property
def supported_features(self):
"""Return the list of supported features."""
return self._support_flags
@property
def extra_state_attributes(self):
"""Platform specific attributes."""
return {
'last_on_operation': self._last_on_operation,
'device_code': self._device_code,
'manufacturer': self._manufacturer,
'supported_models': self._supported_models,
'supported_controller': self._supported_controller,
'commands_encoding': self._commands_encoding
}
async def async_set_temperature(self, **kwargs):
"""Set new target temperatures."""
hvac_mode = kwargs.get(ATTR_HVAC_MODE)
temperature = kwargs.get(ATTR_TEMPERATURE)
if temperature is None:
return
if temperature < self._min_temperature or temperature > self._max_temperature:
_LOGGER.warning('The temperature value is out of min/max range')
return
if self._precision == PRECISION_WHOLE:
self._target_temperature = round(temperature)
else:
self._target_temperature = round(temperature, 1)
if hvac_mode:
await self.async_set_hvac_mode(hvac_mode)
return
if not self._hvac_mode.lower() == HVACMode.OFF:
await self.send_command()
self.async_write_ha_state()
async def async_set_hvac_mode(self, hvac_mode):
"""Set operation mode."""
self._hvac_mode = hvac_mode
if not hvac_mode == HVACMode.OFF:
self._last_on_operation = hvac_mode
await self.send_command()
self.async_write_ha_state()
async def async_set_fan_mode(self, fan_mode):
"""Set fan mode."""
self._current_fan_mode = fan_mode
if not self._hvac_mode.lower() == HVACMode.OFF:
await self.send_command()
self.async_write_ha_state()
async def async_set_swing_mode(self, swing_mode):
"""Set swing mode."""
self._current_swing_mode = swing_mode
if not self._hvac_mode.lower() == HVACMode.OFF:
await self.send_command()
self.async_write_ha_state()
async def async_turn_off(self):
"""Turn off."""
await self.async_set_hvac_mode(HVACMode.OFF)
async def async_turn_on(self):
"""Turn on."""
if self._last_on_operation is not None:
await self.async_set_hvac_mode(self._last_on_operation)
else:
await self.async_set_hvac_mode(self._operation_modes[1])
async def send_command(self):
async with self._temp_lock:
try:
self._on_by_remote = False
operation_mode = self._hvac_mode
fan_mode = self._current_fan_mode
swing_mode = self._current_swing_mode
target_temperature = '{0:g}'.format(self._target_temperature)
if operation_mode.lower() == HVACMode.OFF:
await self._controller.send(self._commands['off'])
return
if 'on' in self._commands:
await self._controller.send(self._commands['on'])
await asyncio.sleep(self._delay)
if self._support_swing == True:
await self._controller.send(
self._commands[operation_mode][fan_mode][swing_mode][target_temperature])
else:
await self._controller.send(
self._commands[operation_mode][fan_mode][target_temperature])
except Exception as e:
_LOGGER.exception(e)
@callback
async def _async_temp_sensor_changed(self, event: Event[EventStateChangedData]) -> None:
"""Handle temperature sensor changes."""
new_state = event.data["new_state"]
if new_state is None:
return
self._async_update_temp(new_state)
self.async_write_ha_state()
@callback
async def _async_humidity_sensor_changed(self, event: Event[EventStateChangedData]) -> None:
"""Handle humidity sensor changes."""
new_state = event.data["new_state"]
if new_state is None:
return
self._async_update_humidity(new_state)
self.async_write_ha_state()
@callback
async def _async_power_sensor_changed(self, event: Event[EventStateChangedData]) -> None:
entity_id = event.data["entity_id"]
old_state = event.data["old_state"]
new_state = event.data["new_state"]
if new_state is None:
return
if old_state is not None and new_state.state == old_state.state:
return
if new_state.state == STATE_ON and self._hvac_mode == HVACMode.OFF:
self._on_by_remote = True
if self._power_sensor_restore_state == True and self._last_on_operation is not None:
self._hvac_mode = self._last_on_operation
else:
self._hvac_mode = STATE_ON
self.async_write_ha_state()
if new_state.state == STATE_OFF:
self._on_by_remote = False
if self._hvac_mode != HVACMode.OFF:
self._hvac_mode = HVACMode.OFF
self.async_write_ha_state()
@callback
def _async_update_temp(self, state):
"""Update thermostat with latest state from temperature sensor."""
try:
if state.state != STATE_UNKNOWN and state.state != STATE_UNAVAILABLE:
self._current_temperature = float(state.state)
except ValueError as ex:
_LOGGER.error("Unable to update from temperature sensor: %s", ex)
@callback
def _async_update_humidity(self, state):
"""Update thermostat with latest state from humidity sensor."""
try:
if state.state != STATE_UNKNOWN and state.state != STATE_UNAVAILABLE:
self._current_humidity = float(state.state)
except ValueError as ex:
_LOGGER.error("Unable to update from humidity sensor: %s", ex)
custom_components/smartir/codes_index.json
+4209
View File
File diff suppressed because it is too large Load Diff
custom_components/smartir/const.py
+59
View File
@@ -0,0 +1,59 @@
"""Constants for the SmartIR config flow and platforms."""
from __future__ import annotations
# ── Sentinels ──────────────────────────────────────────────────────────────────
MANUAL_ENTRY_SENTINEL = "__manual__"
ALL_CONTROLLERS_SENTINEL = "Any"
INDEX_FILENAME = "codes_index.json"
# ── Config-entry data keys ─────────────────────────────────────────────────────
CONF_PLATFORM = "platform"
CONF_DEVICE_CODE = "device_code"
CONF_CONTROLLER_DATA = "controller_data"
CONF_DELAY = "delay"
CONF_DEVICE_NAME = "name"
CONF_TEMPERATURE_SENSOR = "temperature_sensor"
CONF_HUMIDITY_SENSOR = "humidity_sensor"
CONF_POWER_SENSOR = "power_sensor"
CONF_POWER_SENSOR_RESTORE_STATE = "power_sensor_restore_state"
CONF_DEVICE_CLASS = "device_class"
CONF_SOURCE_NAMES = "source_names"
# ── Defaults ───────────────────────────────────────────────────────────────────
DEFAULT_DELAY = "0.5"
# ── Platform choices ───────────────────────────────────────────────────────────
PLATFORMS: dict[str, str] = {
"climate": "Climate (Air Conditioner / Heat Pump)",
"fan": "Fan",
"media_player": "Media Player (TV / Receiver)",
"light": "Light",
}
PLATFORM_SUBDIR: dict[str, str] = {
"climate": "climate",
"fan": "fan",
"media_player": "media_player",
"light": "light",
}
DEFAULT_DEVICE_NAMES: dict[str, str] = {
"climate": "SmartIR Climate",
"fan": "SmartIR Fan",
"media_player": "SmartIR Media Player",
"light": "SmartIR Light",
}
# ── Controller metadata ────────────────────────────────────────────────────────
# Controllers whose device identifier is a Home Assistant remote entity ID.
# These get an EntitySelector in the UI instead of a plain text field.
ENTITY_BASED_CONTROLLERS: frozenset[str] = frozenset({"Broadlink", "Xiaomi"})
CONTROLLER_HINTS: dict[str, str] = {
"Broadlink": "Select the Broadlink remote entity that will send IR commands.",
"Xiaomi": "Select the Xiaomi remote entity that will send IR commands.",
"MQTT": "Enter the MQTT topic to publish IR commands to (e.g. zigbee2mqtt/ir_blaster/set).",
"LOOKin": "Enter the LOOKin device IP address (e.g. 192.168.1.42).",
"ESPHome": "Enter the ESPHome remote service name (e.g. esphome_ir_blaster).",
}
custom_components/smartir/controller.py
+186
View File
@@ -0,0 +1,186 @@
from abc import ABC, abstractmethod
from base64 import b64encode
import binascii
import requests
import logging
import json
from homeassistant.const import ATTR_ENTITY_ID
from . import Helper
_LOGGER = logging.getLogger(__name__)
BROADLINK_CONTROLLER = 'Broadlink'
XIAOMI_CONTROLLER = 'Xiaomi'
MQTT_CONTROLLER = 'MQTT'
LOOKIN_CONTROLLER = 'LOOKin'
ESPHOME_CONTROLLER = 'ESPHome'
ENC_BASE64 = 'Base64'
ENC_HEX = 'Hex'
ENC_PRONTO = 'Pronto'
ENC_RAW = 'Raw'
BROADLINK_COMMANDS_ENCODING = [ENC_BASE64, ENC_HEX, ENC_PRONTO]
XIAOMI_COMMANDS_ENCODING = [ENC_PRONTO, ENC_RAW]
MQTT_COMMANDS_ENCODING = [ENC_RAW]
LOOKIN_COMMANDS_ENCODING = [ENC_PRONTO, ENC_RAW]
ESPHOME_COMMANDS_ENCODING = [ENC_RAW]
def get_controller(hass, controller, encoding, controller_data, delay):
"""Return a controller compatible with the specification provided."""
controllers = {
BROADLINK_CONTROLLER: BroadlinkController,
XIAOMI_CONTROLLER: XiaomiController,
MQTT_CONTROLLER: MQTTController,
LOOKIN_CONTROLLER: LookinController,
ESPHOME_CONTROLLER: ESPHomeController
}
try:
return controllers[controller](hass, controller, encoding, controller_data, delay)
except KeyError:
raise Exception("The controller is not supported.")
class AbstractController(ABC):
"""Representation of a controller."""
def __init__(self, hass, controller, encoding, controller_data, delay):
self.check_encoding(encoding)
self.hass = hass
self._controller = controller
self._encoding = encoding
self._controller_data = controller_data
self._delay = delay
@abstractmethod
def check_encoding(self, encoding):
"""Check if the encoding is supported by the controller."""
pass
@abstractmethod
async def send(self, command):
"""Send a command."""
pass
class BroadlinkController(AbstractController):
"""Controls a Broadlink device."""
def check_encoding(self, encoding):
"""Check if the encoding is supported by the controller."""
if encoding not in BROADLINK_COMMANDS_ENCODING:
raise Exception("The encoding is not supported "
"by the Broadlink controller.")
async def send(self, command):
"""Send a command."""
commands = []
if not isinstance(command, list):
command = [command]
for _command in command:
if self._encoding == ENC_HEX:
try:
_command = binascii.unhexlify(_command)
_command = b64encode(_command).decode('utf-8')
except:
raise Exception("Error while converting "
"Hex to Base64 encoding")
if self._encoding == ENC_PRONTO:
try:
_command = _command.replace(' ', '')
_command = bytearray.fromhex(_command)
_command = Helper.pronto2lirc(_command)
_command = Helper.lirc2broadlink(_command)
_command = b64encode(_command).decode('utf-8')
except:
raise Exception("Error while converting "
"Pronto to Base64 encoding")
commands.append('b64:' + _command)
service_data = {
ATTR_ENTITY_ID: self._controller_data,
'command': commands,
'delay_secs': self._delay
}
await self.hass.services.async_call(
'remote', 'send_command', service_data)
class XiaomiController(AbstractController):
"""Controls a Xiaomi device."""
def check_encoding(self, encoding):
"""Check if the encoding is supported by the controller."""
if encoding not in XIAOMI_COMMANDS_ENCODING:
raise Exception("The encoding is not supported "
"by the Xiaomi controller.")
async def send(self, command):
"""Send a command."""
service_data = {
ATTR_ENTITY_ID: self._controller_data,
'command': self._encoding.lower() + ':' + command
}
await self.hass.services.async_call(
'remote', 'send_command', service_data)
class MQTTController(AbstractController):
"""Controls a MQTT device."""
def check_encoding(self, encoding):
"""Check if the encoding is supported by the controller."""
if encoding not in MQTT_COMMANDS_ENCODING:
raise Exception("The encoding is not supported "
"by the mqtt controller.")
async def send(self, command):
"""Send a command."""
service_data = {
'topic': self._controller_data,
'payload': command
}
await self.hass.services.async_call(
'mqtt', 'publish', service_data)
class LookinController(AbstractController):
"""Controls a Lookin device."""
def check_encoding(self, encoding):
"""Check if the encoding is supported by the controller."""
if encoding not in LOOKIN_COMMANDS_ENCODING:
raise Exception("The encoding is not supported "
"by the LOOKin controller.")
async def send(self, command):
"""Send a command."""
encoding = self._encoding.lower().replace('pronto', 'prontohex')
url = f"http://{self._controller_data}/commands/ir/" \
f"{encoding}/{command}"
await self.hass.async_add_executor_job(requests.get, url)
class ESPHomeController(AbstractController):
"""Controls a ESPHome device."""
def check_encoding(self, encoding):
"""Check if the encoding is supported by the controller."""
if encoding not in ESPHOME_COMMANDS_ENCODING:
raise Exception("The encoding is not supported "
"by the ESPHome controller.")
async def send(self, command):
"""Send a command."""
service_data = {'command': json.loads(command)}
await self.hass.services.async_call(
'esphome', self._controller_data, service_data)
custom_components/smartir/fan.py
+307
View File
@@ -0,0 +1,307 @@
import asyncio
import aiofiles
import json
import logging
import os.path
import voluptuous as vol
from homeassistant.components.fan import (
FanEntity, FanEntityFeature,
PLATFORM_SCHEMA, DIRECTION_REVERSE, DIRECTION_FORWARD)
from homeassistant.const import (
CONF_NAME, STATE_OFF, STATE_ON, STATE_UNKNOWN)
from homeassistant.core import Event, EventStateChangedData, callback
from homeassistant.helpers.event import async_track_state_change, async_track_state_change_event
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item
)
from . import COMPONENT_ABS_DIR, Helper
from .controller import get_controller
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "SmartIR Fan"
DEFAULT_DELAY = 0.5
CONF_UNIQUE_ID = 'unique_id'
CONF_DEVICE_CODE = 'device_code'
CONF_CONTROLLER_DATA = "controller_data"
CONF_DELAY = "delay"
CONF_POWER_SENSOR = 'power_sensor'
SPEED_OFF = "off"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_DEVICE_CODE): cv.positive_int,
vol.Required(CONF_CONTROLLER_DATA): cv.string,
vol.Optional(CONF_DELAY, default=DEFAULT_DELAY): cv.string,
vol.Optional(CONF_POWER_SENSOR): cv.entity_id
})
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the IR Fan platform."""
device_code = config.get(CONF_DEVICE_CODE)
device_files_subdir = os.path.join('codes', 'fan')
device_files_absdir = os.path.join(COMPONENT_ABS_DIR, device_files_subdir)
if not os.path.isdir(device_files_absdir):
os.makedirs(device_files_absdir)
device_json_filename = str(device_code) + '.json'
device_json_path = os.path.join(device_files_absdir, device_json_filename)
if not os.path.exists(device_json_path):
_LOGGER.warning("Couldn't find the device Json file. The component will " \
"try to download it from the GitHub repo.")
try:
codes_source = ("https://raw.githubusercontent.com/"
"smartHomeHub/SmartIR/master/"
"codes/fan/{}.json")
await Helper.downloader(codes_source.format(device_code), device_json_path)
except Exception:
_LOGGER.error("There was an error while downloading the device Json file. " \
"Please check your internet connection or if the device code " \
"exists on GitHub. If the problem still exists please " \
"place the file manually in the proper directory.")
return
try:
async with aiofiles.open(device_json_path, mode='r') as j:
_LOGGER.debug(f"loading json file {device_json_path}")
content = await j.read()
device_data = json.loads(content)
_LOGGER.debug(f"{device_json_path} file loaded")
except Exception:
_LOGGER.error("The device JSON file is invalid")
return
async_add_entities([SmartIRFan(
hass, config, device_data
)])
class SmartIRFan(FanEntity, RestoreEntity):
def __init__(self, hass, config, device_data):
self.hass = hass
self._unique_id = config.get(CONF_UNIQUE_ID)
self._name = config.get(CONF_NAME)
self._device_code = config.get(CONF_DEVICE_CODE)
self._controller_data = config.get(CONF_CONTROLLER_DATA)
self._delay = config.get(CONF_DELAY)
self._power_sensor = config.get(CONF_POWER_SENSOR)
self._manufacturer = device_data['manufacturer']
self._supported_models = device_data['supportedModels']
self._supported_controller = device_data['supportedController']
self._commands_encoding = device_data['commandsEncoding']
self._speed_list = device_data['speed']
self._commands = device_data['commands']
self._speed = SPEED_OFF
self._direction = None
self._last_on_speed = None
self._oscillating = None
self._support_flags = (
FanEntityFeature.SET_SPEED
| FanEntityFeature.TURN_OFF
| FanEntityFeature.TURN_ON)
if (DIRECTION_REVERSE in self._commands and \
DIRECTION_FORWARD in self._commands):
self._direction = DIRECTION_REVERSE
self._support_flags = (
self._support_flags | FanEntityFeature.DIRECTION)
if ('oscillate' in self._commands):
self._oscillating = False
self._support_flags = (
self._support_flags | FanEntityFeature.OSCILLATE)
self._temp_lock = asyncio.Lock()
self._on_by_remote = False
#Init the IR/RF controller
self._controller = get_controller(
self.hass,
self._supported_controller,
self._commands_encoding,
self._controller_data,
self._delay)
async def async_added_to_hass(self):
"""Run when entity about to be added."""
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if last_state is not None:
if 'speed' in last_state.attributes:
self._speed = last_state.attributes['speed']
#If _direction has a value the direction controls appears
#in UI even if SUPPORT_DIRECTION is not provided in the flags
if ('direction' in last_state.attributes and \
self._support_flags & FanEntityFeature.DIRECTION):
self._direction = last_state.attributes['direction']
if 'last_on_speed' in last_state.attributes:
self._last_on_speed = last_state.attributes['last_on_speed']
if self._power_sensor:
async_track_state_change_event(self.hass, self._power_sensor,
self._async_power_sensor_changed)
@property
def unique_id(self):
"""Return a unique ID."""
return self._unique_id
@property
def name(self):
"""Return the display name of the fan."""
return self._name
@property
def state(self):
"""Return the current state."""
if (self._on_by_remote or \
self._speed != SPEED_OFF):
return STATE_ON
return SPEED_OFF
@property
def percentage(self):
"""Return speed percentage of the fan."""
if (self._speed == SPEED_OFF):
return 0
return ordered_list_item_to_percentage(self._speed_list, self._speed)
@property
def speed_count(self):
"""Return the number of speeds the fan supports."""
return len(self._speed_list)
@property
def oscillating(self):
"""Return the oscillation state."""
return self._oscillating
@property
def current_direction(self):
"""Return the direction state."""
return self._direction
@property
def last_on_speed(self):
"""Return the last non-idle speed."""
return self._last_on_speed
@property
def supported_features(self):
"""Return the list of supported features."""
return self._support_flags
@property
def extra_state_attributes(self):
"""Platform specific attributes."""
return {
'last_on_speed': self._last_on_speed,
'device_code': self._device_code,
'manufacturer': self._manufacturer,
'supported_models': self._supported_models,
'supported_controller': self._supported_controller,
'commands_encoding': self._commands_encoding,
}
async def async_set_percentage(self, percentage: int):
"""Set the desired speed for the fan."""
if (percentage == 0):
self._speed = SPEED_OFF
else:
self._speed = percentage_to_ordered_list_item(
self._speed_list, percentage)
if not self._speed == SPEED_OFF:
self._last_on_speed = self._speed
await self.send_command()
self.async_write_ha_state()
async def async_oscillate(self, oscillating: bool) -> None:
"""Set oscillation of the fan."""
self._oscillating = oscillating
await self.send_command()
self.async_write_ha_state()
async def async_set_direction(self, direction: str):
"""Set the direction of the fan"""
self._direction = direction
if not self._speed.lower() == SPEED_OFF:
await self.send_command()
self.async_write_ha_state()
async def async_turn_on(self, percentage: int = None, preset_mode: str = None, **kwargs):
"""Turn on the fan."""
if percentage is None:
percentage = ordered_list_item_to_percentage(
self._speed_list, self._last_on_speed or self._speed_list[0])
await self.async_set_percentage(percentage)
async def async_turn_off(self):
"""Turn off the fan."""
await self.async_set_percentage(0)
async def send_command(self):
async with self._temp_lock:
self._on_by_remote = False
speed = self._speed
direction = self._direction or 'default'
oscillating = self._oscillating
if speed.lower() == SPEED_OFF:
command = self._commands['off']
elif oscillating:
command = self._commands['oscillate']
else:
command = self._commands[direction][speed]
try:
await self._controller.send(command)
except Exception as e:
_LOGGER.exception(e)
@callback
async def _async_power_sensor_changed(self, event: Event[EventStateChangedData]) -> None:
"""Handle power sensor changes."""
entity_id = event.data["entity_id"]
old_state = event.data["old_state"]
new_state = event.data["new_state"]
if new_state is None:
return
if new_state.state == old_state.state:
return
if new_state.state == STATE_ON and self._speed == SPEED_OFF:
self._on_by_remote = True
self._speed = None
self.async_write_ha_state()
if new_state.state == STATE_OFF:
self._on_by_remote = False
if self._speed != SPEED_OFF:
self._speed = SPEED_OFF
self.async_write_ha_state()
custom_components/smartir/light.py
+386
View File
@@ -0,0 +1,386 @@
import asyncio
import aiofiles
import json
import logging
import os.path
import voluptuous as vol
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP_KELVIN,
ColorMode,
LightEntity,
PLATFORM_SCHEMA,
)
from homeassistant.const import (
CONF_NAME,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import callback
from homeassistant.helpers.event import async_track_state_change_event
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from . import COMPONENT_ABS_DIR, Helper
from .controller import get_controller
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "SmartIR Light"
DEFAULT_DELAY = 0.5
CONF_UNIQUE_ID = "unique_id"
CONF_DEVICE_CODE = "device_code"
CONF_CONTROLLER_DATA = "controller_data"
CONF_DELAY = "delay"
CONF_POWER_SENSOR = "power_sensor"
CMD_BRIGHTNESS_INCREASE = "brighten"
CMD_BRIGHTNESS_DECREASE = "dim"
CMD_COLORMODE_COLDER = "colder"
CMD_COLORMODE_WARMER = "warmer"
CMD_POWER_ON = "on"
CMD_POWER_OFF = "off"
CMD_NIGHTLIGHT = "night"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_DEVICE_CODE): cv.positive_int,
vol.Required(CONF_CONTROLLER_DATA): cv.string,
vol.Optional(CONF_DELAY, default=DEFAULT_DELAY): cv.string,
vol.Optional(CONF_POWER_SENSOR): cv.entity_id,
}
)
async def async_setup_platform(
hass,
config,
async_add_entities,
discovery_info=None,
):
"""Set up the IR Light platform."""
device_code = config.get(CONF_DEVICE_CODE)
device_files_subdir = os.path.join("codes", "light")
device_files_absdir = os.path.join(COMPONENT_ABS_DIR, device_files_subdir)
if not os.path.isdir(device_files_absdir):
os.makedirs(device_files_absdir)
device_json_filename = str(device_code) + ".json"
device_json_path = os.path.join(device_files_absdir, device_json_filename)
if not os.path.exists(device_json_path):
_LOGGER.warning(
"Couldn't find the device Json file. The component "
"will try to download it from the Github repo."
)
try:
codes_source = (
"https://raw.githubusercontent.com/"
"smartHomeHub/SmartIR/master/"
"codes/light/{}.json"
)
await Helper.downloader(
codes_source.format(device_code),
device_json_path,
)
except Exception:
_LOGGER.error(
"There was an error while downloading the device Json file. "
"Please check your internet connection or if the device code "
"exists on GitHub. If the problem still exists please "
"place the file manually in the proper directory."
)
return
try:
async with aiofiles.open(device_json_path, mode='r') as j:
_LOGGER.debug(f"loading json file {device_json_path}")
content = await j.read()
device_data = json.loads(content)
_LOGGER.debug(f"{device_json_path} file loaded")
except Exception:
_LOGGER.error("The device JSON file is invalid")
return
async_add_entities([SmartIRLight(hass, config, device_data)])
# find the closest match in a sorted list
def closest_match(value, list):
prev_val = None
for index, entry in enumerate(list):
if entry > (value or 0):
if prev_val is None:
return index
diff_lo = value - prev_val
diff_hi = entry - value
if diff_lo < diff_hi:
return index - 1
return index
prev_val = entry
return len(list) - 1
class SmartIRLight(LightEntity, RestoreEntity):
def __init__(self, hass, config, device_data):
self.hass = hass
self._unique_id = config.get(CONF_UNIQUE_ID)
self._name = config.get(CONF_NAME)
self._device_code = config.get(CONF_DEVICE_CODE)
self._controller_data = config.get(CONF_CONTROLLER_DATA)
self._delay = config.get(CONF_DELAY)
self._power_sensor = config.get(CONF_POWER_SENSOR)
self._manufacturer = device_data["manufacturer"]
self._supported_models = device_data["supportedModels"]
self._supported_controller = device_data["supportedController"]
self._commands_encoding = device_data["commandsEncoding"]
self._brightnesses = device_data["brightness"]
self._colortemps = device_data["colorTemperature"]
self._commands = device_data["commands"]
self._power = STATE_ON
self._brightness = None
self._colortemp = None
self._temp_lock = asyncio.Lock()
self._on_by_remote = False
self._support_color_mode = ColorMode.UNKNOWN
if (
CMD_COLORMODE_COLDER in self._commands
and CMD_COLORMODE_WARMER in self._commands
):
self._colortemp = self.max_color_temp_kelvin
self._support_color_mode = ColorMode.COLOR_TEMP
if CMD_NIGHTLIGHT in self._commands or (
CMD_BRIGHTNESS_INCREASE in self._commands
and CMD_BRIGHTNESS_DECREASE in self._commands
):
self._brightness = 100
self._support_brightness = True
if self._support_color_mode == ColorMode.UNKNOWN:
self._support_color_mode = ColorMode.BRIGHTNESS
else:
self._support_brightness = False
if (
CMD_POWER_OFF in self._commands
and CMD_POWER_ON in self._commands
and self._support_color_mode == ColorMode.UNKNOWN
):
self._support_color_mode = ColorMode.ONOFF
# Init the IR/RF controller
self._controller = get_controller(
self.hass,
self._supported_controller,
self._commands_encoding,
self._controller_data,
self._delay,
)
async def async_added_to_hass(self):
"""Run when entity about to be added."""
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if last_state is not None:
self._power = last_state.state
if ATTR_BRIGHTNESS in last_state.attributes:
self._brightness = last_state.attributes[ATTR_BRIGHTNESS]
if ATTR_COLOR_TEMP_KELVIN in last_state.attributes:
self._colortemp = last_state.attributes[ATTR_COLOR_TEMP_KELVIN]
if self._power_sensor:
async_track_state_change_event(
self.hass, self._power_sensor, self._async_power_sensor_changed
)
@property
def unique_id(self):
"""Return a unique ID."""
return self._unique_id
@property
def name(self):
"""Return the display name of the light."""
return self._name
@property
def supported_color_modes(self):
"""Return the list of supported color modes."""
return [self._support_color_mode]
@property
def color_mode(self):
return self._support_color_mode
@property
def color_temp_kelvin(self):
return self._colortemp
@property
def min_color_temp_kelvin(self):
if self._colortemps:
return self._colortemps[0]
@property
def max_color_temp_kelvin(self):
if self._colortemps:
return self._colortemps[-1]
@property
def is_on(self):
return self._power == STATE_ON or self._on_by_remote
@property
def brightness(self):
return self._brightness
@property
def extra_state_attributes(self):
"""Platform specific attributes."""
return {
"device_code": self._device_code,
"manufacturer": self._manufacturer,
"supported_models": self._supported_models,
"supported_controller": self._supported_controller,
"commands_encoding": self._commands_encoding,
"on_by_remote": self._on_by_remote,
}
async def async_turn_on(self, **params):
did_something = False
# Turn the light on if off
if self._power != STATE_ON and not self._on_by_remote:
self._power = STATE_ON
did_something = True
await self.send_command(CMD_POWER_ON)
if (
ATTR_COLOR_TEMP_KELVIN in params
and ColorMode.COLOR_TEMP == self._support_color_mode
):
target = params.get(ATTR_COLOR_TEMP_KELVIN)
old_color_temp = closest_match(self._colortemp, self._colortemps)
new_color_temp = closest_match(target, self._colortemps)
_LOGGER.debug(
f"Changing color temp from {self._colortemp}K step {old_color_temp} to {target}K step {new_color_temp}"
)
steps = new_color_temp - old_color_temp
did_something = True
if steps < 0:
cmd = CMD_COLORMODE_WARMER
steps = abs(steps)
else:
cmd = CMD_COLORMODE_COLDER
if steps > 0 and cmd:
# If we are heading for the highest or lowest value,
# take the opportunity to resync by issuing enough
# commands to go the full range.
if new_color_temp == len(self._colortemps) - 1 or new_color_temp == 0:
steps = len(self._colortemps)
self._colortemp = self._colortemps[new_color_temp]
await self.send_command(cmd, steps)
if ATTR_BRIGHTNESS in params and self._support_brightness:
# before checking the supported brightnesses, make a special case
# when a nightlight is fitted for brightness of 1
if params.get(ATTR_BRIGHTNESS) == 1 and CMD_NIGHTLIGHT in self._commands:
self._brightness = 1
self._power = STATE_ON
did_something = True
await self.send_command(CMD_NIGHTLIGHT)
elif self._brightnesses:
target = params.get(ATTR_BRIGHTNESS)
old_brightness = closest_match(self._brightness, self._brightnesses)
new_brightness = closest_match(target, self._brightnesses)
did_something = True
_LOGGER.debug(
f"Changing brightness from {self._brightness} step {old_brightness} to {target} step {new_brightness}"
)
steps = new_brightness - old_brightness
if steps < 0:
cmd = CMD_BRIGHTNESS_DECREASE
steps = abs(steps)
else:
cmd = CMD_BRIGHTNESS_INCREASE
if steps > 0 and cmd:
# If we are heading for the highest or lowest value,
# take the opportunity to resync by issuing enough
# commands to go the full range.
if (
new_brightness == len(self._brightnesses) - 1
or new_brightness == 0
):
steps = len(self._colortemps)
did_something = True
self._brightness = self._brightnesses[new_brightness]
await self.send_command(cmd, steps)
# If we did nothing above, and the light is not detected as on
# already issue the on command, even though we think the light
# is on. This is because we may be out of sync due to use of the
# remote when we don't have anything to detect it.
# If we do have such monitoring, avoid issuing the command in case
# on and off are the same remote code.
if not did_something and not self._on_by_remote:
self._power = STATE_ON
await self.send_command(CMD_POWER_ON)
self.async_write_ha_state()
async def async_turn_off(self):
self._power = STATE_OFF
await self.send_command(CMD_POWER_OFF)
self.async_write_ha_state()
async def async_toggle(self):
await (self.async_turn_on() if not self.is_on else self.async_turn_off())
async def send_command(self, cmd, count=1):
if cmd not in self._commands:
_LOGGER.error(f"Unknown command '{cmd}'")
return
_LOGGER.debug(f"Sending {cmd} remote command {count} times.")
remote_cmd = self._commands.get(cmd)
async with self._temp_lock:
self._on_by_remote = False
try:
for _ in range(count):
await self._controller.send(remote_cmd)
except Exception as e:
_LOGGER.exception(e)
@callback
async def _async_power_sensor_changed(self, event):
"""Handle power sensor changes."""
new_state = event.data["new_state"]
if new_state is None:
return
old_state = event.data["old_state"]
if new_state.state == old_state.state:
return
if new_state.state == STATE_ON:
self._on_by_remote = True
self.async_write_ha_state()
if new_state.state == STATE_OFF:
self._on_by_remote = False
self._power = STATE_OFF
self.async_write_ha_state()
custom_components/smartir/manifest.json
+24
View File
@@ -0,0 +1,24 @@
{
"domain": "smartir",
"name": "SmartIR",
"documentation": "https://github.com/smartHomeHub/SmartIR",
"dependencies": [],
"codeowners": ["@smartHomeHub"],
"requirements": ["aiofiles>=0.6.0"],
"homeassistant": "2025.5.0",
"version": "1.18.1",
"updater": {
"version": "1.18.1",
"releaseNotes": "-- Implements new async_track_state_change_event",
"files": [
"__init__.py",
"climate.py",
"media_player.py",
"fan.py",
"light.py",
"controller.py",
"manifest.json",
"services.yaml"
]
}
}
custom_components/smartir/media_player.py
+306
View File
@@ -0,0 +1,306 @@
import asyncio
import aiofiles
import json
import logging
import os.path
import voluptuous as vol
from homeassistant.components.media_player import (
MediaPlayerEntity, PLATFORM_SCHEMA)
from homeassistant.components.media_player.const import (
MediaPlayerEntityFeature, MediaType)
from homeassistant.const import (
CONF_NAME, STATE_OFF, STATE_ON, STATE_UNKNOWN)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from . import COMPONENT_ABS_DIR, Helper
from .controller import get_controller
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "SmartIR Media Player"
DEFAULT_DEVICE_CLASS = "tv"
DEFAULT_DELAY = 0.5
CONF_UNIQUE_ID = 'unique_id'
CONF_DEVICE_CODE = 'device_code'
CONF_CONTROLLER_DATA = "controller_data"
CONF_DELAY = "delay"
CONF_POWER_SENSOR = 'power_sensor'
CONF_SOURCE_NAMES = 'source_names'
CONF_DEVICE_CLASS = 'device_class'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_DEVICE_CODE): cv.positive_int,
vol.Required(CONF_CONTROLLER_DATA): cv.string,
vol.Optional(CONF_DELAY, default=DEFAULT_DELAY): cv.string,
vol.Optional(CONF_POWER_SENSOR): cv.entity_id,
vol.Optional(CONF_SOURCE_NAMES): dict,
vol.Optional(CONF_DEVICE_CLASS, default=DEFAULT_DEVICE_CLASS): cv.string
})
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the IR Media Player platform."""
device_code = config.get(CONF_DEVICE_CODE)
device_files_subdir = os.path.join('codes', 'media_player')
device_files_absdir = os.path.join(COMPONENT_ABS_DIR, device_files_subdir)
if not os.path.isdir(device_files_absdir):
os.makedirs(device_files_absdir)
device_json_filename = str(device_code) + '.json'
device_json_path = os.path.join(device_files_absdir, device_json_filename)
if not os.path.exists(device_json_path):
_LOGGER.warning("Couldn't find the device Json file. The component will " \
"try to download it from the GitHub repo.")
try:
codes_source = ("https://raw.githubusercontent.com/"
"smartHomeHub/SmartIR/master/"
"codes/media_player/{}.json")
await Helper.downloader(codes_source.format(device_code), device_json_path)
except Exception:
_LOGGER.error("There was an error while downloading the device Json file. " \
"Please check your internet connection or if the device code " \
"exists on GitHub. If the problem still exists please " \
"place the file manually in the proper directory.")
return
try:
async with aiofiles.open(device_json_path, mode='r') as j:
_LOGGER.debug(f"loading json file {device_json_path}")
content = await j.read()
device_data = json.loads(content)
_LOGGER.debug(f"{device_json_path} file loaded")
except Exception:
_LOGGER.error("The device JSON file is invalid")
return
async_add_entities([SmartIRMediaPlayer(
hass, config, device_data
)])
class SmartIRMediaPlayer(MediaPlayerEntity, RestoreEntity):
def __init__(self, hass, config, device_data):
self.hass = hass
self._unique_id = config.get(CONF_UNIQUE_ID)
self._name = config.get(CONF_NAME)
self._device_code = config.get(CONF_DEVICE_CODE)
self._controller_data = config.get(CONF_CONTROLLER_DATA)
self._delay = config.get(CONF_DELAY)
self._power_sensor = config.get(CONF_POWER_SENSOR)
self._manufacturer = device_data['manufacturer']
self._supported_models = device_data['supportedModels']
self._supported_controller = device_data['supportedController']
self._commands_encoding = device_data['commandsEncoding']
self._commands = device_data['commands']
self._state = STATE_OFF
self._sources_list = []
self._source = None
self._support_flags = 0
self._device_class = config.get(CONF_DEVICE_CLASS)
#Supported features
if 'off' in self._commands and self._commands['off'] is not None:
self._support_flags = self._support_flags | MediaPlayerEntityFeature.TURN_OFF
if 'on' in self._commands and self._commands['on'] is not None:
self._support_flags = self._support_flags | MediaPlayerEntityFeature.TURN_ON
if 'previousChannel' in self._commands and self._commands['previousChannel'] is not None:
self._support_flags = self._support_flags | MediaPlayerEntityFeature.PREVIOUS_TRACK
if 'nextChannel' in self._commands and self._commands['nextChannel'] is not None:
self._support_flags = self._support_flags | MediaPlayerEntityFeature.NEXT_TRACK
if ('volumeDown' in self._commands and self._commands['volumeDown'] is not None) \
or ('volumeUp' in self._commands and self._commands['volumeUp'] is not None):
self._support_flags = self._support_flags | MediaPlayerEntityFeature.VOLUME_STEP
if 'mute' in self._commands and self._commands['mute'] is not None:
self._support_flags = self._support_flags | MediaPlayerEntityFeature.VOLUME_MUTE
if 'sources' in self._commands and self._commands['sources'] is not None:
self._support_flags = self._support_flags | MediaPlayerEntityFeature.SELECT_SOURCE | MediaPlayerEntityFeature.PLAY_MEDIA
for source, new_name in config.get(CONF_SOURCE_NAMES, {}).items():
if source in self._commands['sources']:
if new_name is not None:
self._commands['sources'][new_name] = self._commands['sources'][source]
del self._commands['sources'][source]
#Sources list
for key in self._commands['sources']:
self._sources_list.append(key)
self._temp_lock = asyncio.Lock()
#Init the IR/RF controller
self._controller = get_controller(
self.hass,
self._supported_controller,
self._commands_encoding,
self._controller_data,
self._delay)
async def async_added_to_hass(self):
"""Run when entity about to be added."""
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if last_state is not None:
self._state = last_state.state
@property
def should_poll(self):
"""Push an update after each command."""
return True
@property
def unique_id(self):
"""Return a unique ID."""
return self._unique_id
@property
def name(self):
"""Return the name of the media player."""
return self._name
@property
def device_class(self):
"""Return the device_class of the media player."""
return self._device_class
@property
def state(self):
"""Return the state of the player."""
return self._state
@property
def media_title(self):
"""Return the title of current playing media."""
return None
@property
def media_content_type(self):
"""Content type of current playing media."""
return MediaType.CHANNEL
@property
def source_list(self):
return self._sources_list
@property
def source(self):
return self._source
@property
def supported_features(self):
"""Flag media player features that are supported."""
return self._support_flags
@property
def extra_state_attributes(self):
"""Platform specific attributes."""
return {
'device_code': self._device_code,
'manufacturer': self._manufacturer,
'supported_models': self._supported_models,
'supported_controller': self._supported_controller,
'commands_encoding': self._commands_encoding,
}
async def async_turn_off(self):
"""Turn the media player off."""
await self.send_command(self._commands['off'])
if self._power_sensor is None:
self._state = STATE_OFF
self._source = None
self.async_write_ha_state()
async def async_turn_on(self):
"""Turn the media player off."""
await self.send_command(self._commands['on'])
if self._power_sensor is None:
self._state = STATE_ON
self.async_write_ha_state()
async def async_media_previous_track(self):
"""Send previous track command."""
await self.send_command(self._commands['previousChannel'])
self.async_write_ha_state()
async def async_media_next_track(self):
"""Send next track command."""
await self.send_command(self._commands['nextChannel'])
self.async_write_ha_state()
async def async_volume_down(self):
"""Turn volume down for media player."""
await self.send_command(self._commands['volumeDown'])
self.async_write_ha_state()
async def async_volume_up(self):
"""Turn volume up for media player."""
await self.send_command(self._commands['volumeUp'])
self.async_write_ha_state()
async def async_mute_volume(self, mute):
"""Mute the volume."""
await self.send_command(self._commands['mute'])
self.async_write_ha_state()
async def async_select_source(self, source):
"""Select channel from source."""
self._source = source
await self.send_command(self._commands['sources'][source])
self.async_write_ha_state()
async def async_play_media(self, media_type, media_id, **kwargs):
"""Support channel change through play_media service."""
if self._state == STATE_OFF:
await self.async_turn_on()
if media_type != MediaType.CHANNEL:
_LOGGER.error("invalid media type")
return
if not media_id.isdigit():
_LOGGER.error("media_id must be a channel number")
return
self._source = "Channel {}".format(media_id)
for digit in media_id:
await self.send_command(self._commands['sources']["Channel {}".format(digit)])
self.async_write_ha_state()
async def send_command(self, command):
async with self._temp_lock:
try:
await self._controller.send(command)
except Exception as e:
_LOGGER.exception(e)
async def async_update(self):
if self._power_sensor is None:
return
power_state = self.hass.states.get(self._power_sensor)
if power_state:
if power_state.state == STATE_OFF:
self._state = STATE_OFF
self._source = None
elif power_state.state == STATE_ON:
self._state = STATE_ON
custom_components/smartir/services.yaml
+4
View File
@@ -0,0 +1,4 @@
check_updates:
description: Check for SmartIR updates.
update_component:
description: Update SmartIR component.