ESP32にスピーカとマイクをつけて実験

 短い音の録音再生（エコー）が可能か実験してみた。今後、raspi経由でいろいろ処理を予定していますが。

　配線の数が多いときは、意外とエナメル線が使いやすい。裏表に通したりしてなるべく干渉しないようにした。PCB等の発注も考えたけれど、配線の変更の可能性もあるし、それなりの料金はかかるので、最小限のユニバーサル基板にしてみた。モジュールが基板からはみ出しているところは、あとでサイズがあいそうな百均のケースに固定できないか、考えたい。
/* esp32c3+INMP441+0802A_LCD+MAX98357A******** INMP441: SD-GPIO6 L/R-Ground WS-GPIO5 SCK-GPIO4 MAX98357A: DIN-GPIO7 DBCLK-GPIO4 LRC-GPIO5 0802A: RS-GPIO0 E-GPIO1 D4-GPIO2 D5-GPIO3 D6-GPIO8 D7-GPIO9 */ #include <Arduino.h> #include <driver/i2s.h> #include <LiquidCrystal.h> #define I2S_BCLK 4 #define I2S_LRC 5 #define I2S_DIN 6 #define I2S_DOUT 7 #define SAMPLE_RATE 16000 #define BUFFER_SIZE 256 #define RECORD_SECONDS 3 LiquidCrystal lcd(0,1,2,3,8,9); int32_t micBuffer[BUFFER_SIZE]; int16_t spkBuffer[BUFFER_SIZE]; int16_t audioBuffer[SAMPLE_RATE * RECORD_SECONDS]; bool recording=false; int recIndex=0; unsigned long lastSound=0; void setup(){ Serial.begin(115200); lcd.begin(8,2); lcd.print("READY"); i2s_config_t i2s_config = { .mode=(i2s_mode_t)(I2S_MODE_MASTER|I2S_MODE_RX|I2S_MODE_TX), .sample_rate=SAMPLE_RATE, .bits_per_sample=I2S_BITS_PER_SAMPLE_32BIT, .channel_format=I2S_CHANNEL_FMT_ONLY_LEFT, .communication_format=I2S_COMM_FORMAT_STAND_I2S, .intr_alloc_flags=0, .dma_buf_count=8, .dma_buf_len=256 }; i2s_pin_config_t pin_config = { .bck_io_num=I2S_BCLK, .ws_io_num=I2S_LRC, .data_out_num=I2S_DOUT, .data_in_num=I2S_DIN }; i2s_driver_install(I2S_NUM_0,&i2s_config,0,NULL); i2s_set_pin(I2S_NUM_0,&pin_config); i2s_set_clk(I2S_NUM_0,16000,I2S_BITS_PER_SAMPLE_32BIT,I2S_CHANNEL_MONO); } void loop(){ size_t bytesRead; i2s_read(I2S_NUM_0,micBuffer,sizeof(micBuffer),&bytesRead,portMAX_DELAY); int samples=bytesRead/4; int32_t energy=0; for(int i=0;i<samples;i++){ int16_t s=micBuffer[i]>>14; energy+=s*s; spkBuffer[i]=s; if(recording){ audioBuffer[recIndex++]=s; } } energy/=samples; if(!recording && energy>200000){ recording=true; recIndex=0; lastSound=millis(); lcd.clear(); lcd.print("REC"); lcd.setCursor(0,1); lcd.print("START"); } if(recording){ if(energy>200000){ lastSound=millis(); } if(millis()-lastSound>2000){ recording=false; lcd.clear(); lcd.print("SAVING"); delay(500); playAudio(); lcd.clear(); lcd.print("READY"); } } } void playAudio(){ lcd.clear(); lcd.print("PLAYING"); size_t written; for(int i=0;i<recIndex;i+=BUFFER_SIZE){ int n=min(BUFFER_SIZE,recIndex-i); i2s_write(I2S_NUM_0,&audioBuffer[i],n*2,&written,portMAX_DELAY); } }

raspiで音声アラームをスケジュールに従い起動

アラームをスケジュールで起動する仕組みを考えてみた。

wavplay.py**********************

import os
import sys
import time
import subprocess
import RPi.GPIO as GPIO 
RELAY_PIN = 27
def play_wav(wav_file):
    file_path = os.path.join("/home/pi", wav_file)
    if not os.path.isfile(file_path):
        return f"File not found: {file_path}"
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(RELAY_PIN, GPIO.OUT)
    try:
        GPIO.output(RELAY_PIN, GPIO.HIGH)
        time.sleep(0.2)
        proc = subprocess.Popen(["aplay", file_path])
        proc.wait()
        return f"Played {wav_file}"
    except Exception as e:
        return f"Error: {e}"
    finally:
        GPIO.output(RELAY_PIN, GPIO.LOW)
        GPIO.cleanup(RELAY_PIN)
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python script.py <waveFilename>")
        sys.exit(1)
    filename = sys.argv[1]
    result = play_wav(filename)
    print(result)

