Compare commits

...

8 Commits

Author SHA1 Message Date
Paulus Schoutsen 8a68e9f5ff Use dedicated POWER_ON, POWER_OFF, MUTE_ON, MUTE_OFF commands 2026-05-14 11:28:27 -04:00
Paulus Schoutsen f1034bb0be Make SELECT_SOURCE conditional on model having source codes
The SR-7000 and SR-7200 models in the infrared-protocols registry
have no SOURCE_* codes, so their source_list is empty. Drop
SELECT_SOURCE from supported_features in that case, and drop the
always-true POWER gate in async_setup_entry while here.

Add SR-7000 as a second parametrized model to the snapshot and a
new test that asserts the feature flag tracks model capability.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 11:28:27 -04:00
Paulus Schoutsen 1dd1b70230 Expand Marantz IR source mapping for non-PM6006 models
The unified Marantz audio code set in infrared-protocols 5.0 adds
source codes used by the other registry models (RC042SR, SR-19,
SR-7000/7200/7300, SR670, AV-9000). Map LD, MD, SAT, TAPE, TV and
VCR1 so those models expose their inputs via select_source.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 11:28:27 -04:00
Paulus Schoutsen d5f5ca15ea Adjust to upcoming changes infrared-protocols 2026-05-14 11:28:27 -04:00
Paulus Schoutsen 0b8d2ee4ba Fixes 2026-05-14 11:28:27 -04:00
Paulus Schoutsen 5a47ca1929 Drop redundant marantz_ir_ prefix from config entry unique ID
Unique IDs are scoped per integration, so the domain prefix was just
noise.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 11:28:27 -04:00
Paulus Schoutsen 3154c04f61 Adapt marantz_infrared to infrared-protocols 3.x API
`Command` moved out of the package root in 3.x (now in
`infrared_protocols.commands`), which broke the test conftest import and
caused both `Check pylint on tests` and `Split tests for full run` CI jobs
to fail at collection time.

