Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 37 additions & 23 deletions infrared_protocols/commands/kaseikyo.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,21 +10,27 @@ class KaseikyoCommand(Command):
"""Kaseikyo format IR command."""

address: int
data: bytes
data: bytes | list[bytes]
error_correction: Callable[[bytes], bytes] | None
base_unit: float

def __init__(
self,
*,
address: int,
data: bytes,
data: bytes | list[bytes],
error_correction: Callable[[bytes], bytes] | None = None,
modulation: int = 38000,
burst_pulse: int = 16,
repeat_count: int = 0,
) -> None:
"""Initialize the Kaseikyo IR command."""
"""Initialize the Kaseikyo IR command.

``data`` is the payload following the 16-bit address. Pass a single
``bytes`` for a single-frame command, or a ``list[bytes]`` to send
several frames (each sharing the same address) as one multi-frame
message separated by an inter-frame gap.
"""
super().__init__(modulation=modulation, repeat_count=repeat_count)
self.address = address
self.data = data
Expand Down Expand Up @@ -55,35 +61,43 @@ def get_raw_timings(self) -> list[int]:
one_low = round(3 * self.base_unit)
frame_time = 130000
trailer_min = 8000
multi_frame_gap = 10000
repeat_frame_gap = max(
frame_time - (leader_high + repeat_low + bit_high), trailer_min
)

timings = [leader_high, -leader_low]

parity = self.address & 0xFFFF
parity ^= parity >> 8
parity ^= parity >> 4
parity &= 0x0F

data_bytes = [
self.address & 0xFF,
(self.address >> 8) & 0xFF,
(self.data[0] & 0xF0) | parity,
*self.data[1:],
]
if self.error_correction:
data_bytes.extend(self.error_correction(bytes(data_bytes)))

for byte in data_bytes:
for _ in range(8):
bit = byte & 1
timings.append(bit_high)
timings.append(-one_low if bit else -zero_low)
byte >>= 1

# End pulse
timings.append(bit_high)
frames = [self.data] if isinstance(self.data, bytes) else self.data

timings: list[int] = []
for i, frame in enumerate(frames):
if i != 0:
# Inter-frame gap for multi-frame commands
timings.append(-multi_frame_gap)
timings.extend([leader_high, -leader_low])

data_bytes = [
self.address & 0xFF,
(self.address >> 8) & 0xFF,
(frame[0] & 0xF0) | parity,
*frame[1:],
]
if self.error_correction:
data_bytes.extend(self.error_correction(bytes(data_bytes)))

for byte in data_bytes:
for _ in range(8):
bit = byte & 1
timings.append(bit_high)
timings.append(-one_low if bit else -zero_low)
byte >>= 1

# End pulse
timings.append(bit_high)

# Add repeat codes if requested
gap = max(frame_time - sum(abs(t) for t in timings), trailer_min)
Expand Down
186 changes: 186 additions & 0 deletions infrared_protocols/commands/panasonic_ac.py
Comment thread
sam0737 marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -0,0 +1,186 @@
"""Panasonic air-conditioner IR protocol.

Panasonic A/C remotes use the Kaseikyo (AEHA) format with the Panasonic vendor
address ``0x2002``. The state is sent as two Kaseikyo frames (a fixed 8-byte
preamble frame followed by a 19-byte payload frame) separated by an inter-frame
gap, so this module builds the state bytes and delegates the physical-layer
encoding to :class:`~infrared_protocols.commands.kaseikyo.KaseikyoCommand`.

The encoder is intentionally generic rather than tied to one regional model: it
exposes the full field set (power, mode, temperature, fan, two swing axes and
nanoeX) so callers can drive whichever combination their unit supports.
"""

from enum import Enum, IntEnum

from .kaseikyo import KaseikyoCommand

PANASONIC_AC_ADDRESS = 0x2002

MIN_TEMP = 16
MAX_TEMP = 30

# Fixed "magic" bytes that frame the state; they must be emitted verbatim.
_FRAME1 = [0x02, 0x20, 0xE0, 0x04, 0x00, 0x00, 0x00, 0x06]
_FRAME2_MAGIC = [0x02, 0x20, 0xE0, 0x04]

_NANOEX_MASK = 0x04
_FEATURE_BASE = 0x02
_SHORT_FRAME_MARKER = 0x80


class PanasonicAcMode(IntEnum):
"""Operation mode, stored in the high nibble of byte 13."""

AUTO = 0x0
DRY = 0x2
COOL = 0x3
HEAT = 0x4


class PanasonicAcFanSpeed(IntEnum):
"""Fan speed, stored in the high nibble of byte 16."""

