377 日記 | 再生モードボタンをつける - TanaRadio

たな

Host

再生モード切り替えのためのボタンとLEDを取り付けました（フェーズ5G完了）。

TanaRadio Pi フェーズ5G 詳細マニュアル

5つ目のボタンで再生モードを選ぶ

0. 今回の目標

フェーズ5Fまでで、すでに次のことができています。

棚ボタンを押す
↓
棚番号が進む
↓
LED1が棚番号ぶん点滅する
↓
選んだ棚のRSSからプレイリストが作られる
↓
mpvで再生される

フェーズ5Gでは、ここに「再生モード選び」を追加します。

再生モードボタンを押す
↓
再生モード番号が進む
↓
LED2がモード番号ぶん点滅する
↓
同じ棚でも、再生対象や並び順が変わる
↓
プレイリストを作り直して再生する

今回入れる再生モードは、まずこの3つです。

1：最新順
2：今日だけ古い順
3：3日前から古い順

今回は、まだ次の2つは入れません。

4：7日前から古い順
5：未聴っぽいもの

「未聴っぽいもの」は、ローカル再生履歴が必要です。ここは5H以降に回すのが正解です。焦って入れると、工作からいきなり業務システム開発になります。沼です。まだ長靴はいりません。

1. 完成後の操作

フェーズ5G完成後の操作はこうなります。

再生／一時停止ボタン：
再生／一時停止

戻るボタン：
前のエピソードへ

次へボタン：
次のエピソードへ

棚ボタン：
棚を1つ進める
LED1が棚番号ぶん点滅

再生モードボタン：
再生モードを1つ進める
LED2がモード番号ぶん点滅

棚は、フェーズ5Eで決めた次の構成をそのまま使います。

1：声日記・生活
   shelf_01_voice_life.txt

2：技術・AI・ものづくり
   shelf_02_tech_ai_mono.txt

3：教育・研究・知の実践
   shelf_03_education_research.txt

4：本・文化・物語
   shelf_04_books_culture.txt

5：TanaRadio
   shelf_05_tanaradio.txt

2. 追加する部品

今回追加するのは、次の2つです。

タクトスイッチ 1個
LED 1個

つまり、5Fまでの構成に加えて、

再生モード選びボタン
再生モード表示LED

を追加します。

5G終了時点では、こうなります。

タクトスイッチ：5個
LED：2個
ロータリーエンコーダー：まだ使わない

ロータリーエンコーダーは次の段階です。ここで欲張らないのが、かなり大事です。

3. GPIO番号

このマニュアルでは、次のGPIO番号で説明します。

再生／一時停止ボタン：GPIO17
戻るボタン：GPIO27
次へボタン：GPIO22
棚ボタン：GPIO23
再生モードボタン：GPIO5

棚表示LED：GPIO24
再生モード表示LED：GPIO6

すでに5Fで別のGPIO番号を使っている場合は、あとで出てくるPythonコードのこの部分だけ変更してください。

PIN_PLAY = 17
PIN_PREV = 27
PIN_NEXT = 22
PIN_SHELF = 23
PIN_MODE = 5

PIN_LED_SHELF = 24
PIN_LED_MODE = 6

4. 配線

4.1 再生モードボタンの配線

新しく追加するタクトスイッチを、次のようにつなぎます。

GPIO5 ─ 1kΩ程度の抵抗 ─ タクトスイッチ ─ GND

これまでのボタンと同じ考え方です。

内部プルアップを使うので、ボタンを押していないときはHIGH、押したときにGNDへ落ちてLOWになります。

4.2 再生モード表示LEDの配線

新しく追加するLEDを、次のようにつなぎます。

GPIO6 ─ 330Ω抵抗 ─ LEDの長い足
LEDの短い足 ─ GND

LEDには向きがあります。

長い足：GPIO側
短い足：GND側

LEDが光らない場合、まず疑うべきはLEDの向きです。コードより先に足を疑ってください。電子工作では、だいたい足か線です。

