add mic recording, rename scripts, extend camera dump
- Rename main.py → play_speaker.py - Add record_mic.py: record audio from camera mic via RTSP to a timestamped WAV in data/ - Add private API camera dump to dump_camera.py - Update README scripts section
This commit is contained in:
78
play_speaker.py
Normal file
78
play_speaker.py
Normal file
@@ -0,0 +1,78 @@
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
import httpx
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
HOST = os.environ["HOST"]
|
||||
API_KEY = os.environ["API_KEY"]
|
||||
USERNAME = os.environ["UNIFI_USERNAME"]
|
||||
PASSWORD = os.environ["UNIFI_PASSWORD"]
|
||||
|
||||
BASE = f"https://{HOST}/proxy/protect/integration/v1"
|
||||
PRIVATE_BASE = f"https://{HOST}/proxy/protect/api"
|
||||
|
||||
CAMERA_NAME = "G6 Pro 360"
|
||||
AUDIO_FILE = "./data/hello.wav"
|
||||
SPEAKER_VOLUME = 50 # 0-100
|
||||
|
||||
|
||||
def set_speaker_volume(camera_id: str, volume: int):
|
||||
# Accept-Encoding must be suppressed at request level — server returns 403 when present
|
||||
with httpx.Client(verify=False, http2=True) as client:
|
||||
res = client.post(
|
||||
f"https://{HOST}/api/auth/login",
|
||||
content=f'{{"username":"{USERNAME}","password":"{PASSWORD}","rememberMe":false}}'.encode(),
|
||||
headers={"Content-Type": "application/json", "Accept-Encoding": ""},
|
||||
)
|
||||
res.raise_for_status()
|
||||
csrf = res.headers.get("x-csrf-token")
|
||||
|
||||
def get_volumes():
|
||||
s = client.get(f"{PRIVATE_BASE}/cameras/{camera_id}").raise_for_status().json()["speakerSettings"]
|
||||
return s["volume"], s["speakerVolume"]
|
||||
|
||||
before = get_volumes()
|
||||
print(f"Speaker volume before: volume={before[0]} speakerVolume={before[1]}")
|
||||
|
||||
client.patch(
|
||||
f"{PRIVATE_BASE}/cameras/{camera_id}",
|
||||
headers={"x-csrf-token": csrf},
|
||||
json={"speakerSettings": {"volume": volume, "speakerVolume": volume}},
|
||||
).raise_for_status()
|
||||
|
||||
after = get_volumes()
|
||||
print(f"Speaker volume after: volume={after[0]} speakerVolume={after[1]}")
|
||||
|
||||
|
||||
def main():
|
||||
with httpx.Client(headers={"X-API-Key": API_KEY}, verify=False) as client:
|
||||
cameras = client.get(f"{BASE}/cameras").raise_for_status().json()
|
||||
camera = next(c for c in cameras if CAMERA_NAME in c["name"])
|
||||
print(f"Camera: {camera['name']} ({camera['id']})")
|
||||
|
||||
set_speaker_volume(camera["id"], SPEAKER_VOLUME)
|
||||
|
||||
session = client.post(f"{BASE}/cameras/{camera['id']}/talkback-session").raise_for_status().json()
|
||||
print(f"Talkback: {session['url']} ({session['codec']} {session['samplingRate']}Hz)")
|
||||
|
||||
proc = subprocess.Popen([
|
||||
"ffmpeg", "-re", "-i", AUDIO_FILE,
|
||||
"-acodec", "libopus",
|
||||
"-ar", str(session["samplingRate"]),
|
||||
"-ac", "1",
|
||||
"-b:a", "64k",
|
||||
"-f", "rtp", session["url"],
|
||||
])
|
||||
|
||||
try:
|
||||
proc.wait()
|
||||
except KeyboardInterrupt:
|
||||
proc.terminate()
|
||||
proc.wait()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user