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