5. いったん電源を切って配線する

配線を追加する前に、Raspberry Piを終了します。

sudo shutdown -h now

電源が落ちたら、電源ケーブルを抜いてから配線します。

配線が終わったら、もう一度電源を入れます。

6. 作業フォルダへ移動する

Raspberry Piが起動したら、ターミナルを開きます。

cd ~/tanaradio5

現在地を確認します。

pwd

次のようになればOKです。

/home/tana/tanaradio5

7. 5Fのコードをバックアップする

フェーズ5Fがうまく動いているので、必ずバックアップします。

もし現在のファイル名が tanaradio_5f.py なら、次のようにします。

cp tanaradio_5f.py tanaradio_5f_ok.py

もしファイル名が違う場合は、その名前に合わせてください。

例：

cp 今動いているファイル名.py 5f_ok.py

これは保険です。
5Gで何か変になっても、5Fに戻れます。工作でもプログラムでも、戻れる場所を作る人は強いです。

8. 必要なパッケージを確認する

すでに入っているはずですが、念のため確認します。

python3 -c "import gpiozero; print('gpiozero OK')"

gpiozero OK と出ればOKです。

もしエラーが出たら、次を実行します。

sudo apt update
sudo apt install -y python3-gpiozero python3-lgpio

mpvも確認します。

mpv --version

バージョン情報が出ればOKです。

9. 新しい5Gコードを作る

新しく tanaradio_5g.py を作ります。

nano tanaradio_5g.py

次のコードをそのまま貼り付けます。

#!/usr/bin/env python3
import json
import socket
import subprocess
import time
import urllib.request
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from signal import pause
from zoneinfo import ZoneInfo

from gpiozero import Button, LED

BASE_DIR = Path(__file__).resolve().parent
PLAYLIST_FILE = BASE_DIR / "playlist_5g.m3u"
STATE_FILE = BASE_DIR / "state_5g.json"
MPV_SOCKET = "/tmp/tanaradio-mpv.sock"

TZ = ZoneInfo("Asia/Tokyo")
MAX_ITEMS = 80

# ===== GPIO番号（BCM） =====
# 5Fまでの配線と違う場合は、ここだけ自分の配線に合わせて変更してください。
PIN_PLAY = 17
PIN_PREV = 27
PIN_NEXT = 22
PIN_SHELF = 23
PIN_MODE = 5

PIN_LED_SHELF = 24
PIN_LED_MODE = 6

SHELVES = [
    {
        "key": "voice",
        "name": "声日記・生活",
        "file": "shelf_01_voice_life.txt",
    },
    {
        "key": "tech",
        "name": "技術・AI・ものづくり",
        "file": "shelf_02_tech_ai_mono.txt",
    },
    {
        "key": "edu",
        "name": "教育・研究・知の実践",
        "file": "shelf_03_education_research.txt",
    },
    {
        "key": "books",
        "name": "本・文化・物語",
        "file": "shelf_04_books_culture.txt",
    },
    {
        "key": "tana",
        "name": "TanaRadio",
        "file": "shelf_05_tanaradio.txt",
    },
]

# 5Gではまず3種類だけにします。
# 「7日前から古い順」「未聴っぽいもの」は、次の段階で足します。
MODES = [
    {
        "key": "latest",
        "name": "最新順",
    },
    {
        "key": "today_oldest",
        "name": "今日だけ古い順",
    },
    {
        "key": "last3_oldest",
        "name": "3日前から古い順",
    },
]


@dataclass
class Episode:
    title: str
    program: str
    pub_dt: datetime
    audio_url: str


mpv_process = None
busy = False


def clamp_index(value, size, default=0):
    if not isinstance(value, int):
        return default
    if value < 0 or value >= size:
        return default
    return value