crontab -e******************

0 17-19 10 11-12 * /usr/bin/python3 /home/pi/wavplay.py PurchaseKerosene.wav

ちなみに、現在のcrontabの一部

5 6 * * * /home/pi/update-cloudflare-ufw.sh >> /home/pi/tmp/ufw-cron.log 2>&1

*/10  * 13-29 2 * curl -s http://localhost:8080/py/collect >/dev/null

0     8 13-29 2 * curl "http://192.168.1.30/setConfig?setA1=21.5&setA2=21.4&setA3=21.5"

１行目cloudflare関連、２，３行目が苗床ヒーター管理

IPoE接続に変えてみた

　 PPPoE接続をしばらく使ってきたが、最近土日や夜間がネット途切れるぐらい混雑してきたので、思い切ってIPoEに変えてみたら夜間でもダウンロード220Mbpsアップロード150Mbpsとだいぶ快適になった。

　ルータをどうするかで迷ったが、最近WIFI-Apを交換したばかりだし、WIFI付きの家庭用ルータはどうしても熱のためか、寿命が短いように思われたので、頑丈な有線ルータということで中古のIX2106をヤフオクで入手。希望小売価格が89000円ということなので、6000円ちょっとで購入でき、コスパはいいのでないかと思う。

　ただ、設定は情報集めがなかなか大変だった。ネットのNECの設定例も、ONU接続用の設定？のような感じだったので、PR400NEの内臓ONUを使うには不安があった。（電話はPR400NEから接続しているので、これを使わざるを得ない）　ChatGPTとGoogleAIの助けを借りてなんとか、解決。

注意点としては

・PR400NEの接続１を停止　sshで設定するために、念のためUSBシリアルコンソール（1500円ほど）用意

・NECのサンプルの変更したところ

 ip-ad subscriber Giga~は削除　※
interface GigaEthernet0.0

  no ip addressip address 192.168.2.2/24
  ipv6 enable
  ipv6 address autoconfig ※のかわり追加
  ipv6 dhcp client dhcpv6-cl 
  no shutdown

ip dhcp profile dhcpv4-sv

  assignable-range 192.168.1.50 192.168.1.70 　::dhcpの範囲を指定できる

  dns-server 192.168.1.1

interface Tunnel0.0
  tunnel mode 4-over-6
  tunnel destination fqdn gw.transix.jp
  tunnel source GigaEthernet1.0 → tunnel source GigaEthernet0.0 に変更
  ip unnumbered GigaEthernet1.0　これはなしで→  ip address 192.0.0.2/29 に変更
  ip tcp adjust-mss auto 
  no shutdown

※内部DNSサーバをうまく扱うことができなかったが、その後、「proxy-dns server ipアドレス priority 優先度」を設定することで、使えるようになった。

　また、bottleサーバでは、cloudflareからの外部アクセス制限をipアドレスで行う方法が使えず、別の方法に変更した。どうしても複雑化しているので細かいところに影響はでてしまうようだ。

※中古のため、縦置きスタンドがなかったため、不要になったアクリルボックスの破片からミニルータでL型に切り抜き、ねじ止めしたら安定するようになる。横置きより縦置きのほうが、熱は逃げやすいと思われるので、縦置きにした。

※念のため、使わずにいた廃品からはずした冷却ファンと使わずにいた温度センサーW1209、それにハードオフの330円の12Vアダプタで冷却もできるようにした。冷却ファンを最初はひもでしばっていたが、どうも不安定なので、あり合わせのアクリル破片をリベットでつないで、縦置き上部に載せられるように、フレームをつくってみた。

※PR400NE側の192.168.2.0/24側に192.168.1.0/24側からアクセスできず、HGWの管理画面が出せず不便だったが、以下の確認することで、なんとか解決できた。

1.PR-400NEの静的ルート: 192.168.1.0/24 宛を 192.168.2.2 へ

2.PR-400NEのipv4パケットフィルタ: 192.168.1.0/24 からの通信を許可

IX2106のマニュアルでは、２つのLANを接続するには、それぞれのインターフェースに ip address 192.168.1.254/24等を記述する例がでていたが、複数のルータが絡む場合はそれぞれきちんと確認が必要だということのようだ。

