) -> None:
"""Handle PLAY MEDIA on given player."""
castplayer = self.castplayers[player_id]
- use_flow_mode = await self.mass.config.get_player_config_value(
- player_id, CONF_FLOW_MODE
- ) or await self.mass.config.get_player_config_value(player_id, CONF_CROSSFADE)
+ use_flow_mode = await self.mass.config.get_player_config_value(player_id, CONF_FLOW_MODE)
url = self.mass.streams.resolve_stream_url(
player_id,
queue_item=queue_item,
"type": "LOAD",
"media": self._create_cc_media_item(queue_item, url),
}
+
# make sure that the media controller app is launched
app_id = ALT_APP_ID if use_flow_mode else DEFAULT_APP_ID
await self._launch_app(castplayer, app_id)
# send queue info to the CC
media_controller = castplayer.cc.media_controller
- await asyncio.to_thread(media_controller.send_message, queuedata, True)
+ await asyncio.to_thread(media_controller.send_message, data=queuedata, inc_session_id=True)
async def enqueue_next_queue_item(self, player_id: str, queue_item: QueueItem) -> None:
"""Handle enqueuing of the next queue item on the player."""
}
media_controller = castplayer.cc.media_controller
queuedata["mediaSessionId"] = media_controller.status.media_session_id
- self.mass.create_task(media_controller.send_message, queuedata, inc_session_id=True)
+ self.mass.create_task(media_controller.send_message, data=queuedata, inc_session_id=True)
self.logger.debug(
"Enqued next track (%s) to player %s",
queue_item.name if queue_item else url,
if castplayer.cc.app_id == app_id:
return # already active
- def launched_callback() -> None:
+ def launched_callback(success: bool, response: dict[str, Any] | None) -> None:
self.mass.loop.call_soon_threadsafe(event.set)
def launch() -> None:
"metadata": cc_item["metadata"],
},
}
- self.mass.create_task(media_controller.send_message, queuedata, True)
+ self.mass.create_task(
+ media_controller.send_message, data=queuedata, inc_session_id=True
+ )
if len(getattr(media_controller.status, "items", [])) < 2:
# In flow mode, all queue tracks are sent to the player as continuous stream.
}
],
}
- self.mass.create_task(media_controller.send_message, msg, inc_session_id=True)
+ self.mass.create_task(media_controller.send_message, data=msg, inc_session_id=True)
from __future__ import annotations
import urllib.error
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
from typing import TYPE_CHECKING, Self
from uuid import UUID
@classmethod
def from_cast_info(cls: Self, cast_info: CastInfo) -> Self:
"""Instantiate ChromecastInfo from CastInfo."""
- return cls(**cast_info._asdict())
+ return cls(**asdict(cast_info))
def update(self, cast_info: CastInfo) -> None:
"""Update ChromecastInfo from CastInfo."""
- for key, value in cast_info._asdict().items():
+ for key, value in asdict(cast_info).items():
if not value:
continue
setattr(self, key, value)
"%s got new media_status for group: %s", self.castplayer.player.display_name, group_uuid
)
- def load_media_failed(self, item, error_code) -> None:
+ def load_media_failed(self, queue_item_id, error_code) -> None:
"""Call when media failed to load."""
- self.prov.logger.warning("Load media failed: %s - error code: %s", item, error_code)
+ self.prov.logger.warning(
+ "Load media failed: %s - error code: %s", queue_item_id, error_code
+ )
def invalidate(self) -> None:
"""