File size: 2,045 Bytes
478eeb0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
import wave
import io
import os
import json
import requests
import yaml
def load_yaml():
"""
Loads a YAML file and extracts the 'model_name' string.
"""
file_path = "../config.yaml"
try:
with open(file_path, 'r') as file:
config = yaml.safe_load(file)
return config
except FileNotFoundError:
print(f"Error: The file '{file_path}' was not found.")
return None
except yaml.YAMLError as e:
print(f"Error parsing YAML file: {e}")
return None
config = load_yaml()
SYNC_PORT = config["SYNC_PORT"]
MOTION_SYNC_URL = f"http://localhost:{SYNC_PORT}/update_duration"
def update_motion_generator_duration(scene_id: str, motion_index: int, duration: float):
"""Sends the actual audio duration to the motion generator's sync server."""
try:
payload = {
"sceneId": scene_id,
"motionIndex": motion_index,
"duration": duration
}
# This sends the data to the server running inside the motion generator script
response = requests.post(MOTION_SYNC_URL, json=payload, timeout=5)
if response.status_code == 200:
print(f"[ sent duration for {scene_id}:{motion_index} ]")
else:
print(f"⚠️ Failed to notify motion generator for {scene_id}_{motion_index}. Status: {response.status_code}, Response: {response.text}")
except requests.exceptions.RequestException as e:
print(f"⚠️ Error connecting to motion generator sync server: {e}")
def calculate_duration_from_bytes(audio_bytes: bytes) -> float:
"""Calculates audio duration in seconds directly from WAV bytes in memory."""
try:
with wave.open(io.BytesIO(audio_bytes), 'rb') as wf:
frames = wf.getnframes()
rate = wf.getframerate()
return frames / float(rate) if rate > 0 else 0.0
except (wave.Error, ZeroDivisionError) as e:
print(f"⚠️ Could not calculate duration from bytes: {e}")
return 0.0
|