from aiohttp import web
from music_assistant.common.helpers.util import get_ip, select_free_port
-from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
+from music_assistant.common.models.config_entries import (
+ ConfigEntry,
+ ConfigValueOption,
+ ConfigValueType,
+)
from music_assistant.common.models.enums import ConfigEntryType, ContentType
from music_assistant.common.models.errors import MediaNotFoundError, QueueEmpty
from music_assistant.common.models.media_items import AudioFormat
CONF_EQ_TREBLE,
CONF_OUTPUT_CHANNELS,
CONF_OUTPUT_CODEC,
+ CONF_PUBLISH_IP,
)
from music_assistant.server.helpers.audio import (
check_audio_support,
get_stream_details,
)
from music_assistant.server.helpers.process import AsyncProcess
+from music_assistant.server.helpers.util import get_ips
from music_assistant.server.helpers.webserver import Webserver
from music_assistant.server.models.core_controller import CoreController
) -> tuple[ConfigEntry, ...]:
"""Return all Config Entries for this core module (if any)."""
default_ip = await get_ip()
+ all_ips = await get_ips()
default_port = await select_free_port(8096, 9200)
return (
ConfigEntry(
"on the given IP and TCP port by players on the local network.",
),
ConfigEntry(
- key=CONF_BIND_IP,
+ key=CONF_PUBLISH_IP,
type=ConfigEntryType.STRING,
default_value=default_ip,
- label="Bind to IP/interface",
- description="Start the streamserver on this specific interface. \n"
- "This IP address is communicated to players where to find this server. "
+ label="Published IP address",
+ description="This IP address is communicated to players where to find this server. "
"Override the default in advanced scenarios, such as multi NIC configurations. \n"
"Make sure that this server can be reached "
"on the given IP and TCP port by players on the local network. \n"
"not be adjusted in regular setups.",
advanced=True,
),
+ ConfigEntry(
+ key=CONF_BIND_IP,
+ type=ConfigEntryType.STRING,
+ default_value="0.0.0.0",
+ options=(ConfigValueOption(x, x) for x in {"0.0.0.0", *all_ips}),
+ label="Bind to IP/interface",
+ description="Start the stream server on this specific interface. \n"
+ "Use 0.0.0.0 to bind to all interfaces, which is the default. \n"
+ "This is an advanced setting that should normally "
+ "not be adjusted in regular setups.",
+ advanced=True,
+ ),
)
async def setup(self, config: CoreConfig) -> None:
)
# start the webserver
self.publish_port = config.get_value(CONF_BIND_PORT)
- self.publish_ip = config.get_value(CONF_BIND_IP)
+ self.publish_ip = config.get_value(CONF_PUBLISH_IP)
await self._server.setup(
- bind_ip=self.publish_ip,
+ bind_ip=config.get_value(CONF_BIND_IP),
bind_port=self.publish_port,
base_url=f"http://{self.publish_ip}:{self.publish_port}",
static_routes=[
MessageType,
SuccessResultMessage,
)
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueOption
from music_assistant.common.models.enums import ConfigEntryType
from music_assistant.common.models.errors import InvalidCommand
from music_assistant.common.models.event import MassEvent
# HA supervisor not present: user is responsible for securing the webserver
# we give the tools to do so by presenting config options
default_ip = await get_ip()
+ all_ips = await get_ips()
default_port = await select_free_port(8095, 9200)
default_base_url = f"http://{default_ip}:{default_port}"
return (
key=CONF_BIND_IP,
type=ConfigEntryType.STRING,
default_value="0.0.0.0",
+ options=(ConfigValueOption(x, x) for x in {"0.0.0.0", *all_ips}),
label="Bind to IP/interface",
description="Start the (web)server on this specific interface. \n"
"Use 0.0.0.0 to bind to all interfaces. \n"
import importlib
import logging
import platform
-import socket
import tempfile
import urllib.error
import urllib.parse
from importlib.metadata import version as pkg_version
from typing import TYPE_CHECKING
+import ifaddr
import memory_tempfile
if TYPE_CHECKING:
return "0.0.0"
-async def get_ips(include_ipv6: bool = False) -> set[str]:
+async def get_ips(include_ipv6: bool = False, ignore_loopback: bool = True) -> set[str]:
"""Return all IP-adresses of all network interfaces."""
def call() -> set[str]:
result: set[str] = set()
- for item in socket.getaddrinfo(socket.gethostname(), None):
- protocol, *_, (ip, *_) = item
- if protocol == socket.AddressFamily.AF_INET or (
- include_ipv6 and protocol == socket.AddressFamily.AF_INET6
- ):
- result.add(ip)
+ adapters = ifaddr.get_adapters()
+ for adapter in adapters:
+ for ip in adapter.ips:
+ if ip.is_IPv6 and not include_ipv6:
+ continue
+ if ip.ip == "127.0.0.1" and ignore_loopback:
+ continue
+ result.add(ip.ip)
return result
return await asyncio.to_thread(call)