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()  # Закрываем соединение с ресурсом, независимо от того, было ли оно успешным или нет