※結局、上記の設定では、ipv6がつながらないことが判明、再度、こちらの設定例１をもとに修正したらうまくいった。192.168.2.0/24というゾーンはつくらないほうがシンプルでいいようだ。ただ、HGWの管理画面を出すには、なぜかipv4ではだめで、ipv6のグローバルアドレス（たまに更新されるらしい）を指定必要だった。AIの意見も聞きながらも、そのままだとうまくいかないところもあったので、やはり、人間の判断も入れながら適宜修正は必要なようだ。

久しぶりの移動運用

 外の気温は6度と、まだ風が冷たかったけれど、比較的電波状況もよく、それほど強い風でもなかったので、久しぶりの移動運用。前回、ダイポールアンテナの実験失敗していたので、再チャレンジ。今回は、NanoVNAで慎重に長さを調整してみたところ、なんとかアンテナチューナで整合できた。末端が地面に近いこともあってか、かなり短めにする必要があった。とりあえず、１エリアの移動局と５９で交信することができた。

 

　（ヒートガンで柔らかくした）塩ビパイプで、不要になった物干しパイプ、壊れた雪はねの柄、つっぱり棒をつなげて作ったポールなので、かなりしなっていますがなんとか持ちこたえています。近くに１ｍ程度の杭があったので、ひもを結ぶのに使わせてもらいました。

　たまたま、河川公園に来ていた方から、声をかけられ、アマチュア無線だという話をしたところ、以前免許を取ったけど今はやってない方のようでした。受信の様子など、見てもらいながら、少しの間、アイボール。

cloudflareのトンネル利用

 これまで、cloudflareのプロキシーは利用していたが、グローバルIpアドレスをルータのものをDNSに設定していたので、Ipアドレスが変更になった途端、外部からアクセスできなくなった。自分では、自動更新の設定なっていたつもりだったけど、よく調べたら、なってなかったようで、結局は、トンネルを作らないとだめだったよう。cloudflareとraspiの間で独自プロトコルでトンネルを作り、常時接続状態にする仕組みのよう。早速、AIに聞きながらraspiにツールをいれ、設定変更。これで、Ipアドレス変更を気にしなくてもよくなり、だいぶ管理は楽になる。

※この方式にしたところ、外部からのアクセスログが正しく記録されなくなってしまった。いろいろ、試行錯誤した結果わかったことは。ヘッダー名: Cloudflareの仕様により、元のIPは CF-Connecting-IP という名前で送られてくるので、それを使うようにする設定が必要らしい。

・sudo a2enmod remoteip

・トンネルでlocalhost:8080でbottleを指定したのがよくなかった。apacheのログを取りたい場合はapache経由のためlocalhost:80にする必要あった。

・ログ形式(apache2.conf)で

LogFormat "%v:%p %a %l %u %t \"%r\" %>s %O \"%{Referer}i\" \"%{User-Agent}i\"" vhost_combined
LogFormat "%a %l %u %t \"%r\" %>s %O \"%{Referer}i\" \"%{User-Agent}i\"" combined
LogFormat "%a %l %u %t \"%r\" %>s %O" common

内部設定(000-default.conf)で

RemoteIPHeader CF-Connecting-IP
RemoteIPInternalProxy 127.0.0.1

変更が必要。

sudo systemctl restart apache2 で再起動。

bottleのformで文字化け

bottleのformのPostで全角文字が文字化けしてしまい、原因がつかめず苦労しました。結局は、対症療法的ですが、以下のようにして解決。

def schedule_edit_post(): 

　 schedules = load_schedule() 

　 new_schedules = [] 

　 for i, s in enumerate(schedules): 

　　　 start = request.forms.get(f"start{i}") 

　　　 end = request.forms.get(f"end{i}") 

　　　 #title = request.forms.get(f"title{i}") 

　　　  title = request.forms.get(f"title{i}").encode('latin1').decode('utf-8')

　いったん、latin1だったものをutf-8にデコードするという、二段構えで解決。bottleのバージョンが古くなってきたから？

raspiでＰＣのタイマー起動2

