Documentation / implementation of Daikin's IR protocol in Python

daikin.py

'''
This module implements the IR protocol for Daikin's air conditioning units.

It is based on this work:
<https://github.com/blafois/Daikin-IR-Reverse>
And contains some additional improvements, such as:

- Implementation of the low layer (IR signal protocol).
  The low-level IR protocol is not explained very well by blafois;
  no mention is made to the following:
  - the pre-frame burst (3.5ms HIGH, 1.7ms LOW)
  - the post-frame HIGH pulse
  - the 35ms inter-frame spacing LOW time
  - the pre-transmission 6-bit burst + 25ms spacing

- Some typos in the encoding of the two timer values.

- Some additional fields and frame types that aren't documented
  by blafois, but are documented by other people in the repo's issues.

Most users will want to call `encode_remote_control_transmission()`
and maybe follow up with `frames_to_signal()` if an IR signal is needed.
The high-level API to decode a frame is `decode_frame_payload()`.
'''

from enum import IntEnum, unique
from typing import ClassVar, Type, get_origin
from dataclasses import dataclass, field, fields
from functools import reduce
import datetime


# LAYER 1
# -------

# this layer handles encoding of the frame bits into a binary IR signal (carrier amplitude).
# here, IR signals are represented as a list of µs durations, where the first duration
# corresponds to a high state, the next one to a low state, etc.

# time constants (µs)

TIME_BURST_HIGH = 3500
TIME_BURST_LOW = 1700
TIME_SPACING_FIRST = 25000
TIME_SPACING_INTER = 35000

TIME_HIGH = 450
TIME_LOW_ZERO = 420
TIME_LOW_ONE = 1286

# encoding into signal

bit_to_signal = lambda bit: \
	[TIME_HIGH, TIME_LOW_ONE if bit else TIME_LOW_ZERO]

def frame_to_signal(data: bytes) -> list[int]:
	'''
	Encodes a single frame to an IR signal.
	this includes the frame starting burst and the final HIGH,
	but no spacing.
	'''
	bits = [bool((byte >> i) & 1) for byte in data for i in range(8)]
	return \
		[TIME_BURST_HIGH, TIME_BURST_LOW] + \
		sum(map(bit_to_signal, bits), []) + \
		[TIME_HIGH]

def frames_to_signal(frames: list[bytes]) -> list[int]:
	'''
	Encodes a series of frames for transmission,
	inserting spacing and the pre-transmission burst.
	'''
	frames = map(frame_to_signal, frames)
	return \
		(bit_to_signal(0) * 6)[:-1] + \
		[TIME_SPACING_FIRST] + \
		reduce(lambda a, b: a + [TIME_SPACING_INTER] + b, frames)


# LAYER 2
# -------

# this layer encodes a code + message into a full frame, complete with header and checksum

HEADER = b'\x11\xDA\x27\x00'

calc_checksum = lambda data: sum(data) & 0xFF

def encode_frame(code: 'FrameCode', payload: bytes) -> bytes:
	'''
	Encodes code + payload into a full frame.
	'''
	data = HEADER + bytes([code]) + payload
	return data + bytes([calc_checksum(data)])

def verify_checksum(frame: bytes) -> bytes:
	'''
	Verifies that the checksum is correct, raising otherwise.
	Returns the frame without the trailing checksum.
	'''
	data, checksum = frame[:-1], frame[-1]
	expected_checksum = calc_checksum(data)
	if checksum != expected_checksum:
		raise ValueError(f'incorrect checksum (expected {expected_checksum:02X}, got {checksum:02X})')
	return data

def decode_frame(frame: bytes) -> tuple['FrameCode', bytes]:
	'''
	Decodes a frame into code + payload, verifying header and checksum.
	'''
	if frame[:4] != HEADER:
		raise ValueError(f'invalid header found: {frame[:4].hex()}')
	data = verify_checksum(frame)[4:]
	return FrameCode.get(data[0]), data[1:]


# UPPER LAYER
# -----------

# this is the higher layer that encodes meaningful data into frame payloads

@unique
class FrameCode(IntEnum):
	''' Type of frame. See the associated class for details. '''

	SETTINGS = 0x00
	DIAGNOSTIC = 0x01

	TIME = 0x42
	COMFORT = 0xC5

	@classmethod
	def get(cls, x: int) -> 'FrameCode':
		try:
			return cls(x)
		except ValueError:
			return x

