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

再生モード切り替えのためのボタンと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 LISTENで開く