eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Refactoring profond de pmoparadise

Browse Source
This commit is contained in:
Eric Coissac
2025-10-21 14:21:17 +02:00
parent f510e59b1a
commit 1567beee2e
19 changed files with 3960 additions and 1165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
Cargo.lock generated
View File

@@ -2476,13 +2476,18 @@ name = "pmoparadise"
version = "0.1.0"
dependencies = [
"anyhow",
"async-stream",
"async-trait",
"axum",
"bytes",
"chrono",
"claxon",
"flacenc",
"futures",
"hex",
"hound",
"pmoaudiocache",
"pmoconfig",
"pmocovers",
"pmodidl",
"pmoplaylist",
@@ -2490,13 +2495,17 @@ dependencies = [
"pmosource",
"pmoupnp",
"reqwest",
"rusqlite",
"serde",
"serde_json",
"serde_yaml",
"sha2",
"symphonia",
"tempfile",
"thiserror 1.0.69",
"tokio",
"tokio-test",
"tokio-util",
"tracing",
"tracing-subscriber",
"url",
@@ -4202,9 +4211,16 @@ dependencies = [
"serde_json",
"url",
"utoipa",
"utoipa-swagger-ui-vendored",
"zip",
]
[[package]]
name = "utoipa-swagger-ui-vendored"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e2eebbbfe4093922c2b6734d7c679ebfebd704a0d7e56dfcb0d05818ce28977d"
[[package]]
name = "uuid"
version = "1.18.1"

751
pmoapp/webapp/src/components/RadioParadiseExplorer.vue
View File

@@ -166,73 +166,221 @@
</div>
</div>
<div class="channel-tracks-section">
<div class="channel-status-section">
<div class="section-header">
<h3>🎧 Channel Tracks (Source API)</h3>
<button class="btn-secondary" @click="fetchChannelTracks" :disabled="channelBrowseLoading">
<span v-if="channelBrowseLoading">⏳ Refreshing…</span>
<span v-else>Refresh Tracks</span>
<h3>📡 Channel Status</h3>
<button class="btn-secondary" @click="fetchChannelStatus" :disabled="channelStatusLoading">
<span v-if="channelStatusLoading">⏳ Refreshing…</span>
<span v-else>Refresh Status</span>
</button>
</div>
<div v-if="channelStatusError" class="error-message">
❌ {{ channelStatusError }}
</div>
<div v-else-if="channelStatus">
<div class="status-grid">
<div class="status-card">
<div class="status-label">Channel</div>
<div class="status-value">{{ channelStatus.slug }}</div>
</div>
<div class="status-card">
<div class="status-label">Active Clients</div>
<div class="status-value">{{ channelStatus.active_clients }}</div>
</div>
<div class="status-card">
<div class="status-label">Queue Length</div>
<div class="status-value">{{ channelStatus.queue_length }}</div>
</div>
<div class="status-card">
<div class="status-label">Update ID</div>
<div class="status-value">{{ channelStatus.update_id }}</div>
</div>
<div class="status-card">
<div class="status-label">Last Change</div>
<div class="status-value">
{{ channelStatus.last_change ? formatTimestamp(new Date(channelStatus.last_change)) : '—' }}
</div>
</div>
<div class="status-card">
<div class="status-label">History Entries</div>
<div class="status-value">
{{ channelStatus.history_entries }} / {{ channelStatus.history_max_tracks }}
</div>
</div>
<div class="status-card">
<div class="status-label">Cache Collection</div>
<div class="status-value">{{ channelStatus.cache_collection_id }}</div>
</div>
<div class="status-card">
<div class="status-label">Cache Tracks</div>
<div class="status-value">
{{ channelStatus.cache_cached_tracks }} / {{ channelStatus.cache_total_tracks }}
</div>
</div>
</div>
<div class="section-meta" v-if="channelStatusLastUpdated">
Last refresh: {{ formatTimestamp(channelStatusLastUpdated) }}
</div>
</div>
</div>
<div class="channel-tracks-section">
<div class="section-header">
<h3>🎧 Live Playlist</h3>
<div class="section-actions">
<button class="btn-tertiary" @click="refreshChannelData" :disabled="channelPlaylistLoading || channelStatusLoading">
<span v-if="channelPlaylistLoading || channelStatusLoading">⏳ Refreshing…</span>
<span v-else>Refresh All</span>
</button>
<button class="btn-secondary" @click="fetchChannelPlaylist" :disabled="channelPlaylistLoading">
<span v-if="channelPlaylistLoading">⏳ Loading…</span>
<span v-else>Refresh Playlist</span>
</button>
</div>
</div>
<div class="section-meta">
<span>Object:</span>
<code>{{ channelBrowse.object_id || channelObjectId(selectedChannel) }}</code>
<span>Items:</span>
<strong>{{ channelBrowse.returned_items }}</strong>
<span v-if="channelBrowse.total">/ {{ channelBrowse.total }}</span>
<span v-if="channelBrowse.update_id">Update ID: {{ channelBrowse.update_id }}</span>
<span v-if="channelBrowseLastUpdated">
Last refresh: {{ formatTimestamp(channelBrowseLastUpdated) }}
<span>Queue length:</span>
<strong>{{ channelPlaylist.queue_length }}</strong>
<span v-if="channelPlaylist.update_id">Update ID: {{ channelPlaylist.update_id }}</span>
<span v-if="channelPlaylistLastUpdated">
Last refresh: {{ formatTimestamp(channelPlaylistLastUpdated) }}
</span>
</div>
<div v-if="channelBrowseError" class="error-message">
❌ {{ channelBrowseError }}
<div v-if="channelPlaylistError" class="error-message">
❌ {{ channelPlaylistError }}
</div>
<div v-else-if="channelBrowseLoading" class="loading-message">
⏳ Loading channel tracks…
<div v-else-if="channelPlaylistLoading" class="loading-message">
⏳ Loading playlist…
</div>
<div v-else>
<div v-if="channelBrowse.containers.length" class="sub-container-notice">
{{ channelBrowse.containers.length }} sub container(s) available.
</div>
<div v-if="channelBrowse.items.length" class="track-grid">
<div v-if="channelPlaylist.items.length" class="track-grid">
<div
v-for="item in channelBrowse.items"
:key="item.id"
:class="['track-card', { active: activeTrackId === item.id }]"
v-for="item in channelPlaylist.items"
:key="trackObjectId(item)"
:class="['track-card', { active: activeTrackId === trackObjectId(item) }]"
>
<div class="track-card-header">
<span :class="trackStatusClass(item.__status)">
{{ trackStatusLabel(item.__status) }}
<span :class="cacheStatusClass(item.cache_status)">
{{ cacheStatusLabel(item.cache_status) }}
</span>
<span v-if="item.update_id" class="track-metadata">update {{ item.update_id }}</span>
<span class="track-metadata">Pending listeners: {{ item.pending_clients }}</span>
</div>
<div class="track-headline">
<div class="track-title">{{ item.title }}</div>
<div class="track-artist">{{ item.artist || item.creator || 'Unknown artist' }}</div>
<div class="track-artist">{{ item.artist || 'Unknown artist' }}</div>
</div>
<div class="track-meta">
<span v-if="item.album">{{ item.album }}</span>
<span v-if="item.resources && item.resources.length && item.resources[0].duration">
⏱ {{ item.resources[0].duration }}
</span>
<span v-if="item.duration_ms">⏱ {{ formatDuration(item.duration_ms) }}</span>
<span v-if="item.elapsed_ms">▶️ @{{ formatDuration(item.elapsed_ms) }}</span>
<span v-if="item.started_at">🕒 {{ formatTimestamp(new Date(item.started_at)) }}</span>
<span v-if="item.cache_status?.size_bytes">💾 {{ formatBytes(item.cache_status.size_bytes) }}</span>
</div>
<div class="track-actions">
<button class="btn-secondary" @click="playTrackItem(item)">▶️ Play Track</button>
<button
class="btn-secondary"
@click="requestCacheForTrack(item)"
:disabled="trackExtrasFor(item).cacheRequestLoading"
>
<span v-if="trackExtrasFor(item).cacheRequestLoading">⏳ Caching…</span>
<span v-else>💾 Request Cache</span>
</button>
<button
class="btn-tertiary"
@click="refreshTrackCacheStatus(item)"
:disabled="trackExtrasFor(item).cacheStatusLoading"
>
<span v-if="trackExtrasFor(item).cacheStatusLoading">⏳ Updating…</span>
<span v-else>🔄 Cache Status</span>
</button>
<button
class="btn-tertiary"
@click="fetchTrackFormats(item)"
:disabled="trackExtrasFor(item).formatsLoading"
>
<span v-if="trackExtrasFor(item).formatsLoading">⏳ Formats…</span>
<span v-else>🎚️ Formats</span>
</button>
<a
v-for="resource in item.resources"
:key="resource.url"
:href="resource.url"
v-if="trackExtrasFor(item).uri"
:href="trackExtrasFor(item).uri"
target="_blank"
class="stream-link"
>
Open resource
Open resolved URI
</a>
</div>
<div v-if="trackExtrasFor(item).cacheRequestMessage" class="inline-success">
✅ {{ trackExtrasFor(item).cacheRequestMessage }}
</div>
<div v-if="trackExtrasFor(item).cacheError" class="inline-error">
❌ {{ trackExtrasFor(item).cacheError }}
</div>
<div v-if="trackExtrasFor(item).resolveError" class="inline-error">
❌ Resolve error: {{ trackExtrasFor(item).resolveError }}
</div>
<div v-if="trackExtrasFor(item).formatsError" class="inline-error">
❌ Formats error: {{ trackExtrasFor(item).formatsError }}
</div>
<div
v-if="trackExtrasFor(item).formats && trackExtrasFor(item).formats.length"
class="formats-list"
>
<div
v-for="format in trackExtrasFor(item).formats"
:key="format.format_id"
class="format-row"
>
<strong>{{ format.format_id }}</strong>
<span>{{ format.mime_type }}</span>
<span v-if="format.sample_rate">{{ format.sample_rate }} Hz</span>
<span v-if="format.bit_depth">{{ format.bit_depth }} bit</span>
<span v-if="format.bitrate">{{ format.bitrate }} kbps</span>
<span v-if="format.channels">{{ format.channels }} ch</span>
</div>
</div>
</div>
</div>
<div v-else class="empty-placeholder">
No cached tracks yet for this channel. Refresh after playback starts.
No tracks currently queued. Try refreshing after playback starts.
</div>
</div>
</div>
<div class="channel-history-section">
<div class="section-header">
<h3>🕰️ Recent History</h3>
<button class="btn-secondary" @click="fetchChannelHistory" :disabled="channelHistoryLoading">
<span v-if="channelHistoryLoading">⏳ Loading…</span>
<span v-else>Refresh History</span>
</button>
</div>
<div class="section-meta" v-if="channelHistoryLastUpdated">
Last refresh: {{ formatTimestamp(channelHistoryLastUpdated) }}
</div>
<div v-if="channelHistoryError" class="error-message">
❌ {{ channelHistoryError }}
</div>
<div v-else-if="channelHistoryLoading" class="loading-message">
⏳ Loading history…
</div>
<div v-else class="history-list">
<div v-if="channelHistory.length === 0" class="empty-placeholder">
No history entries yet.
</div>
<div
v-for="entry in channelHistory"
:key="`${entry.track_id}-${entry.started_at}`"
class="history-item"
>
<div class="history-title">{{ entry.title }}</div>
<div class="history-meta">
<span>{{ entry.artist }}</span>
<span v-if="entry.album">• {{ entry.album }}</span>
<span>• {{ formatDuration(entry.duration_ms) }}</span>
<span>• {{ formatTimestamp(new Date(entry.started_at)) }}</span>
</div>
</div>
</div>
</div>
@@ -379,22 +527,31 @@ const blockSearchLoading = ref(false)
const blockSearchError = ref('')
const channelsError = ref('')
const bitratesError = ref('')
const channelBrowse = ref({
object_id: '',
containers: [],
const channelStatus = ref(null)
const channelStatusLoading = ref(false)
const channelStatusError = ref('')
const channelStatusLastUpdated = ref(null)
const channelPlaylist = ref({
items: [],
returned_containers: 0,
returned_items: 0,
total: 0,
update_id: 0
queue_length: 0,
update_id: 0,
slug: '',
channel_id: selectedChannel.value
})
const channelBrowseLoading = ref(false)
const channelBrowseError = ref('')
const channelBrowseLastUpdated = ref(null)
const channelPlaylistLoading = ref(false)
const channelPlaylistError = ref('')
const channelPlaylistLastUpdated = ref(null)
const channelHistory = ref([])
const channelHistoryLoading = ref(false)
const channelHistoryError = ref('')
const channelHistoryLastUpdated = ref(null)
const trackExtras = ref({})
let refreshTimerId = null
let channelRefreshTimerId = null
const CHANNEL_TRACKS_REFRESH_INTERVAL = 5000
let channelTracksInFlight = false
const CHANNEL_REFRESH_INTERVAL = 7000
// Format duration from milliseconds to MM:SS
function formatDuration(ms) {
@@ -444,6 +601,10 @@ function channelObjectId(channelId) {
return `${SOURCE_ID}:channel:${channelId}`
}
function trackObjectId(item) {
return item?.track_id || item?.id || item?.object_id || ''
}
function playAudio(url) {
if (!url) {
audioError.value = 'No audio URL available'
@@ -496,91 +657,311 @@ async function refreshNowPlaying() {
}
}
function deriveTrackStatus(item) {
const url = item?.resources?.[0]?.url || ''
if (!url) {
return 'pending'
}
if (url.includes('/audio/flac/') && !url.includes('#')) {
return 'cached'
}
if (url.includes('#')) {
return 'downloading'
}
return 'external'
}
function trackStatusLabel(status) {
function cacheStatusLabel(cacheInfo) {
const status = cacheInfo?.status || 'not_cached'
switch (status) {
case 'cached':
return 'Cached'
case 'downloading':
return 'Downloading'
case 'external':
return 'External'
case 'caching':
return cacheInfo?.progress != null
? `Caching ${(cacheInfo.progress * 100).toFixed(0)}%`
: 'Caching'
case 'failed':
return 'Failed'
case 'not_cached':
default:
return 'Pending'
return 'Not cached'
}
}
function trackStatusClass(status) {
function cacheStatusClass(cacheInfo) {
const status = cacheInfo?.status || 'not_cached'
return {
'status-badge': true,
cached: status === 'cached',
downloading: status === 'downloading',
external: status === 'external',
pending: status === 'pending'
caching: status === 'caching',
failed: status === 'failed',
pending: status === 'not_cached'
}
}
async function fetchChannelTracks(options = {}) {
const { silent = false } = options
if (channelTracksInFlight) {
return
function formatBytes(bytes) {
if (typeof bytes !== 'number' || !Number.isFinite(bytes) || bytes <= 0) {
return '0 B'
}
channelTracksInFlight = true
const units = ['B', 'KB', 'MB', 'GB']
let value = bytes
let unitIndex = 0
while (value >= 1024 && unitIndex < units.length - 1) {
value /= 1024
unitIndex += 1
}
return `${value.toFixed(value >= 10 || unitIndex === 0 ? 0 : 1)} ${units[unitIndex]}`
}
async function fetchChannelStatus({ silent = false } = {}) {
if (!silent) {
channelBrowseLoading.value = true
channelStatusLoading.value = true
}
channelBrowseError.value = ''
channelStatusError.value = ''
try {
const params = new URLSearchParams()
params.set('object_id', channelObjectId(selectedChannel.value))
params.set('requested_count', '0')
const response = await fetch(`${SOURCE_API_BASE}/${SOURCE_ID}/browse?${params.toString()}`)
const response = await fetch(`${API_BASE}/channels/${selectedChannel.value}/status`)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const data = await response.json()
channelBrowse.value = {
object_id: data.object_id || channelObjectId(selectedChannel.value),
containers: data.containers ?? [],
items: data.items ?? [],
returned_containers: data.returned_containers ?? (data.containers?.length ?? 0),
returned_items: data.returned_items ?? (data.items?.length ?? 0),
total: data.total ?? ((data.containers?.length ?? 0) + (data.items?.length ?? 0)),
update_id: data.update_id ?? 0
}
// annotate each item with status for UI
channelBrowse.value.items = channelBrowse.value.items.map((item) => ({
...item,
__status: deriveTrackStatus(item)
}))
channelBrowseLastUpdated.value = new Date()
channelStatus.value = await response.json()
channelStatusLastUpdated.value = new Date()
} catch (e) {
channelBrowseError.value = `Failed to load channel tracks: ${e.message}`
console.error('Error fetching channel tracks:', e)
channelStatusError.value = `Failed to load channel status: ${e.message}`
console.error('Error fetching channel status:', e)
} finally {
channelTracksInFlight = false
if (!silent) {
channelBrowseLoading.value = false
channelStatusLoading.value = false
}
}
}
async function fetchChannelPlaylist({ silent = false, limit = 24 } = {}) {
if (!silent) {
channelPlaylistLoading.value = true
}
channelPlaylistError.value = ''
try {
const params = new URLSearchParams()
if (limit != null) {
params.set('limit', String(limit))
}
const response = await fetch(
`${API_BASE}/channels/${selectedChannel.value}/playlist?${params.toString()}`
)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const data = await response.json()
channelPlaylist.value = {
...data,
items: (data.items || []).map((item) => ({
...item,
cache_status: item.cache_status || { status: 'not_cached', progress: 0 }
}))
}
channelPlaylistLastUpdated.value = new Date()
const validIds = new Set(channelPlaylist.value.items.map((item) => trackObjectId(item)).filter(Boolean))
trackExtras.value = Object.fromEntries(
Object.entries(trackExtras.value).filter(([id]) => validIds.has(id))
)
} catch (e) {
channelPlaylistError.value = `Failed to load playlist: ${e.message}`
console.error('Error fetching channel playlist:', e)
} finally {
if (!silent) {
channelPlaylistLoading.value = false
}
}
}
async function fetchChannelHistory({ silent = false, limit = 25 } = {}) {
if (!silent) {
channelHistoryLoading.value = true
}
channelHistoryError.value = ''
try {
const params = new URLSearchParams()
if (limit != null) {
params.set('limit', String(limit))
}
const response = await fetch(
`${API_BASE}/channels/${selectedChannel.value}/history?${params.toString()}`
)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const data = await response.json()
channelHistory.value = data.entries || []
channelHistoryLastUpdated.value = new Date()
} catch (e) {
channelHistoryError.value = `Failed to load history: ${e.message}`
console.error('Error fetching channel history:', e)
} finally {
if (!silent) {
channelHistoryLoading.value = false
}
}
}
async function refreshChannelData({ silent = false } = {}) {
await Promise.all([
fetchChannelStatus({ silent }),
fetchChannelPlaylist({ silent }),
fetchChannelHistory({ silent })
])
}
function updateTrackExtras(trackId, patch) {
if (!trackId) {
return
}
const current = trackExtras.value[trackId] || {
uri: '',
lastResolvedAt: null,
resolving: false,
resolveError: '',
formats: [],
formatsLoading: false,
formatsError: '',
cacheRequestLoading: false,
cacheRequestMessage: '',
cacheError: '',
cacheStatusLoading: false
}
trackExtras.value = {
...trackExtras.value,
[trackId]: {
...current,
...patch
}
}
}
function trackExtrasFor(item) {
const trackId = trackObjectId(item)
return trackExtras.value[trackId] || {}
}
async function resolveTrackUri(trackId) {
if (!trackId) {
throw new Error('Missing track identifier')
}
updateTrackExtras(trackId, { resolving: true, resolveError: '' })
try {
const params = new URLSearchParams()
params.set('object_id', trackId)
const response = await fetch(`${SOURCE_API_BASE}/${SOURCE_ID}/resolve?${params.toString()}`)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const data = await response.json()
updateTrackExtras(trackId, {
resolving: false,
uri: data.uri,
lastResolvedAt: new Date()
})
return data.uri
} catch (e) {
updateTrackExtras(trackId, { resolving: false, resolveError: e.message })
throw e
}
}
async function refreshTrackCacheStatus(item) {
const trackId = trackObjectId(item)
if (!trackId) {
return
}
updateTrackExtras(trackId, { cacheStatusLoading: true, cacheError: '' })
try {
const params = new URLSearchParams()
params.set('object_id', trackId)
const response = await fetch(
`${SOURCE_API_BASE}/${SOURCE_ID}/cache/status?${params.toString()}`
)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const data = await response.json()
const status = data.status || { status: 'not_cached' }
channelPlaylist.value = {
...channelPlaylist.value,
items: channelPlaylist.value.items.map((entry) =>
trackObjectId(entry) === trackId ? { ...entry, cache_status: status } : entry
)
}
updateTrackExtras(trackId, { cacheStatusLoading: false, cacheError: '', cacheStatus: status })
} catch (e) {
updateTrackExtras(trackId, { cacheStatusLoading: false, cacheError: e.message })
console.error('Error refreshing cache status:', e)
}
}
async function requestCacheForTrack(item) {
const trackId = trackObjectId(item)
if (!trackId) {
return
}
updateTrackExtras(trackId, {
cacheRequestLoading: true,
cacheRequestMessage: '',
cacheError: ''
})
try {
const response = await fetch(`${SOURCE_API_BASE}/${SOURCE_ID}/cache`, {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({ object_id: trackId })
})
if (!response.ok) {
const body = await response.json().catch(() => ({}))
const message = body?.error || response.statusText
throw new Error(message)
}
const data = await response.json()
const status = data.status || data?.cache_status
if (status) {
channelPlaylist.value = {
...channelPlaylist.value,
items: channelPlaylist.value.items.map((entry) =>
trackObjectId(entry) === trackId ? { ...entry, cache_status: status } : entry
)
}
}
updateTrackExtras(trackId, {
cacheRequestLoading: false,
cacheRequestMessage: 'Cache request accepted'
})
} catch (e) {
updateTrackExtras(trackId, {
cacheRequestLoading: false,
cacheRequestMessage: '',
cacheError: e.message
})
console.error('Error requesting cache:', e)
} finally {
await refreshTrackCacheStatus(item)
}
}
async function fetchTrackFormats(item) {
const trackId = trackObjectId(item)
if (!trackId) {
return
}
const extras = trackExtrasFor(item)
if (extras.formats?.length && !extras.formatsError) {
return
}
updateTrackExtras(trackId, { formatsLoading: true, formatsError: '' })
try {
const params = new URLSearchParams()
params.set('object_id', trackId)
const response = await fetch(
`${SOURCE_API_BASE}/${SOURCE_ID}/formats?${params.toString()}`
)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const data = await response.json()
updateTrackExtras(trackId, {
formatsLoading: false,
formats: data.formats || []
})
} catch (e) {
updateTrackExtras(trackId, { formatsLoading: false, formatsError: e.message })
console.error('Error fetching track formats:', e)
}
}
// Fetch available channels
async function fetchChannels() {
try {
@@ -635,15 +1016,17 @@ function selectChannel(channelId) {
}
async function changeChannel() {
trackExtras.value = {}
await refreshNowPlaying()
await fetchChannelTracks()
await refreshChannelData()
blockSearchResult.value = null
blockSearchError.value = ''
}
async function changeBitrate() {
trackExtras.value = {}
await refreshNowPlaying()
await fetchChannelTracks()
await refreshChannelData()
blockSearchResult.value = null
blockSearchError.value = ''
}
@@ -744,16 +1127,38 @@ function clearBlockSearch() {
blockSearchError.value = ''
}
function playTrackItem(item) {
const resource = item?.resources?.find(res => res.url)
if (!resource) {
audioError.value = 'No audio resource available for this track'
async function playTrackItem(item) {
const trackId = trackObjectId(item)
if (!trackId) {
audioError.value = 'Unable to determine track identifier'
isPlaying.value = false
return
}
activeTrackId.value = item.id
playAudio(resource.url)
try {
let uri = null
try {
uri = await resolveTrackUri(trackId)
} catch (resolveError) {
console.warn('Falling back to direct resource due to resolve error:', resolveError)
}
if (!uri) {
const fallback = item?.resources?.find((res) => res.url)?.url
uri = fallback
}
if (!uri) {
throw new Error('No audio resource available for this track')
}
activeTrackId.value = trackId
playAudio(uri)
} catch (e) {
audioError.value = e.message
isPlaying.value = false
activeTrackId.value = null
}
}
// Initialize on mount
@@ -761,7 +1166,7 @@ onMounted(async () => {
await fetchChannels()
await fetchBitrates()
await refreshNowPlaying()
await fetchChannelTracks()
await refreshChannelData()
// Auto-refresh every 30 seconds
refreshTimerId = window.setInterval(() => {
@@ -772,10 +1177,10 @@ onMounted(async () => {
// Auto-refresh channel tracks every few seconds
channelRefreshTimerId = window.setInterval(() => {
if (!channelBrowseLoading.value) {
fetchChannelTracks({ silent: true })
if (!channelPlaylistLoading.value && !channelStatusLoading.value) {
refreshChannelData({ silent: true })
}
}, CHANNEL_TRACKS_REFRESH_INTERVAL)
}, CHANNEL_REFRESH_INTERVAL)
})
onUnmounted(() => {
@@ -1326,6 +1731,13 @@ onUnmounted(() => {
color: #9aa0a6;
}
.section-actions {
display: flex;
gap: 10px;
flex-wrap: wrap;
align-items: center;
}
.loading-message {
margin-top: 16px;
color: #9aa0a6;
@@ -1364,6 +1776,36 @@ onUnmounted(() => {
transition: border-color 0.2s, box-shadow 0.2s;
}
.status-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(160px, 1fr));
gap: 12px;
margin-top: 12px;
}
.status-card {
background: rgba(255, 255, 255, 0.03);
border: 1px solid rgba(255, 255, 255, 0.04);
border-radius: 8px;
padding: 12px;
display: flex;
flex-direction: column;
gap: 4px;
}
.status-label {
font-size: 0.75rem;
text-transform: uppercase;
letter-spacing: 0.06em;
color: #9aa0a6;
}
.status-value {
font-size: 1.05rem;
font-weight: 600;
color: #f5f5f5;
}
.track-card-header {
display: flex;
justify-content: space-between;
@@ -1393,22 +1835,22 @@ onUnmounted(() => {
color: #2ecc71;
}
.status-badge.downloading {
.status-badge.caching {
background: rgba(255, 193, 7, 0.15);
border-color: rgba(255, 193, 7, 0.4);
color: #ffc107;
}
.status-badge.external {
background: rgba(0, 212, 255, 0.12);
border-color: rgba(0, 212, 255, 0.4);
color: #00d4ff;
.status-badge.failed {
background: rgba(255, 87, 34, 0.18);
border-color: rgba(255, 87, 34, 0.5);
color: #ff7043;
}
.status-badge.pending {
background: rgba(255, 87, 34, 0.15);
border-color: rgba(255, 87, 34, 0.4);
color: #ff6d3a;
background: rgba(0, 212, 255, 0.12);
border-color: rgba(0, 212, 255, 0.4);
color: #00d4ff;
}
.track-metadata {
@@ -1452,6 +1894,63 @@ onUnmounted(() => {
align-items: center;
}
.inline-error {
margin-top: 6px;
font-size: 0.8rem;
color: #ff6b6b;
}
.inline-success {
margin-top: 6px;
font-size: 0.8rem;
color: #2ecc71;
}
.formats-list {
margin-top: 10px;
border-top: 1px solid rgba(255, 255, 255, 0.08);
padding-top: 8px;
display: flex;
flex-direction: column;
gap: 4px;
font-size: 0.8rem;
color: #9aa0a6;
}
.format-row {
display: flex;
gap: 8px;
flex-wrap: wrap;
}
.history-list {
display: flex;
flex-direction: column;
gap: 12px;
margin-top: 12px;
}
.history-item {
padding: 12px;
border-radius: 8px;
background: rgba(0, 0, 0, 0.25);
border: 1px solid rgba(255, 255, 255, 0.05);
}
.history-title {
font-weight: 600;
color: #f5f5f5;
}
.history-meta {
margin-top: 4px;
font-size: 0.8rem;
color: #9aa0a6;
display: flex;
gap: 6px;
flex-wrap: wrap;
}
.audio-player-container {
margin: 12px 0 24px;
padding: 16px;

32
pmoconfig/src/pmomusic.yaml
View File

@@ -12,7 +12,33 @@ host:
min_level: "INFO"
mediarenderer:
mpd_renderer:
mediaserver:
qobuz:
udn: "uuid:28963b75-4c5f-4da7-b10e-ffafd"
sources:
radio_paradise:
enabled: true
cache:
max_blocks_remembered: 5
track_id_hash_bytes: 512
history:
max_tracks: 100
persistence_backend: sqlite
database_path: /var/lib/pmo/paradise_history.db
activity:
cooling_timeout_seconds: 180
polling:
interval_high_buffer: 120
interval_medium_buffer: 60
interval_low_buffer: 20
backoff_on_error:
initial: 20
max: 300
multiplier: 2.0
stream:
metadata_format: icy
enable_gapless: true
buffer_size_bytes: 65536
api:
base_url: https://api.radioparadise.com
timeout_seconds: 30
user_agent: "PMO-RadioParadise/1.0"

144
pmoparadise/.github/workflows/ci.yml vendored
View File

@@ -1,144 +0,0 @@
name: CI
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main, develop ]
env:
CARGO_TERM_COLOR: always
RUST_BACKTRACE: 1
jobs:
test:
name: Test
runs-on: ${{ matrix.os }}
strategy:
matrix:
os: [ubuntu-latest, macos-latest, windows-latest]
rust: [stable, beta]
steps:
- uses: actions/checkout@v3
- name: Install Rust
uses: dtolnay/rust-toolchain@master
with:
toolchain: ${{ matrix.rust }}
- name: Cache cargo registry
uses: actions/cache@v3
with:
path: ~/.cargo/registry
key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
- name: Cache cargo index
uses: actions/cache@v3
with:
path: ~/.cargo/git
key: ${{ runner.os }}-cargo-git-${{ hashFiles('**/Cargo.lock') }}
- name: Cache cargo build
uses: actions/cache@v3
with:
path: target
key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }}
- name: Run tests (default features)
run: cargo test --verbose
- name: Run tests (per-track feature)
run: cargo test --verbose --features per-track
- name: Run tests (all features)
run: cargo test --verbose --all-features
fmt:
name: Rustfmt
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
with:
components: rustfmt
- name: Check formatting
run: cargo fmt --all -- --check
clippy:
name: Clippy
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
with:
components: clippy
- name: Run clippy (default features)
run: cargo clippy --all-targets -- -D warnings
- name: Run clippy (all features)
run: cargo clippy --all-targets --all-features -- -D warnings
doc:
name: Documentation
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
- name: Build documentation
run: cargo doc --no-deps --all-features
env:
RUSTDOCFLAGS: -D warnings
build:
name: Build
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
- name: Build (default features)
run: cargo build --verbose
- name: Build (no default features)
run: cargo build --verbose --no-default-features
- name: Build (per-track feature)
run: cargo build --verbose --features per-track
- name: Build (all features)
run: cargo build --verbose --all-features
- name: Build release
run: cargo build --release --verbose
coverage:
name: Code Coverage
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
- name: Install cargo-tarpaulin
run: cargo install cargo-tarpaulin
- name: Generate coverage
run: cargo tarpaulin --verbose --all-features --workspace --timeout 120 --out Xml
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
files: ./cobertura.xml
fail_ci_if_error: false

BIN
pmoparadise/.pmomusic_audio/cache.db
View File

Binary file not shown.

13
pmoparadise/Cargo.toml
View File

@@ -19,6 +19,16 @@ tokio = { version = "1", features = ["full"] }
# Sérialisation/Désérialisation JSON
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
serde_yaml = "0.9"
# Helpers
chrono = { version = "0.4", features = ["serde"] }
async-trait = "0.1"
sha2 = "0.10"
hex = "0.4"
tokio-util = { version = "0.7", features = ["io"] }
async-stream = "0.3"
rusqlite = { version = "0.37", features = ["bundled"] }
# Gestion des erreurs
thiserror = "1.0"
@@ -54,6 +64,7 @@ pmosource = { path = "../pmosource" }
# Playlist management for FIFO support
pmoplaylist = { path = "../pmoplaylist" }
pmoconfig = { path = "../pmoconfig" }
# Cache support (OBLIGATOIRE - architecture refactorisée)
pmocovers = { path = "../pmocovers" }
@@ -70,7 +81,7 @@ metadata-only = []
# Active le décodage FLAC par-track
per-track = ["dep:claxon", "dep:hound", "dep:tempfile"]
# Active l'API REST pmoserver
pmoserver = ["dep:pmoserver", "dep:utoipa", "dep:axum"]
pmoserver = ["dep:pmoserver", "dep:utoipa", "dep:axum", "server"]
# Active le media server UPnP (includes pmoserver)
mediaserver = ["dep:pmoupnp", "dep:pmodidl", "dep:uuid", "pmoserver"]
# Feature pour activer le support serveur (cache registry)

48
pmoparadise/src/client.rs
View File

@@ -6,11 +6,30 @@ use reqwest::Client;
use std::time::Duration;
use url::Url;
fn normalize_base_url(base: &str) -> String {
/// Default Radio Paradise API base URL
pub const DEFAULT_API_BASE: &str = "https://api.radioparadise.com/api";
/// Default block base URL pattern
pub const DEFAULT_BLOCK_BASE: &str = "https://apps.radioparadise.com/blocks/chan/0";
/// Default image base URL
pub const DEFAULT_IMAGE_BASE: &str = "https://img.radioparadise.com/";
/// Default timeout for metadata HTTP requests
pub const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 30;
/// Default timeout for large block downloads/streams
pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180;
/// Default User-Agent
pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0";
fn normalize_cover_base_url(base: &str) -> String {
let mut normalized = base.trim().to_string();
if normalized.is_empty() {
return "https://img.radioparadise.com/".to_string();
return DEFAULT_IMAGE_BASE.to_string();
}
if normalized.starts_with("//") {
@@ -38,29 +57,12 @@ fn resolve_cover_with_base(base: &str, cover_path: &str) -> Result<Url> {
return Ok(Url::parse(&url)?);
}
let base = normalize_base_url(base);
let base = normalize_cover_base_url(base);
let base_url = Url::parse(&base)?;
Ok(base_url.join(cover_path)?)
}
/// Default Radio Paradise API base URL
pub const DEFAULT_API_BASE: &str = "https://api.radioparadise.com/api";
/// Default block base URL pattern
pub const DEFAULT_BLOCK_BASE: &str = "https://apps.radioparadise.com/blocks/chan/0";
/// Default image base URL
pub const DEFAULT_IMAGE_BASE: &str = "https://img.radioparadise.com/";
/// Default timeout for metadata HTTP requests
pub const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 30;
/// Default timeout for large block downloads/streams
pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180;
/// Default User-Agent
pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0";
/// Radio Paradise HTTP client
///
@@ -116,7 +118,7 @@ impl RadioParadiseClient {
client,
api_base: DEFAULT_API_BASE.to_string(),
block_base: DEFAULT_BLOCK_BASE.to_string(),
image_base: normalize_base_url(DEFAULT_IMAGE_BASE),
image_base: normalize_cover_base_url(DEFAULT_IMAGE_BASE),
bitrate: Bitrate::default(),
channel: 0,
request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
@@ -223,7 +225,7 @@ impl RadioParadiseClient {
// Set image_base if not provided
if let Some(ref mut base) = block.image_base {
*base = normalize_base_url(base);
*base = normalize_cover_base_url(base);
} else {
block.image_base = Some(self.image_base.clone());
}
@@ -439,7 +441,7 @@ impl ClientBuilder {
} else {
self.block_base.clone()
};
let image_base = normalize_base_url(&self.image_base);
let image_base = normalize_cover_base_url(&self.image_base);
Ok(RadioParadiseClient {
client,

1
pmoparadise/src/lib.rs
View File

@@ -239,6 +239,7 @@
pub mod client;
pub mod error;
pub mod models;
pub mod paradise;
pub mod source;
pub mod stream;

395
pmoparadise/src/paradise/channel.rs Normal file
View File

@@ -0,0 +1,395 @@
//! Channel orchestration primitives.
//!
//! This module wires together configuration, playlists, workers and client
//! tracking for a single Radio Paradise channel. The implementation is still
//! a scaffolding of the final behaviour; commands sent to the worker are
//! logged but not yet executing the full download/buffering pipeline.
use super::config::RadioParadiseConfig;
use super::history::HistoryBackend;
use super::playlist::{PlaylistEntry, SharedPlaylist};
use super::worker::{ParadiseWorker, WorkerCommand};
use crate::client::RadioParadiseClient;
use anyhow::{Context, Result};
use async_stream::try_stream;
use bytes::Bytes;
use futures::{stream::BoxStream, StreamExt};
use pmosource::SourceCacheManager;
use std::fmt;
use std::str::FromStr;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::fs::File;
use tokio::sync::{mpsc, Mutex};
use tokio_util::io::ReaderStream;
use tracing::warn;
/// Logical identifier for a Radio Paradise channel.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ParadiseChannelKind {
Main,
Mellow,
Rock,
Eclectic,
}
impl ParadiseChannelKind {
pub const fn id(self) -> u8 {
match self {
Self::Main => 0,
Self::Mellow => 1,
Self::Rock => 2,
Self::Eclectic => 3,
}
}
pub const fn slug(self) -> &'static str {
match self {
Self::Main => "main",
Self::Mellow => "mellow",
Self::Rock => "rock",
Self::Eclectic => "eclectic",
}
}
pub const fn display_name(self) -> &'static str {
match self {
Self::Main => "Main Mix",
Self::Mellow => "Mellow Mix",
Self::Rock => "Rock Mix",
Self::Eclectic => "Eclectic Mix",
}
}
pub const fn description(self) -> &'static str {
match self {
Self::Main => "Eclectic mix of rock, world, electronica, and more",
Self::Mellow => "Mellower, less aggressive music",
Self::Rock => "Heavier, more guitar-driven music",
Self::Eclectic => "Curated worldwide selection",
}
}
}
impl FromStr for ParadiseChannelKind {
type Err = anyhow::Error;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
match s.to_ascii_lowercase().as_str() {
"main" | "0" => Ok(Self::Main),
"mellow" | "1" => Ok(Self::Mellow),
"rock" | "2" => Ok(Self::Rock),
"eclectic" | "3" => Ok(Self::Eclectic),
other => Err(anyhow::anyhow!("Unknown Radio Paradise channel: {}", other)),
}
}
}
/// Metadata descriptor for a channel.
#[derive(Debug, Clone, Copy)]
pub struct ChannelDescriptor {
pub kind: ParadiseChannelKind,
pub id: u8,
pub slug: &'static str,
pub display_name: &'static str,
pub description: &'static str,
}
impl ChannelDescriptor {
pub const fn new(kind: ParadiseChannelKind) -> Self {
Self {
id: kind.id(),
slug: kind.slug(),
display_name: kind.display_name(),
description: kind.description(),
kind,
}
}
}
pub const ALL_CHANNELS: [ChannelDescriptor; 4] = [
ChannelDescriptor::new(ParadiseChannelKind::Main),
ChannelDescriptor::new(ParadiseChannelKind::Mellow),
ChannelDescriptor::new(ParadiseChannelKind::Rock),
ChannelDescriptor::new(ParadiseChannelKind::Eclectic),
];
/// Public handle to interact with a channel.
#[derive(Clone)]
pub struct ParadiseChannel {
inner: Arc<ParadiseChannelInner>,
}
struct ParadiseChannelInner {
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
config: Arc<RadioParadiseConfig>,
playlist: SharedPlaylist,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
active_clients: AtomicUsize,
worker_tx: mpsc::Sender<WorkerCommand>,
worker: Mutex<Option<ParadiseWorker>>,
}
impl fmt::Debug for ParadiseChannel {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ParadiseChannel")
.field("slug", &self.inner.descriptor.slug)
.field(
"active_clients",
&self.inner.active_clients.load(Ordering::SeqCst),
)
.finish()
}
}
impl ParadiseChannel {
#[allow(clippy::too_many_arguments)]
pub fn new(
descriptor: ChannelDescriptor,
base_client: RadioParadiseClient,
config: Arc<RadioParadiseConfig>,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
) -> Result<Self> {
let client = base_client.clone_with_channel(descriptor.id);
let playlist = SharedPlaylist::new(config.history.max_tracks);
let (worker, worker_tx) = ParadiseWorker::spawn(
descriptor,
client.clone(),
config.clone(),
playlist.clone(),
history.clone(),
cache_manager.clone(),
);
Ok(Self {
inner: Arc::new(ParadiseChannelInner {
descriptor,
client,
config,
playlist,
history,
cache_manager,
active_clients: AtomicUsize::new(0),
worker_tx,
worker: Mutex::new(Some(worker)),
}),
})
}
pub fn descriptor(&self) -> ChannelDescriptor {
self.inner.descriptor
}
pub fn playlist(&self) -> &SharedPlaylist {
&self.inner.playlist
}
pub fn config(&self) -> &Arc<RadioParadiseConfig> {
&self.inner.config
}
pub fn history_backend(&self) -> &Arc<dyn HistoryBackend> {
&self.inner.history
}
pub fn cache_manager(&self) -> Arc<SourceCacheManager> {
self.inner.cache_manager.clone()
}
pub fn client(&self) -> &RadioParadiseClient {
&self.inner.client
}
pub fn active_client_count(&self) -> usize {
self.inner.active_clients.load(Ordering::SeqCst)
}
pub async fn connect_client(
&self,
client_id: impl Into<String>,
) -> Result<ParadiseClientStream> {
let client_id = client_id.into();
self.inner.active_clients.fetch_add(1, Ordering::SeqCst);
if let Err(err) = self
.inner
.worker_tx
.send(WorkerCommand::ClientConnected {
client_id: client_id.clone(),
})
.await
{
self.inner.active_clients.fetch_sub(1, Ordering::SeqCst);
return Err(anyhow::anyhow!("worker unavailable: {}", err));
}
self.inner.playlist.increment_all_pending().await;
self.ensure_started().await?;
Ok(ParadiseClientStream::new(self.clone(), client_id))
}
pub async fn disconnect_client(&self, client_id: impl Into<String>) -> Result<()> {
let client_id = client_id.into();
self.inner.active_clients.fetch_sub(1, Ordering::SeqCst);
self.inner
.worker_tx
.send(WorkerCommand::ClientDisconnected { client_id })
.await
.context("failed to notify worker of client disconnection")?;
Ok(())
}
pub async fn ensure_started(&self) -> Result<()> {
self.inner
.worker_tx
.send(WorkerCommand::EnsureReady)
.await
.context("failed to schedule worker warmup")
}
pub async fn shutdown(&self) -> Result<()> {
self.inner
.worker_tx
.send(WorkerCommand::Shutdown)
.await
.ok();
let mut guard = self.inner.worker.lock().await;
if let Some(worker) = guard.take() {
worker
.wait()
.await
.context("failed to join worker task")
.map(|_| ())
} else {
Ok(())
}
}
pub async fn mark_track_completed(&self, track: &Arc<PlaylistEntry>) {
let remaining = track.decrement_clients();
if remaining > 0 {
return;
}
if let Some(removed) = self
.inner
.playlist
.pop_front_matching(&track.track_id)
.await
{
if let Err(err) = self.inner.history.append(removed.as_history_entry()).await {
warn!(
channel = self.inner.descriptor.slug,
"Failed to persist history entry: {err:?}"
);
}
if let Err(err) = self
.inner
.history
.truncate(self.inner.config.history.max_tracks)
.await
{
warn!(
channel = self.inner.descriptor.slug,
"Failed to truncate history: {err:?}"
);
}
let history_entry = removed.as_history_entry();
self.inner.playlist.push_history_entry(history_entry).await;
}
}
}
/// Placeholder stream handle for per-client playback.
#[derive(Debug, Clone)]
pub struct ParadiseClientStream {
channel: ParadiseChannel,
client_id: String,
}
impl ParadiseClientStream {
fn new(channel: ParadiseChannel, client_id: String) -> Self {
Self { channel, client_id }
}
pub fn client_id(&self) -> &str {
&self.client_id
}
pub fn channel(&self) -> ParadiseChannel {
self.channel.clone()
}
pub fn into_byte_stream(self) -> BoxStream<'static, Result<Bytes, anyhow::Error>> {
let channel = self.channel.clone();
let stream = try_stream! {
channel.ensure_started().await?;
let mut index = 0usize;
loop {
let entries = channel.playlist().active_snapshot().await;
if index >= entries.len() {
channel.ensure_started().await?;
channel.playlist().wait_for_track_count(index).await;
continue;
}
let entry = entries[index].clone();
index += 1;
let audio_pk = entry
.audio_pk
.clone()
.ok_or_else(|| anyhow::anyhow!("Audio not cached yet"))?;
channel
.cache_manager()
.wait_audio_ready(&audio_pk)
.await
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
let file_path = if let Some(path) = entry.file_path.clone() {
path
} else {
channel
.cache_manager()
.audio_file_path(&audio_pk)
.await
.ok_or_else(|| anyhow::anyhow!("Audio file path unavailable"))?
};
let file = File::open(&file_path).await?;
let mut reader = ReaderStream::new(file);
while let Some(chunk) = reader.next().await {
let bytes = chunk?;
yield bytes;
}
channel.mark_track_completed(&entry).await;
}
};
stream.boxed()
}
}
impl Drop for ParadiseClientStream {
fn drop(&mut self) {
let channel = self.channel.clone();
let client_id = self.client_id.clone();
let slug = channel.descriptor().slug;
tokio::spawn(async move {
if let Err(err) = channel.disconnect_client(client_id).await {
warn!(channel = slug, "Failed to disconnect client: {err:?}");
}
});
}
}

317
pmoparadise/src/paradise/config.rs Normal file
View File

@@ -0,0 +1,317 @@
//! Configuration structures for the Radio Paradise orchestration layer.
//!
//! The YAML schema is described in the functional specification. We expose
//! strongly typed structs with sensible defaults so the rest of the crate can
//! depend on a stable configuration shape irrespective of how the data is
//! loaded (embedded defaults, pmoconfig overrides, tests, etc.).
use serde::{Deserialize, Serialize};
use std::time::Duration;
/// Top-level configuration block.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RadioParadiseConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default)]
pub channels: Vec<String>,
#[serde(default)]
pub cache: CacheConfig,
#[serde(default)]
pub history: HistoryConfig,
#[serde(default)]
pub activity: ActivityConfig,
#[serde(default)]
pub polling: PollingConfig,
#[serde(default)]
pub stream: StreamConfig,
#[serde(default)]
pub api: ApiConfig,
}
impl Default for RadioParadiseConfig {
fn default() -> Self {
Self {
enabled: true,
channels: vec![
"main".to_string(),
"mellow".to_string(),
"rock".to_string(),
"eclectic".to_string(),
],
cache: CacheConfig::default(),
history: HistoryConfig::default(),
activity: ActivityConfig::default(),
polling: PollingConfig::default(),
stream: StreamConfig::default(),
api: ApiConfig::default(),
}
}
}
/// Cache related parameters.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheConfig {
#[serde(default = "CacheConfig::default_max_blocks")]
pub max_blocks_remembered: usize,
#[serde(default = "CacheConfig::default_track_id_bytes")]
pub track_id_hash_bytes: usize,
}
impl CacheConfig {
const fn default_max_blocks() -> usize {
5
}
const fn default_track_id_bytes() -> usize {
512
}
}
impl Default for CacheConfig {
fn default() -> Self {
Self {
max_blocks_remembered: Self::default_max_blocks(),
track_id_hash_bytes: Self::default_track_id_bytes(),
}
}
}
/// Persisted history tuning.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HistoryConfig {
#[serde(default = "HistoryConfig::default_max_tracks")]
pub max_tracks: usize,
#[serde(default)]
pub persistence_backend: HistoryBackendKind,
#[serde(default = "HistoryConfig::default_database_path")]
pub database_path: String,
}
impl HistoryConfig {
const fn default_max_tracks() -> usize {
100
}
fn default_database_path() -> String {
"/var/lib/pmo/paradise_history.db".to_string()
}
}
impl Default for HistoryConfig {
fn default() -> Self {
Self {
max_tracks: Self::default_max_tracks(),
persistence_backend: HistoryBackendKind::Sqlite,
database_path: Self::default_database_path(),
}
}
}
impl RadioParadiseConfig {
pub fn load_from_pmoconfig() -> anyhow::Result<Self> {
let cfg = pmoconfig::get_config();
match cfg.get_value(&["sources", "radio_paradise"]) {
Ok(value) => Ok(serde_yaml::from_value(value).unwrap_or_default()),
Err(_) => Ok(Self::default()),
}
}
}
/// Backend selection for history persistence.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum HistoryBackendKind {
#[default]
Sqlite,
Json,
}
/// Activity lifecycle tuning.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActivityConfig {
#[serde(default = "ActivityConfig::default_cooling_timeout")]
pub cooling_timeout_seconds: u64,
}
impl ActivityConfig {
const fn default_cooling_timeout() -> u64 {
180
}
pub fn cooling_timeout(&self) -> Duration {
Duration::from_secs(self.cooling_timeout_seconds)
}
}
impl Default for ActivityConfig {
fn default() -> Self {
Self {
cooling_timeout_seconds: Self::default_cooling_timeout(),
}
}
}
/// Polling strategy configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PollingConfig {
#[serde(default = "PollingConfig::default_interval_high")]
pub interval_high_buffer: u64,
#[serde(default = "PollingConfig::default_interval_medium")]
pub interval_medium_buffer: u64,
#[serde(default = "PollingConfig::default_interval_low")]
pub interval_low_buffer: u64,
#[serde(default)]
pub backoff_on_error: PollingBackoffConfig,
}
impl PollingConfig {
const fn default_interval_high() -> u64 {
120
}
const fn default_interval_medium() -> u64 {
60
}
const fn default_interval_low() -> u64 {
20
}
pub fn high_interval(&self) -> Duration {
Duration::from_secs(self.interval_high_buffer)
}
pub fn medium_interval(&self) -> Duration {
Duration::from_secs(self.interval_medium_buffer)
}
pub fn low_interval(&self) -> Duration {
Duration::from_secs(self.interval_low_buffer)
}
}
impl Default for PollingConfig {
fn default() -> Self {
Self {
interval_high_buffer: Self::default_interval_high(),
interval_medium_buffer: Self::default_interval_medium(),
interval_low_buffer: Self::default_interval_low(),
backoff_on_error: PollingBackoffConfig::default(),
}
}
}
/// Backoff policy for API errors.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PollingBackoffConfig {
#[serde(default = "PollingBackoffConfig::default_initial")]
pub initial: u64,
#[serde(default = "PollingBackoffConfig::default_max")]
pub max: u64,
#[serde(default = "PollingBackoffConfig::default_multiplier")]
pub multiplier: f32,
}
impl PollingBackoffConfig {
const fn default_initial() -> u64 {
20
}
const fn default_max() -> u64 {
300
}
const fn default_multiplier() -> f32 {
2.0
}
}
impl Default for PollingBackoffConfig {
fn default() -> Self {
Self {
initial: Self::default_initial(),
max: Self::default_max(),
multiplier: Self::default_multiplier(),
}
}
}
/// Streaming pipeline configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamConfig {
#[serde(default = "StreamConfig::default_metadata_format")]
pub metadata_format: MetadataFormat,
#[serde(default)]
pub enable_gapless: bool,
#[serde(default = "StreamConfig::default_buffer_size")]
pub buffer_size_bytes: usize,
}
impl StreamConfig {
fn default_metadata_format() -> MetadataFormat {
MetadataFormat::Icy
}
const fn default_buffer_size() -> usize {
64 * 1024
}
}
impl Default for StreamConfig {
fn default() -> Self {
Self {
metadata_format: MetadataFormat::Icy,
enable_gapless: true,
buffer_size_bytes: Self::default_buffer_size(),
}
}
}
/// Metadata transport for streaming.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum MetadataFormat {
Icy,
#[serde(other)]
None,
}
/// Remote API tuning (timeouts, UA, etc.).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ApiConfig {
#[serde(default = "ApiConfig::default_base_url")]
pub base_url: String,
#[serde(default = "ApiConfig::default_timeout")]
pub timeout_seconds: u64,
#[serde(default = "ApiConfig::default_user_agent")]
pub user_agent: String,
}
impl ApiConfig {
fn default_base_url() -> String {
"https://api.radioparadise.com".to_string()
}
const fn default_timeout() -> u64 {
30
}
fn default_user_agent() -> String {
"PMO-RadioParadise/1.0".to_string()
}
pub fn timeout(&self) -> Duration {
Duration::from_secs(self.timeout_seconds)
}
}
impl Default for ApiConfig {
fn default() -> Self {
Self {
base_url: Self::default_base_url(),
timeout_seconds: Self::default_timeout(),
user_agent: Self::default_user_agent(),
}
}
}

326
pmoparadise/src/paradise/history.rs Normal file
View File

@@ -0,0 +1,326 @@
//! History persistence for Radio Paradise playback.
//!
//! The worker pushes every completed track into the history backend while
//! keeping the latest entries available for UPnP browsing. We expose an
//! abstract trait so different storage engines (SQLite, JSON, etc.) can be
//! supported while sharing the same API.
use super::config::{HistoryBackendKind, HistoryConfig};
use crate::models::Song;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::{Arc, Mutex as StdMutex};
use tokio::sync::Mutex;
use tokio::task::spawn_blocking;
/// Serializable record describing a played track.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HistoryEntry {
pub track_id: String,
pub channel_id: u8,
pub started_at: chrono::DateTime<chrono::Utc>,
pub duration_ms: u64,
pub song: SongSnapshot,
}
/// Minimal snapshot of a Radio Paradise song at playback time.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SongSnapshot {
pub title: String,
pub artist: String,
pub album: Option<String>,
pub cover_url: Option<String>,
}
impl SongSnapshot {
pub fn title(&self) -> &str {
&self.title
}
}
impl From<&Song> for SongSnapshot {
fn from(song: &Song) -> Self {
Self {
title: song.title.clone(),
artist: song.artist.clone(),
album: song.album.clone(),
cover_url: song.cover.clone(),
}
}
}
/// Abstract persistence interface.
#[async_trait]
pub trait HistoryBackend: Send + Sync {
async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()>;
async fn recent(&self, limit: usize) -> anyhow::Result<Vec<HistoryEntry>>;
async fn len(&self) -> anyhow::Result<usize>;
async fn truncate(&self, keep: usize) -> anyhow::Result<()>;
}
pub fn history_backend_from_config(
config: &HistoryConfig,
) -> anyhow::Result<Arc<dyn HistoryBackend>> {
match config.persistence_backend {
HistoryBackendKind::Sqlite => {
let backend = SqliteHistoryBackend::new(&config.database_path)?;
Ok(Arc::new(backend))
}
HistoryBackendKind::Json => {
let backend = JsonHistoryBackend::new(&config.database_path);
Ok(Arc::new(backend))
}
}
}
/// Simple JSON file backed history (placeholder implementation).
///
/// The JSON backend is primarily useful for tests and quick setups. The file
/// is stored next to the configured database path with a `.json` extension.
pub struct JsonHistoryBackend {
path: std::path::PathBuf,
entries: Arc<Mutex<Vec<HistoryEntry>>>,
}
impl JsonHistoryBackend {
pub fn new(path: impl AsRef<Path>) -> Self {
let path = path.as_ref().to_path_buf();
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
let entries = if path.exists() {
std::fs::read(&path)
.ok()
.and_then(|bytes| serde_json::from_slice(&bytes).ok())
.unwrap_or_default()
} else {
Vec::new()
};
Self {
path,
entries: Arc::new(Mutex::new(entries)),
}
}
async fn save(&self, entries: &[HistoryEntry]) -> anyhow::Result<()> {
let json = serde_json::to_vec_pretty(entries)?;
tokio::fs::write(&self.path, json).await?;
Ok(())
}
}
#[async_trait]
impl HistoryBackend for JsonHistoryBackend {
async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()> {
let mut entries = self.entries.lock().await;
entries.push(entry);
self.save(&entries).await
}
async fn recent(&self, limit: usize) -> anyhow::Result<Vec<HistoryEntry>> {
let entries = self.entries.lock().await;
let total = entries.len();
let start = total.saturating_sub(limit);
Ok(entries[start..].to_vec())
}
async fn len(&self) -> anyhow::Result<usize> {
Ok(self.entries.lock().await.len())
}
async fn truncate(&self, keep: usize) -> anyhow::Result<()> {
let mut entries = self.entries.lock().await;
if entries.len() > keep {
let drop_count = entries.len() - keep;
entries.drain(0..drop_count);
self.save(&entries).await?;
}
Ok(())
}
}
/// In-memory history backend – useful for tests or ephemeral deployments.
#[derive(Default)]
pub struct MemoryHistoryBackend {
entries: Arc<Mutex<Vec<HistoryEntry>>>,
}
impl MemoryHistoryBackend {
pub fn new() -> Self {
Self::default()
}
}
#[async_trait]
impl HistoryBackend for MemoryHistoryBackend {
async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()> {
let mut entries = self.entries.lock().await;
entries.push(entry);
Ok(())
}
async fn recent(&self, limit: usize) -> anyhow::Result<Vec<HistoryEntry>> {
let entries = self.entries.lock().await;
let total = entries.len();
let start = total.saturating_sub(limit);
Ok(entries[start..].to_vec())
}
async fn len(&self) -> anyhow::Result<usize> {
Ok(self.entries.lock().await.len())
}
async fn truncate(&self, keep: usize) -> anyhow::Result<()> {
let mut entries = self.entries.lock().await;
if entries.len() > keep {
let drop_count = entries.len() - keep;
entries.drain(0..drop_count);
}
Ok(())
}
}
pub struct SqliteHistoryBackend {
conn: Arc<StdMutex<rusqlite::Connection>>,
}
impl SqliteHistoryBackend {
pub fn new(path: impl AsRef<Path>) -> anyhow::Result<Self> {
let path = path.as_ref();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let conn = rusqlite::Connection::open(path)?;
conn.pragma_update(None, "journal_mode", &"WAL")?;
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS paradise_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
track_id TEXT NOT NULL,
channel_id INTEGER NOT NULL,
started_at_ms INTEGER NOT NULL,
duration_ms INTEGER NOT NULL,
title TEXT,
artist TEXT,
album TEXT,
cover_url TEXT
);
CREATE INDEX IF NOT EXISTS idx_history_started_at ON paradise_history(started_at_ms);",
)?;
Ok(Self {
conn: Arc::new(StdMutex::new(conn)),
})
}
fn conn(&self) -> Arc<StdMutex<rusqlite::Connection>> {
self.conn.clone()
}
}
#[async_trait]
impl HistoryBackend for SqliteHistoryBackend {
async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()> {
let conn = self.conn();
spawn_blocking(move || -> anyhow::Result<()> {
let conn = conn.lock().unwrap();
conn.execute(
"INSERT INTO paradise_history (track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
rusqlite::params![
entry.track_id,
entry.channel_id as i64,
entry.started_at.timestamp_millis(),
entry.duration_ms as i64,
entry.song.title,
entry.song.artist,
entry.song.album,
entry.song.cover_url,
],
)?;
Ok(())
})
.await??;
Ok(())
}
async fn recent(&self, limit: usize) -> anyhow::Result<Vec<HistoryEntry>> {
let conn = self.conn();
let limit = limit as i64;
spawn_blocking(move || -> anyhow::Result<Vec<HistoryEntry>> {
let conn = conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url
FROM paradise_history
ORDER BY started_at_ms DESC
LIMIT ?1",
)?;
let mut rows = stmt.query([limit])?;
let mut entries = Vec::new();
while let Some(row) = rows.next()? {
let started_at_ms: i64 = row.get(2)?;
let started_at = DateTime::<Utc>::from_timestamp_millis(started_at_ms)
.ok_or_else(|| anyhow::anyhow!("Invalid timestamp in history"))?;
let entry = HistoryEntry {
track_id: row.get(0)?,
channel_id: row.get::<_, i64>(1)? as u8,
started_at,
duration_ms: row.get::<_, i64>(3)? as u64,
song: SongSnapshot {
title: row.get::<_, Option<String>>(4)?.unwrap_or_default(),
artist: row.get::<_, Option<String>>(5)?.unwrap_or_default(),
album: row.get(6)?,
cover_url: row.get(7)?,
},
};
entries.push(entry);
}
Ok(entries)
})
.await?
}
async fn len(&self) -> anyhow::Result<usize> {
let conn = self.conn();
let count = spawn_blocking(move || -> anyhow::Result<usize> {
let conn = conn.lock().unwrap();
let mut stmt = conn.prepare("SELECT COUNT(*) FROM paradise_history")?;
let count: i64 = stmt.query_row([], |row| row.get(0))?;
Ok(count as usize)
})
.await??;
Ok(count)
}
async fn truncate(&self, keep: usize) -> anyhow::Result<()> {
let conn = self.conn();
spawn_blocking(move || -> anyhow::Result<()> {
let conn = conn.lock().unwrap();
let count: i64 =
conn.query_row("SELECT COUNT(*) FROM paradise_history", [], |row| {
row.get(0)
})?;
let keep = keep as i64;
if count <= keep {
return Ok(());
}
let to_remove = count - keep;
conn.execute(
"DELETE FROM paradise_history
WHERE id IN (
SELECT id FROM paradise_history
ORDER BY started_at_ms ASC
LIMIT ?1
)",
rusqlite::params![to_remove],
)?;
Ok(())
})
.await??;
Ok(())
}
}

33
pmoparadise/src/paradise/mod.rs Normal file
View File

@@ -0,0 +1,33 @@
//! Internal orchestration layer for dynamic Radio Paradise streaming.
//!
//! This module implements the high level structures described in the
//! Radio Paradise functional specification:
//! - `ParadiseChannel`: lifecycle and state machine for a single RP channel.
//! - `ParadiseWorker`: async task responsible for polling/downloading blocks.
//! - `ParadiseClientStream`: per-client audio stream with independent cursor.
//! - Shared caches and history storage hooked into existing PMO components.
//!
//! The implementation is split across several submodules to keep concerns
//! isolated (configuration, playlist management, history persistence, etc.).
//! The goal of this scaffolding is to provide a clear, testable surface for
//! the eventual end-to-end integration with the UPnP server and HTTP routes.
mod channel;
mod config;
mod history;
mod playlist;
mod worker;
pub use channel::{
ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream, ALL_CHANNELS,
};
pub use config::{
ActivityConfig, ApiConfig, CacheConfig, HistoryConfig, PollingConfig, RadioParadiseConfig,
StreamConfig,
};
pub use history::{
history_backend_from_config, HistoryBackend, HistoryEntry, JsonHistoryBackend,
MemoryHistoryBackend,
};
pub use playlist::PlaylistEntry;
pub use worker::{ParadiseWorker, WorkerCommand};

293
pmoparadise/src/paradise/playlist.rs Normal file
View File

@@ -0,0 +1,293 @@
//! Shared playlist structures for Radio Paradise channels.
//!
//! This module keeps track of the active queue and history for a Radio
//! Paradise channel. Each playlist entry knows how many clients still need
//! to consume it before the worker can evict it.
use super::history::{HistoryEntry, SongSnapshot};
use crate::models::Song;
use chrono::{DateTime, Utc};
use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::SystemTime;
use tokio::sync::{Notify, RwLock};
/// Metadata stored for an active track.
#[derive(Debug)]
pub struct PlaylistEntry {
pub track_id: String,
pub channel_id: u8,
pub song: Arc<Song>,
pub started_at: DateTime<Utc>,
pub duration_ms: u64,
pub audio_pk: Option<String>,
pub file_path: Option<PathBuf>,
pending_clients: AtomicUsize,
}
impl PlaylistEntry {
#[allow(clippy::too_many_arguments)]
pub fn new(
track_id: String,
channel_id: u8,
song: Arc<Song>,
started_at: DateTime<Utc>,
duration_ms: u64,
audio_pk: Option<String>,
file_path: Option<PathBuf>,
pending_clients: usize,
) -> Self {
Self {
track_id,
channel_id,
song,
started_at,
duration_ms,
audio_pk,
file_path,
pending_clients: AtomicUsize::new(pending_clients),
}
}
pub fn as_history_entry(&self) -> HistoryEntry {
HistoryEntry {
track_id: self.track_id.clone(),
channel_id: self.channel_id,
started_at: self.started_at,
duration_ms: self.duration_ms,
song: SongSnapshot::from(self.song.as_ref()),
}
}
pub fn pending_clients(&self) -> usize {
self.pending_clients.load(Ordering::SeqCst)
}
pub fn set_pending_clients(&self, value: usize) {
self.pending_clients.store(value, Ordering::SeqCst);
}
pub fn increment_clients(&self) -> usize {
self.pending_clients.fetch_add(1, Ordering::SeqCst) + 1
}
pub fn decrement_clients(&self) -> usize {
let mut current = self.pending_clients.load(Ordering::SeqCst);
loop {
if current == 0 {
return 0;
}
match self.pending_clients.compare_exchange(
current,
current - 1,
Ordering::SeqCst,
Ordering::SeqCst,
) {
Ok(_) => return current - 1,
Err(actual) => current = actual,
}
}
}
}
#[derive(Default)]
struct PlaylistState {
active: VecDeque<Arc<PlaylistEntry>>,
history: VecDeque<HistoryEntry>,
max_history: usize,
}
impl PlaylistState {
fn new(max_history: usize) -> Self {
Self {
active: VecDeque::new(),
history: VecDeque::new(),
max_history,
}
}
fn active_len(&self) -> usize {
self.active.len()
}
fn push_active(&mut self, entry: Arc<PlaylistEntry>) {
self.active.push_back(entry);
}
fn active_snapshot(&self) -> Vec<Arc<PlaylistEntry>> {
self.active.iter().cloned().collect()
}
fn pop_front_if_ready(&mut self) -> Option<Arc<PlaylistEntry>> {
if let Some(front) = self.active.front() {
if front.pending_clients() == 0 {
return self.active.pop_front();
}
}
None
}
fn pop_front_matching(&mut self, track_id: &str) -> Option<Arc<PlaylistEntry>> {
if let Some(front) = self.active.front() {
if front.track_id == track_id && front.pending_clients() == 0 {
return self.active.pop_front();
}
}
None
}
fn push_history(&mut self, entry: HistoryEntry) {
self.history.push_back(entry);
self.trim_history();
}
fn recent_history(&self, limit: usize) -> Vec<HistoryEntry> {
let total = self.history.len();
let start = total.saturating_sub(limit);
self.history.iter().skip(start).cloned().collect()
}
fn trim_history(&mut self) {
while self.history.len() > self.max_history {
self.history.pop_front();
}
}
fn clear(&mut self) -> bool {
let changed = !self.active.is_empty() || !self.history.is_empty();
if changed {
self.active.clear();
self.history.clear();
}
changed
}
fn increment_all(&self) {
for entry in &self.active {
entry.increment_clients();
}
}
}
struct SharedPlaylistInner {
state: RwLock<PlaylistState>,
notify: Notify,
update_id: AtomicU32,
last_change: RwLock<Option<SystemTime>>,
}
#[derive(Clone)]
pub struct SharedPlaylist(Arc<SharedPlaylistInner>);
impl SharedPlaylist {
pub fn new(max_history: usize) -> Self {
Self(Arc::new(SharedPlaylistInner {
state: RwLock::new(PlaylistState::new(max_history)),
notify: Notify::new(),
update_id: AtomicU32::new(0),
last_change: RwLock::new(None),
}))
}
async fn touch(&self) {
self.0.update_id.fetch_add(1, Ordering::SeqCst);
let mut last_change = self.0.last_change.write().await;
*last_change = Some(SystemTime::now());
}
pub async fn push_active(&self, entry: Arc<PlaylistEntry>) {
let mut guard = self.0.state.write().await;
guard.push_active(entry);
drop(guard);
self.touch().await;
self.0.notify.notify_waiters();
}
pub async fn active_len(&self) -> usize {
let guard = self.0.state.read().await;
guard.active_len()
}
pub async fn active_snapshot(&self) -> Vec<Arc<PlaylistEntry>> {
let guard = self.0.state.read().await;
guard.active_snapshot()
}
pub async fn clear(&self) {
let mut guard = self.0.state.write().await;
let changed = guard.clear();
drop(guard);
if changed {
self.touch().await;
self.0.notify.notify_waiters();
}
}
pub async fn wait_for_track_count(&self, current_len: usize) {
loop {
let len = {
let guard = self.0.state.read().await;
guard.active_len()
};
if len > current_len {
break;
}
self.0.notify.notified().await;
}
}
pub async fn pop_front_if_ready(&self) -> Option<Arc<PlaylistEntry>> {
let mut guard = self.0.state.write().await;
let result = guard.pop_front_if_ready();
drop(guard);
if result.is_some() {
self.touch().await;
self.0.notify.notify_waiters();
}
result
}
pub async fn pop_front_matching(&self, track_id: &str) -> Option<Arc<PlaylistEntry>> {
let mut guard = self.0.state.write().await;
let result = guard.pop_front_matching(track_id);
drop(guard);
if result.is_some() {
self.touch().await;
self.0.notify.notify_waiters();
}
result
}
pub async fn push_history_entry(&self, entry: HistoryEntry) {
let mut guard = self.0.state.write().await;
guard.push_history(entry);
drop(guard);
self.touch().await;
}
pub async fn recent_history(&self, limit: usize) -> Vec<HistoryEntry> {
let guard = self.0.state.read().await;
guard.recent_history(limit)
}
pub async fn increment_all_pending(&self) {
let guard = self.0.state.read().await;
guard.increment_all();
}
pub fn update_id(&self) -> u32 {
self.0.update_id.load(Ordering::SeqCst)
}
pub async fn last_change(&self) -> Option<SystemTime> {
self.0.last_change.read().await.clone()
}
}

758
pmoparadise/src/paradise/worker.rs Normal file
View File

@@ -0,0 +1,758 @@
//! Background worker for Radio Paradise channels.
//!
//! The worker handles API polling, block ingestion, caching and playlist
//! maintenance. It keeps the channel state in sync with connected clients
//! and ensures fresh content is available according to the specification.
use super::channel::ChannelDescriptor;
use super::config::RadioParadiseConfig;
use super::history::HistoryBackend;
use super::playlist::{PlaylistEntry, SharedPlaylist};
use crate::client::RadioParadiseClient;
use crate::models::{Block, Song};
use anyhow::{anyhow, Context, Result};
use bytes::Bytes;
use chrono::Utc;
use futures::stream;
use pmosource::{SourceCacheManager, TrackMetadata};
use sha2::{Digest, Sha256};
use std::collections::{HashSet, VecDeque};
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::{sleep, Duration};
use tokio_util::io::StreamReader;
use tracing::{debug, error, info, warn};
use url::Url;
/// Commands sent to the background worker.
#[derive(Debug)]
pub enum WorkerCommand {
EnsureReady,
ClientConnected { client_id: String },
ClientDisconnected { client_id: String },
RefreshBlock,
Shutdown,
}
/// Handle to the spawned worker task.
pub struct ParadiseWorker {
descriptor: ChannelDescriptor,
join_handle: JoinHandle<()>,
}
impl ParadiseWorker {
#[allow(clippy::too_many_arguments)]
pub fn spawn(
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
config: Arc<RadioParadiseConfig>,
playlist: SharedPlaylist,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
) -> (Self, mpsc::Sender<WorkerCommand>) {
let (tx, mut rx) = mpsc::channel(32);
let join_handle = tokio::spawn(async move {
info!(channel = descriptor.slug, "Starting Radio Paradise worker");
let mut state =
WorkerState::new(descriptor, client, config, playlist, history, cache_manager);
loop {
if let Some(task) = state.scheduled_task.as_mut() {
let kind = task.kind;
let mut pending_command: Option<Option<WorkerCommand>> = None;
tokio::select! {
cmd = rx.recv() => {
pending_command = Some(cmd);
}
_ = &mut task.sleep => {
state.scheduled_task = None;
if let Err(err) = state.handle_scheduled_task(kind).await {
error!(channel = state.descriptor.slug, "Worker scheduled task error: {err:?}");
state.on_error(err);
}
}
}
if let Some(Some(cmd)) = pending_command {
if let Err(err) = state.handle_command(cmd).await {
error!(
channel = state.descriptor.slug,
"Worker command error: {err:?}"
);
state.on_error(err);
}
if state.shutdown {
break;
}
} else if let Some(None) = pending_command {
// Command channel closed, terminate
break;
}
} else {
match rx.recv().await {
Some(cmd) => {
if let Err(err) = state.handle_command(cmd).await {
error!(
channel = state.descriptor.slug,
"Worker command error: {err:?}"
);
state.on_error(err);
}
if state.shutdown {
break;
}
}
None => break,
}
}
}
info!(channel = state.descriptor.slug, "Worker stopped");
});
(
Self {
descriptor,
join_handle,
},
tx,
)
}
pub async fn wait(self) -> Result<()> {
if let Err(err) = self.join_handle.await {
if err.is_cancelled() {
warn!(
channel = self.descriptor.slug,
"Worker task cancelled: {err}"
);
return Ok(());
}
return Err(anyhow!("Worker join error: {}", err));
}
Ok(())
}
}
struct WorkerState {
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
config: Arc<RadioParadiseConfig>,
playlist: SharedPlaylist,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
active_clients: usize,
status: ChannelLifecycle,
processed_blocks: HashSet<u64>,
recent_blocks: VecDeque<u64>,
next_block_hint: Option<u64>,
scheduled_task: Option<ScheduledTask>,
backoff: BackoffState,
shutdown: bool,
}
impl WorkerState {
fn new(
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
config: Arc<RadioParadiseConfig>,
playlist: SharedPlaylist,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
) -> Self {
Self {
descriptor,
client,
config,
playlist,
history,
cache_manager,
active_clients: 0,
status: ChannelLifecycle::Idle,
processed_blocks: HashSet::new(),
recent_blocks: VecDeque::new(),
next_block_hint: None,
scheduled_task: None,
backoff: BackoffState::new(),
shutdown: false,
}
}
async fn handle_command(&mut self, cmd: WorkerCommand) -> Result<()> {
debug!(channel = self.descriptor.slug, ?cmd, "Worker command");
match cmd {
WorkerCommand::EnsureReady => {
self.ensure_ready().await?;
}
WorkerCommand::ClientConnected { .. } => {
self.active_clients = self.active_clients.saturating_add(1);
self.enter_active();
self.ensure_ready().await?;
}
WorkerCommand::ClientDisconnected { .. } => {
self.active_clients = self.active_clients.saturating_sub(1);
if self.active_clients == 0 {
self.enter_cooling();
}
}
WorkerCommand::RefreshBlock => {
self.fetch_next_block().await?;
}
WorkerCommand::Shutdown => {
self.shutdown = true;
self.cancel_scheduled_task();
}
}
if !self.shutdown {
self.maybe_schedule_poll().await;
}
Ok(())
}
async fn handle_scheduled_task(&mut self, kind: ScheduledTaskKind) -> Result<()> {
match kind {
ScheduledTaskKind::Poll => {
self.fetch_next_block().await?;
self.maybe_schedule_poll().await;
}
ScheduledTaskKind::Cooling => {
debug!(
channel = self.descriptor.slug,
"Cooling timeout reached -> idle"
);
self.status = ChannelLifecycle::Idle;
self.next_block_hint = None;
self.playlist.clear().await;
self.processed_blocks.clear();
self.recent_blocks.clear();
}
}
Ok(())
}
fn on_error(&mut self, err: anyhow::Error) {
warn!(channel = self.descriptor.slug, "Worker error: {err:?}");
let delay = self
.backoff
.next_delay(&self.config.polling.backoff_on_error);
self.schedule_task(ScheduledTaskKind::Poll, delay);
}
fn enter_active(&mut self) {
if !matches!(self.status, ChannelLifecycle::Active) {
debug!(
channel = self.descriptor.slug,
"Channel entering Active state"
);
}
self.status = ChannelLifecycle::Active;
if matches!(self.scheduled_task_kind(), Some(ScheduledTaskKind::Cooling)) {
self.cancel_scheduled_task();
}
self.backoff.reset();
}
fn enter_cooling(&mut self) {
if matches!(self.status, ChannelLifecycle::Idle) {
return;
}
debug!(
channel = self.descriptor.slug,
"Channel entering Cooling state"
);
self.status = ChannelLifecycle::Cooling;
let duration = Duration::from_secs(self.config.activity.cooling_timeout_seconds.max(1));
self.schedule_task(ScheduledTaskKind::Cooling, duration);
}
async fn ensure_ready(&mut self) -> Result<()> {
if !matches!(self.status, ChannelLifecycle::Active) {
self.enter_active();
}
let has_tracks = self.playlist.active_len().await > 0;
if !has_tracks {
debug!(
channel = self.descriptor.slug,
"Playlist empty – fetching now playing"
);
let now_playing = self.client.now_playing().await?;
self.process_block(now_playing.block).await?;
}
Ok(())
}
async fn fetch_next_block(&mut self) -> Result<()> {
if !matches!(self.status, ChannelLifecycle::Active) {
debug!(
channel = self.descriptor.slug,
"Skipping poll while not active"
);
return Ok(());
}
let event_id = self.next_block_hint;
let block = self.client.get_block(event_id).await?;
self.process_block(block).await?;
Ok(())
}
async fn process_block(&mut self, block: Block) -> Result<()> {
if self.is_recent_block(block.event) {
debug!(
channel = self.descriptor.slug,
event = block.event,
"Skipping already processed block"
);
self.next_block_hint = Some(block.end_event);
return Ok(());
}
info!(
channel = self.descriptor.slug,
event = block.event,
"Processing Radio Paradise block"
);
let _ = &self.history;
let block_url = Url::parse(&block.url)?;
let block_bytes = self
.client
.download_block(&block_url)
.await
.context("Failed to download block")?;
let decoded = decode_block_audio(block_bytes.to_vec())?;
let ordered_songs = block.songs_ordered();
let total_frames = decoded.samples.len() / decoded.channels;
for (position, (song_index, song)) in ordered_songs.iter().enumerate() {
let track = self
.process_song(
&block,
song_index,
song,
position,
&ordered_songs,
total_frames,
&decoded,
)
.await?;
self.playlist.push_active(track.clone()).await;
}
self.record_processed_block(block.event);
self.next_block_hint = Some(block.end_event);
self.backoff.reset();
Ok(())
}
async fn process_song(
&self,
block: &Block,
song_index: &usize,
song: &Song,
position: usize,
ordered_songs: &[(usize, &Song)],
total_frames: usize,
decoded: &DecodedBlock,
) -> Result<Arc<PlaylistEntry>> {
let duration_ms = song_duration_ms(block, ordered_songs, position);
let start_frame = ms_to_frames(song.elapsed, decoded.sample_rate);
let end_frame = if position + 1 < ordered_songs.len() {
ms_to_frames(ordered_songs[position + 1].1.elapsed, decoded.sample_rate)
} else {
total_frames
};
if end_frame <= start_frame || end_frame > total_frames {
warn!(
channel = self.descriptor.slug,
song_index = song_index,
"Invalid frame range for song, skipping"
);
return Err(anyhow!("Invalid frame range"));
}
let channels = decoded.channels;
let start = start_frame * channels;
let end = end_frame * channels;
let slice = decoded
.samples
.get(start..end)
.ok_or_else(|| anyhow!("Sample slice out of bounds"))?;
let track_samples = slice.to_vec();
let flac_bytes = encode_samples_to_flac(
track_samples,
decoded.channels,
decoded.sample_rate,
decoded.bits_per_sample,
)
.await
.context("Failed to encode song to FLAC")?;
let track_id = self.compute_track_id(&flac_bytes);
let placeholder_uri = format!("{}#{}", block.url, song_index);
let mut metadata = TrackMetadata {
original_uri: placeholder_uri.clone(),
cached_audio_pk: None,
cached_cover_pk: None,
};
if let Some(cover_pk) = self.cache_cover(block, song).await? {
metadata.cached_cover_pk = Some(cover_pk);
}
let flac_len = flac_bytes.len() as u64;
let reader = StreamReader::new(stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
flac_bytes,
))]));
let audio_pk = self
.cache_manager
.cache_audio_from_reader(&track_id, reader, Some(flac_len))
.await
.map_err(|e| anyhow!("Cache audio error: {e}"))?;
metadata.cached_audio_pk = Some(audio_pk.clone());
self.cache_manager
.update_metadata(track_id.clone(), metadata)
.await;
let file_path = self.cache_manager.audio_file_path(&audio_pk).await;
let entry = Arc::new(PlaylistEntry::new(
track_id,
self.descriptor.id,
Arc::new(song.clone()),
Utc::now(),
duration_ms,
Some(audio_pk),
file_path,
self.active_clients,
));
Ok(entry)
}
async fn cache_cover(&self, block: &Block, song: &Song) -> Result<Option<String>> {
if let Some(ref cover_path) = song.cover {
let cover_url =
resolve_cover_url(block.image_base.as_deref(), &self.client, cover_path)
.context("Invalid cover URL")?;
match self.cache_manager.cache_cover(cover_url.as_str()).await {
Ok(pk) => return Ok(Some(pk)),
Err(err) => {
warn!(channel = self.descriptor.slug, "Cover cache error: {err}");
}
}
}
Ok(None)
}
fn compute_track_id(&self, flac_bytes: &[u8]) -> String {
let slice_len = flac_bytes.len().min(self.config.cache.track_id_hash_bytes);
let mut hasher = Sha256::new();
hasher.update(&flac_bytes[..slice_len]);
let hash = hasher.finalize();
format!("rp:{}:{}", self.descriptor.id, hex::encode(hash))
}
async fn maybe_schedule_poll(&mut self) {
if !matches!(self.status, ChannelLifecycle::Active) {
return;
}
let buffer_len = self.playlist.active_len().await;
let interval = if buffer_len > 3 {
self.config.polling.high_interval()
} else if buffer_len >= 2 {
self.config.polling.medium_interval()
} else {
self.config.polling.low_interval()
};
self.schedule_task(ScheduledTaskKind::Poll, interval);
}
fn schedule_task(&mut self, kind: ScheduledTaskKind, duration: Duration) {
self.scheduled_task = Some(ScheduledTask {
kind,
sleep: Box::pin(sleep(duration)),
});
}
fn cancel_scheduled_task(&mut self) {
self.scheduled_task = None;
}
fn scheduled_task_kind(&self) -> Option<ScheduledTaskKind> {
self.scheduled_task.as_ref().map(|task| task.kind)
}
fn record_processed_block(&mut self, event: u64) {
self.processed_blocks.insert(event);
self.recent_blocks.push_back(event);
let max = self.config.cache.max_blocks_remembered.max(1);
while self.recent_blocks.len() > max {
if let Some(ev) = self.recent_blocks.pop_front() {
self.processed_blocks.remove(&ev);
}
}
}
fn is_recent_block(&self, event: u64) -> bool {
self.processed_blocks.contains(&event)
}
}
struct ScheduledTask {
kind: ScheduledTaskKind,
sleep: Pin<Box<tokio::time::Sleep>>,
}
#[derive(Clone, Copy)]
enum ScheduledTaskKind {
Poll,
Cooling,
}
#[derive(Clone, Copy, Debug)]
enum ChannelLifecycle {
Idle,
Cooling,
Active,
}
struct BackoffState {
current: Option<Duration>,
}
impl BackoffState {
fn new() -> Self {
Self { current: None }
}
fn reset(&mut self) {
self.current = None;
}
fn next_delay(&mut self, config: &super::config::PollingBackoffConfig) -> Duration {
let next = match self.current {
Some(current) => {
let multiplied = (current.as_secs_f32() * config.multiplier).round() as u64;
Duration::from_secs(multiplied.min(config.max))
}
None => Duration::from_secs(config.initial),
};
self.current = Some(next);
next
}
}
struct DecodedBlock {
samples: Vec<i32>,
channels: usize,
sample_rate: u32,
bits_per_sample: u32,
}
fn song_duration_ms(block: &Block, ordered: &[(usize, &Song)], position: usize) -> u64 {
let song = ordered[position].1;
if song.duration > 0 {
return song.duration;
}
if let Some((_, next_song)) = ordered.get(position + 1) {
return next_song.elapsed.saturating_sub(song.elapsed);
}
block.length.saturating_sub(song.elapsed)
}
fn ms_to_frames(ms: u64, sample_rate: u32) -> usize {
((ms as u128 * sample_rate as u128) / 1000) as usize
}
fn decode_block_audio(data: Vec<u8>) -> anyhow::Result<DecodedBlock> {
use symphonia::core::audio::SampleBuffer;
use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL};
use symphonia::core::errors::Error as SymphoniaError;
use symphonia::core::formats::FormatOptions;
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
let cursor = std::io::Cursor::new(data);
let mss = MediaSourceStream::new(Box::new(cursor), Default::default());
let hint = Hint::new();
let probed = symphonia::default::get_probe()
.format(
&hint,
mss,
&FormatOptions::default(),
&MetadataOptions::default(),
)
.map_err(|e| anyhow!("Failed to probe format: {e}"))?;
let mut format = probed.format;
let track = format
.tracks()
.iter()
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
.ok_or_else(|| anyhow!("No audio track found"))?;
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions::default())
.map_err(|e| anyhow!("Failed to create decoder: {e}"))?;
let channels = track
.codec_params
.channels
.ok_or_else(|| anyhow!("Missing channel info"))?
.count();
let sample_rate = track
.codec_params
.sample_rate
.ok_or_else(|| anyhow!("Missing sample rate"))?;
let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16);
let mut samples_i32 = Vec::new();
let track_id = track.id;
loop {
let packet = match format.next_packet() {
Ok(packet) => packet,
Err(SymphoniaError::ResetRequired) => {
decoder.reset();
continue;
}
Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
break;
}
Err(e) => return Err(anyhow!("Decode error: {e}")),
};
if packet.track_id() != track_id {
continue;
}
match decoder.decode(&packet) {
Ok(decoded) => {
let spec = *decoded.spec();
let duration = decoded.capacity() as u64;
let mut sample_buf = SampleBuffer::<i32>::new(duration, spec);
sample_buf.copy_interleaved_ref(decoded);
samples_i32.extend_from_slice(sample_buf.samples());
}
Err(SymphoniaError::DecodeError(_)) => continue,
Err(e) => return Err(anyhow!("Decode error: {e}")),
}
}
if samples_i32.is_empty() {
return Err(anyhow!("No samples decoded"));
}
let (normalized_samples, target_bits): (Vec<i32>, u32) = match bits_per_sample {
0..=16 => {
let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect();
(samples, 16)
}
17..=24 => {
let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect();
(samples, 24)
}
_ => (samples_i32, 32),
};
Ok(DecodedBlock {
samples: normalized_samples,
channels,
sample_rate,
bits_per_sample: target_bits,
})
}
async fn encode_samples_to_flac(
samples: Vec<i32>,
channels: usize,
sample_rate: u32,
bits_per_sample: u32,
) -> anyhow::Result<Vec<u8>> {
tokio::task::spawn_blocking(move || {
use flacenc::bitsink::ByteSink;
use flacenc::component::BitRepr;
use flacenc::error::Verify;
let config = flacenc::config::Encoder::default()
.into_verified()
.map_err(|e| anyhow!("FLAC config error: {e:?}"))?;
let source = flacenc::source::MemSource::from_samples(
&samples,
channels,
bits_per_sample as usize,
sample_rate as usize,
);
let flac_stream = flacenc::encode_with_fixed_block_size(&config, source, config.block_size)
.map_err(|e| anyhow!("FLAC encode error: {e:?}"))?;
let mut sink = ByteSink::new();
flac_stream
.write(&mut sink)
.map_err(|e| anyhow!("FLAC write error: {e:?}"))?;
Ok::<_, anyhow::Error>(sink.into_inner())
})
.await?
}
fn resolve_cover_url(
image_base: Option<&str>,
client: &RadioParadiseClient,
cover: &str,
) -> Result<Url> {
if cover.starts_with("http://") || cover.starts_with("https://") {
return Url::parse(cover).map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}"));
}
if cover.starts_with("//") {
let url = format!("https:{cover}");
return Url::parse(&url).map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}"));
}
if let Some(base) = image_base {
match Url::parse(base).and_then(|base_url| base_url.join(cover)) {
Ok(url) => return Ok(url),
Err(err) => {
debug!("Failed to join cover '{cover}' with base '{base}': {err}");
}
}
}
client
.cover_url(cover)
.map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}"))
}