複数の予約できるようにアップデートしてみた。（一部<textarea>タグを半角にする必要あり） **********bottle****** #PC on off schedule SCHEDULE_FILE = "/home/pi/schedule.json" def load_schedule(): if not os.path.exists(SCHEDULE_FILE): return {} with open(SCHEDULE_FILE, "r") as f: return json.load(f) def save_schedule(s): tmp = SCHEDULE_FILE + ".tmp" with open(tmp, "w") as f: json.dump(s, f, indent=2) os.replace(tmp, SCHEDULE_FILE) # 現在のschedule取得 @app.get("/py/schedule") @local_only def get_schedule(): response.content_type = "application/json" return load_schedule() @app.get("/py/schedule/edit") @local_only def schedule_edit(): week_labels = ["日", "月", "火", "水", "木", "金", "土"] s = load_schedule() html = "<html><head><meta charset='UTF-8'></head><body><h2>スケジュール設定</h2><form method='post'>" for i, sched in enumerate(s): start = sched["start"] end = sched["end"] title = sched["title"] weekdays = sched["weekdays"] html += f""" <span>sched{i+1}</span>　開始:<input type="datetime-local" id="start{i}" name="start{i}" value="{start.replace(' ', 'T') if start else ''}" data-index="{i}">　　終了: <input type="datetime-local" id="end{i}" name="end{i}" value="{end.replace(' ', 'T') if end else ''}" data-index="{i}">　<input type="text" name="title{i}" id="title{i}" value="{title}" data-index="{i}"> 　曜日:""" for d, label in enumerate(week_labels): checked = "checked" if weekdays[d] else "" html += f""" <label> <input type="checkbox" name="weekday{i}_{d}" {checked}>{label} </label> """ html += "<br><br>" html += r""" <select id="targetIndex"> """ for j in range(len(s)): html += f""" <option value="{j}">sched{j+1}</option> """ html += """ </select> 　＜ｔｅｘｔａｒｅａ id="pasteArea" rows="3" cols="60" placeholder="ここに貼り付け"></ｔｅｘｔａａｒｅａ> 　<button type="button" onclick="parsePaste()">反映</button> <h4>新しいスケジュール</h4>　 開始:<input type="datetime-local" name="start_new">　終了:<input type="datetime-local" name="end_new"> 　<input type="submit" value="保存"> </form> <script> document.addEventListener("DOMContentLoaded", function () { function RepeatedReservFlg(index) { let repeated = false; for (let d = 0; d < 7; d++) { const cb = document.getElementsByName("weekday" + index + "_" + d)[0]; if (cb && cb.checked) { repeated = true; break; } } return repeated; } function checkDateTime(input,i) { if (!input.value) return; const inputDate = new Date(input.value); const now = new Date(); if (inputDate < now && !RepeatedReservFlg(i) ) { input.style.backgroundColor = "#d3d3d3"; // グレー } else { input.style.backgroundColor = ""; // 元に戻す } } // すべてのdatetime-localを取得 const inputs = document.querySelectorAll("input[type='datetime-local']"); inputs.forEach(function(input) { const index = input.dataset.index; // 初期表示時チェック checkDateTime(input,index); // 変更時にもチェック input.addEventListener("change", function() { checkDateTime(input,index); }); }); }); function toLocal(dt) { const y = dt.getFullYear(); const m = ("0"+(dt.getMonth()+1)).slice(-2); const d = ("0"+dt.getDate()).slice(-2); const h = ("0"+dt.getHours()).slice(-2); const min = ("0"+dt.getMinutes()).slice(-2); return `${y}-${m}-${d}T${h}:${min}`; } function parsePaste() { const text = document.getElementById("pasteArea").value.trim(); const lines = text.split("\\n"); let start, end,title=""; // パターン① 両方に日付あり let r1 = /(\d{4})\/(\d{2})\/(\d{2}).*?(\d{2}):(\d{2}).*?(\d{4})\/(\d{2})\/(\d{2}).*?(\d{2}):(\d{2})/; // パターン② 終了に日付なし let r2 = /(\d{4})\/(\d{2})\/(\d{2}).*?(\d{2}):(\d{2}).*?[～~\-]\s*(\d{2}):(\d{2})/; let m; if (m = text.match(r1)) { start = new Date(m[1], m[2]-1, m[3], m[4], m[5]); end = new Date(m[6], m[7]-1, m[8], m[9], m[10]); const lineIndex = lines.findIndex(line => r1.test(line)); if (lineIndex >= 0 && lines[lineIndex + 1]) title = lines[lineIndex + 1].trim(); } else if (m = text.match(r2)) { start = new Date(m[1], m[2]-1, m[3], m[4], m[5]); end = new Date(m[1], m[2]-1, m[3], m[6], m[7]); // 日跨ぎ対応 if (end <= start) end.setDate(end.getDate() + 1); const lineIndex = lines.findIndex(line => r2.test(line)); if (lineIndex >= 0 && lines[lineIndex + 2]) title = lines[lineIndex + 2].trim(); } else { alert("日付形式が認識できません"); return; } // 開始4分前 start.setMinutes(start.getMinutes() - 4); let duration = (end - start) / 60000; // 25％余裕 let margin = Math.min(30, Math.max(5, Math.ceil(duration * 0.25))); end.setMinutes(end.getMinutes() + margin); const index = document.getElementById("targetIndex").value; document.getElementById("start"+index).value = toLocal(start); document.getElementById("end"+index).value = toLocal(end); document.getElementById("title"+index).value = title; } </script> </body></html>""" return html @app.post("/py/schedule/edit") @local_only def schedule_edit_post(): schedules = load_schedule() new_schedules = [] for i, s in enumerate(schedules): start = request.forms.get(f"start{i}") end = request.forms.get(f"end{i}") title = request.forms.get(f"title{i}") title = title.encode('latin1').decode('utf-8') if title else "" # ★ 曜日配列を取得 weekdays = [] for d in range(7): val = request.forms.get(f"weekday{i}_{d}") weekdays.append(bool(val)) # チェックされていればTrue if start and end: start = start.replace("T", " ") + ":00" end = end.replace("T", " ") + ":00" new_schedules.append({ "start": start, "end": end, "title": title, "weekdays": weekdays }) else: # 繰り返しだけ（日時なし）も許可するならここで追加 if any(weekdays): new_schedules.append({ "start": "", "end": "", "title": title, "weekdays": weekdays }) save_schedule(new_schedules) return """ <html><head><meta charset="UTF-8"></head><body> 保存しました<br> <a href="/py/schedule/edit">戻る</a> </body></html> """ *********gpio_control.py*********** import RPi.GPIO as GPIO import time import subprocess import json from datetime import datetime, timedelta import os IN_PIN = 22 # タッチセンサー入力 OUT_PIN = 4 # ON/OFFパルス出力 SCHEDULE_FILE = "/home/pi/schedule.json" # タイマー用JSON PC_IP = "192.168.*.*" # PINGチェックするPC GPIO.setmode(GPIO.BCM) GPIO.setup(IN_PIN, GPIO.IN, pull_up_down=GPIO.PUD_DOWN) GPIO.setup(OUT_PIN, GPIO.OUT, initial=GPIO.LOW) def trigger(tm): GPIO.output(OUT_PIN, GPIO.HIGH) time.sleep(tm) GPIO.output(OUT_PIN, GPIO.LOW) print(f"{datetime.now()} Trigger executed") def is_pc_on(): try: result = subprocess.run( ["ping", "-c", "1", "-W", "1", PC_IP], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) return result.returncode == 0 except: return False def load_schedule(): if not os.path.exists(SCHEDULE_FILE): return [] with open(SCHEDULE_FILE) as f: return json.load(f) def check_timer(schedules): now = datetime.now() now_time = now.strftime("%H:%M:%S") # Python weekday(): 月=0 ... 日=6 today_index = (now.weekday() + 1) % 7 # あなた形式：日=0 for sched in schedules: weekdays = sched.get("weekdays", [False]*7) # ?? 繰り返しスケジュール if any(weekdays): if not weekdays[today_index]: continue start_time = sched["start"][-8:] if sched["start"] else None end_time = sched["end"][-8:] if sched["end"] else None if start_time and end_time: if start_time <= now_time <= end_time: return True # ?? 単発スケジュール else: start = datetime.strptime(sched["start"], "%Y-%m-%d %H:%M:%S") end = datetime.strptime(sched["end"], "%Y-%m-%d %H:%M:%S") if start <= now <= end: return True return False prevflg = False # 直前のON/OFF状態 try: while True: schedules = load_schedule() timer_flag = check_timer(schedules) touch_flag = GPIO.input(IN_PIN) == GPIO.HIGH curflg = timer_flag or touch_flag if curflg != prevflg: if curflg == False : if is_pc_on(): #touch_flg:falseでも、すぐにはping onにならないので反応せず trigger(5.0) else : if is_pc_on()==False: trigger(0.1) prevflg = curflg time.sleep(1) # 1秒ごとにチェック finally: GPIO.cleanup()

・複数にすることで、複雑度が増して解決が難しかった。そこで、タイマーにしても、タッチセンサーにしても、統一したflagを使うことにして、単純化したらうまくいった。flagがon->offになる場合は、pingでPCをチェックしoffならそのまま、onならoffにするためのHigh信号triggerとし、flagがoff->onになる場合は、pingによりPCチェックをして、onならそのまま、offならonにするためのHigh信号triggerとする。

・offにするためのHigh信号triggerは、長めの1秒にしてやる必要があった。

・曜日による繰り返し設定もできるようにした。(2026/3/5)