Compare commits

...

2 Commits

Author SHA1 Message Date
Paulus Schoutsen f493f666ae Add buttons platform to Marantz IR Remote (PM6006)
Adds the three commands not exposed by the media player as button
entities: Speaker A/B, Source Direct, and Loudness. Each press alternates
the RC-5 toggle bit shared with the media player.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-09 11:03:06 -04:00
Paulus Schoutsen bb15ac5469 Add Marantz IR Remote integration for the PM6006
Adds a new integration that proxies Marantz amplifier IR codes through
an existing infrared transmitter entity (e.g. via ESPHome). Initial
release ships a media player covering power, volume, mute, and source
selection (CD, Coax, Network, Optical, Phono, Recorder, Tuner).

Codes are sourced from the infrared-protocols library; analog inputs
use standard RC-5, while Optical / Coax / Network use captured Marantz
"Pre-Code" raw 38 kHz timings.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-09 11:03:05 -04:00
24 changed files with 1384 additions and 2 deletions
+1
View File
@@ -355,6 +355,7 @@ homeassistant.components.lunatone.*
homeassistant.components.lutron.*
homeassistant.components.madvr.*
homeassistant.components.manual.*
homeassistant.components.marantz_infrared.*
homeassistant.components.mastodon.*
homeassistant.components.matrix.*
homeassistant.components.matter.*
Generated
+2
View File
@@ -1040,6 +1040,8 @@ CLAUDE.md @home-assistant/core
/tests/components/lyric/ @timmo001
/homeassistant/components/madvr/ @iloveicedgreentea
/tests/components/madvr/ @iloveicedgreentea
/homeassistant/components/marantz_infrared/ @home-assistant/core
/tests/components/marantz_infrared/ @home-assistant/core
/homeassistant/components/mastodon/ @fabaff @andrew-codechimp
/tests/components/mastodon/ @fabaff @andrew-codechimp
/homeassistant/components/matrix/ @PaarthShah
+5
View File
@@ -0,0 +1,5 @@
{
"domain": "marantz",
"name": "Marantz",
"integrations": ["marantz", "marantz_infrared"]
}
@@ -0,0 +1,37 @@
"""Marantz IR Remote integration for Home Assistant."""
from dataclasses import dataclass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
PLATFORMS = [Platform.BUTTON, Platform.MEDIA_PLAYER]
@dataclass
class MarantzIrRuntimeData:
"""Runtime data for a Marantz IR config entry.
The RC-5 toggle bit must alternate between distinct key presses so
the receiver can distinguish a new press from a held-down repeat.
The toggle is tracked at the device level (one value per config
entry) so all entities — buttons and the media player — share it.
"""
toggle: int = 0
type MarantzIrConfigEntry = ConfigEntry[MarantzIrRuntimeData]
async def async_setup_entry(hass: HomeAssistant, entry: MarantzIrConfigEntry) -> bool:
"""Set up Marantz IR from a config entry."""
entry.runtime_data = MarantzIrRuntimeData()
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: MarantzIrConfigEntry) -> bool:
"""Unload a Marantz IR config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@@ -0,0 +1,80 @@
"""Button platform for Marantz IR integration.
Only commands that aren't already exposed by the media player live here:
speaker A/B, source-direct toggle, and loudness toggle.
"""
from dataclasses import dataclass
from infrared_protocols.codes.marantz.pm6006 import MarantzPM6006Code
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import MarantzIrConfigEntry
from .const import CONF_INFRARED_ENTITY_ID, CONF_MODEL, MarantzModel
from .entity import MarantzIrEntity
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class MarantzIrButtonEntityDescription(ButtonEntityDescription):
"""Describes Marantz IR button entity."""
command_code: MarantzPM6006Code
PM6006_BUTTON_DESCRIPTIONS: tuple[MarantzIrButtonEntityDescription, ...] = (
MarantzIrButtonEntityDescription(
key="speaker_ab",
translation_key="speaker_ab",
command_code=MarantzPM6006Code.SPEAKER_AB,
),
MarantzIrButtonEntityDescription(
key="source_direct",
translation_key="source_direct",
command_code=MarantzPM6006Code.SOURCE_DIRECT,
),
MarantzIrButtonEntityDescription(
key="loudness",
translation_key="loudness",
command_code=MarantzPM6006Code.LOUDNESS,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: MarantzIrConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Marantz IR buttons from config entry."""
infrared_entity_id = entry.data[CONF_INFRARED_ENTITY_ID]
model = entry.data[CONF_MODEL]
if model == MarantzModel.PM6006:
async_add_entities(
MarantzIrButton(entry, infrared_entity_id, description)
for description in PM6006_BUTTON_DESCRIPTIONS
)
class MarantzIrButton(MarantzIrEntity, ButtonEntity):
"""Marantz IR button entity."""
entity_description: MarantzIrButtonEntityDescription
def __init__(
self,
entry: MarantzIrConfigEntry,
infrared_entity_id: str,
description: MarantzIrButtonEntityDescription,
) -> None:
"""Initialize Marantz IR button."""
super().__init__(entry, infrared_entity_id, unique_id_suffix=description.key)
self.entity_description = description
async def async_press(self) -> None:
"""Press the button."""
await self._send_command(self.entity_description.command_code)
@@ -0,0 +1,77 @@
"""Config flow for Marantz IR integration."""
from typing import Any
import voluptuous as vol
from homeassistant.components.infrared import (
DOMAIN as INFRARED_DOMAIN,
async_get_emitters,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.selector import (
EntitySelector,
EntitySelectorConfig,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import CONF_INFRARED_ENTITY_ID, CONF_MODEL, DOMAIN, MarantzModel
MODEL_NAMES: dict[MarantzModel, str] = {
MarantzModel.PM6006: "PM6006",
}
class MarantzIrConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle config flow for Marantz IR."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
emitter_entity_ids = async_get_emitters(self.hass)
if not emitter_entity_ids:
return self.async_abort(reason="no_emitters")
if user_input is not None:
entity_id = user_input[CONF_INFRARED_ENTITY_ID]
model = user_input[CONF_MODEL]
await self.async_set_unique_id(f"marantz_ir_{model}_{entity_id}")
self._abort_if_unique_id_configured()
ent_reg = er.async_get(self.hass)
entry = ent_reg.async_get(entity_id)
entity_name = (
entry.name or entry.original_name or entity_id if entry else entity_id
)
model_name = MODEL_NAMES[MarantzModel(model)]
title = f"Marantz {model_name} via {entity_name}"
return self.async_create_entry(title=title, data=user_input)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_MODEL): SelectSelector(
SelectSelectorConfig(
options=[model.value for model in MarantzModel],
translation_key=CONF_MODEL,
mode=SelectSelectorMode.DROPDOWN,
)
),
vol.Required(CONF_INFRARED_ENTITY_ID): EntitySelector(
EntitySelectorConfig(
domain=INFRARED_DOMAIN,
include_entities=emitter_entity_ids,
)
),
}
),
)
@@ -0,0 +1,13 @@
"""Constants for the Marantz IR integration."""
from enum import StrEnum
DOMAIN = "marantz_infrared"
CONF_INFRARED_ENTITY_ID = "infrared_entity_id"
CONF_MODEL = "model"
class MarantzModel(StrEnum):
"""Supported Marantz models."""
PM6006 = "pm6006"
@@ -0,0 +1,97 @@
"""Common entity for Marantz IR integration."""
import logging
from types import ModuleType
from infrared_protocols.codes.marantz import pm6006
from homeassistant.components.infrared import async_send_command
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import Event, EventStateChangedData, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import async_track_state_change_event
from . import MarantzIrConfigEntry
from .const import CONF_MODEL, DOMAIN, MarantzModel
_LOGGER = logging.getLogger(__name__)
# Each supported model points at the library module that exposes its codes
# and the ``MODEL_ID`` / ``MODEL_NAME`` constants used for the device
# registry entry.
_MODEL_MODULES: dict[MarantzModel, ModuleType] = {
MarantzModel.PM6006: pm6006,
}
class MarantzIrEntity(Entity):
"""Marantz IR base entity."""
_attr_has_entity_name = True
def __init__(
self,
entry: MarantzIrConfigEntry,
infrared_entity_id: str,
unique_id_suffix: str,
) -> None:
"""Initialize Marantz IR entity."""
self._infrared_entity_id = infrared_entity_id
self._runtime_data = entry.runtime_data
self._attr_unique_id = f"{entry.entry_id}_{unique_id_suffix}"
model_module = _MODEL_MODULES[MarantzModel(entry.data[CONF_MODEL])]
self._make_command = model_module.make_command
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
name=f"Marantz {model_module.MODEL_NAME}",
manufacturer="Marantz",
model=model_module.MODEL_ID,
)
async def async_added_to_hass(self) -> None:
"""Subscribe to infrared entity state changes."""
await super().async_added_to_hass()
@callback
def _async_ir_state_changed(event: Event[EventStateChangedData]) -> None:
"""Handle infrared entity state changes."""
new_state = event.data["new_state"]
ir_available = (
new_state is not None and new_state.state != STATE_UNAVAILABLE
)
if ir_available != self.available:
_LOGGER.info(
"Infrared entity %s used by %s is %s",
self._infrared_entity_id,
self.entity_id,
"available" if ir_available else "unavailable",
)
self._attr_available = ir_available
self.async_write_ha_state()
self.async_on_remove(
async_track_state_change_event(
self.hass, [self._infrared_entity_id], _async_ir_state_changed
)
)
ir_state = self.hass.states.get(self._infrared_entity_id)
self._attr_available = (
ir_state is not None and ir_state.state != STATE_UNAVAILABLE
)
async def _send_command(self, code: pm6006.MarantzPM6006Code) -> None:
"""Send an IR command using the Marantz protocol.
Flips the RC-5 toggle bit before each frame so the receiver
treats consecutive presses as new presses, not as a held repeat.
"""
self._runtime_data.toggle ^= 1
await async_send_command(
self.hass,
self._infrared_entity_id,
self._make_command(code, toggle=self._runtime_data.toggle),
context=self._context,
)
@@ -0,0 +1,11 @@
{
"domain": "marantz_infrared",
"name": "Marantz Infrared",
"codeowners": ["@home-assistant/core"],
"config_flow": true,
"dependencies": ["infrared"],
"documentation": "https://www.home-assistant.io/integrations/marantz_infrared",
"integration_type": "device",
"iot_class": "assumed_state",
"quality_scale": "silver"
}
@@ -0,0 +1,100 @@
"""Media player platform for Marantz IR integration."""
from infrared_protocols.codes.marantz.pm6006 import MarantzPM6006Code
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import MarantzIrConfigEntry
from .const import CONF_INFRARED_ENTITY_ID, CONF_MODEL, MarantzModel
from .entity import MarantzIrEntity
PARALLEL_UPDATES = 1
# The Optical button on the amp toggles between the two optical inputs and
# the receiver remembers which one was last used, so we cannot deterministically
# pick between Optical 1 and Optical 2 over IR. We expose a single Optical
# entry that just sends the toggle and let the user press again to switch.
SOURCE_TO_CODE: dict[str, MarantzPM6006Code] = {
"CD": MarantzPM6006Code.SOURCE_CD,
"Coax": MarantzPM6006Code.SOURCE_COAX,
"Network": MarantzPM6006Code.SOURCE_NETWORK,
"Optical": MarantzPM6006Code.SOURCE_OPTICAL,
"Phono": MarantzPM6006Code.SOURCE_PHONO,
"Recorder": MarantzPM6006Code.SOURCE_CDR,
"Tuner": MarantzPM6006Code.SOURCE_TUNER,
}
async def async_setup_entry(
hass: HomeAssistant,
entry: MarantzIrConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Marantz IR media player from config entry."""
infrared_entity_id = entry.data[CONF_INFRARED_ENTITY_ID]
model = entry.data[CONF_MODEL]
if model == MarantzModel.PM6006:
async_add_entities([MarantzIrAmplifierMediaPlayer(entry, infrared_entity_id)])
class MarantzIrAmplifierMediaPlayer(MarantzIrEntity, MediaPlayerEntity):
"""Marantz IR amplifier media player entity."""
_attr_name = None
_attr_assumed_state = True
_attr_device_class = MediaPlayerDeviceClass.RECEIVER
_attr_source_list = list(SOURCE_TO_CODE)
_attr_supported_features = (
MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.SELECT_SOURCE
)
def __init__(self, entry: MarantzIrConfigEntry, infrared_entity_id: str) -> None:
"""Initialize Marantz IR amplifier media player."""
super().__init__(entry, infrared_entity_id, unique_id_suffix="media_player")
self._attr_state = MediaPlayerState.ON
async def async_turn_on(self) -> None:
"""Send the power toggle and assume the amplifier is now on.
Marantz integrated amplifiers expose only a single POWER toggle
over IR — there are no discrete on/off codes — so turn-on and
turn-off send the same frame and rely on assumed_state.
"""
await self._send_command(MarantzPM6006Code.POWER)
self._attr_state = MediaPlayerState.ON
self.async_write_ha_state()
async def async_turn_off(self) -> None:
"""Send the power toggle and assume the amplifier is now off."""
await self._send_command(MarantzPM6006Code.POWER)
self._attr_state = MediaPlayerState.OFF
self.async_write_ha_state()
async def async_volume_up(self) -> None:
"""Send volume up command."""
await self._send_command(MarantzPM6006Code.VOLUME_UP)
async def async_volume_down(self) -> None:
"""Send volume down command."""
await self._send_command(MarantzPM6006Code.VOLUME_DOWN)
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command."""
await self._send_command(MarantzPM6006Code.MUTE)
async def async_select_source(self, source: str) -> None:
"""Select an input source."""
await self._send_command(SOURCE_TO_CODE[source])
self._attr_source = source
self.async_write_ha_state()
@@ -0,0 +1,110 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
This integration does not provide additional actions.
appropriate-polling:
status: exempt
comment: |
This integration does not poll.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup:
status: exempt
comment: |
This integration only proxies commands through an existing infrared
entity, so there is no separate connection to validate during setup.
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: |
This integration does not register custom actions.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow:
status: exempt
comment: |
This integration does not require authentication.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
This integration does not support discovery.
discovery:
status: exempt
comment: |
This integration is configured manually via config flow.
docs-data-update:
status: exempt
comment: |
This integration does not fetch data from devices.
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: |
Each config entry creates a single device.
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: |
No entities should be disabled by default.
entity-translations: done
exception-translations:
status: exempt
comment: |
This integration does not raise exceptions.
icon-translations:
status: exempt
comment: |
This integration does not use custom icons.
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: |
This integration does not have repairable issues.
stale-devices:
status: exempt
comment: |
Each config entry manages exactly one device.
# Platinum
async-dependency:
status: exempt
comment: |
This integration has no external dependencies.
inject-websession:
status: exempt
comment: |
This integration does not make HTTP requests.
strict-typing: done
@@ -0,0 +1,42 @@
{
"config": {
"abort": {
"already_configured": "This Marantz device has already been configured with this transmitter.",
"no_emitters": "No infrared transmitter entities found. Please set up an infrared device first."
},
"step": {
"user": {
"data": {
"infrared_entity_id": "Infrared transmitter",
"model": "Model"
},
"data_description": {
"infrared_entity_id": "The infrared transmitter entity to use for sending commands.",
"model": "The Marantz model to control."
},
"description": "Select the Marantz model and the infrared transmitter entity to use for controlling your Marantz device.",
"title": "Set up Marantz IR Remote"
}
}
},
"entity": {
"button": {
"loudness": {
"name": "Loudness"
},
"source_direct": {
"name": "Source direct"
},
"speaker_ab": {
"name": "Speaker A/B"
}
}
},
"selector": {
"model": {
"options": {
"pm6006": "PM6006"
}
}
}
}
+1
View File
@@ -427,6 +427,7 @@ FLOWS = {
"lyric",
"madvr",
"mailgun",
"marantz_infrared",
"mastodon",
"matter",
"mcp",
+14 -2
View File
@@ -4020,8 +4020,20 @@
},
"marantz": {
"name": "Marantz",
"integration_type": "virtual",
"supported_by": "denonavr"
"integrations": {
"marantz": {
"integration_type": "virtual",
"config_flow": false,
"supported_by": "denonavr",
"name": "Marantz"
},
"marantz_infrared": {
"integration_type": "device",
"config_flow": true,
"iot_class": "assumed_state",
"name": "Marantz Infrared"
}
}
},
"martec": {
"name": "Martec",
Generated
+10
View File
@@ -3305,6 +3305,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.marantz_infrared.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.mastodon.*]
check_untyped_defs = true
disallow_incomplete_defs = true
@@ -0,0 +1 @@
"""Tests for the Marantz Infrared integration."""
@@ -0,0 +1,108 @@
"""Common fixtures for the Marantz Infrared tests."""
from collections.abc import Generator
from unittest.mock import patch
from infrared_protocols import Command as InfraredCommand
import pytest
from homeassistant.components.infrared import (
DATA_COMPONENT as INFRARED_DATA_COMPONENT,
DOMAIN as INFRARED_DOMAIN,
InfraredEntity,
)
from homeassistant.components.marantz_infrared import PLATFORMS
from homeassistant.components.marantz_infrared.const import (
CONF_INFRARED_ENTITY_ID,
CONF_MODEL,
DOMAIN,
MarantzModel,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
MOCK_INFRARED_ENTITY_ID = "infrared.test_ir_transmitter"
class MockInfraredEntity(InfraredEntity):
"""Mock infrared entity for testing."""
_attr_has_entity_name = True
_attr_name = "Test IR transmitter"
def __init__(self, unique_id: str) -> None:
"""Initialize mock entity."""
self._attr_unique_id = unique_id
self.send_command_calls: list[InfraredCommand] = []
async def async_send_command(self, command: InfraredCommand) -> None:
"""Mock send command."""
self.send_command_calls.append(command)
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
entry_id="01JTEST0000000000000000000",
title="Marantz PM6006 via Test IR transmitter",
data={
CONF_MODEL: MarantzModel.PM6006,
CONF_INFRARED_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
},
unique_id=f"marantz_ir_pm6006_{MOCK_INFRARED_ENTITY_ID}",
)
@pytest.fixture
def mock_infrared_entity() -> MockInfraredEntity:
"""Return a mock infrared entity."""
return MockInfraredEntity("test_ir_transmitter")
@pytest.fixture
def platforms() -> list[Platform]:
"""Return platforms to set up."""
return PLATFORMS
@pytest.fixture
def mock_make_marantz_amplifier_command() -> Generator[None]:
"""Patch make_command to return the MarantzPM6006Code directly.
This allows tests to assert on the high-level code enum value
rather than the raw RC-5 timings.
"""
with patch(
"infrared_protocols.codes.marantz.pm6006.make_command",
side_effect=lambda code, **kwargs: code,
):
yield
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_infrared_entity: MockInfraredEntity,
mock_make_marantz_amplifier_command: None,
platforms: list[Platform],
) -> MockConfigEntry:
"""Set up the Marantz Infrared integration for testing."""
assert await async_setup_component(hass, INFRARED_DOMAIN, {})
await hass.async_block_till_done()
infrared_component = hass.data[INFRARED_DATA_COMPONENT]
await infrared_component.async_add_entities([mock_infrared_entity])
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.marantz_infrared.PLATFORMS", platforms):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
@@ -0,0 +1,151 @@
# serializer version: 1
# name: test_entities[button.marantz_amplifier_pm6006_loudness-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.marantz_amplifier_pm6006_loudness',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Loudness',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Loudness',
'platform': 'marantz_infrared',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'loudness',
'unique_id': '01JTEST0000000000000000000_loudness',
'unit_of_measurement': None,
})
# ---
# name: test_entities[button.marantz_amplifier_pm6006_loudness-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Marantz Amplifier PM6006 Loudness',
}),
'context': <ANY>,
'entity_id': 'button.marantz_amplifier_pm6006_loudness',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_entities[button.marantz_amplifier_pm6006_source_direct-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.marantz_amplifier_pm6006_source_direct',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Source direct',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Source direct',
'platform': 'marantz_infrared',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'source_direct',
'unique_id': '01JTEST0000000000000000000_source_direct',
'unit_of_measurement': None,
})
# ---
# name: test_entities[button.marantz_amplifier_pm6006_source_direct-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Marantz Amplifier PM6006 Source direct',
}),
'context': <ANY>,
'entity_id': 'button.marantz_amplifier_pm6006_source_direct',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_entities[button.marantz_amplifier_pm6006_speaker_a_b-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.marantz_amplifier_pm6006_speaker_a_b',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Speaker A/B',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Speaker A/B',
'platform': 'marantz_infrared',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'speaker_ab',
'unique_id': '01JTEST0000000000000000000_speaker_ab',
'unit_of_measurement': None,
})
# ---
# name: test_entities[button.marantz_amplifier_pm6006_speaker_a_b-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Marantz Amplifier PM6006 Speaker A/B',
}),
'context': <ANY>,
'entity_id': 'button.marantz_amplifier_pm6006_speaker_a_b',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
@@ -0,0 +1,73 @@
# serializer version: 1
# name: test_entities[media_player.marantz_amplifier_pm6006-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'source_list': list([
'CD',
'Coax',
'Network',
'Optical',
'Phono',
'Recorder',
'Tuner',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'media_player',
'entity_category': None,
'entity_id': 'media_player.marantz_amplifier_pm6006',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <MediaPlayerDeviceClass.RECEIVER: 'receiver'>,
'original_icon': None,
'original_name': None,
'platform': 'marantz_infrared',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 3464>,
'translation_key': None,
'unique_id': '01JTEST0000000000000000000_media_player',
'unit_of_measurement': None,
})
# ---
# name: test_entities[media_player.marantz_amplifier_pm6006-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'assumed_state': True,
'device_class': 'receiver',
'friendly_name': 'Marantz Amplifier PM6006',
'source_list': list([
'CD',
'Coax',
'Network',
'Optical',
'Phono',
'Recorder',
'Tuner',
]),
'supported_features': <MediaPlayerEntityFeature: 3464>,
}),
'context': <ANY>,
'entity_id': 'media_player.marantz_amplifier_pm6006',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
@@ -0,0 +1,82 @@
"""Tests for the Marantz Infrared button platform."""
from infrared_protocols.codes.marantz.pm6006 import MarantzPM6006Code
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import MockInfraredEntity
from .utils import check_availability_follows_ir_entity
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture
def platforms() -> list[Platform]:
"""Return platforms to set up."""
return [Platform.BUTTON]
@pytest.mark.usefixtures("init_integration")
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test all button entities are created with correct attributes."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
device_entry = device_registry.async_get_device(
identifiers={("marantz_infrared", mock_config_entry.entry_id)}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id
@pytest.mark.parametrize(
("entity_id", "expected_code"),
[
("button.marantz_amplifier_pm6006_speaker_a_b", MarantzPM6006Code.SPEAKER_AB),
(
"button.marantz_amplifier_pm6006_source_direct",
MarantzPM6006Code.SOURCE_DIRECT,
),
("button.marantz_amplifier_pm6006_loudness", MarantzPM6006Code.LOUDNESS),
],
)
@pytest.mark.usefixtures("init_integration")
async def test_button_press_sends_correct_code(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
entity_id: str,
expected_code: MarantzPM6006Code,
) -> None:
"""Test pressing a button sends the correct IR code."""
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert len(mock_infrared_entity.send_command_calls) == 1
assert mock_infrared_entity.send_command_calls[0] == expected_code
@pytest.mark.usefixtures("init_integration")
async def test_button_availability_follows_ir_entity(
hass: HomeAssistant,
) -> None:
"""Test button becomes unavailable when IR entity is unavailable."""
entity_id = "button.marantz_amplifier_pm6006_loudness"
await check_availability_follows_ir_entity(hass, entity_id)
@@ -0,0 +1,134 @@
"""Tests for the Marantz Infrared config flow."""
import pytest
from homeassistant.components.infrared import (
DATA_COMPONENT as INFRARED_DATA_COMPONENT,
DOMAIN as INFRARED_DOMAIN,
)
from homeassistant.components.marantz_infrared.const import (
CONF_INFRARED_ENTITY_ID,
CONF_MODEL,
DOMAIN,
MarantzModel,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from .conftest import MOCK_INFRARED_ENTITY_ID, MockInfraredEntity
from tests.common import MockConfigEntry
@pytest.fixture
async def setup_infrared(
hass: HomeAssistant, mock_infrared_entity: MockInfraredEntity
) -> None:
"""Set up the infrared component with a mock entity."""
assert await async_setup_component(hass, INFRARED_DOMAIN, {})
await hass.async_block_till_done()
component = hass.data[INFRARED_DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
@pytest.mark.usefixtures("setup_infrared")
async def test_user_flow_success(
hass: HomeAssistant,
) -> None:
"""Test successful user config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_MODEL: MarantzModel.PM6006,
CONF_INFRARED_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Marantz PM6006 via Test IR transmitter"
assert result["data"] == {
CONF_MODEL: MarantzModel.PM6006,
CONF_INFRARED_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
}
assert result["result"].unique_id == f"marantz_ir_pm6006_{MOCK_INFRARED_ENTITY_ID}"
@pytest.mark.usefixtures("setup_infrared")
async def test_user_flow_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test user flow aborts when entry is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_MODEL: MarantzModel.PM6006,
CONF_INFRARED_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_user_flow_no_emitters(hass: HomeAssistant) -> None:
"""Test user flow aborts when no infrared emitters exist."""
assert await async_setup_component(hass, INFRARED_DOMAIN, {})
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_emitters"
@pytest.mark.usefixtures("setup_infrared")
@pytest.mark.parametrize(
("entity_name", "expected_title"),
[
(None, "Marantz PM6006 via Test IR transmitter"),
("AC IR emitter", "Marantz PM6006 via AC IR emitter"),
],
)
async def test_user_flow_title_from_entity_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
entity_name: str | None,
expected_title: str,
) -> None:
"""Test config entry title uses the entity name."""
entity_registry.async_update_entity(MOCK_INFRARED_ENTITY_ID, name=entity_name)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_MODEL: MarantzModel.PM6006,
CONF_INFRARED_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == expected_title
@@ -0,0 +1,19 @@
"""Tests for the Marantz Infrared integration setup."""
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_setup_and_unload_entry(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
"""Test setting up and unloading a config entry."""
entry = init_integration
assert entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
@@ -0,0 +1,186 @@
"""Tests for the Marantz Infrared media player platform."""
from infrared_protocols.codes.marantz.pm6006 import MarantzPM6006Code
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_INPUT_SOURCE_LIST,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_UP,
)
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import MockInfraredEntity
from .utils import check_availability_follows_ir_entity
from tests.common import MockConfigEntry, snapshot_platform
MEDIA_PLAYER_ENTITY_ID = "media_player.marantz_amplifier_pm6006"
@pytest.fixture
def platforms() -> list[Platform]:
"""Return platforms to set up."""
return [Platform.MEDIA_PLAYER]
@pytest.mark.usefixtures("init_integration")
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the media player entity is created with correct attributes."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
device_entry = device_registry.async_get_device(
identifiers={("marantz_infrared", mock_config_entry.entry_id)}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id
@pytest.mark.parametrize(
("service", "service_data", "expected_code"),
[
(SERVICE_TURN_ON, {}, MarantzPM6006Code.POWER),
(SERVICE_TURN_OFF, {}, MarantzPM6006Code.POWER),
(SERVICE_VOLUME_UP, {}, MarantzPM6006Code.VOLUME_UP),
(SERVICE_VOLUME_DOWN, {}, MarantzPM6006Code.VOLUME_DOWN),
(SERVICE_VOLUME_MUTE, {"is_volume_muted": True}, MarantzPM6006Code.MUTE),
],
)
@pytest.mark.usefixtures("init_integration")
async def test_media_player_action_sends_correct_code(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
service: str,
service_data: dict[str, bool],
expected_code: MarantzPM6006Code,
) -> None:
"""Test each media player action sends the correct IR code."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, **service_data},
blocking=True,
)
assert len(mock_infrared_entity.send_command_calls) == 1
assert mock_infrared_entity.send_command_calls[0] == expected_code
@pytest.mark.parametrize(
("source", "expected_code"),
[
("CD", MarantzPM6006Code.SOURCE_CD),
("Recorder", MarantzPM6006Code.SOURCE_CDR),
("Phono", MarantzPM6006Code.SOURCE_PHONO),
("Tuner", MarantzPM6006Code.SOURCE_TUNER),
("Coax", MarantzPM6006Code.SOURCE_COAX),
("Network", MarantzPM6006Code.SOURCE_NETWORK),
("Optical", MarantzPM6006Code.SOURCE_OPTICAL),
],
)
@pytest.mark.usefixtures("init_integration")
async def test_media_player_select_source(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
source: str,
expected_code: MarantzPM6006Code,
) -> None:
"""Test selecting a source sends the correct IR code and updates state."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, ATTR_INPUT_SOURCE: source},
blocking=True,
)
assert mock_infrared_entity.send_command_calls == [expected_code]
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state is not None
assert state.attributes[ATTR_INPUT_SOURCE] == source
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == [
"CD",
"Coax",
"Network",
"Optical",
"Phono",
"Recorder",
"Tuner",
]
@pytest.mark.usefixtures("init_integration")
async def test_turn_on_off_update_assumed_state(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""Both turn-on and turn-off send POWER but assume opposite states."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state is not None
assert state.state == "off"
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state is not None
assert state.state == "on"
assert mock_infrared_entity.send_command_calls == [
MarantzPM6006Code.POWER,
MarantzPM6006Code.POWER,
]
@pytest.mark.usefixtures("init_integration")
async def test_media_player_availability_follows_ir_entity(
hass: HomeAssistant,
) -> None:
"""Test media player becomes unavailable when IR entity is unavailable."""
await check_availability_follows_ir_entity(hass, MEDIA_PLAYER_ENTITY_ID)
async def test_toggle_flips_between_commands(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""The RC-5 toggle bit must alternate so the receiver sees distinct presses."""
for expected_toggle in (1, 0, 1, 0):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_UP,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
)
assert init_integration.runtime_data.toggle == expected_toggle
assert len(mock_infrared_entity.send_command_calls) == 4
@@ -0,0 +1,30 @@
"""Tests for the Marantz Infrared integration."""
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from .conftest import MOCK_INFRARED_ENTITY_ID
async def check_availability_follows_ir_entity(
hass: HomeAssistant,
entity_id: str,
) -> None:
"""Check that entity becomes unavailable when IR entity is unavailable."""
state = hass.states.get(entity_id)
assert state is not None
assert state.state != STATE_UNAVAILABLE
hass.states.async_set(MOCK_INFRARED_ENTITY_ID, STATE_UNAVAILABLE)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_UNAVAILABLE
hass.states.async_set(MOCK_INFRARED_ENTITY_ID, "2026-01-01T00:00:00.000")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state is not None
assert state.state != STATE_UNAVAILABLE