484
pmoparadise/src/pmoserver_ext.rs
View File

@@ -3,22 +3,31 @@
//! Ce module fournit un trait d'extension pour ajouter facilement l'API Radio Paradise
//! à un serveur pmoserver.
use crate::{models::Bitrate, Block, NowPlaying, RadioParadiseClient};
use crate::paradise::{ParadiseChannel, PlaylistEntry};
use crate::{models::Bitrate, Block, NowPlaying, RadioParadiseClient, RadioParadiseSource};
use axum::{
body::Body,
extract::{Path, Query, State},
http::StatusCode,
http::{HeaderMap, HeaderName, HeaderValue, StatusCode},
response::IntoResponse,
routing::get,
Json, Router,
};
use chrono::{DateTime, Utc};
use futures::StreamExt;
use pmosource::api::CacheStatusInfo;
use pmosource::CacheStatus;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::RwLock;
use utoipa::{OpenApi, ToSchema};
use tracing::error;
use utoipa::{IntoParams, OpenApi, ToSchema};
/// État partagé pour l'API Radio Paradise
#[derive(Clone)]
pub struct RadioParadiseState {
client: Arc<RwLock<RadioParadiseClient>>,
source: Arc<RadioParadiseSource>,
}
const MAX_CHANNEL_ID: u8 = 3;
@@ -30,13 +39,53 @@ struct ParadiseQuery {
bitrate: Option<u8>,
}
#[derive(Debug, Default, Deserialize)]
#[serde(default)]
struct StreamQuery {
channel: Option<u8>,
client_id: Option<String>,
}
#[derive(Debug, Default, Deserialize, IntoParams)]
#[serde(default)]
#[into_params(parameter_in = Query)]
struct ListLimitQuery {
/// Nombre maximum d'éléments à retourner (0 = tous)
#[serde(default)]
limit: Option<usize>,
}
impl RadioParadiseState {
pub async fn new() -> anyhow::Result<Self> {
let client = RadioParadiseClient::new()
.await
.map_err(|e| anyhow::anyhow!("Failed to create RadioParadise client: {}", e))?;
#[cfg(feature = "server")]
let source = RadioParadiseSource::from_registry_default(client.clone())
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
#[cfg(not(feature = "server"))]
let source = {
let base_dir = std::env::temp_dir().join("pmoparadise_api");
let cover_dir = base_dir.join("covers");
let audio_dir = base_dir.join("audio");
std::fs::create_dir_all(&cover_dir)?;
std::fs::create_dir_all(&audio_dir)?;
let cover_cache = Arc::new(pmocovers::cache::new_cache(
cover_dir.to_string_lossy().as_ref(),
256,
)?);
let audio_cache = Arc::new(pmoaudiocache::cache::new_cache(
audio_dir.to_string_lossy().as_ref(),
256,
)?);
RadioParadiseSource::new_default(client.clone(), cover_cache, audio_cache)
};
Ok(Self {
client: Arc::new(RwLock::new(client)),
source: Arc::new(source),
})
}
@@ -69,6 +118,19 @@ impl RadioParadiseState {
Ok(client)
}
fn channel_for_id(&self, channel_id: u8) -> Result<Arc<ParadiseChannel>, StatusCode> {
if channel_id > MAX_CHANNEL_ID {
return Err(StatusCode::BAD_REQUEST);
}
self.source
.channel(channel_id)
.ok_or(StatusCode::SERVICE_UNAVAILABLE)
}
pub fn source(&self) -> Arc<RadioParadiseSource> {
self.source.clone()
}
}
/// Information sur un canal Radio Paradise
@@ -345,6 +407,145 @@ pub struct BitrateInfo {
pub name: String,
}
/// Statut opérationnel d'un canal Radio Paradise
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ChannelStatusResponse {
/// ID numérique du canal
pub channel_id: u8,
/// Slug du canal (main, mellow, ...)
pub slug: String,
/// Nom complet du canal
pub name: String,
/// Description
pub description: String,
/// Nombre de clients connectés au flux
pub active_clients: usize,
/// Nombre de morceaux présents dans la file d'attente
pub queue_length: usize,
/// Valeur courante d'update_id
pub update_id: u32,
/// Dernière modification (RFC3339)
pub last_change: Option<String>,
/// Nombre total d'entrées en historique (persisté)
pub history_entries: usize,
/// Limite configurée pour l'historique
pub history_max_tracks: usize,
/// Le canal est-il activé dans la configuration ?
pub configured: bool,
/// Identifiant de collection pour le cache
pub cache_collection_id: String,
/// Nombre total de pistes connues du cache
pub cache_total_tracks: usize,
/// Nombre de pistes déjà en cache
pub cache_cached_tracks: usize,
}
/// Entrée détaillée de la file d'attente
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ChannelPlaylistEntry {
/// Position dans la file
pub index: usize,
/// ID unique de la piste
pub track_id: String,
/// ID du canal
pub channel_id: u8,
/// Titre du morceau
pub title: String,
/// Artiste
pub artist: String,
/// Album
pub album: Option<String>,
/// URL de couverture (si disponible)
pub cover_url: Option<String>,
/// Durée du morceau en ms
pub duration_ms: u64,
/// Offset dans le block (ms)
pub elapsed_ms: u64,
/// Horodatage prévu/démarré (RFC3339)
pub started_at: String,
/// Nombre de clients restants à servir
pub pending_clients: usize,
/// Note éventuelle (0-10)
pub rating: Option<f32>,
/// Année éventuelle
pub year: Option<u32>,
/// Statut de cache
pub cache_status: CacheStatusInfo,
}
impl ChannelPlaylistEntry {
fn from_entry(entry: &Arc<PlaylistEntry>, index: usize, cache_status: CacheStatusInfo) -> Self {
let song = entry.song.as_ref();
Self {
index,
track_id: entry.track_id.clone(),
channel_id: entry.channel_id,
title: song.title.clone(),
artist: song.artist.clone(),
album: song.album.clone(),
cover_url: song.cover.clone(),
duration_ms: entry.duration_ms,
elapsed_ms: song.elapsed,
started_at: entry.started_at.to_rfc3339(),
pending_clients: entry.pending_clients(),
rating: song.rating,
year: song.year,
cache_status,
}
}
}
/// Réponse pour la file d'attente d'un canal
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ChannelPlaylistResponse {
/// ID du canal
pub channel_id: u8,
/// Slug du canal
pub slug: String,
/// Update ID du playlist
pub update_id: u32,
/// Taille totale de la file au moment de la capture
pub queue_length: usize,
/// Entrées retournées
pub items: Vec<ChannelPlaylistEntry>,
}
/// Entrée d'historique d'écoute
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ChannelHistoryEntry {
/// ID unique de la piste
pub track_id: String,
/// ID du canal
pub channel_id: u8,
/// Titre
pub title: String,
/// Artiste
pub artist: String,
/// Album
pub album: Option<String>,
/// URL de couverture
pub cover_url: Option<String>,
/// Début de lecture (RFC3339)
pub started_at: String,
/// Durée en ms
pub duration_ms: u64,
}
/// Réponse pour l'historique d'un canal
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ChannelHistoryResponse {
/// ID du canal
pub channel_id: u8,
/// Slug du canal
pub slug: String,
/// Nombre total d'entrées disponibles
pub total_available: usize,
/// Nombre d'entrées retournées dans cette réponse
pub returned: usize,
/// Entrées
pub entries: Vec<ChannelHistoryEntry>,
}
/// GET /bitrates - Liste les bitrates disponibles
#[utoipa::path(
get,
@@ -381,6 +582,262 @@ async fn get_bitrates() -> Json<Vec<BitrateInfo>> {
Json(bitrates)
}
/// GET /channels/{channel_id}/status - Statut détaillé d'un canal
#[utoipa::path(
get,
path = "/channels/{channel_id}/status",
params(
("channel_id" = u8, Path, description = "Channel ID (0-3)")
),
responses(
(status = 200, description = "Statut du canal", body = ChannelStatusResponse),
(status = 400, description = "Canal invalide"),
(status = 503, description = "Canal indisponible"),
(status = 500, description = "Erreur interne lors de la récupération du statut")
),
tag = "Radio Paradise"
)]
async fn get_channel_status(
State(state): State<RadioParadiseState>,
Path(channel_id): Path<u8>,
) -> Result<Json<ChannelStatusResponse>, StatusCode> {
let channel = state.channel_for_id(channel_id)?;
let descriptor = channel.descriptor();
let playlist = channel.playlist();
let queue_length = playlist.active_len().await;
let update_id = playlist.update_id();
let last_change = playlist
.last_change()
.await
.map(|ts| DateTime::<Utc>::from(ts).to_rfc3339());
let history_len = channel.history_backend().len().await.map_err(|e| {
error!(
channel = descriptor.slug,
"Failed to retrieve history size: {e:?}"
);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let cache_stats = channel.cache_manager().statistics().await;
let config = channel.config().clone();
let configured = config.channels.iter().any(|slug| slug == descriptor.slug);
let status = ChannelStatusResponse {
channel_id,
slug: descriptor.slug.to_string(),
name: descriptor.display_name.to_string(),
description: descriptor.description.to_string(),
active_clients: channel.active_client_count(),
queue_length,
update_id,
last_change,
history_entries: history_len,
history_max_tracks: config.history.max_tracks,
configured,
cache_collection_id: cache_stats.collection_id,
cache_total_tracks: cache_stats.total_tracks,
cache_cached_tracks: cache_stats.cached_tracks,
};
Ok(Json(status))
}
/// GET /channels/{channel_id}/playlist - File d'attente du canal
#[utoipa::path(
get,
path = "/channels/{channel_id}/playlist",
params(
("channel_id" = u8, Path, description = "Channel ID (0-3)"),
ListLimitQuery
),
responses(
(status = 200, description = "File d'attente courante", body = ChannelPlaylistResponse),
(status = 400, description = "Canal invalide"),
(status = 503, description = "Canal indisponible"),
(status = 500, description = "Erreur lors de la récupération de la file d'attente")
),
tag = "Radio Paradise"
)]
async fn get_channel_playlist(
State(state): State<RadioParadiseState>,
Path(channel_id): Path<u8>,
Query(query): Query<ListLimitQuery>,
) -> Result<Json<ChannelPlaylistResponse>, StatusCode> {
let channel = state.channel_for_id(channel_id)?;
let descriptor = channel.descriptor();
let playlist = channel.playlist();
let snapshot = playlist.active_snapshot().await;
let total_len = snapshot.len();
let limit = query.limit.filter(|limit| *limit > 0).unwrap_or(total_len);
let cache_manager = channel.cache_manager();
let mut items = Vec::new();
for (index, entry) in snapshot.into_iter().enumerate().take(limit) {
let cache_status = match cache_manager.get_cache_status(&entry.track_id).await {
Ok(status) => status,
Err(err) => CacheStatus::Failed {
error: err.to_string(),
},
};
items.push(ChannelPlaylistEntry::from_entry(
&entry,
index,
CacheStatusInfo::from(cache_status),
));
}
let response = ChannelPlaylistResponse {
channel_id,
slug: descriptor.slug.to_string(),
update_id: playlist.update_id(),
queue_length: total_len,
items,
};
Ok(Json(response))
}
/// GET /channels/{channel_id}/history - Historique récent du canal
#[utoipa::path(
get,
path = "/channels/{channel_id}/history",
params(
("channel_id" = u8, Path, description = "Channel ID (0-3)"),
ListLimitQuery
),
responses(
(status = 200, description = "Historique récent", body = ChannelHistoryResponse),
(status = 400, description = "Canal invalide"),
(status = 503, description = "Canal indisponible"),
(status = 500, description = "Erreur lors de la récupération de l'historique")
),
tag = "Radio Paradise"
)]
async fn get_channel_history(
State(state): State<RadioParadiseState>,
Path(channel_id): Path<u8>,
Query(query): Query<ListLimitQuery>,
) -> Result<Json<ChannelHistoryResponse>, StatusCode> {
let channel = state.channel_for_id(channel_id)?;
let descriptor = channel.descriptor();
let backend = channel.history_backend().clone();
let limit = query.limit.unwrap_or(50);
let entries_raw = backend.recent(limit).await.map_err(|e| {
error!(
channel = descriptor.slug,
"Failed to retrieve channel history: {e:?}"
);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let total_available = backend.len().await.map_err(|e| {
error!(
channel = descriptor.slug,
"Failed to count channel history entries: {e:?}"
);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let entries: Vec<ChannelHistoryEntry> = entries_raw
.into_iter()
.map(|entry| ChannelHistoryEntry {
track_id: entry.track_id,
channel_id: entry.channel_id,
title: entry.song.title,
artist: entry.song.artist,
album: entry.song.album,
cover_url: entry.song.cover_url,
started_at: entry.started_at.to_rfc3339(),
duration_ms: entry.duration_ms,
})
.collect();
let response = ChannelHistoryResponse {
channel_id,
slug: descriptor.slug.to_string(),
total_available,
returned: entries.len(),
entries,
};
Ok(Json(response))
}
/// GET /stream - Stream audio pour un canal donné
#[utoipa::path(
get,
path = "/stream",
params(
("channel" = Option<u8>, Query, description = "Channel ID (0-3)"),
("client_id" = Option<String>, Query, description = "Identifiant personnalisé du client")
),
responses(
(status = 200, description = "Flux audio FLAC (gapless)", content_type = "audio/flac"),
(status = 400, description = "Paramètres invalides"),
(status = 503, description = "Canal indisponible")
),
tag = "Radio Paradise"
)]
async fn stream_channel(
State(state): State<RadioParadiseState>,
Query(params): Query<StreamQuery>,
) -> Result<impl IntoResponse, StatusCode> {
let channel_id = params.channel.unwrap_or(0);
let channel = state.channel_for_id(channel_id)?;
let client_id = params.client_id.clone().unwrap_or_else(|| {
format!(
"api-{}-{}",
channel.descriptor().slug,
Utc::now().timestamp_micros()
)
});
let client_stream = channel.connect_client(client_id).await.map_err(|e| {
error!("Failed to create streaming client: {e:?}");
StatusCode::SERVICE_UNAVAILABLE
})?;
let stream = client_stream
.into_byte_stream()
.map(|chunk| chunk.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)));
let body = Body::from_stream(stream);
let mut headers = HeaderMap::new();
headers.insert(
axum::http::header::CONTENT_TYPE,
HeaderValue::from_static("audio/flac"),
);
headers.insert(
axum::http::header::CACHE_CONTROL,
HeaderValue::from_static("no-cache"),
);
headers.insert(
HeaderName::from_static("icy-name"),
HeaderValue::from_static("Radio Paradise"),
);
headers.insert(
HeaderName::from_static("icy-genre"),
HeaderValue::from_static("Eclectic"),
);
headers.insert(
HeaderName::from_static("icy-description"),
HeaderValue::from_static("PMO Radio Paradise relay"),
);
headers.insert(
HeaderName::from_static("icy-metaint"),
HeaderValue::from_static("0"),
);
Ok((headers, body))
}
/// Documentation OpenAPI pour l'API Radio Paradise
#[derive(OpenApi)]
#[openapi(
@@ -394,14 +851,24 @@ async fn get_bitrates() -> Json<Vec<BitrateInfo>> {
get_current_block,
get_block_by_id,
get_channels,
get_bitrates
get_channel_status,
get_channel_playlist,
get_channel_history,
get_bitrates,
stream_channel
),
components(schemas(
NowPlayingResponse,
BlockResponse,
SongInfo,
ChannelInfo,
BitrateInfo
BitrateInfo,
ChannelStatusResponse,
ChannelPlaylistEntry,
ChannelPlaylistResponse,
ChannelHistoryEntry,
ChannelHistoryResponse,
CacheStatusInfo
)),
tags(
(name = "Radio Paradise", description = "Endpoints pour Radio Paradise streaming")
@@ -416,7 +883,11 @@ pub fn create_api_router(state: RadioParadiseState) -> Router {
.route("/block/current", get(get_current_block))
.route("/block/{event_id}", get(get_block_by_id))
.route("/channels", get(get_channels))
.route("/channels/{channel_id}/status", get(get_channel_status))
.route("/channels/{channel_id}/playlist", get(get_channel_playlist))
.route("/channels/{channel_id}/history", get(get_channel_history))
.route("/bitrates", get(get_bitrates))
.route("/stream", get(stream_channel))
.with_state(state)
}
@@ -430,6 +901,9 @@ pub trait RadioParadiseExt {
/// # Routes créées
///
/// - API: `/api/radioparadise/*`
/// - `/now-playing`
/// - `/block/*`
/// - `/stream`
/// - Swagger: `/swagger-ui/radioparadise`
async fn init_radioparadise(&mut self) -> anyhow::Result<RadioParadiseState>;
}

1086
pmoparadise/src/source.rs
View File

File diff suppressed because it is too large Load Diff

2
pmoserver/Cargo.toml
View File

@@ -20,4 +20,4 @@ axum-server = "0.7.2"
axum-embed = "0.1.0"
rust-embed = "8.7.2"
utoipa = { version = "5.4.0", features = ["axum_extras"] }
utoipa-swagger-ui = { version = "9.0.2", features = ["axum"] }
utoipa-swagger-ui = { version = "9.0.2", features = ["axum", "vendored"] }

394
pmosource/src/api.rs
View File

@@ -24,7 +24,7 @@ use axum::{
extract::{Path, Query},
http::{header, StatusCode},
response::{IntoResponse, Response},
routing::{delete, get},
routing::{get, post},
Json, Router,
};
@@ -32,7 +32,9 @@ use axum::{
use serde::{Deserialize, Serialize};
#[cfg(feature = "server")]
use crate::{MusicSource, SourceCapabilities, SourceStatistics};
use crate::{
AudioFormat, CacheStatus, MusicSource, MusicSourceError, SourceCapabilities, SourceStatistics,
};
#[cfg(feature = "server")]
use std::sync::Arc;
@@ -222,6 +224,151 @@ pub struct BrowseItemInfo {
pub resources: Vec<BrowseItemResourceInfo>,
}
/// Paramètres génériques pour cibler un objet d'une source
#[cfg(feature = "server")]
#[derive(Debug, Deserialize, utoipa::IntoParams)]
#[into_params(parameter_in = Query)]
pub struct ObjectQuery {
/// ID de l'objet (container ou item)
pub object_id: String,
}
/// Réponse pour la résolution d'URI d'un objet
#[cfg(feature = "server")]
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct ResolveUriResponse {
/// ID de l'objet demandé
pub object_id: String,
/// URI résolue (cache ou origine)
pub uri: String,
}
/// États possibles pour le cache
#[cfg(feature = "server")]
#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum CacheStatusState {
NotCached,
Caching,
Cached,
Failed,
}
/// Informations détaillées sur le cache d'un objet
#[cfg(feature = "server")]
#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct CacheStatusInfo {
/// État du cache
pub status: CacheStatusState,
/// Progression (0.0 - 1.0)
pub progress: Option<f32>,
/// Taille en octets si connue
pub size_bytes: Option<u64>,
/// Message d'erreur éventuel
pub error: Option<String>,
}
#[cfg(feature = "server")]
impl From<CacheStatus> for CacheStatusInfo {
fn from(status: CacheStatus) -> Self {
match status {
CacheStatus::NotCached => Self {
status: CacheStatusState::NotCached,
progress: Some(0.0),
size_bytes: None,
error: None,
},
CacheStatus::Caching { progress } => Self {
status: CacheStatusState::Caching,
progress: Some(progress),
size_bytes: None,
error: None,
},
CacheStatus::Cached { size_bytes } => Self {
status: CacheStatusState::Cached,
progress: Some(1.0),
size_bytes: Some(size_bytes),
error: None,
},
CacheStatus::Failed { error } => Self {
status: CacheStatusState::Failed,
progress: None,
size_bytes: None,
error: Some(error),
},
}
}
}
/// Corps de requête pour déclencher la mise en cache
#[cfg(feature = "server")]
#[derive(Debug, Deserialize, utoipa::ToSchema)]
pub struct CacheRequest {
/// ID de l'objet à mettre en cache
pub object_id: String,
}
/// Réponse standard pour les endpoints de cache
#[cfg(feature = "server")]
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct CacheStatusResponse {
/// ID de l'objet
pub object_id: String,
/// Informations de cache
pub status: CacheStatusInfo,
}
/// Paramètres pour récupérer les formats disponibles
#[cfg(feature = "server")]
#[derive(Debug, Deserialize, utoipa::IntoParams)]
#[into_params(parameter_in = Query)]
pub struct FormatsQuery {
/// ID de l'objet ciblé
pub object_id: String,
}
/// Description d'un format audio disponible
#[cfg(feature = "server")]
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct AudioFormatInfo {
/// Identifiant technique du format
pub format_id: String,
/// MIME type (`audio/flac`, `audio/mpeg`, ...)
pub mime_type: String,
/// Fréquence d'échantillonnage (Hz)
pub sample_rate: Option<u32>,
/// Profondeur de bits
pub bit_depth: Option<u8>,
/// Débit en kbps (lossy)
pub bitrate: Option<u32>,
/// Nombre de canaux
pub channels: Option<u8>,
}
#[cfg(feature = "server")]
impl From<AudioFormat> for AudioFormatInfo {
fn from(format: AudioFormat) -> Self {
Self {
format_id: format.format_id,
mime_type: format.mime_type,
sample_rate: format.sample_rate,
bit_depth: format.bit_depth,
bitrate: format.bitrate,
channels: format.channels,
}
}
}
/// Réponse contenant les formats disponibles pour un objet
#[cfg(feature = "server")]
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct AudioFormatsResponse {
/// ID de l'objet
pub object_id: String,
/// Liste des formats supportés
pub formats: Vec<AudioFormatInfo>,
}
#[cfg(feature = "server")]
impl From<&DidlContainer> for BrowseContainerInfo {
fn from(container: &DidlContainer) -> Self {
@@ -651,6 +798,234 @@ async fn browse_source(
}
}
/// Résout l'URI réelle d'un objet (cache ou origine)
#[cfg(feature = "server")]
#[utoipa::path(
get,
path = "/{id}/resolve",
params(
("id" = String, Path, description = "ID de la source"),
ObjectQuery
),
responses(
(status = 200, description = "URI résolue", body = ResolveUriResponse),
(status = 404, description = "Source ou objet introuvable", body = ErrorResponse),
(status = 500, description = "Erreur lors de la résolution de l'URI", body = ErrorResponse),
),
tag = "sources"
)]
async fn resolve_source_uri(
Path(id): Path<String>,
Query(params): Query<ObjectQuery>,
) -> impl IntoResponse {
match get_source(&id).await {
Some(source) => match source.resolve_uri(&params.object_id).await {
Ok(uri) => {
let response = ResolveUriResponse {
object_id: params.object_id,
uri,
};
(StatusCode::OK, Json(response)).into_response()
}
Err(MusicSourceError::ObjectNotFound(_)) => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "Object not found".to_string(),
}),
)
.into_response(),
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: format!("Failed to resolve URI: {}", e),
}),
)
.into_response(),
},
None => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: format!("Source '{}' not found", id),
}),
)
.into_response(),
}
}
/// Récupère le statut du cache pour un objet
#[cfg(feature = "server")]
#[utoipa::path(
get,
path = "/{id}/cache/status",
params(
("id" = String, Path, description = "ID de la source"),
ObjectQuery
),
responses(
(status = 200, description = "Statut du cache", body = CacheStatusResponse),
(status = 404, description = "Source ou objet introuvable", body = ErrorResponse),
(status = 501, description = "Cache non supporté", body = ErrorResponse),
(status = 500, description = "Erreur lors de la récupération du statut du cache", body = ErrorResponse),
),
tag = "sources"
)]
async fn get_source_cache_status(
Path(id): Path<String>,
Query(params): Query<ObjectQuery>,
) -> impl IntoResponse {
match get_source(&id).await {
Some(source) => match source.get_cache_status(&params.object_id).await {
Ok(status) => {
let response = CacheStatusResponse {
object_id: params.object_id,
status: CacheStatusInfo::from(status),
};
(StatusCode::OK, Json(response)).into_response()
}
Err(MusicSourceError::ObjectNotFound(_)) => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "Object not found".to_string(),
}),
)
.into_response(),
Err(MusicSourceError::NotSupported(msg)) => (
StatusCode::NOT_IMPLEMENTED,
Json(ErrorResponse { error: msg }),
)
.into_response(),
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: format!("Failed to get cache status: {}", e),
}),
)
.into_response(),
},
None => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: format!("Source '{}' not found", id),
}),
)
.into_response(),
}
}
/// Demande la mise en cache d'un objet
#[cfg(feature = "server")]
#[utoipa::path(
post,
path = "/{id}/cache",
params(
("id" = String, Path, description = "ID de la source")
),
request_body = CacheRequest,
responses(
(status = 200, description = "Requête de cache enregistrée", body = CacheStatusResponse),
(status = 404, description = "Source ou objet introuvable", body = ErrorResponse),
(status = 501, description = "Cache non supporté", body = ErrorResponse),
(status = 500, description = "Erreur lors de la mise en cache", body = ErrorResponse),
),
tag = "sources"
)]
async fn request_source_cache(
Path(id): Path<String>,
Json(payload): Json<CacheRequest>,
) -> impl IntoResponse {
match get_source(&id).await {
Some(source) => match source.cache_item(&payload.object_id).await {
Ok(status) => {
let response = CacheStatusResponse {
object_id: payload.object_id,
status: CacheStatusInfo::from(status),
};
(StatusCode::OK, Json(response)).into_response()
}
Err(MusicSourceError::ObjectNotFound(_)) => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "Object not found".to_string(),
}),
)
.into_response(),
Err(MusicSourceError::NotSupported(msg)) => (
StatusCode::NOT_IMPLEMENTED,
Json(ErrorResponse { error: msg }),
)
.into_response(),
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: format!("Failed to cache item: {}", e),
}),
)
.into_response(),
},
None => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: format!("Source '{}' not found", id),
}),
)
.into_response(),
}
}
/// Récupère les formats audio disponibles pour un objet
#[cfg(feature = "server")]
#[utoipa::path(
get,
path = "/{id}/formats",
params(
("id" = String, Path, description = "ID de la source"),
FormatsQuery
),
responses(
(status = 200, description = "Formats disponibles", body = AudioFormatsResponse),
(status = 404, description = "Source ou objet introuvable", body = ErrorResponse),
(status = 500, description = "Erreur lors de la récupération des formats", body = ErrorResponse),
),
tag = "sources"
)]
async fn get_source_formats(
Path(id): Path<String>,
Query(params): Query<FormatsQuery>,
) -> impl IntoResponse {
match get_source(&id).await {
Some(source) => match source.get_available_formats(&params.object_id).await {
Ok(formats) => {
let response = AudioFormatsResponse {
object_id: params.object_id,
formats: formats.into_iter().map(AudioFormatInfo::from).collect(),
};
(StatusCode::OK, Json(response)).into_response()
}
Err(MusicSourceError::ObjectNotFound(_)) => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "Object not found".to_string(),
}),
)
.into_response(),
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: format!("Failed to get available formats: {}", e),
}),
)
.into_response(),
},
None => (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: format!("Source '{}' not found", id),
}),
)
.into_response(),
}
}
/// Désenregistre une source musicale
///
/// Supprime une source du registre par son ID.
@@ -718,6 +1093,10 @@ pub fn create_sources_router() -> Router {
.route("/{id}/root", get(get_source_root))
.route("/{id}/browse", get(browse_source))
.route("/{id}/image", get(get_source_image))
.route("/{id}/resolve", get(resolve_source_uri))
.route("/{id}/cache/status", get(get_source_cache_status))
.route("/{id}/cache", post(request_source_cache))
.route("/{id}/formats", get(get_source_formats))
}
/// Structure pour la documentation OpenAPI de base
@@ -736,6 +1115,10 @@ pub fn create_sources_router() -> Router {
get_source_root,
browse_source,
get_source_image,
resolve_source_uri,
get_source_cache_status,
request_source_cache,
get_source_formats,
unregister_source_handler,
),
components(
@@ -749,6 +1132,13 @@ pub fn create_sources_router() -> Router {
BrowseItemResourceInfo,
BrowseItemInfo,
SourceBrowseResponse,
ResolveUriResponse,
CacheStatusState,
CacheStatusInfo,
CacheStatusResponse,
CacheRequest,
AudioFormatInfo,
AudioFormatsResponse,
ErrorResponse,
)
),

10
pmosource/src/cache.rs
View File

@@ -282,6 +282,16 @@ impl SourceCacheManager {
collection_id: self.collection_id.clone(),
}
}
/// Récupère le chemin de fichier pour une piste audio en cache
///
/// Retourne `None` si le fichier n'est pas encore disponible.
pub async fn audio_file_path(&self, pk: &str) -> Option<std::path::PathBuf> {
match self.audio_cache.get(pk).await {
Ok(path) => Some(path),
Err(_) => None,
}
}
}
/// Statistiques du cache pour une source