def load_state():
    default_state = {
        "shelf": 0,
        "mode": 2,  # 初期値は「3日前から古い順」
    }

    if not STATE_FILE.exists():
        return default_state

    try:
        data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
        return {
            "shelf": clamp_index(data.get("shelf"), len(SHELVES), 0),
            "mode": clamp_index(data.get("mode"), len(MODES), 2),
        }
    except Exception:
        return default_state


def save_state(state):
    STATE_FILE.write_text(
        json.dumps(state, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def read_feed_urls(shelf_file):
    path = BASE_DIR / shelf_file

    if not path.exists():
        raise FileNotFoundError(f"{shelf_file} が見つかりません。")

    urls = []
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        if line.startswith("#"):
            continue
        urls.append(line)

    if not urls:
        raise ValueError(f"{shelf_file} にRSS URLがありません。")

    return urls


def download_rss(url):
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": "TanaRadio-Pi/5G",
        },
    )

    with urllib.request.urlopen(request, timeout=25) as response:
        return response.read()


def parse_pubdate(text):
    if not text:
        return datetime(1970, 1, 1, tzinfo=timezone.utc).astimezone(TZ)

    try:
        dt = parsedate_to_datetime(text)
    except Exception:
        try:
            dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
        except Exception:
            dt = datetime(1970, 1, 1, tzinfo=timezone.utc)

    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)

    return dt.astimezone(TZ)


def find_audio_url(item):
    enclosure = item.find("enclosure")
    if enclosure is not None:
        audio_url = enclosure.attrib.get("url")
        if audio_url:
            return audio_url

    for child in item:
        if child.tag.endswith("content"):
            audio_url = child.attrib.get("url")
            if audio_url:
                return audio_url

    return ""


def find_text(item, tag_name):
    element = item.find(tag_name)
    if element is not None and element.text:
        return element.text.strip()
    return ""


def extract_episodes_from_rss(rss_data):
    root = ET.fromstring(rss_data)
    channel = root.find("channel")

    if channel is None:
        raise ValueError("RSSのchannelが見つかりません。")

    program = channel.findtext("title", default="").strip()
    episodes = []

    for item in channel.findall("item"):
        audio_url = find_audio_url(item)
        if not audio_url:
            continue

        title = find_text(item, "title") or "(no title)"
        pub_text = find_text(item, "pubDate")
        pub_dt = parse_pubdate(pub_text)

        episodes.append(
            Episode(
                title=title,
                program=program,
                pub_dt=pub_dt,
                audio_url=audio_url,
            )
        )

    return episodes


def collect_episodes(shelf):
    urls = read_feed_urls(shelf["file"])
    all_episodes = []

    print(f"棚: {shelf['name']}")
    print(f"RSS数: {len(urls)}")

    for url in urls:
        try:
            print(f"RSS取得: {url}")
            rss_data = download_rss(url)
            episodes = extract_episodes_from_rss(rss_data)
            all_episodes.extend(episodes)
            print(f"  取得エピソード数: {len(episodes)}")
        except Exception as e:
            print(f"  このRSSは読み飛ばします: {e}")

    return all_episodes


def select_episodes_by_mode(episodes, mode):
    now = datetime.now(TZ)

    if mode["key"] == "latest":
        selected = sorted(episodes, key=lambda e: e.pub_dt, reverse=True)

    elif mode["key"] == "today_oldest":
        today = now.date()
        selected = [
            e for e in episodes
            if e.pub_dt.date() == today
        ]
        selected = sorted(selected, key=lambda e: e.pub_dt)

    elif mode["key"] == "last3_oldest":
        since = now - timedelta(days=3)
        selected = [
            e for e in episodes
            if e.pub_dt >= since
        ]
        selected = sorted(selected, key=lambda e: e.pub_dt)

    else:
        selected = sorted(episodes, key=lambda e: e.pub_dt, reverse=True)

    return selected[:MAX_ITEMS]


