音频直播Icecast服务器以及liquidsoap推流

icecast是一个音频直播流媒体服务器，支持Ogg(Vorbis和Theora)，Opus，WebM和MP3流。他可以被用来创建网络电台。
官方网站
Gitlab源码地址

Icecast本身只是个直播服务器，一般使用方式是使用推流客户端（例如IceS，liquidsoap等）推流到icecast server，然后用户从icecast收听。

Icecast server安装使用

1. 下载icecast包，下载地址
2. 解压，进入解压后目录icecast-2.4.4，执行./configure并根据提示安装缺失的包
3. 成功后执行make && make install安装完成
4. 会自动生成一个配置文件在/usr/local/etc/icecast.xml，编辑配置文件，主要是服务ip、端口、流地址和密码
5. 启动命令su - appuser -c "icecast -b -c /usr/local/etc/icecast.xml"，即可在后台启动

IceS推流

IceS作为推流端使用简便，问题是指支持Ogg Vorbis编码和MP3编码

1. 下载ices，下载地址
2. 解压，进入解压后目录ices-2.0.2，执行./configure并根据提示安装缺失的包
3. 成功后执行make && make install安装完成
4. 几个示例配置文件在当前目录下的conf目录中，ices-xxx.xml
5. 复制ices-playlist.xml模板，修改icecast服务器参数，包括ip、端口、密码和stream名字
6. 修改playlist.txt文件，每行一个ogg vorbis编码的音频文件路径
7. 执行ices conf/ices-playlist.xml，即可把流推到icecast的服务器了
8. 使用icecast服务器的ip:port/streamname就可以开始收听广播

liquidsoap推流

liquidsoap是使用ocaml编译的成宿，编译安装好之后会生成liquidsoap指令，使用其自定义的脚本语言可以编写复杂的推流方式以及使用外部编解码器
由于支持外部编解码器，所以理论上liquidsoap可以支持所有格式和编码
官方网站
Github源码地址

安装流程

# 安装依赖包
yum install ocaml bubblewrap libao bzip2
# faad用于decode来源是audio/aac格式
wget https://li.nux.ro/download/nux/dextop/el7/x86_64/faad2-libs-2.7-5.el7.nux.x86_64.rpm
rpm -Uvh faad2-libs-2.7-5.el7.nux.x86_64.rpm
wget https://li.nux.ro/download/nux/dextop/el7/x86_64/faad2-devel-2.7-5.el7.nux.x86_64.rpm
rpm -Uvh faad2-devel-2.7-5.el7.nux.x86_64.rpm
# fdkaac用户编码aac格式输出
rpm --import http://li.nux.ro/download/nux/RPM-GPG-KEY-nux.ro
rpm -Uvh http://li.nux.ro/download/nux/dextop/el7/x86_64/nux-dextop-release-0-5.el7.nux.noarch.rpm
yum install -y ffmpeg ffmpeg-devel fdk-aac-devel

# 安装opam包管理工具
sh <(curl -sL https://raw.githubusercontent.com/ocaml/opam/master/shell/install.sh)
# 初始化环境，安装liquidsoap包
opam init
opam switch create 4.08.0
opam depext taglib mad lame vorbis cry samplerate faad opus fdkaac liquidsoap 
opam install taglib mad lame vorbis cry samplerate faad opus fdkaac liquidsoap

使用方式

可以使用一行式指令，如下会把从http://c16.radioboss.fm/stream/5直播流，转码成opus编码推给icecast服务器

liquidsoap '
  output.icecast(%opus(bitrate=16),
     host = "172.31.34.113", port = 8000,
     password = "hackme", mount = "mystream.ogg",
     mksafe(input.https("https://c16.radioboss.fm/stream/5")))'

python-shout

这个包是用python语言实现的读取数据流，发送到icecast server的工具，使用简单，不包含转码，需要手动转码

pip install python-shout

下方脚本是连续下载url中音频存为mp3，然后使用ffmpeg转换为opus编码的ogg文件，最后把ogg文件读取推送到icecast server

import logging
import os
import threading
import urllib

logging.basicConfig(format='%(asctime)s [PID:%(process)d] %(name)s %(threadName)s \"%(filename)s\"[:%(lineno)d] '
                           '%(levelname)s %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
logger.info('loading %s', __name__)

import datetime
import shout
import time
import queue

from pydub import AudioSegment


class Converter:

    def __init__(self, url, delete_file=True) -> None:
        self.url = url
        self.delete_file = delete_file
        self.convert_q = queue.Queue()
        self.send_q = queue.Queue()
        self.read_thread = threading.Thread(target=self.write_to_temp_file)
        self.convert_thread = threading.Thread(target=self.convert_opus)
        self.send_thread = threading.Thread(target=self.send_to_icecast)

    def write_to_temp_file(self):
        response = urllib.request.urlopen(self.url)
        while True:
            f = open(self.get_file_name("1111rdio0001"), 'wb')
            video_file_size_start = 0
            video_file_size_end = 100 * 1024  # end in 30 k
            block_size = 1024
            logger.info(f"ready to read from url={response.url} to file={f.name}")
            st = time.time()
            while True:
                try:
                    # buffer = response.read(block_size)
                    buffer = response.readline()
                    if not buffer:
                        break
                    video_file_size_start += len(buffer)
                    if video_file_size_start > video_file_size_end:
                        break
                    f.write(buffer)

                except Exception as e:
                    logger.exception(e)
            et = time.time()
            logger.info(f"read {video_file_size_start / 1024}k data in {round(et - st, 2)}s")
            f.close()
            self.convert_q.put(f.name)

    def convert_opus(self):
        while True:
            fa = self.convert_q.get(block=True)
            opus_fa = "re_" + fa.split(".")[0] + ".ogg"
            AudioSegment.from_mp3(fa).export(opus_fa, format='opus', bitrate="16k")
            self.send_q.put(opus_fa)
            if self.delete_file and os.path.exists(fa):
                os.remove(fa)

    def send_to_icecast(self):
        s = shout.Shout()
        s.host = "172.31.34.113"
        s.port = 8000
        s.password = "hackme"
        s.mount = "/mystream.ogg"
        # s.format = "mp3"
        
        s.open()

        while True:
            try:

                total = 0
                st = time.time()
                fa = self.send_q.get(block=True)
                logger.info("send_to_icecast opening file %s" % fa)
                f = open(fa, 'rb')
                # s.set_metadata({'song': fa})

                nbuf = f.read(4096)
                while 1:
                    buf = nbuf
                    nbuf = f.read(4096)
                    total = total + len(buf)
                    if len(buf) == 0:
                        break
                    s.send(buf)
                    # s.sync()
                f.close()

                et = time.time()
                br = total * 0.008 / (et - st)
                logger.info("Sent %d bytes in %d seconds (%f kbps)" % (total, et - st, br))
                if self.delete_file and os.path.exists(fa):
                    os.remove(fa)
            except Exception:
                logger.exception("fail to send data to icecast")
        # s.close()

    def get_file_name(self, radio_id, suffix="mp3"):
        return f"{radio_id}_{int(datetime.datetime.now().timestamp() * 1000)}.{suffix}"

    def start(self):
        self.read_thread.start()
        self.convert_thread.start()
        self.send_thread.start()


if __name__ == "__main__":
    # Converter("http://s3.voscast.com:10026/;stream").start()
    Converter("https://onlineradiobox.com/json/ng/christovibes/play?platform=web").start()