LineageOSというものがあるのでそれを焼く。以下に手順が書いてある。
イメージをビルドしたい場合は以下に従う。lineage-22.2-20260806-nightly-porg-signed.zipというものができるので、adb reboot recoveryで再起動してから、それをadbでsideloadする。
部分ビルドしたい場合は以下のような感じ。mより後はAndroid.dpに記載のターゲット。
$ source build/envsetup.sh && breakfast porg_tab && croot $ m android.hardware.audio@6.0-impl
WSL立ち上げるとPCが重い。メモリも6GBくらい食っていた。まともにPC使えないので掃除した。
psコマンドでリソース食っている原因を確認。k8sがソース食っていた。そういえば昔入れたな。。
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
2337 root 20 0 3107080 663908 105088 S 113.1 8.2 13:38.90 kubelite
1509 hayashi 20 0 62.4g 662388 58496 S 22.0 8.2 5:29.46 node
6375 root 20 0 1780156 47232 32000 S 19.0 0.6 1:59.35 calico-node
113 root 20 0 526812 7276 1408 S 12.1 0.1 0:47.59 snapfuse
こんな感じで掃除するとそこそこ軽くなった。
$ sudo snap remove microk8s ubuntu-desktop-installer gtk-common-themes $ sudo apt purge *k8s* $ sudo apt autoremove
WebSocketの使い方メモ
const WebSocket = require("ws"); // サーバを作成 const wss = new WebSocket.Server({ port: PORT }); // 接続時の処理 wss.on("connection", (ws) => { let id = clientIdCounter++; const client = { ws, name: "Anonymous", roomId: null, userId: id, }; // クライアントを保持しておく(必須ではない) clients.set(ws, client); // クライアントに自分のIDを返却 client.ws.send( JSON.stringify({ type: "yourId", id: client.userId, }) ); // ここにメッセージ受信した際の処理を書いておく ws.on("message", (message) => { ... }
// IPアドレスとポート指定してWebSocketを作成 ws = new WebSocket("ws://" + ip.value) // Openした際の処理 ws.onopen = () => { ws.send(JSON.stringify({ type: "getRooms" })) } // サーバからメッセージを受信した際の処理 ws.onmessage = (e) => { const msg = JSON.parse(e.data) logger(log_mp, "Recv: " + msg.type); if (msg.type === "yourId") { myUserId = msg.id logger_add(log_mp, "Your ID: " + myUserId); } }
// メッセージ送信処理。WebSocketがOpenしているか確認してから送信
if (ws && ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({
type: "joinRoom",
.vscodeignoreがないのが原因。中身は以下のような感じ。
# .vscodeignore **/*.ts **/*.map **/.gitignore **/tsconfig.json **/webpack.config.js **/vsc-extension-quickstart.md **/changelog.md **/node_modules/** !node_modules/vscode-languageclient/** # 必要なら例外で残す !node_modules/vscode-jsonrpc/**
Pythonでマルチプロセス通信を行う場合はmultiprocessingのManagerやQueueを使用すると便利。Managerは共有の変数、Queueはトリガーとして使用できる。
import multiprocessing import time class Device: def __init__(self, shared_state, command_queue): self.shared_state = shared_state # 状態共有用 self.command_queue = command_queue # コマンド受信用 self.running = True def run(self): counter = 0 while self.running: # コマンド受信チェック（非ブロッキング） try: cmd = self.command_queue.get_nowait() if cmd == "reset": print("[Device] Reset command received") counter = 0 elif cmd == "stop": print("[Device] Stop command received") self.running = False except Exception: pass # 状態更新 counter += 1 self.shared_state["device_status"] = f"running: {counter}" time.sleep(1) class Main: def __init__(self, shared_state, command_queue): self.shared_state = shared_state self.command_queue = command_queue def run(self): for i in range(8): state = self.shared_state.get("device_status", "unknown") print(f"[Main] Device state: {state}") # 3秒後にresetコマンド送信 if i == 3: print("[Main] Sending reset command") self.command_queue.put("reset") time.sleep(1) if __name__ == "__main__": manager = multiprocessing.Manager() shared_state = manager.dict() command_queue = multiprocessing.Queue() # Deviceプロセス起動 device = Device(shared_state, command_queue) device_process = multiprocessing.Process(target=device.run) device_process.start() try: main = Main(shared_state, command_queue) main.run() finally: # 停止コマンド送信 command_queue.put("stop") device_process.join() print("Stopped Device process.")
想定するノイズは、バッファアンダーランのようなノイズ。 検証対象としてはノイズ検出しやすそうな穏やかな曲調のカノン。以下からダウンロード。
以下のスクリプトで、バッファアンダーランを想定して一部バッファを削除してノイズを想定した音源を作成。
import numpy as np import librosa import soundfile as sf import random def inject_audio_dropouts(input_path, output_path, num_dropouts=5, dropout_ms=50, dropout_interval_ms=500, random_drop=False, remove=False): y, sr = librosa.load(input_path, sr=None, mono=True) samples_per_drop = int(sr * (dropout_ms / 1000.0)) print(f"Loaded audio file: {input_path} with sample rate: {sr} Hz samples_per_drop: {samples_per_drop}") if random_drop: for i in range(num_dropouts): start = random.randint(0, len(y) - samples_per_drop) if remove: # サンプル削除（欠落） y = np.concatenate([y[:start], y[start+samples_per_drop:]]) else: # 無音化 y[start:start+samples_per_drop] = 0 print(f"\rDropout {i+1}/{num_dropouts} at sample {start}", end='') else: for i in range(num_dropouts): start = i * int(sr * (dropout_interval_ms / 1000.0)) if start + samples_per_drop > len(y): break if remove: # サンプル削除（欠落） y = np.concatenate([y[:start], y[start+samples_per_drop:]]) else: # 無音化 y[start:start+samples_per_drop] = 0 print(f"\rDropout {i+1}/{num_dropouts} at sample {start}", end='') sf.write(output_path, y, sr) # 使用例 inject_audio_dropouts("Canon.wav", "Canon_with_dropouts.wav", num_dropouts=10000, dropout_ms=40, dropout_interval_ms=200, random_drop=False, remove=True)
ノイズ検出は以下のコードで行う。スペクトル変化量でノイズ検出する。
import librosa import numpy as np import plotly.graph_objects as go import os def detect_noise_offline( input_file, spectral_threshold=0.02, frame_length=1024, hop_length=512 ): """ WAV/音声ファイルを読み込み、スペクトル変化量でノイズ検出し、 PlotlyでHTMLグラフを保存する。 Parameters ---------- input_file : str 入力音声ファイル（wavなど） spectral_threshold : float スペクトル変化量の絶対値閾値 frame_length : int STFTのフレーム長 hop_length : int STFTのホップ長 """ # ==== 音声読み込み ==== y, sr = librosa.load(input_file, sr=None, mono=True) # ==== STFT ==== S = np.abs(librosa.stft(y, n_fft=frame_length, hop_length=hop_length)) # ==== フレーム間のスペクトル差分 ==== diffs = [] for i in range(1, S.shape[1]): diff_val = np.mean(np.abs(S[:, i] - S[:, i - 1])) diffs.append(diff_val) times = librosa.frames_to_time(np.arange(1, S.shape[1]), sr=sr, hop_length=hop_length) # ==== ノイズ検出 ==== detections = [t for t, d in zip(times, diffs) if d > spectral_threshold] # ==== Plotly グラフ作成 ==== fig = go.Figure() # スペクトル変化量 fig.add_trace(go.Scatter( x=times, y=diffs, mode='lines', name='Spectral Change' )) # 閾値線 fig.add_trace(go.Scatter( x=[times[0], times[-1]], y=[spectral_threshold, spectral_threshold], mode='lines', line=dict(color='red', dash='dash'), name='Threshold' )) # 検出ポイント fig.add_trace(go.Scatter( x=detections, y=[spectral_threshold] * len(detections), mode='markers', marker=dict(color='orange', size=8, symbol='x'), name='Detected Noise' )) fig.update_layout( title=f"Spectral Change and Noise Detection - {os.path.basename(input_file)}", xaxis_title="Time (s)", yaxis_title="Spectral Difference (abs)", template="plotly_white" ) # ==== HTML保存 ==== output_html = os.path.splitext(input_file)[0] + ".html" fig.write_html(output_html) # ==== ログ出力 ==== for det_time in detections: print(f"Detect! {det_time:.3f}s") print(f"✅ グラフを保存しました: {output_html}") return detections # ==== 使用例 ==== # detect_noise_offline("Canon.wav", spectral_threshold=0.02)
ノイズ有無でのグラフの比較は以下のとおり。ノイズありのほうが値が大きくなっており、0.4あたりを閾値にすればノイズ検出できそう。
リアルタイムで検出する場合は以下。
import sounddevice as sd import numpy as np import librosa import time import sys # ==== 設定 ==== DEVICE_ID = 1 # Noneでデフォルトマイク, またはデバイス番号 FRAME_DURATION = 0.05 # 秒 SPECTRAL_THRESHOLD = 0.13 # ← 絶対値で設定（例: 0.02） FRAME_LENGTH = 1024 HOP_LENGTH = 512 prev_spec = None def audio_callback(indata, frames, time_info, status): global prev_spec if status: print(status, file=sys.stderr) # モノラル化 y = indata[:, 0] # STFT S = np.abs(librosa.stft(y, n_fft=FRAME_LENGTH, hop_length=HOP_LENGTH)) if prev_spec is not None: diff = np.mean(np.abs(S - prev_spec)) # バー表示 bar = "#" * int(diff * 100) sys.stdout.write(f"\r[{bar:<50}] {diff:.4f}") sys.stdout.flush() # 閾値判定（絶対値） if diff > SPECTRAL_THRESHOLD: print(f"\nDetect! {time.strftime('%H:%M:%S')} diff={diff:.4f}") prev_spec = S def main(): print(f"録音開始 (閾値={SPECTRAL_THRESHOLD}, Ctrl+Cで終了)") with sd.InputStream( device=DEVICE_ID, channels=1, callback=audio_callback, samplerate=44100, blocksize=int(44100 * FRAME_DURATION) ): while True: time.sleep(0.1) if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\n終了")
