audio
June 28, 2023
Вывод аудио потока в numpy
Подключаемся к каналу. и выводим в консоль в формате numpy.
добавлены комментарии, исключительные события (сообщения об ошибках), запись лога о статусе ответа ресурса.
import requests # Импортируем модуль requests для выполнения HTTP-запросов
import numpy as np # Импортируем модуль NumPy для работы с массивами
from datetime import datetime # Импортируем класс datetime из модуля datetime для работы с датой и временем
url = "https://icecast-vgtrk.cdnvideo.ru/vestifm_mp3_192kbps" # URL ресурса, к которому осуществляется подключение
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
} # Заголовки запроса, в данном случае указывается User-Agent для эмуляции браузера
try:
# Подключаемся к потоку аудио
response = requests.get(url, headers=headers, stream=True) # Выполняем GET-запрос к ресурсу с переданными заголовками и параметром stream=True
status_code = response.status_code # Получаем HTTP-код ответа
log_file_path = 'log.txt' # Путь к файлу журнала логов
with open(log_file_path, 'r+') as log_file: # Открываем файл журнала логов в режиме чтения и записи
current_logs = log_file.read() # Читаем текущее содержимое файла
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Получаем текущую дату и время в формате "%Y-%m-%d %H:%M:%S"
log_entry = f"{timestamp} - Status Code: {status_code}\n" # Формируем запись для журнала логов
log_file.seek(0, 0) # Переходим в начало файла
log_file.write(log_entry + current_logs) # Записываем новую запись перед старыми данными
if response.status_code == 200: # Если HTTP-код ответа равен 200 (успешный ответ)
chunk_size = 1024 # Размер блока данных, которые будут читаться из потока
stream = response.iter_content(chunk_size=chunk_size) # Получаем итератор по данным из потока
for data in stream:
# Конвертируем байты в массив NumPy
samples = np.frombuffer(data, dtype=np.int16) # Преобразуем данные из байтов в массив NumPy с типом int16
print(samples) # Выводим значения на экран
else: # Если HTTP-код ответа не равен 200
print("Ошибка подключения ресурса:", status_code)
except requests.exceptions.RequestException as e: # Обрабатываем ошибки связанные с выполнением запроса
print("Ошибка подключения к ресурсу:", str(e))
except IOError: # Обрабатываем ошибки ввода-вывода при работе с файлами
print("Ошибка открытия файла.")
finally:
response.close() # Закрываем соединение с ресурсом, независимо от того, было ли оно успешным или нет