らじる★らじるの再生や録音を自動化

Last modified: 2019-08-18

らじる★らじるの再生や録音をWebブラウザを用いずに自動で行うことを目指す。Webブラウザでサイトを訪問して聴くこともできるが、CPUに負荷がかかる上に自動化に向かないため、別の方法を探す。

配信の方式について

らじる★らじるではconfig_web.xmlというファイルが公開されており、これをダウンロードして構文解析することでリスナーのいる場所に合った放送局と指定チャンネルから音楽プレーヤなどのソフトウェアが対応する配信URL(m3u8形式)が得られるようになっている。

放送局の一覧:

  • 札幌
  • 仙台
  • 東京
  • 名古屋
  • 大阪
  • 広島
  • 松山
  • 福岡

チャンネルの一覧

  • ラジオ第一
  • ラジオ第二
  • FM

この配信方式は “HTTP Live Streaming (HLS)” と呼ばれており、細切れのオーディオデータを複数用意してプレイリストとしてのm3u8ファイルを配信用URLとして用いる。動画(映像あり)の配信でも同様のやり方で実現できる。通信のプロトコルは専用のものではなくHTTPで処理され、受信側のソフトウェアはm3u8ファイルを開いたときにその中に記述された細切れのファイル群を順にダウンロードした上で再生や書き出しを行える必要がある。

エリアとチャンネルを指定してm3u8ファイルのURLを取得するスクリプト

下のスクリプトに

  • エリア名 (sapporo,sendai,tokyo,nagoya,osaka,hiroshima,matsuyama,fukuokaのいずれか)
  • チャンネル名 (r1,r2,fmのいずれか)

の2つの文字列を順に引数として指定して実行すると、m3u8ファイルのURLを標準出力に出力する。

Lua(LuaJIT)版とPython版を作成したが、いずれもExpatライブラリの機能を用いたものとなっており、高速に動作し、出力結果も同じものとなる。

以下のスクリプトは公開される汎用的なコードとして書かれており、実行の度にWebサーバからconfig_web.xmlを必ずダウンロードする。これを改造して同ファイルやその出力結果(m3u8ファイルのURL)をローカルのストレージやRAMディスク(/dev/shm/以下)に保存して、2回目以降の実行時にそれが存在した場合にダウンロードをせずに再利用する(失敗したら削除して保存されていない場合と同様に処理する)ようにすることもできる。ただし、ここではそのようなコードを貼り付けることはしない。

Lua版

実行するにはRaspbianの

  • luajit (Luaスクリプトを高速に実行)
  • lua-sec (HTTPS URLからのダウンロード処理に使用)
  • lua-expat (XMLファイルの構文解析に使用)

のパッケージの追加インストールが必要。

$ sudo apt install luajit lua-sec lua-expat
ファイル名:radiru-get-streamurl.lua ライセンス:GPLv3+
#! /usr/bin/luajit

-- radiru-get-streamurl.lua
-- This script requires "lua-sec" and "lua-expat"
--
-- (C) 2019 kakurasan
-- Licensed under GPLv3+

local https = require 'ssl.https'
local ltn12 = require 'ltn12'
local lxp = require 'lxp'


local api = {
    xml = 'https://www.nhk.or.jp/radio/config/config_web.xml',
    areas = {'sapporo', 'sendai', 'tokyo', 'nagoya',
             'osaka', 'hiroshima', 'matsuyama', 'fukuoka'},
    stations = {'r1', 'r2', 'fm'},
}

local parser_data = {
    hls_elem_name = nil,
    is_area_elem = false,
    is_specified_area = false,
    is_hls_elem = false,
    area = nil,
    m3u8 = nil,
}

local callbacks = {
    StartElement = function(parser, name)
        if parser_data.is_specified_area then
            if name == parser_data.hls_elem_name then
                parser_data.is_hls_elem = true
            end
        else
            parser_data.is_area_elem = name == 'area'
        end
    end,
    EndElement = function(parser, name)
        if parser_data.is_specified_area and name == 'data' then
            parser_data.is_specified_area = false
        elseif parser_data.is_hls_elem and name == parser_data.hls_elem_name then
            parser_data.is_hls_elem = false
        elseif parser_data.is_area_elem and name == 'area' then
            parser_data.is_area_elem = false
        end
    end,
    CharacterData = function(parser, data)
        if parser_data.is_hls_elem then
            parser_data.m3u8 = data
        elseif parser_data.is_area_elem and data == parser_data.area then
            parser_data.is_specified_area = true
        end
    end
}