AUTO = 0xA
LOW = 0x3
MEDIUM_LOW = 0x4
MEDIUM = 0x5
MEDIUM_HIGH = 0x6
HIGH = 0x7


class PanasonicAcSwingAxis1(IntEnum):
"""Swing positions for protocol slot 1 (low nibble of byte 16).

The physical louver this drives depends on the unit: on window units it is
the (single) horizontal louver, while on split units it is the vertical
louver. The position names follow the IRremoteESP8266 vertical-swing labels
and are axis-relative, so the caller maps them to the real direction.
"""

AUTO = 0xF
HIGHEST = 0x1
HIGH = 0x2
MIDDLE = 0x3
LOW = 0x4
LOWEST = 0x5


class PanasonicAcSwingAxis2(IntEnum):
"""Swing positions for protocol slot 2 (low nibble of byte 17).

On split units this is the horizontal louver; many units (e.g. single-louver
window units) leave it at :attr:`AUTO`. The position names follow the
IRremoteESP8266 horizontal-swing labels and are axis-relative.
"""

AUTO = 0xD
MIDDLE = 0x6
FULL_LEFT = 0x9
LEFT = 0xA
RIGHT = 0xB
FULL_RIGHT = 0xC


class PanasonicAcToggle(Enum):
"""Short-frame toggle command, holding its two payload bytes (13, 14)."""

QUIET = (0x81, 0x33)
POWERFUL = (0x86, 0x35)


def _checksum(state: list[int], start: int, end: int) -> int:
"""Sum bytes ``state[start..end]`` (inclusive) modulo 256."""
return sum(state[start : end + 1]) & 0xFF


def _to_frames(state: list[int]) -> list[bytes]:
"""Split a full state byte list into Kaseikyo per-frame payloads.

Each frame's first two bytes are the Kaseikyo address (``0x2002``), so the
payload passed to :class:`KaseikyoCommand` is the state with those two
address bytes dropped from each section.
"""
return [bytes(state[2:8]), bytes(state[10:])]


class PanasonicAcCommand(KaseikyoCommand):
"""Panasonic air-conditioner full-state IR command."""

def __init__(
self,
*,
mode: PanasonicAcMode,
temperature: float,
fan: PanasonicAcFanSpeed = PanasonicAcFanSpeed.AUTO,
power: bool = True,
swing_axis1: PanasonicAcSwingAxis1 = PanasonicAcSwingAxis1.AUTO,
swing_axis2: PanasonicAcSwingAxis2 = PanasonicAcSwingAxis2.AUTO,
nanoex: bool = False,
modulation: int = 38000,
) -> None:
"""Build a full Panasonic A/C state command.

``temperature`` is in degrees Celsius and is stored as ``round(°C × 2)``
in byte 14, preserving the protocol's 0.5 °C step. It must be within
:data:`MIN_TEMP`..:data:`MAX_TEMP`.
"""
if not MIN_TEMP <= temperature <= MAX_TEMP:
raise ValueError(
f"temperature {temperature} out of range {MIN_TEMP}..{MAX_TEMP}"
)

state = [
*_FRAME1,
*_FRAME2_MAGIC,
0x00,
(mode << 4) | (0x01 if power else 0x00),
round(temperature * 2),
0x80,
(fan << 4) | swing_axis1,
swing_axis2,
0x00,
0x0E,
0xE0,
0x00,
0x00,
0x81,
0x00,
_FEATURE_BASE | (_NANOEX_MASK if nanoex else 0x00),
]
state.append(_checksum(state, 8, 25))

super().__init__(
address=PANASONIC_AC_ADDRESS,
data=_to_frames(state),
modulation=modulation,
)


class PanasonicAcToggleCommand(KaseikyoCommand):
"""Panasonic air-conditioner short toggle command (Quiet / Powerful)."""

def __init__(
self,
*,
toggle: PanasonicAcToggle,
modulation: int = 38000,
) -> None:
"""Build a short Quiet/Powerful toggle command.

These are dedicated 16-byte frames that carry no mode/temperature/fan/
swing state; the unit keeps whatever it was already running.
"""
state = [
*_FRAME1,
*_FRAME2_MAGIC,
_SHORT_FRAME_MARKER,
*toggle.value,
]
state.append(_checksum(state, 8, 14))

super().__init__(
address=PANASONIC_AC_ADDRESS,
data=_to_frames(state),
modulation=modulation,
)
108 changes: 108 additions & 0 deletions tests/commands/test_panasonic_ac.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
"""Tests for the Panasonic air-conditioner IR commands."""

import pytest