NOTICE = 'if this is legitimate, pass strict=False to disable this check, and open a bug report'

class FrameBase:
	'''
	Base class for a class associated with a frame type.

	## Encode / decode API

	Such classes have two special class properties:
	`_code` (the FrameCode associated with this kind of
	payload) and `_layout` (the bit layout of the payload).

	The layout is specified as a list of `(width, field_name)`
	tuples, where `width` is the width in bits of the field,
	and `field_name` is the name of the class property that
	the value is stored at. The tuples are specified in the
	order in which they appear at the frame (which is allowed
	to be different from the order in which the fields are
	defined in the class).

	Using these special properties, `self.encode()` and
	`cls.decode(frame)` can be used to [de]serialize a frame
	payload into a class instance.

	## Special fields

	There is a one-to-one mapping between payloads and instances,
	so every bit in the payload is represented by some property.

	For bits that have no known use and are expected to be always
	zero, we give them a `_zeroP_W` name (P being the bit
	position of the field and W being its width in bits). These
	are keyword only parameters, default to zero and are NOT part
	of the stable API, as they may be removed or renamed at any point.
	`decode()` has a `strict` option (enabled by default) that raises
	whenever some non-zero value is found in these fields. The fields
	are declared last, to reduce pollution in repr().

	For bits that have been seen set not always set to zero, but
	whose meaning isn't well known, we give them `_unknownP` names.
	'''
	_code: ClassVar[FrameCode]
	_layout: ClassVar[list[tuple[int, str]]]

	def encode(self, full_frame=True) -> bytes:
		''' Encode into a frame (or just the payload, if full_frame=False). '''
		bits, n = 0, 0
		for w, name in type(self)._layout:
			value = int(getattr(self, name))
			assert not (value >> w), f'field {name}={value} is out of range'
			bits |= value << n
			n += w
		assert n % 8 == 0
		frame = bits.to_bytes(n // 8, 'little')
		if full_frame:
			frame = encode_frame(self._code, frame)
		return frame

	@classmethod
	def decode(cls, payload: bytes, strict=True):
		''' Decode a frame's payload into an instance of this class. '''
		size = sum(w for w, _ in cls._layout)
		assert len(payload) * 8 == size, \
			f'invalid payload length (expected {size//8} bytes, got {len(payload)})'
		bits = int.from_bytes(payload, 'little')
		fields = {}
		for w, name in cls._layout:
			value, bits = bits & ~((~0) << w), bits >> w
			if strict and name.startswith('_zero') and value:
				raise ValueError(f'field {name} was supposed to be zero, but it is {value}. {NOTICE}')
			ann = cls.__annotations__[name]
			ann = get_origin(ann) or ann
			try:
				value = ann(value)
			except ValueError as err:
				if strict:
					raise ValueError(f'field {name} has invalid value. {NOTICE}') from err
			fields[name] = value
		return cls(**fields)

	def __repr__(self):
		params = [ f.name for f in fields(self) ]
		params = [ k for k in params if not (k.startswith('_zero') and getattr(self, k) == 0) ]
		params = ', '.join(f'{k}={repr(getattr(self, k))}' for k in params)
		return f'{type(self).__name__}({params})'

@unique
class Mode(IntEnum):
	''' Operating mode of the AC unit. '''
	AUTO = 0
	''' Automatic '''
	# value 1 not observed
	DRY = 2
	''' Dehumidifier '''
	COLD = 3
	''' Cooler '''
	HEAT = 4
	''' Heater '''
	# value 5 not observed
	FAN = 6
	''' Fan only '''

@unique
class FanSpeed(IntEnum):
	''' Fan speed setting. '''

	SPEED_1 = 3
	''' Manual speed 1 of 5 '''
	SPEED_2 = 4
	''' Manual speed 2 of 5 '''
	SPEED_3 = 5
	''' Manual speed 3 of 5 '''
	SPEED_4 = 6
	''' Manual speed 4 of 5 '''
	SPEED_5 = 7
	''' Manual speed 5 of 5 '''

	AUTO = 0xA
	''' Automatic '''
	SILENT = 0xB
	''' Silent '''

@dataclass(repr=False)
class SettingsFrame(FrameBase):
	''' Main frame, containing most of the AC settings. '''
	_code = FrameCode.SETTINGS
	_layout = [
		(1, 'power'),
		(1, 'timer_on'),
		(1, 'timer_off'),
		(1, '_unknown3'),
		(4, 'mode'),

		(8, 'temperature'),
		(8, '_zero16_8'),
		(4, 'swing_vertical'),
		(4, 'fan_speed'),
		(4, 'swing_horizontal'),
		(4, '_zero36_4'),

		(12, 'timer_on_duration'),
		(12, 'timer_off_duration'),

		(1, 'powerful_mode'),
		(4, '_zero65_4'),
		(1, 'silent_mode'),
		(10, '_zero70_10'),

		(1, '_unknown80'),
		(3, '_zero81_3'),
		(4, '_unknown84'),

		(1, '_zero88_1'),
		(1, 'eco_sensing'),
		(1, 'economy_mode'),
		(4, '_zero91_4'),
		(1, '_unknown95'),

		(8, '_zero96_8'),
	]

	power: bool
	''' Whether the unit is ON or OFF. '''
	mode: Mode = Mode.AUTO

	temperature: int = 0x32
	'''
	Temperature, in units of HALF a Celsius degree.
	For example, value 47 indicates 23.5°C.
	If the remote is in Farenheit mode, it converts the temperature to Celsius and rounds.
	Note: in DRY and FAN modes, the temperature transmitted is 25°C (it is not relevant).
	'''
	fan_speed: FanSpeed = FanSpeed.AUTO
	''' Fan speed '''
	swing_vertical: int = 0
	''' Fan vertical swing: 0 if disabled, 0xF if enabled '''
	swing_horizontal: int = 0
	''' Fan horizontal swing: 0 if disabled, 0xF if enabled '''
	powerful_mode: bool = False
	'''
	Whether powerful mode is enabled or not.
	If enabled, the unit will stay in maximum power for 20 minutes and then
	revert to the settings transmitted in the rest of the frame.
	'''
	silent_mode: bool = False
	''' Silent outdoor unit feature '''
	economy_mode: bool = False
	''' Whether economy mode is enabled or not. '''
	eco_sensing: bool = False
	''' Whether eco sensing is enabled or not (seems to be mutually exclusive with economy mode). '''

	timer_on: bool = False
	''' Whether the ON timer is enabled or not. '''
	timer_off: bool = False
	''' Whether the OFF timer is enabled or not. '''
	timer_on_duration: int = 0x600
	''' Duration for the ON timer, in minutes. When the timer is not enabled, value 0x600 (1536) is observed here. '''
	timer_off_duration: int = 0x600
	''' Duration for the OFF timer, in minutes. When the timer is not enabled, value 0x600 (1536) is observed here. '''

	_unknown3: bool = field(kw_only=True, default=True)
	''' always seems to be 1 '''
	_unknown80: bool = field(kw_only=True, default=True)
	''' 1 in the repo's example, 0 in our model '''
	_unknown84: int = field(kw_only=True, default=12)
	''' 12 in the repo's example (4 bits) '''
	_unknown95: bool = field(kw_only=True, default=True)
	''' 1 in the repo's example, 0 in our model '''

	_zero16_8: int = field(kw_only=True, default=0)
	_zero36_4: int = field(kw_only=True, default=0)
	_zero65_4: int = field(kw_only=True, default=0)
	_zero70_10: int = field(kw_only=True, default=0)
	_zero81_3: int = field(kw_only=True, default=0)
	_zero88_1: int = field(kw_only=True, default=0)
	_zero91_4: int = field(kw_only=True, default=0)
	_zero96_8: int = field(kw_only=True, default=0)

@dataclass(repr=False)
class DiagnosticFrame(FrameBase):
	'''
	Daikin air cons has a diagnostic feature where remote sends out all codes but the AC unit long beeps when the diagnostic code matches.
	See <https://www.daikin.com/products/ac/services/error_codes> for the feature.
	'''
	_code = FrameCode.DIAGNOSTIC
	_layout = [
		(8, 'code'),
		(8, '_zero8_8'),
	]

	code: int
	''' 8-bit value encoding the error code, see `error_code_to_string` and `string_to_error_code` '''

	_zero8_8: int = field(kw_only=True, default=0)

@dataclass(repr=False)
class TimeFrame(FrameBase):
	''' Frame containing the remote's current time. '''
	_code = FrameCode.TIME
	_layout = [
		(12, 'time'),
		(4, '_zero12_4'),
	]

	time: int = 0
	'''
	The time of the day, in minutes since 00:00.
	This was originally documented as being always zero.
	See `time_to_minutes` and `minutes_to_time` for conversion and examples.
	'''

	_zero12_4: int = field(kw_only=True, default=0)

@dataclass(repr=False)
class ComfortFrame(FrameBase):
	''' Frame containing the comfort mode settings. '''
	_code = FrameCode.COMFORT
	_layout = [
		(12, '_zero0_12'),
		(1, 'comfort_mode'),
		(3, '_zero13_3'),
	]

	comfort_mode: bool = False
	''' Whether comfort mode is enabled or not. '''

	_zero0_12: int = field(kw_only=True, default=0)
	_zero13_3: int = field(kw_only=True, default=0)


# CONVENIENCE API

frame_classes = [SettingsFrame, DiagnosticFrame, TimeFrame, ComfortFrame]
frame_classes: dict[FrameCode, Type[FrameBase]] = \
	{ cls._code: cls for cls in frame_classes }

def decode_frame_payload(frame: bytes, strict=True) -> FrameBase:
	'''
	Convenience function to decode a frame into an instance of the appropriate class.
	Unlike `decode_frame`, this function will fail if the code is not implemented.
	'''
	code, data = decode_frame(frame)
	return frame_classes[code].decode(data, strict=strict)

def encode_remote_control_transmission(
	power: bool,
	mode: int = 0,
	temperature: int = 0x30,
	fan_speed: int = 0xA,
	swing_vertical: int = 0,
	swing_horizontal: int = 0,
	powerful_mode: bool = False,
	silent_mode: bool = False,
	economy_mode: bool = False,
	eco_sensing: bool = False,
	comfort_mode: bool = False,
	time: int = 0,
	timer_on: bool = False,
	timer_off: bool = False,
	timer_on_duration: int = 0x600,
	timer_off_duration: int = 0x600,
	*,
	_unknown3: bool = True,
	_unknown80: bool = True,
	_unknown84: int = 12,
	_unknown95: bool = True,
) -> list[bytes]:
	'''
	Encode a full transmission by a remote control. When a button is
	pressed on the remote control, multiple frames are transferred at
	once. This method is a convenience function so that the frames don't
	have to be formatted separately. See the corresponding classes for
	more documentation on the parameters.
	'''
	return [
		ComfortFrame(comfort_mode=comfort_mode).encode(),
		TimeFrame(time=time).encode(),
		SettingsFrame(
			power=power,
			mode=mode,
			temperature=temperature,
			fan_speed=fan_speed,
			swing_vertical=swing_vertical,
			swing_horizontal=swing_horizontal,
			powerful_mode=powerful_mode,
			silent_mode=silent_mode,
			economy_mode=economy_mode,
			eco_sensing=eco_sensing,
			timer_on=timer_on,
			timer_off=timer_off,
			timer_on_duration=timer_on_duration,
			timer_off_duration=timer_off_duration,
			_unknown3=_unknown3,
			_unknown80=_unknown80,
			_unknown84=_unknown84,
			_unknown95=_unknown95,
		).encode(),
	]

def time_to_minutes(time: datetime.time) -> int:
	''' Convenience function to convert a time of the day into a "minute count" for transmission. '''
	return time.hour * 60 + time.minute

def minutes_to_time(time: int) -> datetime.time:
	'''
	Convenience function to convert from a "minute count" to a standard time object.
	For example, value 1095 corresponds to time 18:15.
	'''
	return datetime.time(*divmod(time, 60))

CODE_1_CHARS = "0ACEHFJLPU987654"
CODE_2_CHARS = "0123456789AHCJEF"

def error_code_to_string(code: int) -> str:
	''' Decodes an 8-bit error code into a readable string. '''
	return CODE_1_CHARS[code >> 4] + CODE_2_CHARS[code & 0b1111]

def string_to_error_code(code: str) -> int:
	''' Encodes an error code string into its 8-bit value for transmission. '''
	assert len(code) == 2
	return CODE_1_CHARS.index(code[0]) << 4 | CODE_2_CHARS.index(code[1])