function gen_usage()
    usage = 'USAGE: %s [AREA] ["r1", "r2", or "fm"]\nAvailable AREA:\n '
    for i, area in ipairs(api.areas) do
      usage = usage .. ('%s%s'):format(area, (i < #api.areas) and ', ' or '')
    end
    return usage:format(arg[0]) .. '\n'
end

do
    -- Check for number of arguments
    if #arg ~= 2 then
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    -- Check for area
    local area = arg[1]
    local area_is_valid = false
    for _, valid_area in ipairs(api.areas) do
        if area == valid_area then
            area_is_valid = true
        end
    end
    if not area_is_valid then
        io.stderr:write(('ERROR: Invalid area (%s)\n\n'):format(area))
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    -- Check for station
    local station = arg[2]
    local station_is_valid = false
    for _, valid_station in ipairs(api.stations) do
        if station == valid_station then
            station_is_valid = true
        end
    end
    if not station_is_valid then
        io.stderr:write(('ERROR: Invalid station (%s)\n\n'):format(station))
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    -- Download/parse XML
    local xmldata = {}
    local _, code, _, _ = https.request {
        url = api.xml,
        sink = ltn12.sink.table(xmldata),
    }
    if code ~= 200 then
        io.stderr:write(('ERROR: Could not download XML (status %s)\n'):format(code))
        os.exit(false)
    end
    local p = lxp.new(callbacks)
    parser_data.hls_elem_name = station .. 'hls'
    parser_data.area = area
    p:parse(table.concat(xmldata))
    p:close()
    if parser_data.m3u8 then
        io.stdout:write(parser_data.m3u8)
    else
        io.stderr:write('ERROR: Could not find m3u8 URL.')
        os.exit(false)
    end
end

Python版

こちらはPythonの標準の機能のみを用いているため、追加のパッケージは不要。

ファイル名:radiru-get-streamurl.py ライセンス:GPLv3+
#! /usr/bin/python

# radiru-get-streamurl.py
# (C) 2019 kakurasan
# Licensed under GPLv3+

from __future__ import print_function

import xml.parsers.expat as expat
import sys
try:
    from urllib.parse import urlparse as parse_url
    from http.client import HTTPSConnection
except ImportError:
    from urlparse import urlparse as parse_url
    from httplib import HTTPSConnection


class Api:
    xml = 'https://www.nhk.or.jp/radio/config/config_web.xml'
    areas = ('sapporo', 'sendai', 'tokyo', 'nagoya',
             'osaka', 'hiroshima', 'matsuyama', 'fukuoka')
    stations = ('r1', 'r2', 'fm')


class ConfigWebParser:
    xml = None
    p = expat.ParserCreate()
    hls_elem_name = None
    is_area_elem = False
    is_specified_area = False
    is_hls_elem = False
    area = None
    m3u8 = None

    def __init__(self, area, station):
        self.xml = get_xml()
        self.hls_elem_name = station + 'hls'
        self.area = area
        self.p.StartElementHandler = self._expat_start_elem
        self.p.EndElementHandler = self._expat_end_elem
        self.p.CharacterDataHandler = self._expat_char_data

    def _expat_start_elem(self, name, attrs):
        if self.is_specified_area:
            if name == self.hls_elem_name:
                self.is_hls_elem = True
        else:
            self.is_area_elem = name == 'area'

    def _expat_end_elem(self, name):
        if self.is_specified_area and name == 'data':
            self.is_specified_area = False
        elif self.is_hls_elem and name == self.hls_elem_name:
            self.is_hls_elem = False
        elif self.is_area_elem and name == 'area':
            self.is_area_elem = False

    def _expat_char_data(self, data):
        if self.is_hls_elem:
            self.m3u8 = data
        elif self.is_area_elem and data == self.area:
            self.is_specified_area = True

    def get_m3u8(self):
        self.p.Parse(self.xml)
        if self.m3u8:
            return self.m3u8
        else:
            return None


def get_xml():
    o = parse_url(Api.xml)
    conn = HTTPSConnection(o.netloc)
    conn.request('GET', o.path)
    res = conn.getresponse()
    try:
        if res.status != 200:
            sys.exit(
              'ERROR: Could not download XML (status {})'.format(res.status))
        xmldata = res.read()
    finally:
        conn.close()
    return xmldata


def gen_usage():
    usage = 'USAGE: {} [AREA] ["r1", "r2", or "fm"]\nAvailable AREA:\n '
    for i, area in enumerate(Api.areas):
        usage += '{}{}'.format(area, ', ' if i < len(Api.areas) - 1 else '')
    return usage.format(__file__)


def main():
    # Check for number of arguments
    if len(sys.argv) != 3:
        sys.exit(gen_usage())

    # Check for area
    area = sys.argv[1]
    area_is_valid = False
    for valid_area in Api.areas:
        if area == valid_area:
            area_is_valid = True
    if not area_is_valid:
        print('ERROR: Invalid area ({})\n'.format(area), file=sys.stderr)
        sys.exit(gen_usage())

    # Check for station
    station = sys.argv[2]
    station_is_valid = False
    for valid_station in Api.stations:
        if station == valid_station:
            station_is_valid = True
    if not station_is_valid:
        print('ERROR: Invalid station ({})\n'.format(station), file=sys.stderr)
        sys.exit(gen_usage())

    # Download/parse XML
    streamurl = ConfigWebParser(area, station).get_m3u8()
    if streamurl:
        print(streamurl, end='')
    else:
        sys.exit('ERROR: Could not find m3u8 URL.')


if __name__ == '__main__':
    main()

上のスクリプトを用いた再生と録音の例

上のスクリプトを用いて、自動化できる形で再生や録音を行うスクリプトの例を扱う。

以下の全てのスクリプトはm3u8ファイルの場所を取得するスクリプトの場所を得るための処理を含んでおり、streamurl_helperという名前の変数にこれが入るようになっているが、公開されるスクリプトとして汎用的な書き方にしたものであり、この取得処理をなくして代わりに上のスクリプトの絶対パスを代入する処理に置き換えたほうが若干低負荷で動作する。

再生を行うスクリプト例

上のスクリプトと同じディレクトリに配置し、両方に実行属性を付けた上で

  • エリア名
  • チャンネル名
  • 再生を続ける時間 (秒数, “[分]:[秒]“, “[時]:[分]:[秒]“)

を順に引数として指定して実行すると、mpvを用いて指定した長さだけ指定したチャンネルのラジオを再生する。

このスクリプトはGNU/LinuxのPCでラジオを聴くのに使うこともできる。

Lua版

radiru-get-streamurl.luaと同じディレクトリに配置する。

ファイル名:play-radiru.lua ライセンス:GPLv3+
#! /usr/bin/luajit

-- play-radiru.lua
-- This script requires "radiru-get-streamurl.lua"
--
-- (C) 2019 kakurasan
-- Licensed under GPLv3+

local helper_name = 'radiru-get-streamurl.lua'

function gen_usage()
    usage = 'USAGE: %s [AREA] ["r1", "r2", or "fm"]\n[DURATION]\n'
    return usage:format(arg[0])
end

do
    -- Check for number of arguments
    if #arg ~= 3 then
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    local area, station, duration = arg[1], arg[2], arg[3]
    local f = io.popen(('realpath "%s"'):format(arg[0]))
    local scriptpath = f:read()
    f:close()
    f = io.popen(('dirname "%s"'):format(scriptpath))
    local scriptdir = f:read()
    f:close()
    local streamurl_helper = ('%s/%s'):format(scriptdir, helper_name)

    -- Get stream URL
    f = io.popen(('"%s" %s %s'):format(streamurl_helper, area, station))
    local streamurl = f:read('*all')
    f:close()

    -- Play
    os.execute(('mpv --really-quiet --end %s %s'):format(duration, streamurl))
end

Python版

radiru-get-streamurl.pyと同じディレクトリに配置する。

ファイル名:play-radiru.py ライセンス:GPLv3+
#! /usr/bin/python

# play-radiru.py
# This script requires "radiru-get-streamurl.py"
#
# (C) 2019 kakurasan
# Licensed under GPLv3+

from __future__ import print_function

import subprocess
import sys
import os


helper_name = 'radiru-get-streamurl.py'


def gen_usage():
    usage = 'USAGE: {} [AREA] ["r1", "r2", or "fm"] [DURATION]'
    return usage.format(__file__)


def main():
    # Check for number of arguments
    if len(sys.argv) != 4:
        sys.exit(gen_usage())

    area, station, duration = sys.argv[1:]
    streamurl_helper = os.path.join(os.path.dirname(__file__), helper_name)

    # Get stream URL
    try:
        streamurl = subprocess.check_output((streamurl_helper, area, station))
    except subprocess.CalledProcessError as e:
        sys.exit(e.output)

    # Play
    subprocess.call(('mpv', '--really-quiet', '--end', duration, streamurl))


if __name__ == '__main__':
    main()

録音を行うスクリプト例

m3u8ファイルの場所を取得するスクリプトと同じディレクトリに配置し、両方に実行属性を付けた上で

  • エリア名
  • チャンネル名
  • 録音を続ける時間
  • 出力されるファイルの場所(重要:拡張子は.aac.m4aにする)

を順に引数として指定して実行すると、ffmpegを用いて指定した長さだけ指定したチャンネルのラジオを録音する。

!!拡張子を間違えると正しくファイルが書き出されないので注意!!.aacの拡張子を指定して書き出されたファイルには、プレーヤのソフトウェアから認識される長さが実際と異なるという問題があり、.m4aの拡張子の場合は録音中に書き込み途中のファイルを開いて再生することができない。.aacの拡張子て書き出されたファイルはffmpeg.m4aの拡張子のファイルに無劣化で変換できるため、これも踏まえて好みの書き出し方で運用する。

Lua版

radiru-get-streamurl.luaと同じディレクトリに配置する。

ファイル名:rec-radiru.lua ライセンス:GPLv3+
#! /usr/bin/luajit

-- rec-radiru.lua
-- This script requires "radiru-get-streamurl.lua"
--
-- (C) 2019 kakurasan
-- Licensed under GPLv3+

local helper_name = 'radiru-get-streamurl.lua'

function gen_usage()
    usage = 'USAGE: %s [AREA] ["r1", "r2", or "fm"]\n[DURATION] [OUTPUT]\n'
    return usage:format(arg[0])
end

do
    -- Check for number of arguments
    if #arg ~= 4 then
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    local area, station, duration, outfile = arg[1], arg[2], arg[3], arg[4]
    local f = io.popen(('realpath "%s"'):format(arg[0]))
    local scriptpath = f:read()
    f:close()
    f = io.popen(('dirname "%s"'):format(scriptpath))
    local scriptdir = f:read()
    f:close()
    local streamurl_helper = ('%s/%s'):format(scriptdir, helper_name)

    -- Get stream URL
    f = io.popen(('"%s" %s %s'):format(streamurl_helper, area, station))
    local streamurl = f:read('*all')
    f:close()

    -- Record
    os.execute(
      ('ffmpeg -v quiet -i %s -t %s -acodec copy -y "%s"'):format(
       streamurl, duration, outfile))
end

Python版

radiru-get-streamurl.pyと同じディレクトリに配置する。

ファイル名:rec-radiru.py ライセンス:GPLv3+
#! /usr/bin/python

# rec-radiru.py
# This script requires "radiru-get-streamurl.py"
#
# (C) 2019 kakurasan
# Licensed under GPLv3+

from __future__ import print_function

import subprocess
import sys
import os


helper_name = 'radiru-get-streamurl.py'


def gen_usage():
    usage = 'USAGE: {} [AREA] ["r1", "r2", or "fm"] [DURATION] [OUTPUT]'
    return usage.format(__file__)


def main():
    # Check for number of arguments
    if len(sys.argv) != 5:
        sys.exit(gen_usage())

    area, station, duration, outfile = sys.argv[1:]
    streamurl_helper = os.path.join(os.path.dirname(__file__), helper_name)

    # Get stream URL
    try:
        streamurl = subprocess.check_output((streamurl_helper, area, station))
    except subprocess.CalledProcessError as e:
        sys.exit(e.output)

    # Record
    subprocess.call(('ffmpeg', '-v', 'quiet', '-i', streamurl,
                     '-t', duration, '-acodec', 'copy', '-y', outfile))


if __name__ == '__main__':
    main()

crontabの設定例

上記の配信URL取得スクリプトと再生/録音スクリプトが既に正常に動く状態で配置されているという前提でのcrontabの設定例。

tokyoの部分は各自のエリアに合わせて変更する。

# 月曜から金曜の6:20からラジオ第一を1時間再生する
20 6 * * mon,tue,wed,thu,fri /path/to/play-radiru.lua tokyo r1 "1:0:0"
# HDMIオーディオを使用して音を出す場合はディスプレイの電源を強制的にオンにする
19 6 * * mon,tue,wed,thu,fri /usr/bin/vcgencmd display_power 1 >/dev/null
# ラジオの再生が終了するときにディスプレイの電源をオフにする場合の指定
20 7 * * mon,tue,wed,thu,fri /usr/bin/vcgencmd display_power 0 >/dev/null

# 日曜の17:00からFM放送を1時間30分録音
# 録音時間が極端に長くなければファイルサイズは大きくても数十MiBという規模で
# RAMディスクに一時保存もできる
0 17 * * sun /path/to/rec-radiru.lua tokyo fm "1:30:0" /dev/shm/fm.aac

これらのスクリプトを設定することで

  • 決まった時間帯にラジオを再生(朝の目覚まし,ラジオ体操,ニュースなど)
  • リアルタイムで聴けない時間帯の放送や聞き逃したくない放送を録音

といったことが自動化できる。

書き出されたファイルを無劣化で切り取りたい場合、ffmpeg-ssオプションで開始位置を指定して-acodec copy付きでファイルを書き出してからmpvなどで再生する試行錯誤をして、次にそのファイルに対して-tオプションで終了位置を指定して書き出して再生またはAudacityでの波形チェックをする試行錯誤を行い、納得のいく切り取り結果が得られたら保存用に移動やコピーをする。

上の設定例のコメントにも書いているが、手元の環境では、録音の際には/dev/shm/以下に一時的に保存されるようにしてmicroSDカードへの書き込みを防ぎ、後でPCからFileZillaなどでダウンロードして必要に応じて切り取りや保存用のコピーを行うようにしている。