While here, drop the ModuleType-based per-model dispatch in `entity.py`
that Copilot flagged as untyped. The library no longer exposes the
display-only `MODEL_ID`/`MODEL_NAME` constants the dispatch relied on,
so PM6006 is imported directly and the human-readable model name lives
in a typed `dict[MarantzModel, str]` in the integration.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 11:28:27 -04:00
Paulus Schoutsen e0f08e10f7 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-14 11:28:27 -04:00
21 changed files with 1290 additions and 2 deletions
+1
View File
@@ -358,6 +358,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
@@ -1045,6 +1045,8 @@ CLAUDE.md @home-assistant/core
/tests/components/lyric/ @timmo001
/homeassistant/components/madvr/ @iloveicedgreentea
/tests/components/madvr/ @iloveicedgreentea
/homeassistant/components/marantz_infrared/ @balloob
/tests/components/marantz_infrared/ @balloob
/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.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 of a config entry 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,69 @@
"""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.selector import (
EntitySelector,
EntitySelectorConfig,
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import CONF_INFRARED_EMITTER_ENTITY_ID, CONF_MODEL, DOMAIN, MODELS
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_EMITTER_ENTITY_ID]
model = user_input[CONF_MODEL]
await self.async_set_unique_id(f"{model}_{entity_id}")
self._abort_if_unique_id_configured()
return self.async_create_entry(title=MODELS[model].name, data=user_input)
model_options = [
SelectOptionDict(value=slug, label=model.name)
for slug, model in MODELS.items()
]
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_MODEL): SelectSelector(
SelectSelectorConfig(
options=model_options,
mode=SelectSelectorMode.DROPDOWN,
)
),
vol.Required(CONF_INFRARED_EMITTER_ENTITY_ID): EntitySelector(
EntitySelectorConfig(
domain=INFRARED_DOMAIN,
include_entities=emitter_entity_ids,
)
),
}
),
)
@@ -0,0 +1,13 @@
"""Constants for the Marantz IR integration."""
from infrared_protocols.codes.marantz import models as marantz_models
from homeassistant.util import slugify
DOMAIN = "marantz_infrared"
CONF_INFRARED_EMITTER_ENTITY_ID = "infrared_emitter_entity_id"
CONF_MODEL = "model"
MODELS: dict[str, marantz_models.MarantzModel] = {
slugify(model.name): model for model in marantz_models.ALL_MODELS
}
@@ -0,0 +1,90 @@
"""Common entity for Marantz IR integration."""
import logging
from infrared_protocols.codes.marantz import models as marantz_models
from infrared_protocols.codes.marantz.audio import MarantzAudioCode
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, MODELS
_LOGGER = logging.getLogger(__name__)
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}"
lib_model = MODELS[entry.data[CONF_MODEL]]
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
name=f"Marantz {lib_model.name}",
manufacturer="Marantz",
# Generic catch-all entries aren't a specific physical product.
model=None if lib_model is marantz_models.GENERIC else lib_model.name,
)
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: MarantzAudioCode) -> 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,
code.to_command(toggle=self._runtime_data.toggle),
context=self._context,
)
@@ -0,0 +1,11 @@
{
"domain": "marantz_infrared",
"name": "Marantz Infrared",
"codeowners": ["@balloob"],
"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,150 @@
"""Media player platform for Marantz IR integration."""
from dataclasses import dataclass
from typing import Any
from infrared_protocols.codes.marantz.audio import MarantzAudioCode
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.restore_state import ExtraStoredData, RestoreEntity
from . import MarantzIrConfigEntry
from .const import CONF_INFRARED_EMITTER_ENTITY_ID, CONF_MODEL, MODELS
from .entity import MarantzIrEntity
PARALLEL_UPDATES = 1
SOURCE_TO_CODE: dict[str, MarantzAudioCode] = {
"cd": MarantzAudioCode.SOURCE_CD,
"coax": MarantzAudioCode.SOURCE_COAX,
"laserdisc": MarantzAudioCode.SOURCE_LD,
"md": MarantzAudioCode.SOURCE_MD,
"network": MarantzAudioCode.SOURCE_NETWORK,
"optical": MarantzAudioCode.SOURCE_OPTICAL,
"phono": MarantzAudioCode.SOURCE_PHONO,
"recorder": MarantzAudioCode.SOURCE_CDR,
"satellite": MarantzAudioCode.SOURCE_SAT,
"tape": MarantzAudioCode.SOURCE_TAPE,
"tuner": MarantzAudioCode.SOURCE_TUNER,
"tv": MarantzAudioCode.SOURCE_TV,
"vcr": MarantzAudioCode.SOURCE_VCR1,
}
@dataclass
class _MarantzAmplifierExtraData(ExtraStoredData):
"""Persisted assumed-state data for a Marantz amplifier.
Stored separately from the entity state because while the amplifier is
OFF, ``MediaPlayerEntity.state_attributes`` strips ``source`` / mute,
so a restart in the OFF state would otherwise lose them.
"""
source: str | None
is_volume_muted: bool | None
def as_dict(self) -> dict[str, Any]:
"""Serialize for the restore-state store."""
return {"source": self.source, "is_volume_muted": self.is_volume_muted}
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_EMITTER_ENTITY_ID]
async_add_entities([MarantzIrAmplifierMediaPlayer(entry, infrared_entity_id)])
class MarantzIrAmplifierMediaPlayer(MarantzIrEntity, MediaPlayerEntity, RestoreEntity):
"""Marantz IR amplifier media player entity."""
_attr_name = None
_attr_assumed_state = True
_attr_device_class = MediaPlayerDeviceClass.RECEIVER
_attr_translation_key = "receiver"
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")
codes = MODELS[entry.data[CONF_MODEL]].codes
self._source_to_code = {
source: code for source, code in SOURCE_TO_CODE.items() if code in codes
}
self._attr_source_list = list(self._source_to_code)
features = (
MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.VOLUME_MUTE
)
if self._source_to_code:
features |= MediaPlayerEntityFeature.SELECT_SOURCE
self._attr_supported_features = features
@property
def extra_restore_state_data(self) -> ExtraStoredData:
"""Persist source and mute regardless of ON/OFF state."""
return _MarantzAmplifierExtraData(
source=self._attr_source,
is_volume_muted=self._attr_is_volume_muted,
)
async def async_added_to_hass(self) -> None:
"""Restore last known assumed state, source, and mute."""
await super().async_added_to_hass()
if (last_state := await self.async_get_last_state()) is not None and (
last_state.state in (MediaPlayerState.ON, MediaPlayerState.OFF)
):
self._attr_state = MediaPlayerState(last_state.state)
if (extra := await self.async_get_last_extra_data()) is not None:
data = extra.as_dict()
if (source := data.get("source")) in self._source_to_code:
self._attr_source = source
if (muted := data.get("is_volume_muted")) is not None:
self._attr_is_volume_muted = bool(muted)
async def async_turn_on(self) -> None:
"""Send discrete power-on command."""
await self._send_command(MarantzAudioCode.POWER_ON)
self._attr_state = MediaPlayerState.ON
self.async_write_ha_state()
async def async_turn_off(self) -> None:
"""Send discrete power-off command."""
await self._send_command(MarantzAudioCode.POWER_OFF)
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(MarantzAudioCode.VOLUME_UP)
async def async_volume_down(self) -> None:
"""Send volume down command."""
await self._send_command(MarantzAudioCode.VOLUME_DOWN)
async def async_mute_volume(self, mute: bool) -> None:
"""Send discrete mute-on or mute-off command."""
await self._send_command(
MarantzAudioCode.MUTE_ON if mute else MarantzAudioCode.MUTE_OFF
)
self._attr_is_volume_muted = mute
self.async_write_ha_state()
async def async_select_source(self, source: str) -> None:
"""Select an input source."""
await self._send_command(self._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,47 @@
{
"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_emitter_entity_id": "Infrared transmitter",
"model": "Model"
},
"data_description": {
"infrared_emitter_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": {
"media_player": {
"receiver": {
"state_attributes": {
"source": {
"state": {
"cd": "CD",
"coax": "Coax",
"laserdisc": "LaserDisc",
"md": "MD",
"network": "Network",
"optical": "Optical",
"phono": "Phono",
"recorder": "Recorder",
"satellite": "Satellite",
"tape": "Tape",
"tuner": "Tuner",
"tv": "TV",
"vcr": "VCR"
}
}
}
}
}
}
}
+1
View File
@@ -430,6 +430,7 @@ FLOWS = {
"lyric",
"madvr",
"mailgun",
"marantz_infrared",
"mastodon",
"matter",
"mcp",
+14 -2
View File
@@ -4032,8 +4032,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
@@ -3337,6 +3337,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,126 @@
"""Common fixtures for the Marantz Infrared tests."""
from collections.abc import Generator
from unittest.mock import patch
from infrared_protocols.codes.marantz.audio import MarantzAudioCode
from infrared_protocols.commands 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_EMITTER_ENTITY_ID,
CONF_MODEL,
DOMAIN,
MODELS,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
from tests.common import MockConfigEntry
MOCK_INFRARED_ENTITY_ID = "infrared.test_ir_transmitter"
MOCK_MODEL = "pm6006_integrated_amplifier"
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 model(request: pytest.FixtureRequest) -> str:
"""Return the Marantz model slug to use for the config entry.
Override with ``@pytest.mark.parametrize("model", [...], indirect=True)``.
"""
return getattr(request, "param", MOCK_MODEL)
@pytest.fixture
def mock_config_entry(model: str) -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
entry_id="01JTEST0000000000000000000",
title=MODELS[model].name,
data={
CONF_MODEL: model,
CONF_INFRARED_EMITTER_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
},
unique_id=f"{model}_{MOCK_INFRARED_ENTITY_ID}",
)
def media_player_entity_id(model: str) -> str:
"""Return the expected media_player entity_id for a model slug."""
return f"media_player.marantz_{slugify(MODELS[model].name)}"
@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_marantz_to_command() -> Generator[None]:
"""Make ``MarantzAudioCode.to_command`` return the code itself.
This lets tests assert on the high-level code enum value rather
than on the raw RC-5 timings.
"""
def _identity(self: MarantzAudioCode, repeat_count: int = 0, *, toggle: int = 0):
return self
with patch.object(MarantzAudioCode, "to_command", _identity):
yield
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_infrared_entity: MockInfraredEntity,
mock_marantz_to_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,127 @@
# serializer version: 1
# name: test_entities[pm6006_integrated_amplifier][media_player.marantz_pm6006_integrated_amplifier-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_pm6006_integrated_amplifier',
'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': 'receiver',
'unique_id': '01JTEST0000000000000000000_media_player',
'unit_of_measurement': None,
})
# ---
# name: test_entities[pm6006_integrated_amplifier][media_player.marantz_pm6006_integrated_amplifier-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'assumed_state': True,
'device_class': 'receiver',
'friendly_name': 'Marantz PM6006 Integrated Amplifier',
'source_list': list([
'cd',
'coax',
'network',
'optical',
'phono',
'recorder',
'tuner',
]),
'supported_features': <MediaPlayerEntityFeature: 3464>,
}),
'context': <ANY>,
'entity_id': 'media_player.marantz_pm6006_integrated_amplifier',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_entities[sr_7000_receiver][media_player.marantz_sr_7000_receiver-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
}),
'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_sr_7000_receiver',
'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: 1416>,
'translation_key': 'receiver',
'unique_id': '01JTEST0000000000000000000_media_player',
'unit_of_measurement': None,
})
# ---
# name: test_entities[sr_7000_receiver][media_player.marantz_sr_7000_receiver-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'assumed_state': True,
'device_class': 'receiver',
'friendly_name': 'Marantz SR-7000 Receiver',
'supported_features': <MediaPlayerEntityFeature: 1416>,
}),
'context': <ANY>,
'entity_id': 'media_player.marantz_sr_7000_receiver',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
@@ -0,0 +1,130 @@
"""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_EMITTER_ENTITY_ID,
CONF_MODEL,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
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: "pm6006_integrated_amplifier",
CONF_INFRARED_EMITTER_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "PM6006 Integrated Amplifier"
assert result["data"] == {
CONF_MODEL: "pm6006_integrated_amplifier",
CONF_INFRARED_EMITTER_ENTITY_ID: MOCK_INFRARED_ENTITY_ID,
}
assert (
result["result"].unique_id
== f"pm6006_integrated_amplifier_{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: "pm6006_integrated_amplifier",
CONF_INFRARED_EMITTER_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(
("model", "expected_title"),
[
("generic_amplifier", "Generic Amplifier"),
("pm6006_integrated_amplifier", "PM6006 Integrated Amplifier"),
],
)
async def test_user_flow_title_from_model(
hass: HomeAssistant, model: str, expected_title: str
) -> None:
"""Test config entry title is the model 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: model,
CONF_INFRARED_EMITTER_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,297 @@
"""Tests for the Marantz Infrared media player platform."""
from typing import Any
from infrared_protocols.codes.marantz.audio import MarantzAudioCode
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.infrared import (
DATA_COMPONENT as INFRARED_DATA_COMPONENT,
DOMAIN as INFRARED_DOMAIN,
)
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_INPUT_SOURCE_LIST,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_UP,
MediaPlayerEntityFeature,
MediaPlayerState,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
from .conftest import MockInfraredEntity, media_player_entity_id
from .utils import check_availability_follows_ir_entity
from tests.common import (
MockConfigEntry,
mock_restore_cache_with_extra_data,
snapshot_platform,
)
MEDIA_PLAYER_ENTITY_ID = "media_player.marantz_pm6006_integrated_amplifier"
@pytest.fixture
def platforms() -> list[Platform]:
"""Return platforms to set up."""
return [Platform.MEDIA_PLAYER]
@pytest.mark.parametrize(
"model",
["pm6006_integrated_amplifier", "sr_7000_receiver"],
indirect=True,
)
@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(
("model", "has_select_source"),
[
("pm6006_integrated_amplifier", True),
("sr_7000_receiver", False),
],
indirect=["model"],
)
@pytest.mark.usefixtures("init_integration")
async def test_select_source_feature_matches_model(
hass: HomeAssistant,
model: str,
has_select_source: bool,
) -> None:
"""SELECT_SOURCE is advertised only when the model has source codes."""
state = hass.states.get(media_player_entity_id(model))
assert state is not None
features = state.attributes["supported_features"]
assert bool(features & MediaPlayerEntityFeature.SELECT_SOURCE) is has_select_source
@pytest.mark.parametrize(
("service", "service_data", "expected_code"),
[
(SERVICE_TURN_ON, {}, MarantzAudioCode.POWER_ON),
(SERVICE_TURN_OFF, {}, MarantzAudioCode.POWER_OFF),
(SERVICE_VOLUME_UP, {}, MarantzAudioCode.VOLUME_UP),
(SERVICE_VOLUME_DOWN, {}, MarantzAudioCode.VOLUME_DOWN),
(SERVICE_VOLUME_MUTE, {"is_volume_muted": True}, MarantzAudioCode.MUTE_ON),
(SERVICE_VOLUME_MUTE, {"is_volume_muted": False}, MarantzAudioCode.MUTE_OFF),
],
)
@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: MarantzAudioCode,
) -> 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", MarantzAudioCode.SOURCE_CD),
("recorder", MarantzAudioCode.SOURCE_CDR),
("phono", MarantzAudioCode.SOURCE_PHONO),
("tuner", MarantzAudioCode.SOURCE_TUNER),
("coax", MarantzAudioCode.SOURCE_COAX),
("network", MarantzAudioCode.SOURCE_NETWORK),
("optical", MarantzAudioCode.SOURCE_OPTICAL),
],
)
@pytest.mark.usefixtures("init_integration")
async def test_media_player_select_source(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
source: str,
expected_code: MarantzAudioCode,
) -> 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:
"""Turn-on sends POWER_ON and turn-off sends POWER_OFF."""
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 == [
MarantzAudioCode.POWER_OFF,
MarantzAudioCode.POWER_ON,
]
@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 _setup_with_restore(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_infrared_entity: MockInfraredEntity,
restored: State,
extra_data: dict[str, Any],
) -> None:
"""Seed the restore cache (state + extra data) and set up the integration."""
mock_restore_cache_with_extra_data(hass, [(restored, extra_data)])
assert await async_setup_component(hass, INFRARED_DOMAIN, {})
await hass.async_block_till_done()
await hass.data[INFRARED_DATA_COMPONENT].async_add_entities([mock_infrared_entity])
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
@pytest.mark.parametrize(
"restored_state",
[MediaPlayerState.ON, MediaPlayerState.OFF],
)
async def test_restores_state_source_and_mute(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_infrared_entity: MockInfraredEntity,
mock_marantz_to_command: None,
restored_state: MediaPlayerState,
) -> None:
"""State, source, and mute survive a restart even from the OFF state.
Source/mute are persisted via extra-restore data so the OFF case
(where the base class strips them from state attributes) still
restores them the user sees the previously-selected source the
moment they turn the amp back on.
"""
await _setup_with_restore(
hass,
mock_config_entry,
mock_infrared_entity,
State(MEDIA_PLAYER_ENTITY_ID, restored_state),
extra_data={"source": "phono", "is_volume_muted": True},
)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state is not None
assert state.state == restored_state.value
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 state.attributes[ATTR_INPUT_SOURCE] == "phono"
assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is True
@pytest.mark.usefixtures("init_integration")
async def test_initial_state_unknown_when_no_restore(hass: HomeAssistant) -> None:
"""With no previous state to restore, the entity reports unknown."""
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_INPUT_SOURCE) is None
assert state.attributes.get(ATTR_MEDIA_VOLUME_MUTED) is None
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