def write_playlist(episodes):
    lines = ["#EXTM3U"]

    for episode in episodes:
        title = f"{episode.program} - {episode.title}".replace("\n", " ")
        pub = episode.pub_dt.strftime("%Y-%m-%d %H:%M")
        lines.append(f"# {pub} {title}")
        lines.append(f"#EXTINF:-1,{title}")
        lines.append(episode.audio_url)

    PLAYLIST_FILE.write_text(
        "\n".join(lines) + "\n",
        encoding="utf-8",
    )

    print(f"{PLAYLIST_FILE.name} を作成しました。")
    print(f"再生対象: {len(episodes)} 件")


def playlist_has_audio():
    if not PLAYLIST_FILE.exists():
        return False

    for line in PLAYLIST_FILE.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            return True

    return False


def stop_mpv():
    global mpv_process

    if mpv_process is not None and mpv_process.poll() is None:
        mpv_process.terminate()
        try:
            mpv_process.wait(timeout=3)
        except subprocess.TimeoutExpired:
            mpv_process.kill()

    mpv_process = None


def start_mpv():
    global mpv_process

    stop_mpv()

    if not playlist_has_audio():
        print("再生できる音声URLがないため、mpvは起動しません。")
        print("今日モードで0件の場合などは正常です。")
        return

    socket_path = Path(MPV_SOCKET)
    if socket_path.exists():
        try:
            socket_path.unlink()
        except Exception:
            pass

    cmd = [
        "mpv",
        "--no-video",
        "--really-quiet",
        f"--input-ipc-server={MPV_SOCKET}",
        str(PLAYLIST_FILE),
    ]

    mpv_process = subprocess.Popen(cmd)
    time.sleep(0.5)
    print("mpvを起動しました。")


def send_mpv_command(command):
    try:
        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
            client.connect(MPV_SOCKET)
            message = json.dumps({"command": command}) + "\n"
            client.send(message.encode("utf-8"))
    except Exception as e:
        print(f"mpvへ命令を送れませんでした: {e}")


def blink(led, count):
    count = max(1, int(count))
    for _ in range(count):
        led.on()
        time.sleep(0.18)
        led.off()
        time.sleep(0.18)


def rebuild_playlist_and_restart():
    state = load_state()
    shelf = SHELVES[state["shelf"]]
    mode = MODES[state["mode"]]

    print("")
    print("=== プレイリストを作り直します ===")
    print(f"棚: {state['shelf'] + 1}. {shelf['name']}")
    print(f"再生モード: {state['mode'] + 1}. {mode['name']}")

    episodes = collect_episodes(shelf)
    selected = select_episodes_by_mode(episodes, mode)
    write_playlist(selected)
    start_mpv()


def run_exclusive(action):
    global busy

    if busy:
        print("処理中です。少し待ってください。")
        return

    busy = True
    try:
        action()
    except Exception as e:
        print(f"エラー: {e}")
    finally:
        busy = False


def on_play():
    print("再生／一時停止")
    send_mpv_command(["cycle", "pause"])


def on_prev():
    print("前のエピソードへ")
    send_mpv_command(["playlist-prev"])


def on_next():
    print("次のエピソードへ")
    send_mpv_command(["playlist-next"])


def on_shelf():
    def action():
        state = load_state()
        state["shelf"] = (state["shelf"] + 1) % len(SHELVES)
        save_state(state)

        shelf = SHELVES[state["shelf"]]
        print(f"棚を変更: {state['shelf'] + 1}. {shelf['name']}")
        blink(led_shelf, state["shelf"] + 1)
        rebuild_playlist_and_restart()

    run_exclusive(action)


def on_mode():
    def action():
        state = load_state()
        state["mode"] = (state["mode"] + 1) % len(MODES)
        save_state(state)

        mode = MODES[state["mode"]]
        print(f"再生モードを変更: {state['mode'] + 1}. {mode['name']}")
        blink(led_mode, state["mode"] + 1)
        rebuild_playlist_and_restart()

    run_exclusive(action)


