"""
Utilitaires reseau et conversion audio pour l'infrastructure serveur de Mirza.
Fournit la detection d'IP locale, l'introspection du reseau Tailscale,
et l'encapsulation de flux PCM en trames WAV RIFF pour la diffusion reseau.
"""
from __future__ import annotations
import json
import os
import socket
import struct
import subprocess
from .logger import getMirzaLogger
logger = getMirzaLogger("mirza.core.network")
[docs]
def pcm16ToWavBytes(pcmData: bytes, sampleRate: int = 16000) -> bytes:
"""
Encapsule un flux audio brut PCM 16-bit mono dans un conteneur WAV standard RIFF.
Parameters
----------
pcmData : bytes
Flux d'octets binaires representant les echantillons PCM 16-bit signes.
sampleRate : int, optional
Frequence d'echantillonnage en Hertz (defaut: 16000).
Returns
-------
bytes
Fichier WAV complet avec en-tete de 44 octets concatene aux donnees PCM.
"""
header = bytearray(44)
header[0:4] = b"RIFF"
struct.pack_into("<I", header, 4, 36 + len(pcmData))
header[8:12] = b"WAVE"
header[12:16] = b"fmt "
struct.pack_into("<IHHIIHH", header, 16, 16, 1, 1, sampleRate, sampleRate * 2, 2, 16)
header[36:40] = b"data"
struct.pack_into("<I", header, 40, len(pcmData))
return bytes(header) + pcmData
[docs]
def getLocalIp() -> str:
"""
Detecte l'adresse IP locale de la machine hote sur le reseau local (LAN/Wi-Fi).
Returns
-------
str
Adresse IPv4 locale (ex: "192.168.1.42"), ou "127.0.0.1" en cas d'echec.
"""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0.2)
s.connect(("8.8.8.8", 80))
localIp = s.getsockname()[0]
s.close()
return localIp
except Exception as exc:
logger.debug("Echec de resolution IP locale: %s", exc)
return "127.0.0.1"
[docs]
def getTailscaleInfo() -> tuple[str | None, str | None]:
"""
Detecte l'adresse IP Tailscale IPv4 et le nom DNS de la machine.
Returns
-------
tuple[str | None, str | None]
(adresse IPv4 Tailscale, nom DNS MagicDNS). None quand le service est
inactif.
"""
userSocket = os.path.expanduser("~/.local/share/tailscale/tailscaled.sock")
cmdBase = []
if os.path.exists(userSocket):
cmdBase = ["tailscale", f"--socket={userSocket}"]
else:
cmdBase = ["tailscale"]
tailscaleIp: str | None = None
dnsName: str | None = None
try:
res = subprocess.run(
[*cmdBase, "ip", "-4"],
capture_output=True,
text=True,
timeout=1.5,
check=False,
)
if res.returncode == 0 and res.stdout.strip():
tailscaleIp = res.stdout.strip().split("\n")[0]
except Exception as exc:
logger.debug("Echec detection IP Tailscale: %s", exc)
try:
res = subprocess.run(
[*cmdBase, "status", "--json"],
capture_output=True,
text=True,
timeout=1.5,
check=False,
)
if res.returncode == 0 and res.stdout.strip():
data = json.loads(res.stdout)
dns = data.get("Self", {}).get("DNSName", "").rstrip(".")
if dns:
dnsName = dns
except Exception as exc:
logger.debug("Echec detection DNS Tailscale: %s", exc)
return tailscaleIp, dnsName
[docs]
def getTailscaleIp() -> str | None:
"""
Detecte l'adresse IP Tailscale IPv4 de la machine.
Returns
-------
str | None
Adresse IPv4 Tailscale si le service est actif et connecte, None sinon.
"""
ip, _ = getTailscaleInfo()
return ip