TanaRadio

たな

私立工科系大学教員(科学史)の日々の出来事や雑感。日記,メモ,ブログなど TanaRadio Blog ( https://tanakahi.hatenablog.com/ )

  1. 378 日記 | ロータリーエンコーダーで音量調節

    2d ago

    378 日記 | ロータリーエンコーダーで音量調節

    まず,ロータリーエンコーダーをDIP化基板にハンダ付けして,それからこれを使って音量調整できるようにしました(フェーズ5I完了)。 フェーズ5I 詳細マニュアル ロータリーエンコーダーで音量調整する 0. 今回の目標 今回追加する機能はこれです。 ロータリーエンコーダーを右に回す ↓ 音量アップ ロータリーエンコーダーを左に回す ↓ 音量ダウン 今回は、エンコーダーの押し込みによるミュートは扱いません。 たなさんの手元のロータリーエンコーダーは3端子タイプなので、 A C / GND / COM B の3本だけを使います。 1. 今回使うGPIO フェーズ5Gでは、次の構成で進んでいる前提にします。 再生/一時停止ボタン:GPIO17 戻るボタン:GPIO27 次へボタン:GPIO22 棚選びボタン:GPIO23 再生モードボタン:GPIO5 棚LED:GPIO24 モードLED:GPIO6 mpv IPCソケット: /tmp/tanaradio-mpv.sock 今回は、ロータリーエンコーダー用に次の2本を追加します。 エンコーダーA:GPIO20 エンコーダーB:GPIO21 エンコーダーC / COM:GND 物理ピンで言うと、Raspberry Piの40ピンヘッダーでは次の並びが使いやすいです。 GPIO20:物理ピン38 GND :物理ピン39 GPIO21:物理ピン40 ここが気持ちよく並んでいます。電子工作の神様が一瞬だけ優しい顔をしています。 gpiozeroの公式ドキュメントでも、ロータリーエンコーダーはA/BをGPIOへ、CつまりcommonをGNDへ接続する形で説明されています。 2. 配線する前にRaspberry Piを終了する まず、現在動いている tanaradio_5g.py があれば Ctrl + C で止めます。 そのあと、Raspberry Piを終了します。 sudo shutdown -h now 緑LEDの点滅が落ち着き、完全に止まったら電源ケーブルを抜きます。 GPIO配線は、必ず電源を抜いてから行います。 特に今回はGPIOが増えるので、ここは慎重でよいです。急がない。ラジオは逃げません。 3. ロータリーエンコーダーを配線する ロータリーエンコーダーの3端子を、次のようにつなぎます。 ロータリーエンコーダー A → GPIO20 ロータリーエンコーダー C → GND ロータリーエンコーダー B → GPIO21 DIP化基板に A、C、B のような表記があれば、それに従います。 もし表記が分かりにくい場合は、中央の端子が C / COM / GND であることが多いです。ただし、ここは部品によって違う可能性もあります。基板に印字があるなら印字を優先してください。 重要なのはこれです。 ロータリーエンコーダーには 5V をつながない 3.3V も今回はつながない C / COM は GND につなぐ A と B を GPIO20 / GPIO21 につなぐ 4. 起動して作業フォルダへ移動する 配線が終わったら、Raspberry Piの電源を入れます。 起動したらターミナルを開きます。 cd ~/tanaradio5 確認します。 pwd 次のように表示されればOKです。 /home/tana/tanaradio5 5. 必要パッケージを確認する 念のため、gpiozero関係を入れておきます。 sudo apt update sudo apt install -y python3-gpiozero python3-lgpio すでに入っていれば、そのままで大丈夫です。 6. ロータリーエンコーダー単体テスト まず、音量制御とは関係なく、エンコーダーを回したらPythonが反応するかだけ確認します。 nano test_encoder_5i.py 次を貼り付けます。 #!/usr/bin/env python3 from gpiozero import RotaryEncoder from signal import pause ENC_A_GPIO = 20 ENC_B_GPIO = 21 encoder = RotaryEncoder( ENC_A_GPIO, ENC_B_GPIO, max_steps=0, bounce_time=0.002 ) def rotated_clockwise(): print("右回し") def rotated_counter_clockwise(): print("左回し") encoder.when_rotated_clockwise = rotated_clockwise encoder.when_rotated_counter_clockwise = rotated_counter_clockwise print("ロータリーエンコーダーのテストを開始します。") print("つまみを左右に回してください。") print("終了するには Ctrl+C を押してください。") pause() 保存します。 Ctrl + O Enter Ctrl + X 実行できるようにします。 chmod +x test_encoder_5i.py 実行します。 ./test_encoder_5i.py つまみを右に回して、 右回し 左に回して、 左回し と出れば成功です。 終了は、 Ctrl + C です。 8. 5Gを動かしたまま音量制御テストをする 次に、tanaradio_5g.py はそのまま使い、別のテストプログラムからmpvへ音量命令を送ります。 mpvはIPCソケットで外部制御できます。mpvの公式マニュアルでも、外部プログラムから制御する場合は --input-ipc-server を使う方法が説明されています。 また、mpvの音量操作は add volume 5 / add volume -5 のようなコマンドで行えます。 まず、1つ目のターミナルで5Gを起動します。 cd ~/tanaradio5 ./tanaradio_5g.py 音が出る状態にしておきます。 次に、別のターミナルを開いて、同じフォルダへ移動します。 cd ~/tanaradio5 音量テスト用ファイルを作ります。 nano test_volume_5i.py 次を貼り付けます。 #!/usr/bin/env python3 import json import os import socket import time from gpiozero import RotaryEncoder from signal import pause MPV_SOCKET = "/tmp/tanaradio-mpv.sock" ENC_A_GPIO = 20 ENC_B_GPIO = 21 VOLUME_STEP = 5 MIN_VOLUME_INTERVAL = 0.04 last_volume_time = 0.0 encoder = RotaryEncoder( ENC_A_GPIO, ENC_B_GPIO, max_steps=0, bounce_time=0.002 ) def send_mpv_command(*command): if not os.path.exists(MPV_SOCKET): print("mpvのソケットが見つかりません。") print("先に別ターミナルで ./tanaradio_5g.py を起動してください。") return try: data = json.dumps({"command": list(command)}) + "\n" with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: client.connect(MPV_SOCKET) client.sendall(data.encode("utf-8")) try: response = client.recv(4096) if response: print(response.decode("utf-8").strip()) except Exception: pass except Exception as e: print(f"mpvコマンド送信エラー: {e}") def change_volume(delta): global last_volume_time now = time.monotonic() if now - last_volume_time 0: print("音量アップ") else: print("音量ダウン") send_mpv_command("add", "volume", delta) def volume_up(): change_volume(VOLUME_STEP) def volume_down(): change_volume(-VOLUME_STEP) encoder.when_rotated_clockwise = volume_up encoder.when_rotated_counter_clockwise = volume_down print("ロータリーエンコーダー音量テストを開始します。") print("別ターミナルで tanaradio_5g.py が動いている必要があります。") print("つまみを右に回すと音量アップ、左に回すと音量ダウンです。") print("終了するには Ctrl+C を押してください。") pause() 保存します。 Ctrl + O Enter Ctrl + X 実行できるようにします。 chmod +x test_volume_5i.py 実行します。 ./test_volume_5i.py 5Gで音声を流したまま、つまみを回します。 右回し → 音量アップ 左回し → 音量ダウン になれば、フェーズ5Iの核心部分は成功です。 9. 5Gコードを5Iとしてコピーする ここから本体に統合します。 まず、5Gコードをコピーします。 cp tanaradio_5g.py tanaradio_5i.py バックアップも作っておきます。 cp tanaradio_5g.py tanaradio_5g_backup.py 編集します。 nano tanaradio_5i.py 10. RotaryEncoder をimportに追加する コードの上の方に、たぶん次のような行があります。 from gpiozero import Button, LED これを次のようにします。 from gpiozero import Button, LED, RotaryEncoder すでに RotaryEncoder が入っていたら、そのままでOKです。 11. GPIO設定を追加する GPIO番号を定義している部分に、次を追加します。 ENC_A_GPIO = 20 ENC_B_GPIO = 21 VOLUME_STEP = 5 MIN_VOLUME_INTERVAL = 0.04 last_volume_time = 0.0 VOLUME_STEP = 5 は、1クリックごとに音量を5ずつ変える設定です。 大きすぎると感じたら、あとで次のように小さくできます。 VOLUME_STEP = 3 または、 VOLUME_STEP = 2 12. 音量変更関数を追加する ボタンの処理関数が並んでいるあたりに、次を追加します。 def change_volume(delta): global last_volume_time now = time.monotonic() if now - last_volume_time 0: print(f"音量アップ +{delta}") else: print(f"音量ダウン {delta}") send_mpv_command("add", "volume", delta) def volume_up(): change_volume(VOLUME_STEP) def volume_down(): change_volume(-VOLUME_STEP) ここで使っている send_mpv_command() は、5Gですでに使っているmpv制御関数です。 もし5Gコード内で関数名が違っていて、たとえば、 send_command(...) や、 mpv_command(...) のような名前になっている場合は、この部分だけ5G側の関数名に合わせてください。 つまり、この行です。 send_mpv_command("add", "volume", delta) ここを、5Gで使っている名前に合わせます。 13. エンコーダーを作成してイベントを登録する ボタンを作っている場所を探します。 たとえば次のような場所です。 play_button = Button(...) prev_button = Button(...) next_button = Button(...) shelf_button

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

    3d ago

    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 = 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 No

    18 min
  3. 376 日記 | 棚選択ボタンとLEDランプを追加

    5d ago

    376 日記 | 棚選択ボタンとLEDランプを追加

    ボタンで棚を選択できるようにしました。選んでいる棚番号はLEDランプの点滅回数で表示します(フェーズ5F完了)。 # TanaRadio Pi フェーズ5F 詳細マニュアル ## 棚選びボタンとLED1を追加する ## 0. 今回の目標 フェーズ5Eでは、棚ファイルを指定してプレイリストを作り、mpvで再生できるようになりました。 フェーズ5Fでは、その「棚の切り替え」を物理ボタンで行えるようにします。 今回の動きは次の通りです。 ```text 棚選びボタンを押す ↓ 棚番号が1つ進む ↓ LED1が棚番号ぶん点滅する ↓ 選ばれた棚ファイルから playlist.m3u を作る ↓ mpvをその棚で再生し直す ``` 今回はまだ、再生モード選びボタンは追加しません。 再生モードは、5Eまでと同じく、 ```text 過去3日分 古い順 ``` を基本にします。 --- # 1. 今回使う棚構成 フェーズ5Fでは、フェーズ5Eで作った次の棚構成を使います。 ```text 1: 声日記・生活 (voice) -> shelf_01_voice_life.txt 2: 技術・AI・ものづくり (tech) -> shelf_02_tech_ai_mono.txt 3: 教育・研究・知の実践 (edu) -> shelf_03_education_research.txt 4: 本・文化・物語 (books) -> shelf_04_books_culture.txt 5: TanaRadio (tana) -> shelf_05_tanaradio.txt ``` 棚選びボタンを押すと、この順番で進みます。 ```text 声日記・生活 ↓ 技術・AI・ものづくり ↓ 教育・研究・知の実践 ↓ 本・文化・物語 ↓ TanaRadio ↓ 声日記・生活 ``` LED1の点滅回数は、次の対応です。 ```text 1回点滅:声日記・生活 2回点滅:技術・AI・ものづくり 3回点滅:教育・研究・知の実践 4回点滅:本・文化・物語 5回点滅:TanaRadio ``` --- # 2. 今回追加する部品 今回追加する部品は2つです。 ```text タクトスイッチ 1個 LED 1個 ``` 用途は次の通りです。 ```text タクトスイッチ:棚選びボタン LED:棚番号表示LED ``` 既存の3ボタンは、そのまま使います。 ```text 再生/一時停止 戻る 次へ ``` つまり、フェーズ5F終了時点では、次の構成になります。 ```text ボタン1:再生/一時停止 ボタン2:戻る ボタン3:次へ ボタン4:棚選び LED1:棚番号表示 ``` --- # 3. GPIO番号 このマニュアルでは、次のGPIO番号を使います。 ```text 再生/一時停止:GPIO17 戻る:GPIO27 次へ:GPIO22 棚選びボタン:GPIO23 棚番号LED:GPIO24 ``` Raspberry Piの物理ピン番号では、次の対応です。 ```text GPIO17:物理ピン11 GPIO27:物理ピン13 GPIO22:物理ピン15 GPIO23:物理ピン16 GPIO24:物理ピン18 GND:物理ピン6, 9, 14, 20, 25 など ``` 注意点です。 ```text GPIO番号と物理ピン番号は違います。 ``` コードでは「GPIO番号」を使います。 たとえば、 ```text GPIO23 ``` は、物理ピン番号では、 ```text 16番ピン ``` です。 ここを取り違えると、ボタンを押しても無反応になります。静かな失敗です。電子工作ではいちばん地味に困るやつです。 --- # 4. 電源を切る 配線を触る前に、必ずRaspberry Piを終了します。 ターミナルで次を実行します。 ```bash sudo shutdown -h now ``` 完全に電源が切れてから、USB-C電源を抜きます。 --- # 5. 棚選びボタンを配線する 棚選びボタンは、次のようにつなぎます。 ```text GPIO23 ─ 1kΩ抵抗 ─ タクトスイッチ ─ GND ``` この構成では、Python側で内部プルアップを使います。 そのため、動作は次のようになります。 ```text ボタンを押していない:HIGH ボタンを押した:LOW ``` これまでのボタンと同じ考え方です。 --- # 6. 棚番号LEDを配線する LEDは、次のようにつなぎます。 ```text GPIO24 ─ 330Ω抵抗 ─ LEDの長い足 LEDの短い足 ─ GND ``` LEDの足の目安です。 ```text 長い足:プラス側 短い足:GND側 ``` 光らない場合は、まずLEDの向きを疑ってください。 LEDは、向きが逆だと基本的に光りません。 壊れたわけではなく、ただ黙っているだけです。 --- # 7. Raspberry Piを起動する 配線が終わったら、Raspberry Piの電源を入れます。 ターミナルを開き、作業フォルダに移動します。 ```bash cd ~/tanaradio5 ``` 現在地を確認します。 ```bash pwd ``` 次のように表示されればOKです。 ```text /home/tana/tanaradio5 ``` ユーザー名が違う場合は、そこだけ変わります。 --- # 8. 必要なパッケージを確認する GPIOとmpvを使うため、念のため次を実行します。 ```bash sudo apt update sudo apt install -y python3-gpiozero python3-lgpio mpv ``` すでに入っている場合は、そのままで大丈夫です。 --- # 9. 棚ファイルを確認する 今回使う棚ファイルがあるか確認します。 ```bash ls shelf_*.txt ``` 次の5つが見えればOKです。 ```text 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 ``` それぞれの中身も確認します。 ```bash cat shelf_01_voice_life.txt cat shelf_02_tech_ai_mono.txt cat shelf_03_education_research.txt cat shelf_04_books_culture.txt cat shelf_05_tanaradio.txt ``` RSS URLが1行ずつ並んでいればOKです。 例です。 ```text https://listen.style/p/xxxx/rss https://listen.style/p/yyyy/rss https://listen.style/p/zzzz/rss ``` 空の棚ファイルがある場合、その棚に切り替えたときにエラーになります。 テスト段階では、空の棚にも一時的に動作確認用のRSS URLを1本入れておくと安心です。 --- # 10. 5F用Pythonファイルを作る 新しいPythonファイルを作ります。 ```bash nano tanaradio_5f.py ``` 次のコードをそのまま貼り付けます。 重要です。 最初の行は必ず、次で始めてください。 ```text #!/usr/bin/env python3 ``` この前に、余計な文字を入れないでください。 ```python #!/usr/bin/env python3 import argparse import json import os import signal 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 threading import Lock from gpiozero import Button, LED # ========================= # GPIO設定 # 5Dまでと違うGPIOを使っていた場合は、ここだけ直してください。 # BCM番号です。物理ピン番号ではありません。 # ========================= PLAY_GPIO = 17 # 再生/一時停止ボタン PREV_GPIO = 27 # 戻るボタン NEXT_GPIO = 22 # 次へボタン SHELF_GPIO = 23 # 棚選びボタン(今回追加) SHELF_LED_GPIO = 24 # 棚番号表示LED(今回追加) # ========================= # ファイル設定 # ========================= BASE_DIR = Path(__file__).resolve().parent PLAYLIST_FILE = BASE_DIR / "playlist.m3u" STATE_FILE = BASE_DIR / "state_5f.json" MPV_SOCKET = "/tmp/tanaradio_5f_mpv.sock" # フェーズ5Fでは、まだ再生モードボタンは追加しません。 # 5Eまでと同じく「過去3日分・古い順」を基本にします。 DAYS_BACK = 3 # ========================= # 棚設定 # フェーズ5Eで作った棚構成に合わせています。 # ========================= SHELVES = [ { "label": "声日記・生活", "short": "voice", "file": "shelf_01_voice_life.txt", }, { "label": "技術・AI・ものづくり", "short": "tech", "file": "shelf_02_tech_ai_mono.txt", }, { "label": "教育・研究・知の実践", "short": "edu", "file": "shelf_03_education_research.txt", }, { "label": "本・文化・物語", "short": "books", "file": "shelf_04_books_culture.txt", }, { "label": "TanaRadio", "short": "tana", "file": "shelf_05_tanaradio.txt", }, ] @dataclass class Episode: title: str audio_url: str pub_date: datetime | None feed_title: str mpv_process: subprocess.Popen | None = None current_shelf_index = 0 shelf_lock = Lock() shelf_led: LED | None = None def read_lines(path: Path) -> list[str]: if not path.exists(): return [] lines = [] for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line: continue if line.startswith("#"): continue lines.append(line) return lines def load_state() -> int: if not STATE_FILE.exists(): return 0 try: data = json.loads(STATE_FILE.read_text(encoding="utf-8")) index = int(data.get("shelf_index", 0)) if 0 None: data = { "shelf_index": index, "shelf_label": SHELVES[index]["label"], "shelf_short": SHELVES[index]["short"], "shelf_file": SHELVES[index]["file"], "saved_at": datetime.now().isoformat(timespec="seconds"), } STATE_FILE.write_text( json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8", ) def download_rss(url: str) -> bytes: request = urllib.request.Request( url, headers={"User-Agent": "TanaRadio-Pi/5F"}, ) with urllib.request.urlopen(request, timeout=25) as response: return response.read() def get_child_text(parent: ET.Element, tag_name: str) -> str: element = parent.find(tag_name) if element is not None and element.text: return element.text

    26 min
  4. 6d ago

    375 日記 | 棚を選択してプレイリストを作成

    5つの棚(ポッドキャストのグループ)を作成し,それを選んでプレイリストを作成できるようにしました(フェーズ5E完了)。 フェーズ5E 詳細マニュアル 棚ファイルを作り、コマンドで棚を切り替える 0. 今回の目標 フェーズ5Eでやることは、これです。 棚ごとのRSSリストを作る ↓ コマンドで棚を指定する ↓ 指定した棚のRSSだけを読む ↓ 過去3日分・古い順で playlist.m3u を作る ↓ mpv または tanaradio_5d.py で再生する 今回はまだ、以下はやりません。 棚選びボタン LED点滅表示 再生モード選び ローカル再生履歴 ロータリーエンコーダー ここは焦らなくてよいです。 5Eの目的は、「棚」という考え方を、まずファイルで実現することです。 ステップ1:作業フォルダへ移動する Raspberry Piでターミナルを開きます。 cd ~/tanaradio5 確認します。 pwd 次のように表示されればOKです。 /home/ユーザー名/tanaradio5 ファイルも確認します。 ls 少なくとも、次のようなファイルがあるはずです。 feeds.txt make_playlist_5c.py playlist.m3u tanaradio_5d.py ステップ2:今の状態をバックアップする 5Dまで動いているので、念のためバックアップします。 cp make_playlist_5c.py make_playlist_5c_backup.py cp tanaradio_5d.py tanaradio_5d_backup.py cp playlist.m3u playlist_backup_5d.m3u 確認します。 ls バックアップファイルが見えればOKです。 make_playlist_5c_backup.py tanaradio_5d_backup.py playlist_backup_5d.m3u 5Eでは tanaradio_5d.py は基本的に変更しません。 ここ、大事です。5Dで動いたものはそのまま残します。動いたものは宝です。Linux作業では、動いたものを消すと、静かに深い後悔が来ます。 ステップ3:棚ファイルを作る まず、今回決めた5つの棚に対応するファイルを作ります。 touch shelf_01_voice_life.txt touch shelf_02_tech_ai_mono.txt touch shelf_03_education_research.txt touch shelf_04_books_culture.txt touch shelf_05_tanaradio.txt 確認します。 ls shelf_*.txt 次のように表示されれば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 棚の意味は次の通りです。 1:声日記・生活 2:技術・AI・ものづくり 3:教育・研究・知の実践 4:本・文化・物語 5:TanaRadio ステップ4:まず1つの棚にRSSを入れる 最初から全部きれいに分類しなくてよいです。 まずは、動作確認のために、これまで使っていた feeds.txt を1番の棚にコピーします。 cp feeds.txt shelf_01_voice_life.txt 確認します。 cat shelf_01_voice_life.txt RSS URLが表示されればOKです。 あとで、実際の分類に合わせて中身を編集します。 たとえば、1番の棚を編集するには、 nano shelf_01_voice_life.txt 次のように、コメントを入れながらRSS URLを書けます。 # 声日記・生活 # で始まる行は、あとで作るプログラムでは無視されます。 つまり、番組名メモを書いておいて大丈夫です。 ステップ5:5E用スクリプトを作る 新しいPythonファイルを作ります。 nano make_playlist_5e.py 次のコードをそのまま貼り付けてください。 #!/usr/bin/env python3 import sys import urllib.request import xml.etree.ElementTree as ET from pathlib import Path from datetime import datetime, timedelta, timezone from email.utils import parsedate_to_datetime from zoneinfo import ZoneInfo PLAYLIST_FILE = "playlist.m3u" # 5Eでは、まだ再生モード選びはしない。 # 5Cと同じく「過去3日分・古い順」で固定する。 DAYS = 3 JST = ZoneInfo("Asia/Tokyo") SHELVES = [ { "id": "1", "key": "voice", "name": "声日記・生活", "file": "shelf_01_voice_life.txt", "aliases": ["life", "koe", "nikki"], }, { "id": "2", "key": "tech", "name": "技術・AI・ものづくり", "file": "shelf_02_tech_ai_mono.txt", "aliases": ["ai", "mono", "monozukuri"], }, { "id": "3", "key": "edu", "name": "教育・研究・知の実践", "file": "shelf_03_education_research.txt", "aliases": ["education", "research", "study"], }, { "id": "4", "key": "books", "name": "本・文化・物語", "file": "shelf_04_books_culture.txt", "aliases": ["book", "culture", "story"], }, { "id": "5", "key": "tana", "name": "TanaRadio", "file": "shelf_05_tanaradio.txt", "aliases": ["tanaradio"], }, ] def show_shelves(): print("使える棚:") for shelf in SHELVES: print( f" {shelf['id']}: {shelf['name']} " f"({shelf['key']}) -> {shelf['file']}" ) print() print("例:") print(" python3 make_playlist_5e.py 1") print(" python3 make_playlist_5e.py tech") print(" python3 make_playlist_5e.py shelf_05_tanaradio.txt") def resolve_shelf(argument): if argument is None: # 引数なしの場合は1番棚を使う return SHELVES[0] arg = argument.strip() for shelf in SHELVES: candidates = [shelf["id"], shelf["key"]] + shelf["aliases"] if arg in candidates: return shelf # .txt が指定された場合は、任意の棚ファイルとして扱う path = Path(arg) if path.suffix == ".txt" or path.exists(): return { "id": "?", "key": "custom", "name": path.stem, "file": str(path), "aliases": [], } raise ValueError(f"棚を見つけられませんでした: {argument}") def read_feed_urls(shelf_file): path = Path(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): print(f"RSSを取得します: {url}") request = urllib.request.Request( url, headers={ "User-Agent": "TanaRadio-Pi/5E" } ) with urllib.request.urlopen(request, timeout=20) as response: return response.read() def get_text(element, tag_name): child = element.find(tag_name) if child is not None and child.text: return child.text.strip() return "" def find_audio_url(item): # 通常のPodcast RSSでは enclosure に音声URLが入る enclosure = item.find("enclosure") if enclosure is not None: audio_url = enclosure.attrib.get("url") if audio_url: return audio_url # 念のため media:content 的な形式にも軽く対応する for child in item: if child.tag.endswith("content"): audio_url = child.attrib.get("url") if audio_url: return audio_url return None def parse_pub_date(item): pub_date_text = get_text(item, "pubDate") if not pub_date_text: return None try: dt = parsedate_to_datetime(pub_date_text) # タイムゾーン情報がない場合はUTC扱いにする if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(JST) except Exception: return None def parse_rss(rss_data, feed_url): root = ET.fromstring(rss_data) channel = root.find("channel") if channel is None: raise ValueError("RSSのchannelが見つかりません。") program_title = get_text(channel, "title") if not program_title: program_title = "番組名不明" items = channel.findall("item") episodes = [] for item in items: episode_title = get_text(item, "title") audio_url = find_audio_url(item) pub_dt = parse_pub_date(item) if not audio_url: continue if pub_dt is None: continue episodes.append( { "program_title": program_title, "episode_title": episode_title, "audio_url": audio_url, "pub_dt": pub_dt, "feed_url": feed_url, } ) return episodes def collect_episodes(feed_urls): all_episodes = [] for url in feed_urls: try: rss_data = download_rss(url) episodes = parse_rss(rss_data, url) all_episodes.extend(episodes) print(f" 取得できたエピソード数: {len(episodes)} 件") except Exception as e: print(" このRSSの取得または解析でエラーが発生しました。") print(f" {e}") print(" 次のRSSへ進みます。") return all_episodes def remove_duplicates(episodes): seen = set() unique = [] for ep in episodes: audio_url = ep["audio_url"] if audio_url in seen: continue seen.add(audio_url) unique.append(ep) return unique def filter_recent_episodes(episodes): now = datetime.now(JST) cutoff = now - timedelta(days=DAYS) recent = [] for ep in episodes: if ep["pub_dt"] >= cutoff: recent.append(ep) return recent def make_playlist(episodes,

    30 min
  5. Jul 4

    374 日記 | 物理ボタンでプレイリスト再生を制御

    物理ボタンでプレイリスト再生の制御ができるようになりました(フェーズ5D完了)。 *以下のtanaradio_5d.pyのコードは,修正済みのものです(この収録後も1回修正しました)。 フェーズ5D 詳細マニュアル 既存3ボタンでRSSプレイリストを操作する 0. 今回の目標 今回やることは、これです。 フェーズ5Cで playlist.m3u を作る ↓ mpvで playlist.m3u を再生する ↓ GPIOの3ボタンでmpvを操作する 3つのボタンの役割は、今回は次のようにします。 再生/一時停止ボタン:再生・一時停止 戻るボタン:前のエピソードへ 次へボタン:次のエピソードへ mpvへの命令は次の3つです。 再生/一時停止:cycle pause 戻る:playlist-prev 次へ:playlist-next 構想メモでは、戻るボタンに「1回押し=先頭へ、2回押し=前へ、長押し=10秒戻る」という発展案があります。 ただし、5Dではまだ入れません。ここで欲張ると、ボタン判定の沼に入ります。沼は楽しいですが、今日はまだ岸辺でお茶を飲みます。 ステップ1:作業フォルダへ移動する Raspberry Piでターミナルを開きます。 cd ~/tanaradio5 確認します。 pwd 次のような表示になればOKです。 /home/ユーザー名/tanaradio5 ステップ2:5Cまでのファイルを確認する ls 少なくとも、次のようなファイルがあるはずです。 feeds.txt make_playlist_5c.py playlist.m3u スクリプト名は、たなさんが5Cで作った名前に合わせてください。 以下では、5Cのスクリプト名を make_playlist_5c.py として進めます。 ステップ3:5Cのプレイリストをもう一度作る まず、5Cのスクリプトを実行して、最新の playlist.m3u を作ります。 ./make_playlist_5c.py または、実行権限を付けていない場合は、 python3 make_playlist_5c.py うまくいけば、たとえば次のような表示になります。 playlist.m3u を作成しました。 音声URL数: 12 件 完了しました。 確認します。 ls -l playlist.m3u playlist.m3u が存在し、サイズが0でなければOKです。 中身も少し見ます。 head playlist.m3u 次のように、音声URLが入っていればOKです。 #EXTM3U # エピソードタイトル https://... ステップ4:mpvだけで再生できるか確認する ボタン操作に入る前に、まず普通に再生します。 mpv playlist.m3u 音が出ればOKです。 この状態では、キーボードで操作できます。 スペースキー:一時停止/再開 Enter:次へ q:終了 確認できたら、q を押してmpvを終了します。 ここで音が出ない場合は、5Dには進まず、音声出力側を先に確認してください。 ステップ5:GPIOライブラリを確認する 今回はPythonからGPIOボタンを読むために gpiozero を使います。 まず確認します。 python3 -c "import gpiozero; print('gpiozero OK')" 次のように出ればOKです。 gpiozero OK もしエラーが出たら、次を実行します。 sudo apt update sudo apt install -y python3-gpiozero python3-lgpio インストール後、もう一度確認します。 python3 -c "import gpiozero; print('gpiozero OK')" ステップ6:GPIO番号を確認する 今回のコードでは、フェーズ4で使ってきた3つのGPIOを次のように使います。 GPIO17:再生/一時停止 GPIO27:戻る、つまり前のエピソードへ GPIO22:次へ、つまり次のエピソードへ 配線はこれまで通り、基本形は次です。構想メモでも、ボタン入力には保険として抵抗を残す方針になっています。 GPIO ─ 1kΩ程度の抵抗 ─ ボタン ─ GND つまり、ボタンを押すとGPIOがGNDにつながる形です。 Python側では pull_up=True にします。 ステップ7:5D用スクリプトを作る 新しいファイルを作ります。 nano tanaradio_5d.py 次のコードをそのまま貼り付けます。 #!/usr/bin/env python3 from gpiozero import Button from signal import pause import json import os import signal import socket import subprocess import sys import time from pathlib import Path # ===== 設定ここから ===== BASE_DIR = Path(__file__).resolve().parent PLAYLIST_FILE = BASE_DIR / "playlist.m3u" SOCKET_PATH = "/tmp/tanaradio-mpv.sock" # フェーズ4で使った既存3ボタンのGPIO番号 PIN_PLAY_PAUSE = 17 # 再生/一時停止 PIN_BACK = 27 # 戻る:前のエピソードへ PIN_NEXT = 22 # 次へ:次のエピソードへ # ===== 設定ここまで ===== mpv_process = None def check_playlist(): if not PLAYLIST_FILE.exists(): print(f"{PLAYLIST_FILE} が見つかりません。") print("先にフェーズ5Cのスクリプトを実行して playlist.m3u を作ってください。") sys.exit(1) text = PLAYLIST_FILE.read_text(encoding="utf-8", errors="ignore").strip() if not text: print(f"{PLAYLIST_FILE} は空です。") sys.exit(1) def start_mpv(): global mpv_process if os.path.exists(SOCKET_PATH): os.remove(SOCKET_PATH) print("mpvを起動します。") print(f"プレイリスト: {PLAYLIST_FILE}") mpv_process = subprocess.Popen( [ "mpv", "--no-video", f"--input-ipc-server={SOCKET_PATH}", str(PLAYLIST_FILE), ], cwd=str(BASE_DIR), ) # mpvのIPCソケットができるまで少し待つ for _ in range(50): if os.path.exists(SOCKET_PATH): print("mpv操作用ソケットを確認しました。") return time.sleep(0.1) print("mpv操作用ソケットが見つかりません。") print("mpvが起動できていない可能性があります。") def send_mpv_command(command): payload = json.dumps({"command": command}).encode("utf-8") + b"\n" try: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: client.settimeout(1.0) client.connect(SOCKET_PATH) client.sendall(payload) # mpvからの返事を少し読む。 # これをしないと、mpv側に Broken pipe が出ることがある。 try: client.recv(4096) except socket.timeout: pass except FileNotFoundError: print("mpvのソケットが見つかりません。mpvが終了している可能性があります。") except ConnectionRefusedError: print("mpvに接続できませんでした。") except Exception as e: print("mpvへの命令送信でエラーが出ました。") print(e) def play_pause(): print("ボタン: 再生/一時停止") send_mpv_command(["cycle", "pause"]) def back_episode(): print("ボタン: 戻る:前のエピソードへ") send_mpv_command(["playlist-prev"]) def next_episode(): print("ボタン: 次へ") send_mpv_command(["playlist-next"]) def setup_buttons(): print("GPIOボタンを準備します。") print(f"再生/一時停止: GPIO{PIN_PLAY_PAUSE}") print(f"戻る: GPIO{PIN_BACK}") print(f"次へ: GPIO{PIN_NEXT}") play_button = Button(PIN_PLAY_PAUSE, pull_up=True, bounce_time=0.08) back_button = Button(PIN_BACK, pull_up=True, bounce_time=0.08) next_button = Button(PIN_NEXT, pull_up=True, bounce_time=0.08) play_button.when_pressed = play_pause back_button.when_pressed = back_episode next_button.when_pressed = next_episode # main側で保持するために返す return play_button, back_button, next_button def cleanup(signum=None, frame=None): global mpv_process print("\n終了します。") if mpv_process is not None and mpv_process.poll() is None: mpv_process.terminate() try: mpv_process.wait(timeout=5) except subprocess.TimeoutExpired: mpv_process.kill() if os.path.exists(SOCKET_PATH): try: os.remove(SOCKET_PATH) except OSError: pass sys.exit(0) def main(): signal.signal(signal.SIGINT, cleanup) signal.signal(signal.SIGTERM, cleanup) check_playlist() start_mpv() # 重要: # setup_buttons() の戻り値を変数に入れて保持する。 # これをしないと、環境によってはボタンが反応しなくなることがある。 buttons = setup_buttons() print("準備完了です。") print("3つの物理ボタンで操作できます。終了は Ctrl + C です。") pause() if __name__ == "__main__": main() 保存します。 Ctrl + O Enter Ctrl + X ステップ8:実行権限を付ける chmod +x tanaradio_5d.py ステップ9:5Dを実行する ./tanaradio_5d.py うまくいくと、次のような表示になります。 mpvを起動します。 プレイリスト: /home/ユーザー名/tanaradio5/playlist.m3u mpv操作用ソケットを確認しました。 GPIOボタンを準備します。 再生/一時停止: GPIO17 戻る: GPIO27 次へ: GPIO22 準備完了です。 3つの物理ボタンで操作できます。終了は Ctrl + C です。 この状態で、音声が流れ始めます。 ステップ10:3つのボタンを試す 10-1. 再生/一時停止ボタン GPIO17につながっているボタンを押します。 ターミナルに次のように出ればOKです。 ボタン: 再生/一時停止 音声が一時停止します。 もう一度押すと再開します。 10-2. 次へボタン GPIO22につながっているボタンを押します。 ボタン: 次へ 次のエピソードへ進めばOKです。 10-3. 戻るボタン GPIO27につながっているボ

    26 min
  6. Jul 2

    372 日記 | 複数RSSで過去3日分を古い順に再生

    複数RSSで過去3日分を古い順に再生できるようになりまいした(フェーズ5C完了)。*NHKの番組を再生している部分はカットしました。 フェーズ5C 詳細マニュアル 複数RSSから「過去3日分・古い順」のプレイリストを作る 0. 今回の目標 5Cでやることは、これです。 複数のRSS URLを feeds.txt に書く ↓ PythonでRSSを順番に読む ↓ 各RSSからエピソードを集める ↓ 過去3日分だけに絞る ↓ 公開日時の古い順に並べる ↓ playlist.m3u を作る ↓ mpvで連続再生する 今回も、まだやらないことがあります。 棚選び 再生モード選び LED表示 追加ボタン ロータリーエンコーダー 再生履歴 ここ、大事です。 5Cは「複数RSS化だけ」です。まだラジオの操作部品は増やしません。焦らないのが勝ち筋です。 ステップ1:作業フォルダへ移動する Raspberry Piでターミナルを開きます。 cd ~/tanaradio5 確認します。 pwd 次のような表示ならOKです。 /home/pi/tanaradio5 ユーザー名が違う場合は、pi の部分は別名になります。 ステップ2:5Bまでのファイルを確認する ls おそらく、次のようなファイルがあるはずです。 feed.txt make_playlist_5a.py make_playlist_5b.py playlist.m3u 5Cでは、5Bのファイルを壊さず、新しく次の2つを作ります。 feeds.txt make_playlist_5c.py 5Bはそのまま残します。 動いたものは残す。これは電子工作でもプログラムでも鉄則です。動いたものを消すと、あとで泣きます。小さく泣くならまだしも、Linux相手だとわりと本気で泣きます。 ステップ3:複数RSS用の feeds.txt を作る まず、RSS URLを複数書くためのファイルを作ります。 nano feeds.txt 中には、RSS URLを1行に1本ずつ書きます。 例です。 https://listen.style/p/xxxxx/rss https://listen.style/p/yyyyy/rss https://listen.style/p/zzzzz/rss 最初は欲張らず、まずは2本でよいです。 https://listen.style/p/xxxxx/rss https://listen.style/p/yyyyy/rss 保存します。 Ctrl + O Enter Ctrl + X 確認します。 cat feeds.txt RSS URLが複数行で表示されればOKです。 ステップ4:5C用のPythonスクリプトを作る 新しいファイルを作ります。 nano make_playlist_5c.py 次のコードをそのまま貼り付けてください。 #!/usr/bin/env python3 import urllib.request import xml.etree.ElementTree as ET from pathlib import Path from datetime import datetime, timedelta, timezone from email.utils import parsedate_to_datetime from zoneinfo import ZoneInfo FEEDS_FILE = "feeds.txt" PLAYLIST_FILE = "playlist.m3u" # 5Bと同じく「過去3日分」 DAYS = 3 JST = ZoneInfo("Asia/Tokyo") def read_feed_urls(): path = Path(FEEDS_FILE) if not path.exists(): raise FileNotFoundError(f"{FEEDS_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"{FEEDS_FILE} にRSS URLが書かれていません。") return urls def download_rss(url): print(f"RSSを取得します: {url}") request = urllib.request.Request( url, headers={ "User-Agent": "TanaRadio-Pi/5C" } ) with urllib.request.urlopen(request, timeout=20) as response: return response.read() def get_text(element, tag_name): child = element.find(tag_name) if child is not None and child.text: return child.text.strip() return "" def find_audio_url(item): # 通常のPodcast RSSでは enclosure に音声URLが入る enclosure = item.find("enclosure") if enclosure is not None: audio_url = enclosure.attrib.get("url") if audio_url: return audio_url # 念のため media:content 的な形式にも軽く対応する for child in item: if child.tag.endswith("content"): audio_url = child.attrib.get("url") if audio_url: return audio_url return None def parse_pub_date(item): pub_date_text = get_text(item, "pubDate") if not pub_date_text: return None try: dt = parsedate_to_datetime(pub_date_text) # タイムゾーン情報がない場合はUTC扱いにする if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(JST) except Exception: return None def parse_rss(rss_data, feed_url): root = ET.fromstring(rss_data) channel = root.find("channel") if channel is None: raise ValueError("RSSのchannelが見つかりません。") program_title = get_text(channel, "title") if not program_title: program_title = "番組名不明" items = channel.findall("item") episodes = [] for item in items: episode_title = get_text(item, "title") audio_url = find_audio_url(item) pub_dt = parse_pub_date(item) if not audio_url: continue if pub_dt is None: continue episodes.append({ "program_title": program_title, "episode_title": episode_title, "audio_url": audio_url, "pub_dt": pub_dt, "feed_url": feed_url, }) return episodes def collect_episodes(feed_urls): all_episodes = [] for url in feed_urls: try: rss_data = download_rss(url) episodes = parse_rss(rss_data, url) all_episodes.extend(episodes) print(f" 取得できたエピソード数: {len(episodes)} 件") except Exception as e: print(" このRSSの取得または解析でエラーが発生しました。") print(f" {e}") print(" 次のRSSへ進みます。") return all_episodes def filter_recent_episodes(episodes): now = datetime.now(JST) cutoff = now - timedelta(days=DAYS) recent = [] for ep in episodes: if ep["pub_dt"] >= cutoff: recent.append(ep) return recent def remove_duplicates(episodes): seen = set() unique = [] for ep in episodes: audio_url = ep["audio_url"] if audio_url in seen: continue seen.add(audio_url) unique.append(ep) return unique def make_playlist(episodes): if not episodes: raise ValueError( f"過去{DAYS}日分のエピソードが見つかりませんでした。" ) # 古い順に並べる episodes.sort(key=lambda ep: ep["pub_dt"]) playlist_lines = ["#EXTM3U"] for ep in episodes: pub_text = ep["pub_dt"].strftime("%Y-%m-%d %H:%M") program = ep["program_title"] title = ep["episode_title"] url = ep["audio_url"] playlist_lines.append(f"# {pub_text} / {program} / {title}") playlist_lines.append(url) Path(PLAYLIST_FILE).write_text( "\n".join(playlist_lines) + "\n", encoding="utf-8" ) print(f"{PLAYLIST_FILE} を作成しました。") print(f"プレイリスト内のエピソード数: {len(episodes)} 件") def main(): try: feed_urls = read_feed_urls() print(f"RSS URL数: {len(feed_urls)} 本") episodes = collect_episodes(feed_urls) print(f"全取得エピソード数: {len(episodes)} 件") episodes = remove_duplicates(episodes) print(f"重複除去後: {len(episodes)} 件") episodes = filter_recent_episodes(episodes) print(f"過去{DAYS}日分: {len(episodes)} 件") make_playlist(episodes) print("完了しました。") except Exception as e: print("エラーが発生しました。") print(e) if __name__ == "__main__": main() 保存します。 Ctrl + O Enter Ctrl + X ステップ5:実行権限をつける chmod +x make_playlist_5c.py ただし、実行はまず次の形でやるのがおすすめです。 python3 make_playlist_5c.py 以前のように、Pythonファイルをシェルとして実行してしまう事故を避けるためです。 python3 を前につければ、確実にPythonとして実行されます。 ステップ6:プレイリストを作る 実行します。 python3 make_playlist_5c.py うまくいくと、次のような表示になります。 RSS URL数: 2 本 RSSを取得します: https://listen.style/p/xxxxx/rss 取得できたエピソード数: 10 件 RSSを取得します: https://listen.style/p/yyyyy/rss 取得できたエピソード数: 10 件 全取得エピソード数: 20 件 重複除去後: 20 件 過去3日分: 5 件 playlist.m3u を作成しました。 プレイリスト内のエピソード数: 5 件 完了しました。 ここで見るポイントは、次の3つです。 RSS URL数 過去3日分 プレイリスト内のエピソード数 プレイリスト内のエピソード数 が1件以上なら成功です。 ステップ7:playlist.m3u の中身を確認する head -n 30 playlist.m3u 次のように表示されればOKです。 #EXTM3U # 2026-07-01 08:30 / 番組A / エピソードタイトル https://... # 2026-07-01 10:15 / 番組B / エピソードタイトル https://... ここで確認したいのは、複数番組が混ざっているかどうかです。 たとえば、 番組A 番組B 番組A 番組C のように並んでいれば、5Cらしい動きになっています。 単にRSSを順番に再生しているのではなく、複数RSSのエピソードを集めて、公開

    12 min

About

私立工科系大学教員(科学史)の日々の出来事や雑感。日記,メモ,ブログなど TanaRadio Blog ( https://tanakahi.hatenablog.com/ )