def print_status():
    state = load_state()
    shelf = SHELVES[state["shelf"]]
    mode = MODES[state["mode"]]

    print("")
    print("TanaRadio Pi フェーズ5G")
    print("========================")
    print(f"現在の棚: {state['shelf'] + 1}. {shelf['name']}")
    print(f"現在の再生モード: {state['mode'] + 1}. {mode['name']}")
    print("")
    print("操作:")
    print("  再生ボタン: 再生／一時停止")
    print("  戻るボタン: 前のエピソード")
    print("  次へボタン: 次のエピソード")
    print("  棚ボタン: 棚を進める。LED1が棚番号ぶん点滅")
    print("  モードボタン: 再生モードを進める。LED2がモード番号ぶん点滅")
    print("  Ctrl + C: 終了")
    print("")


# ===== GPIOオブジェクト =====
btn_play = Button(PIN_PLAY, pull_up=True, bounce_time=0.05)
btn_prev = Button(PIN_PREV, pull_up=True, bounce_time=0.05)
btn_next = Button(PIN_NEXT, pull_up=True, bounce_time=0.05)
btn_shelf = Button(PIN_SHELF, pull_up=True, bounce_time=0.05)
btn_mode = Button(PIN_MODE, pull_up=True, bounce_time=0.05)

led_shelf = LED(PIN_LED_SHELF)
led_mode = LED(PIN_LED_MODE)


def main():
    btn_play.when_pressed = on_play
    btn_prev.when_pressed = on_prev
    btn_next.when_pressed = on_next
    btn_shelf.when_pressed = on_shelf
    btn_mode.when_pressed = on_mode

    print_status()

    state = load_state()
    blink(led_shelf, state["shelf"] + 1)
    time.sleep(0.4)
    blink(led_mode, state["mode"] + 1)

    run_exclusive(rebuild_playlist_and_restart)

    try:
        pause()
    except KeyboardInterrupt:
        print("\n終了します。")
    finally:
        stop_mpv()
        led_shelf.off()
        led_mode.off()


if __name__ == "__main__":
    main()

保存します。

Ctrl + O
Enter
Ctrl + X

10. 実行権限をつける

chmod +x tanaradio_5g.py

11. 棚ファイルがあるか確認する

次を実行します。

ls shelf_*.txt

次の5つが表示されればOKです。

shelf_01_voice_life.txt
shelf_02_tech_ai_mono.txt
shelf_03_education_research.txt
shelf_04_books_culture.txt
shelf_05_tanaradio.txt

中身も軽く確認します。

head shelf_01_voice_life.txt

RSS URLが入っていればOKです。

12. 5Gを実行する

まず、既存のmpvが残っている場合は止めます。

pkill mpv

そして実行します。

./tanaradio_5g.py

うまくいくと、ターミナルに次のような表示が出ます。

TanaRadio Pi フェーズ5G
========================
現在の棚: 1. 声日記・生活
現在の再生モード: 3. 3日前から古い順

操作:
  再生ボタン: 再生／一時停止
  戻るボタン: 前のエピソード
  次へボタン: 次のエピソード
  棚ボタン: 棚を進める。LED1が棚番号ぶん点滅
  モードボタン: 再生モードを進める。LED2がモード番号ぶん点滅
  Ctrl + C: 終了

その後、RSSを取得し、プレイリストが作られ、mpvが起動します。

=== プレイリストを作り直します ===
棚: 1. 声日記・生活
再生モード: 3. 3日前から古い順
RSS数: ...
playlist_5g.m3u を作成しました。
再生対象: ... 件
mpvを起動しました。

音が出れば、まず成功です。

13. 動作テスト

13.1 起動時のLED点滅

起動直後に、

LED1：棚番号ぶん点滅
LED2：再生モード番号ぶん点滅

します。

初期状態では、再生モードは3番にしてあります。

LED2が3回点滅
→ 3日前から古い順

13.2 再生モードボタンを押す

再生モードボタンを押します。

1回押すごとに、次のように進みます。