from infrared_protocols.commands.panasonic_ac import (
PanasonicAcCommand,
PanasonicAcFanSpeed,
PanasonicAcMode,
PanasonicAcSwingAxis1,
PanasonicAcSwingAxis2,
PanasonicAcToggle,
PanasonicAcToggleCommand,
)

# Known-good frames from the reverse-engineered protocol spec. Each frame
# includes its two leading Kaseikyo address bytes (0x02 0x20).
FULL_COOL_24_FRAMES: list[list[int]] = [
[0x02, 0x20, 0xE0, 0x04, 0x00, 0x00, 0x00, 0x06],
# fmt: off
[
0x02, 0x20, 0xE0, 0x04, 0x00, 0x31, 0x30, 0x80, 0xAF, 0x0D,
0x00, 0x0E, 0xE0, 0x00, 0x00, 0x81, 0x00, 0x02, 0x14,
],
# fmt: on
]
QUIET_FRAMES: list[list[int]] = [
[0x02, 0x20, 0xE0, 0x04, 0x00, 0x00, 0x00, 0x06],
[0x02, 0x20, 0xE0, 0x04, 0x80, 0x81, 0x33, 0x3A],
]


def _decode_frames(timings: list[int]) -> list[list[int]]:
"""Decode raw Kaseikyo timings back into per-frame byte lists.

Splits on the inter-frame gap, drops each frame's leader and trailing end
pulse, then packs the LSB-first bits into bytes. A bit is a 1 when its space
is clearly longer than the bit mark, so decoding is independent of the exact
base-unit timings. The decoded bytes include each frame's address bytes.
"""
frames: list[list[int]] = [[]]
for value in timings:
if value == -10000:
frames.append([])
else:
frames[-1].append(value)

decoded: list[list[int]] = []
for frame in frames:
body = frame[2:-1]
mark = body[0]
bits = [1 if -body[i + 1] > 2 * mark else 0 for i in range(0, len(body), 2)]
decoded.append(
[sum(bits[i + k] << k for k in range(8)) for i in range(0, len(bits), 8)]
)
return decoded


def test_full_command_frames() -> None:
"""Test the full state command encodes to the expected frame bytes."""
command = PanasonicAcCommand(mode=PanasonicAcMode.COOL, temperature=24.0)
assert _decode_frames(command.get_raw_timings()) == FULL_COOL_24_FRAMES
assert command.modulation == 38000
assert command.repeat_count == 0


def test_toggle_command_frames() -> None:
"""Test the Quiet toggle command encodes to the expected frame bytes."""
command = PanasonicAcToggleCommand(toggle=PanasonicAcToggle.QUIET)
assert _decode_frames(command.get_raw_timings()) == QUIET_FRAMES


def test_power_off_clears_power_bit() -> None:
"""Test power=False clears bit 0 of byte 13."""
on = PanasonicAcCommand(mode=PanasonicAcMode.HEAT, temperature=20.0)
off = PanasonicAcCommand(mode=PanasonicAcMode.HEAT, temperature=20.0, power=False)
assert _decode_frames(on.get_raw_timings())[1][5] & 0x0F == 1
assert _decode_frames(off.get_raw_timings())[1][5] & 0x0F == 0


def test_nanoex_sets_feature_bit() -> None:
"""Test nanoeX flips bit 0x04 of byte 25."""
without = PanasonicAcCommand(mode=PanasonicAcMode.AUTO, temperature=26.0)
with_nanoex = PanasonicAcCommand(
mode=PanasonicAcMode.AUTO, temperature=26.0, nanoex=True
)
assert _decode_frames(without.get_raw_timings())[1][17] == 0x02
assert _decode_frames(with_nanoex.get_raw_timings())[1][17] == 0x06


def test_swing_axes_land_in_expected_nibbles() -> None:
"""Test the two swing axes encode into bytes 16 and 17."""
command = PanasonicAcCommand(
mode=PanasonicAcMode.COOL,
temperature=24.0,
fan=PanasonicAcFanSpeed.HIGH,
swing_axis1=PanasonicAcSwingAxis1.LOWEST,
swing_axis2=PanasonicAcSwingAxis2.FULL_LEFT,
)
frame2 = _decode_frames(command.get_raw_timings())[1]
assert frame2[8] == (PanasonicAcFanSpeed.HIGH << 4) | PanasonicAcSwingAxis1.LOWEST
assert frame2[9] == PanasonicAcSwingAxis2.FULL_LEFT


@pytest.mark.parametrize("temperature", [15.0, 30.5, 31.0])
def test_temperature_out_of_range_raises(temperature: float) -> None:
"""Test an out-of-range temperature raises ValueError."""
with pytest.raises(ValueError, match="out of range"):
PanasonicAcCommand(mode=PanasonicAcMode.COOL, temperature=temperature)