radio.py

from random import random

from yandex_music import Track


class Radio:
	def __init__(self, client):
		self.client = client
		self.station_id = None
		self.station_from = None

		self.play_id = None
		self.index = 0
		self.current_track = None
		self.station_tracks = None

	def start_radio(self, station_id, station_from) -> Track:
		self.station_id = station_id
		self.station_from = station_from

		# get first 5 tracks
		self.__update_radio_batch(None)

		# setup current track
		self.current_track = self.__update_current_track()
		return self.current_track

	def play_next(self) -> Track:
		# send prev track finalize info
		self.__send_play_end_track(self.current_track, self.play_id)
		self.__send_play_end_radio(self.current_track, self.station_tracks.batch_id)

		# get next index
		self.index += 1
		if self.index >= len(self.station_tracks.sequence):
			# get next 5 tracks. Set index to 0
			self.__update_radio_batch(self.current_track.track_id)

		# setup next track
		self.current_track = self.__update_current_track()
		return self.current_track

	def __update_radio_batch(self, queue=None):
		self.index = 0
		self.station_tracks = self.client.rotor_station_tracks(self.station_id, queue=queue)
		self.__send_start_radio(self.station_tracks.batch_id)

	def __update_current_track(self):
		self.play_id = self.__generate_play_id()
		track = self.client.tracks([self.station_tracks.sequence[self.index].track.track_id])[0]
		self.__send_play_start_track(track, self.play_id)
		self.__send_play_start_radio(track, self.station_tracks.batch_id)
		return track

	def __send_start_radio(self, batch_id):
		self.client.rotor_station_feedback_radio_started(
			station=self.station_id,
			from_=self.station_from,
			batch_id=batch_id
		)

	def __send_play_start_track(self, track, play_id):
		total_seconds = track.duration_ms / 1000
		self.client.play_audio(
			from_="desktop_win-home-playlist_of_the_day-playlist-default",
			track_id=track.id,
			album_id=track.albums[0].id,
			play_id=play_id,
			track_length_seconds=0,
			total_played_seconds=0,
			end_position_seconds=total_seconds
		)

	def __send_play_start_radio(self, track, batch_id):
		self.client.rotor_station_feedback_track_started(
			station=self.station_id,
			track_id=track.id,
			batch_id=batch_id
		)

	def __send_play_end_track(self, track, play_id):
		# played_seconds = 5.0
		played_seconds = track.duration_ms / 1000
		total_seconds = track.duration_ms / 1000
		self.client.play_audio(
			from_="desktop_win-home-playlist_of_the_day-playlist-default",
			track_id=track.id,
			album_id=track.albums[0].id,
			play_id=play_id,
			track_length_seconds=int(total_seconds),
			total_played_seconds=played_seconds,
			end_position_seconds=total_seconds
		)

	def __send_play_end_radio(self, track, batch_id):
		played_seconds = track.duration_ms / 1000
		self.client.rotor_station_feedback_track_finished(
			station=self.station_id,
			track_id=track.id,
			total_played_seconds=played_seconds,
			batch_id=batch_id

		)
		# self.client.rotor_station_feedback_skip(
		# 	station=self.station_id,
		# 	track_id=track.track_id,
		# 	total_played_seconds=played_seconds,
		# 	batch_id=batch_id
		# )
		pass

	@staticmethod
	def __generate_play_id():
		return "%s-%s-%s" % (int(random() * 1000), int(random() * 1000), int(random() * 1000))

radio_example.py

from math import floor
from random import random
from time import sleep

from yandex_music import Client

from radio import Radio

# API instance
client = Client(token="YOUR_TOKEN_HERE")

# Get random station
_stations = client.rotor_stations_list()
_station_random_index = floor(len(_stations) * random())
_station = _stations[_station_random_index].station
_station_id = f'{_station.id.type}:{_station.id.tag}'
_station_from = _station.id_for_from

# Radio instance
radio = Radio(client)

# start radio and get first track
first_track = radio.start_radio(_station_id, _station_from)
print("[Radio] First track is:", first_track)

# get new track every 5 sec
while True:
	sleep(5)
	next_track = radio.play_next()
	print("[Radio] Next track is:", next_track)

stream_example.py

from time import sleep

from yandex_music import Client

from radio import Radio

# API instance
client = Client(token="YOUR_API_KEY_HERE")

# get some track
track = client.tracks(['2816574:303266'])[0]
album = track.albums[0]
artist = track.artists[0]

# stream by track
_station_id, _station_from = f'track:{track.id}', 'track'

# stream by album
# _station_id, _station_from = f'album:{album.id}', 'album'

# stream by artist
# _station_id, _station_from = f'artist:{artist.id}', 'artist'

# Radio instance
radio = Radio(client)

# start radio and get first track
first_track = radio.start_radio(_station_id, _station_from)
print("[Radio] First track is:", first_track)

# get new track every 5 sec
while True:
	sleep(5)
	next_track = radio.play_next()
	print("[Radio] Next track is:", next_track)