diff --git a/config/azuracast.json.example b/config/azuracast.json.example new file mode 100644 index 00000000..e2bc20e9 --- /dev/null +++ b/config/azuracast.json.example @@ -0,0 +1,11 @@ +[ + { + "type": "azuracast", + "enable": true, + "name": "azura", + "data": { + "url": "ws://192.168.0.101", + "station": "my-station-name" + } + } +] diff --git a/package-lock.json b/package-lock.json index 59c57139..9a561271 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11746,9 +11746,9 @@ "peer": true }, "node_modules/ws": { - "version": "8.18.0", - "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.0.tgz", - "integrity": "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw==", + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-RKW2aJZMXeMxVpnZ6bck+RswznaxmzdULiBr6KY7XkTnW8uvt0iT9H5DkHUChXrc+uurzwa0rVI16n/Xzjdz1w==", "engines": { "node": ">=10.0.0" }, diff --git a/src/backend/common/infrastructure/config/source/sources.ts b/src/backend/common/infrastructure/config/source/sources.ts index 77cfbbd9..728b594c 100644 --- a/src/backend/common/infrastructure/config/source/sources.ts +++ b/src/backend/common/infrastructure/config/source/sources.ts @@ -1,3 +1,4 @@ +import { AzuracastSourceAIOConfig, AzuracastSourceConfig } from "./azuracast.js"; import { ChromecastSourceAIOConfig, ChromecastSourceConfig } from "./chromecast.js"; import { DeezerSourceAIOConfig, DeezerSourceConfig } from "./deezer.js"; import { JellyApiSourceAIOConfig, JellyApiSourceConfig, JellySourceAIOConfig, JellySourceConfig } from "./jellyfin.js"; @@ -38,7 +39,8 @@ export type SourceConfig = | ChromecastSourceConfig | MusikcubeSourceConfig | MPDSourceConfig - | VLCSourceConfig; + | VLCSourceConfig + | AzuracastSourceConfig; export type SourceAIOConfig = SpotifySourceAIOConfig @@ -60,4 +62,5 @@ export type SourceAIOConfig = | ChromecastSourceAIOConfig | MusikcubeSourceAIOConfig | MPDSourceAIOConfig - | VLCSourceAIOConfig; + | VLCSourceAIOConfig + | AzuracastSourceAIOConfig; diff --git a/src/backend/common/vendor/azuracast/AzuracastApiClient.ts b/src/backend/common/vendor/azuracast/AzuracastApiClient.ts new file mode 100644 index 00000000..c6fd50c0 --- /dev/null +++ b/src/backend/common/vendor/azuracast/AzuracastApiClient.ts @@ -0,0 +1,85 @@ +import { childLogger } from "@foxxmd/logging"; +import { URLData } from "../../../../core/Atomic.js"; +import { joinedUrl, normalizeWSAddress } from "../../../utils/NetworkUtils.js"; +import { AbstractApiOptions } from "../../infrastructure/Atomic.js"; +import { AzuracastData, AzuraStationResponse } from "../../infrastructure/config/source/azuracast.js"; +import AbstractApiClient from "../AbstractApiClient.js"; +import { WS, CloseEvent, ErrorEvent, RetryEvent } from 'iso-websocket' + + +export class AzuracastApiClient extends AbstractApiClient { + + declare config: AzuracastData + + urlData: URLData; + + wsNowPlaying: AzuraStationResponse + wsCurrenTime: number = 0; + socket!: WS; + + constructor(name: any, config: AzuracastData, options: AbstractApiOptions) { + super('Azuracast API', name, config, options); + + this.urlData = normalizeWSAddress(config.url); + } + + connectWS() { + + const url = joinedUrl(this.urlData.url, '/api/live/nowplaying/websocket'); + const socket = new WebSocket(url); + + socket.onopen = (e) => { + socket.send(JSON.stringify({ + subs: { + [`station:${this.config.station}`]: {"recover": true} + } + })); + }; + + socket.onerror = (e) => { + this.logger.error(e); + } + + // Handle a now-playing event from a station. Update your now-playing data accordingly. + function handleSseData(ssePayload, useTime = true) { + const jsonData = ssePayload.data; + + if (useTime && 'current_time' in jsonData) { + this.wsCurrenTime = jsonData.current_time; + } + + this.wsNowPlaying = jsonData.np as AzuraStationResponse; + } + + socket.onmessage = (e) => { + + const jsonData = JSON.parse(e.data as string); + + if ('connect' in jsonData) { + const connectData = jsonData.connect; + + if ('data' in connectData) { + // Legacy SSE data + connectData.data.forEach( + (initialRow) => handleSseData(initialRow) + ); + } else { + // New Centrifugo time format + if ('time' in connectData) { + this.wsCurrenTime = Math.floor(connectData.time / 1000); + } + + // New Centrifugo cached NowPlaying initial push. + for (const subName in connectData.subs) { + const sub = connectData.subs[subName]; + if ('publications' in sub && sub.publications.length > 0) { + sub.publications.forEach((initialRow) => handleSseData(initialRow, false)); + } + } + } + } else if ('pub' in jsonData) { + handleSseData(jsonData.pub); + } + }; + } +} \ No newline at end of file diff --git a/src/backend/sources/AzuracastSource.ts b/src/backend/sources/AzuracastSource.ts new file mode 100644 index 00000000..02efd426 --- /dev/null +++ b/src/backend/sources/AzuracastSource.ts @@ -0,0 +1,269 @@ +import { MemoryPositionalSource } from "./MemoryPositionalSource.js"; +import { sleep } from "../utils.js"; +import { RecentlyPlayedOptions } from "./AbstractSource.js"; +import { childLogger, Logger } from "@foxxmd/logging"; +import { EventEmitter } from "events"; +import { WS, CloseEvent, ErrorEvent, RetryEvent } from 'iso-websocket' +import pEvent from 'p-event'; +import { PlayObject, URLData } from "../../core/Atomic.js"; +import { UpstreamError } from "../common/errors/UpstreamError.js"; +import { + FormatPlayObjectOptions, + InternalConfig, + PlayerStateData, + PlayPlatformId, + REPORTED_PLAYER_STATUSES, + SINGLE_USER_PLATFORM_ID, +} from "../common/infrastructure/Atomic.js"; +import { AzuracastSourceConfig, AzuraNowPlayingResponse, AzuraStationResponse } from "../common/infrastructure/config/source/azuracast.js"; +import { isPortReachable, normalizeWSAddress } from "../utils/NetworkUtils.js"; +import { PlayerStateOptions } from "./PlayerState/AbstractPlayerState.js"; +import { AzuracastPlayerState } from "./PlayerState/AzuracastPlayerState.js"; + + +export class AzuracastSource extends MemoryPositionalSource { + + declare config: AzuracastSourceConfig; + + urlData!: URLData; + + wsNowPlaying: AzuraStationResponse + wsCurrenTime: number = 0; + client!: WS; + + + constructor(name: any, config: AzuracastSourceConfig, internal: InternalConfig, emitter: EventEmitter) { + const { + data = {} + } = config; + const { + ...rest + } = data; + super('azuracast', name, { ...config, data: { ...rest } }, internal, emitter); + + const { + data: { + url, + } = {} + } = config; + this.requiresAuth = false; + this.canPoll = true; + } + + protected async doBuildInitData(): Promise { + const { + data: { + url + } = {} + } = this.config; + if (url === null || url === undefined || url === '') { + throw new Error('url must be defined'); + } + this.urlData = normalizeWSAddress(url, { defaultPath: '/api/live/nowplaying/websocket' }); + const normal = this.urlData.normal; + this.logger.verbose(`Config URL: '${url ?? '(None Given)'}' => Normalized: '${normal}'`) + if (!normal.includes('ws://') && !normal.includes('wss://')) { + throw new Error(`Server URL must be start with with ws:// or wss://`); + } + this.client = new WS(this.urlData.url.toString(), { + automaticOpen: false, + retry: { + retries: 0 + } + }); + const wsLogger = childLogger(this.logger, 'WS'); + this.client.addEventListener('retry', (e) => { + wsLogger.verbose(`Retrying connection, attempt ${e.attempt}`, { labels: 'WS' }); + }); + this.client.addEventListener('close', (e) => { + wsLogger.warn(`Connection was closed: ${e.code} => ${e.reason}`, { labels: 'WS' }); + if (e.reason.includes('unauthenticated')) { + this.authed = false; + } + }); + this.client.addEventListener('open', (e) => { + wsLogger.verbose(`Connection was established.`, { labels: 'WS' }); + // if (this.authed) { + // // was a reconnect, try auto authenticating + // wsLogger.verbose('Resending auth message after (probably) reconnection...'); + // this.client.send(JSON.stringify(this.getAuthPayload())); + // } + }); + this.client.addEventListener('error', (e) => { + if (e.message.includes('Connection failed after')) { + this.connectionOK = false; + //this.authed = false; + } + const hint = e.error?.cause?.message ?? undefined; + wsLogger.error(new Error(`Communication with server failed${hint !== undefined ? ` (${hint})` : ''}`, { cause: e.error })); + }); + + this.client.addEventListener('message', (e) => { + this.parseWSData(getMessageData(e)); + // if (isAuthenticateResponse(data)) { + // wsLogger.verbose(`${!data.options.authenticated ? 'NOT ' : ''}Authenticated for Muiskcube ${data.options.environment.app_version} with API v${data.options.environment.api_version}`); + // } + }); + return true; + } + + private parseWSPayload(payload: any, useTime = true) { + const jsonData = payload.data; + + if (useTime && 'current_time' in jsonData) { + this.wsCurrenTime = jsonData.current_time; + } + + this.wsNowPlaying = jsonData.np as AzuraStationResponse; + } + + private parseWSData(jsonData: any) { + + if ('connect' in jsonData) { + const connectData = jsonData.connect; + + if ('data' in connectData) { + // Legacy SSE data + connectData.data.forEach( + (initialRow) => this.parseWSPayload(initialRow) + ); + } else { + // New Centrifugo time format + if ('time' in connectData) { + this.wsCurrenTime = Math.floor(connectData.time / 1000); + } + + // New Centrifugo cached NowPlaying initial push. + for (const subName in connectData.subs) { + const sub = connectData.subs[subName]; + if ('publications' in sub && sub.publications.length > 0) { + sub.publications.forEach((initialRow) => this.parseWSPayload(initialRow, false)); + } + } + } + } else if ('pub' in jsonData) { + this.parseWSPayload(jsonData.pub); + } + } + + protected async doCheckConnection(): Promise { + try { + try { + await isPortReachable(this.urlData.port, { host: this.urlData.url.hostname }); + this.logger.verbose(`${this.urlData.url.hostname}:${this.urlData.port} is reachable.`); + } catch (e) { + throw e; + } + + this.client.open(); + const opened = await pEvent(this.client, 'open'); + return true; + } catch (e) { + this.client.close(); + const hint = e.error?.cause?.message ?? undefined; + throw new Error(`Could not connect to Azuracast server${hint !== undefined ? ` (${hint})` : ''}`, { cause: e.error ?? e }); + } + } + + onPollPostAuthCheck = async (): Promise => { + this.logger.verbose(`Listening for activity on Station ${this.config.data.station}`); + this.client.send(JSON.stringify({ + subs: { + [`station:${this.config.data.station}`]: { "recover": true } + } + })); + return true; + } + + // TODO return based on user intervention + protected isStationValidListen = () => { + if(this.wsNowPlaying === undefined) { + this.logger.debug({leaf: `Station ${this.config.data.station}`}, `No data returned yet`); + return false; + } + if(!this.wsNowPlaying.is_online && this.config.data.monitorWhenLive) { + this.logger.debug({leaf: `Station ${this.config.data.station}`}, `Currently offline`); + return false; + } + if(this.config.data.monitorWhenListeners !== undefined) { + if(this.config.data.monitorWhenListeners === true && this.wsNowPlaying.listeners.current === 0) { + this.logger.debug({leaf: `Station ${this.config.data.station}`}, `No listeners`); + return false; + } + if(typeof this.config.data.monitorWhenListeners === 'number' && this.wsNowPlaying.listeners.current < this.config.data.monitorWhenListeners) { + this.logger.debug({leaf: `Station ${this.config.data.station}`}, `Requries ${this.config.data.monitorWhenListeners} listeners to be active but currently only ${this.wsNowPlaying.listeners.current}`); + return false; + } + } + return true; + } + + getRecentlyPlayed = async (options: RecentlyPlayedOptions = {}) => { + if (this.client.readyState !== this.client.OPEN) { + throw new Error('WS connection is no longer open.'); + } + + let play: PlayObject | undefined; + const online = this.isStationValidListen(); + + if(this.isStationValidListen() && this.wsNowPlaying.now_playing !== undefined) { + play = formatPlayObj(this.wsNowPlaying.now_playing); + } + + const playerState: PlayerStateData = { + platformId: SINGLE_USER_PLATFORM_ID, + status: online ? REPORTED_PLAYER_STATUSES.playing : REPORTED_PLAYER_STATUSES.stopped, + play, + position: online && play !== undefined ? play.meta.trackProgressPosition : undefined + } + + return this.processRecentPlays([playerState]); + } + + getNewPlayer = (logger: Logger, id: PlayPlatformId, opts: PlayerStateOptions) => new AzuracastPlayerState(logger, id, opts); +} + +const formatPlayObj = (obj: AzuraNowPlayingResponse, options: FormatPlayObjectOptions = {}): PlayObject => { + + const { + song, + duration, + elapsed, + remaining, + } = obj; + + const { + text, + artist, + title, + album + } = song; + + const track: string = title ?? text; + + return { + data: { + artists: artist !== undefined && artist !== '' ? [artist] : [], + album: album !== '' ? album : undefined, + track, + duration + }, + meta: { + trackProgressPosition: elapsed + } + } +} + +const getMessageData = (e: any): T => { + return JSON.parse(e.data) as T; +} + +const isCloseEvent = (e: Event): e is CloseEvent => { + return e.type === 'close'; +} +const isErrorEvent = (e: Event): e is ErrorEvent => { + return e.type === 'error'; +} +const isRetryEvent = (e: Event): e is RetryEvent => { + return e.type === 'retry'; +} diff --git a/src/backend/sources/MusikcubeSource.ts b/src/backend/sources/MusikcubeSource.ts index d39bdc75..b1d17f52 100644 --- a/src/backend/sources/MusikcubeSource.ts +++ b/src/backend/sources/MusikcubeSource.ts @@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto"; import normalizeUrl from 'normalize-url'; import pEvent from 'p-event'; import { URL } from "url"; -import { PlayObject } from "../../core/Atomic.js"; +import { PlayObject, URLData } from "../../core/Atomic.js"; import { UpstreamError } from "../common/errors/UpstreamError.js"; import { FormatPlayObjectOptions, @@ -23,6 +23,7 @@ import { import { sleep } from "../utils.js"; import { RecentlyPlayedOptions } from "./AbstractSource.js"; import { MemoryPositionalSource } from "./MemoryPositionalSource.js"; +import { normalizeWSAddress } from "../utils/NetworkUtils.js"; const CLIENT_STATE = { 0: 'connecting', @@ -34,7 +35,7 @@ const CLIENT_STATE = { export class MusikcubeSource extends MemoryPositionalSource { declare config: MusikcubeSourceConfig; - url: URL; + url: URLData; client!: WS; @@ -56,46 +57,23 @@ export class MusikcubeSource extends MemoryPositionalSource { } = {} } = config; this.deviceId = device_id ?? name; - this.url = MusikcubeSource.parseConnectionUrl(url); + this.url = normalizeWSAddress(url, {defaultPort: 7905}); this.requiresAuth = true; this.canPoll = true; } - static parseConnectionUrl(valRaw: string) { - let val = valRaw.trim(); - if(!val.match(/^(?:wss?|https?):/i)) { - val = `ws://${val}`; - } - const normal = normalizeUrl(val, {removeTrailingSlash: false}) - const url = new URL(normal); - - // default WS - if (url.protocol === 'https:') { - url.protocol = 'wss:'; - } else if (url.protocol === 'http:') { - url.protocol = 'ws:'; - } else { - url.protocol = 'ws:' - } - - if (url.port === null || url.port === '') { - url.port = '7905'; - } - return url; - } - protected async doBuildInitData(): Promise { const { data: { url } = {} } = this.config; - const normal = this.url.toString(); + const normal = this.url.normal; this.logger.verbose(`Config URL: '${url ?? '(None Given)'}' => Normalized: '${normal}'`) if (!normal.includes('ws://') && !normal.includes('wss://')) { - throw new Error(`Server URL must be start with with ws:// or wss://`); + throw new Error(`Server URL must start with ws:// or wss://`); } - this.client = new WS(this.url.toString(), { + this.client = new WS(this.url.url.toString(), { automaticOpen: false, retry: { retries: 0 diff --git a/src/backend/sources/PlayerState/AzuracastPlayerState.ts b/src/backend/sources/PlayerState/AzuracastPlayerState.ts new file mode 100644 index 00000000..772808ba --- /dev/null +++ b/src/backend/sources/PlayerState/AzuracastPlayerState.ts @@ -0,0 +1,16 @@ +import { Logger } from "@foxxmd/logging"; +import { PlayPlatformId, REPORTED_PLAYER_STATUSES } from "../../common/infrastructure/Atomic.js"; +import { AbstractPlayerState, PlayerStateOptions } from "./AbstractPlayerState.js"; +import { GenericPlayerState } from "./GenericPlayerState.js"; +import { PositionalPlayerState } from "./PositionalPlayerState.js"; + +export class AzuracastPlayerState extends PositionalPlayerState { + constructor(logger: Logger, platformId: PlayPlatformId, opts?: PlayerStateOptions) { + super(logger, platformId, {allowedDrift: 17000, rtTruth: true, ...(opts || {})}); + this.gracefulEndBuffer = this.allowedDrift / 1000; + } + + protected isSessionStillPlaying(position: number): boolean { + return this.reportedStatus === REPORTED_PLAYER_STATUSES.playing; + } +} diff --git a/src/backend/sources/ScrobbleSources.ts b/src/backend/sources/ScrobbleSources.ts index 6dfc1d45..8505a8a6 100644 --- a/src/backend/sources/ScrobbleSources.ts +++ b/src/backend/sources/ScrobbleSources.ts @@ -3,6 +3,7 @@ import { childLogger, Logger } from '@foxxmd/logging'; import EventEmitter from "events"; import { ConfigMeta, InternalConfig, isSourceType, SourceType, sourceTypes } from "../common/infrastructure/Atomic.js"; import { AIOConfig, SourceDefaults } from "../common/infrastructure/config/aioConfig.js"; +import { AzuracastData, AzuracastSourceConfig } from "../common/infrastructure/config/source/azuracast.js"; import { ChromecastSourceConfig } from "../common/infrastructure/config/source/chromecast.js"; import { DeezerData, DeezerSourceConfig } from "../common/infrastructure/config/source/deezer.js"; import { @@ -31,6 +32,7 @@ import { WildcardEmitter } from "../common/WildcardEmitter.js"; import { parseBool, readJson } from "../utils.js"; import { validateJson } from "../utils/ValidationUtils.js"; import AbstractSource from "./AbstractSource.js"; +import { AzuracastSource } from "./AzuracastSource.js"; import { ChromecastSource } from "./ChromecastSource.js"; import DeezerSource from "./DeezerSource.js"; import JellyfinApiSource from "./JellyfinApiSource.js"; @@ -169,6 +171,9 @@ export default class ScrobbleSources { case 'vlc': this.schemaDefinitions[type] = getTypeSchemaFromConfigGenerator("VLCSourceConfig"); break; + case 'azuracast': + this.schemaDefinitions[type] = getTypeSchemaFromConfigGenerator("AzuracastSourceConfig"); + break; } } return this.schemaDefinitions[type]; @@ -501,6 +506,25 @@ export default class ScrobbleSources { data: ytm as YTMusicData }); } + break; + case 'azuracast': + const azura = { + station: process.env.AZURA_STATION, + url: process.env.AZURA_URL, + monitorWhenListeners: process.env.AZURA_LISTENERS_NUM, + monitorWhenLive: process.env.AZURA_LIVE, + apiKey: process.env.AZURA_KEY + } + if (!Object.values(azura).every(x => x === undefined)) { + configs.push({ + type: 'azuracast', + name: 'unnamed', + source: 'ENV', + mode: 'single', + configureAs: defaultConfigureAs, + data: azura as unknown as AzuracastData + }); + } break; default: break; @@ -681,6 +705,9 @@ export default class ScrobbleSources { case 'vlc': newSource = await new VLCSource(name, compositeConfig as VLCSourceConfig, this.internalConfig, this.emitter); break; + case 'azuracast': + newSource = await new AzuracastSource(name, compositeConfig as AzuracastSourceConfig, this.internalConfig, this.emitter); + break; default: break; } diff --git a/src/backend/utils/NetworkUtils.ts b/src/backend/utils/NetworkUtils.ts index 0d60b1b1..5192fc10 100644 --- a/src/backend/utils/NetworkUtils.ts +++ b/src/backend/utils/NetworkUtils.ts @@ -79,6 +79,50 @@ export const normalizeWebAddress = (val: string): URLData => { } } +export const normalizeWSAddress = (val: string, options: {defaultPort?: number | string, defaultPath?: string} = {}): URLData => { + let cleanUserUrl = val.trim(); + const results = parseRegexSingle(QUOTES_UNWRAP_REGEX, val); + if (results !== undefined && results.groups && results.groups.length > 0) { + cleanUserUrl = results.groups[0]; + } + if(!cleanUserUrl.match(/^(?:wss?|https?):/i)) { + cleanUserUrl = `ws://${cleanUserUrl}`; + } + const normal = normalizeUrl(val, {removeTrailingSlash: false}) + const url = new URL(normal); + + // default WS + if (url.protocol === 'https:') { + url.protocol = 'wss:'; + } else if (url.protocol === 'http:') { + url.protocol = 'ws:'; + } else if(url.protocol === '') { + url.protocol = 'ws:' + } + + const {defaultPort, defaultPath} = options; + + let port: number; + if(url.port === null || url.port === '') { + if(defaultPort !== undefined) { + url.port = defaultPort.toString(); + port = parseInt(url.port); + } else { + port = url.protocol === 'ws:' ? 80 : 443; + } + } + + if(url.pathname === '/' && defaultPath !== undefined) { + url.pathname = defaultPath; + } + + return { + url, + normal: url.toString(), + port + } +} + export const generateBaseURL = (userUrl: string | undefined, defaultPort: number | string): URL => { const urlStr = userUrl ?? `http://localhost:${defaultPort}`; let cleanUserUrl = urlStr.trim();