1：最新順
↓
2：今日だけ古い順
↓
3：3日前から古い順
↓
1：最新順

ボタンを押すと、LED2がモード番号ぶん点滅します。

LED2が1回点滅
→ 最新順

LED2が2回点滅
→ 今日だけ古い順

LED2が3回点滅
→ 3日前から古い順

その後、プレイリストが作り直され、mpvが再起動します。

13.3 棚ボタンも確認する

棚ボタンも、5Fと同じように動くか確認します。

棚ボタンを押す
↓
LED1が棚番号ぶん点滅
↓
選んだ棚でプレイリスト作成
↓
再生

ここが動けば、5Fの機能は壊れていません。

14. 再生モードの意味

1：最新順

RSSから取れたエピソードを、新しいものから順に並べます。

新しい
↓
古い

新着チェック向きです。

2：今日だけ古い順

今日公開されたエピソードだけを対象にします。

今日の古いもの
↓
今日の新しいもの

今日のエピソードがない場合は、

再生対象: 0 件
再生できる音声URLがないため、mpvは起動しません。
今日モードで0件の場合などは正常です。

と表示されます。

これはエラーではありません。
単に「今日はまだ該当エピソードがない」という意味です。

3：3日前から古い順

現在時刻から見て、過去3日以内のエピソードを対象にします。

3日前
↓
2日前
↓
昨日
↓
今日

これは「追いつき再生」向きです。
TanaRadio Piの日常利用では、このモードがいちばんラジオらしいと思います。

15. 作られるファイル

5Gでは、新しく次のファイルが作られます。

playlist_5g.m3u
state_5g.json

現在の棚と再生モードに応じて作られるmpv用プレイリストです。

中身を確認するには、

head playlist_5g.m3u
state_5g.json

現在選ばれている棚番号と再生モード番号を保存するファイルです。

確認するには、

cat state_5g.json

例：

{
  "shelf": 0,
  "mode": 2
}

Python内部では0から数えるため、

"shelf": 0 → 棚1
"mode": 2 → モード3

という意味になります。

ここは少しややこしいですが、正常です。コンピュータは0から数えたがる生き物です。

16. 終了方法

プログラムを止めるには、ターミナルで次を押します。

Ctrl + C

すると、

終了します。

と出て、mpvも止まります。

もし音だけ残ってしまった場合は、次で止めます。

pkill mpv

18. 成功条件

フェーズ5Gの成功条件は、次の5つです。

1. 5つ目のボタンを押すと、再生モードが切り替わる
2. LED2が再生モード番号ぶん点滅する
3. 最新順・今日だけ古い順・3日前から古い順が切り替わる
4. 棚ボタンとLED1もこれまで通り動く
5. 再生／一時停止・戻る・次へボタンも動く

これができれば、フェーズ5Gは成功です。

19. 今回はここで止めてよいです

5Gまで来ると、TanaRadio Piはかなり「声の本棚」らしくなっています。

どの棚を聴くか
どの時間順で聴くか

この2つを、物理ボタンで選べるようになったからです。

次の段階は、おそらく次のどちらかです。

フェーズ5H：
ローカル再生履歴を作る

フェーズ5I：
ロータリーエンコーダーで音量調整する

私のおすすめは、まだ再生履歴には行かず、先にロータリーエンコーダーを音量用として追加することです。理由は単純で、音量ダイヤルが付くと一気にラジオ感が出るからです。やる気が出る工作は、正義です。

#声日記　 #TanaRadioPi

サマリー

TanaRadio Piのフェーズ5Gでは、再生モード切り替え用の5つ目のボタンとLEDが追加されました。これにより、「最新順」「今日だけ古い順」「3日前から古い順」の3種類の再生モードを物理ボタンで選択可能になり、LEDの点滅で現在のモードが視覚的に確認できます。既存の棚選択機能も正常に動作し、TanaRadio Piは「声の本棚」としてさらに進化しました。

フェーズ5Gの開始と目標