rasp-python-1

=====================================================================================

<< BACK >>

=====================================================================================

ws／Http　Server

import os
import ssl
import json
import time
import asyncio
import threading
import websockets
import sys
import subprocess
from http.server import HTTPServer, SimpleHTTPRequestHandler
from gpiozero import AngularServo, LED
from gpiozero.pins.pigpio import PiGPIOFactory

# --- 1. グローバル変数の宣言と初期化 ---
# 変数名だけ用意
factory = None
servo_ud = None
servo_rl = None
ssr = None
httpd = None
running = True
is_homing = False  
light_ = 0
rl_ang, rl_dir = 0, 0
ud_ang, ud_dir = 0, 0

# --- 2. HTTPハンドラ（ブラウザに画面を出す役目） ---
class RestrictedHTTPHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        allowed_extensions = {'.html', '.css', '.js', '.ico'}
        _, ext = os.path.splitext(self.path)
        if self.path == "/" or ext.lower() in allowed_extensions:
            return super().do_GET()
        else:
            self.send_error(403, "Access Denied.")

# --- 3. モーター制御ロジック（別スレッドで動く） ---
def motor_loop(servo, limit):
    global ud_ang, rl_ang, ud_dir, rl_dir, running, is_homing
    step = 0.5
    is_ud = (servo == servo_ud)
    
    servo.detach() # 最初は脱力

    while running:
        cur_dir = ud_dir if is_ud else rl_dir
        if cur_dir != 0:
            angle = ud_ang if is_ud else rl_ang
            angle += (cur_dir * step)
            angle = max(-limit, min(limit, angle))
            servo.angle = angle
            if is_ud: ud_ang = angle
            else: rl_ang = angle
            time.sleep(0.05)
            continue 
        else:
            if not is_homing and servo.value is not None:
                servo.detach()
            time.sleep(0.1)

# --- 4. WebSocket ハンドラー（スマホの命令をさばく） ---
async def ws_handler(websocket, path=None):
    global light_, rl_ang, rl_dir, ud_ang, ud_dir, running, is_homing
    try:
        async for message in websocket:
            data = json.loads(message)
            cmd = data.get("cmd")
            if cmd == 'get_pos':
                pos_str = f"{int(-ud_ang)}/{int(-rl_ang)}"
                await websocket.send(json.dumps(pos_str))
            elif cmd in ['up', 'down', 'left', 'right', 'stop']:
                if cmd == 'up':      ud_dir = -1
                elif cmd == 'down':    ud_dir = 1
                elif cmd == 'right':   rl_dir = -1
                elif cmd == 'left':    rl_dir = 1
                elif cmd == 'stop':    ud_dir = rl_dir = 0
            elif cmd == 'light':
                light_ ^= 1
                ssr.on() if light_ else ssr.off()
            elif cmd == 'home':
                is_homing = True 
                rl_ang = ud_ang = rl_dir = ud_dir = 0
                servo_ud.angle = 0
                servo_rl.angle = 0
                time.sleep(1.5) 
                is_homing = False 
                pos_str = "0/0"
                await websocket.send(json.dumps(pos_str))                
    except Exception:
        pass

# --- 5. メイン実行部（ここが「本番」の順番） ---
if __name__ == '__main__':
    # ① まずハードウェアを準備する
    factory = PiGPIOFactory()
    servo_ud = AngularServo(21, min_angle=-90, max_angle=90, min_pulse_width=0.0005, max_pulse_width=0.0024, pin_factory=factory, initial_angle=0)
    servo_rl = AngularServo(20, min_angle=-90, max_angle=90, min_pulse_width=0.0005, max_pulse_width=0.0024, pin_factory=factory, initial_angle=0)
    ssr = LED(26) 
    
    time.sleep(1.5) # 最初の位置合わせ完了を待つ

    # ② 次に「モータースレッド」を裏側で走らせる
    # これをサーバー起動（③）の「前」に書くのが重要！
    threading.Thread(target=motor_loop, args=(servo_ud, 60), daemon=True).start()
    threading.Thread(target=motor_loop, args=(servo_rl, 70), daemon=True).start()
    
    # ③ サーバーの設定と起動
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    base_path = os.path.dirname(os.path.abspath(__file__))
    ssl_ctx.load_cert_chain(os.path.join(base_path, 'localhost.pem'), os.path.join(base_path, 'localhost-key.pem'))

    async def start_servers():
        global httpd
        async with websockets.serve(ws_handler, "0.0.0.0", 8700, ssl=ssl_ctx):
            # HTTPSサーバーをさらに別スレッドで走らせる
            httpd = HTTPServer(('', 443), RestrictedHTTPHandler)
            httpd.socket = ssl_ctx.wrap_socket(httpd.socket, server_side=True)
            threading.Thread(target=httpd.serve_forever, daemon=True).start()
            
            print("--- System Ready ---")
            while running: await asyncio.sleep(1)

    # 実行
    try:
        loop.run_until_complete(start_servers())
    except KeyboardInterrupt:
        print("\nStopping...")
        running = False
        if httpd:
            stop_thread = threading.Thread(target=httpd.shutdown)
            stop_thread.start()
            stop_thread.join()
    finally:
        if httpd: httpd.server_close()
        servo_ud.close()
        servo_rl.close()
        ssr.off()
        sys.exit(0)

if name == ‘main’:

「このファイルが直接実行された時だけ、この下の処理を動かせ」**という決まり文句
なぜこれが必要なのか？
Pythonのファイルは、2つの使われ方をします。

直接実行される: python inter_02.py と打ち込んで動かす。
インポートされる: 他のプログラムから import inter_02 として、中にある関数だけを借りる。

もしこの if 文がないと、他のファイルから関数を借りようとしただけで、HTTPSサーバーやWSSサーバーが勝手に起動してしまいます。

__name__ と '__main__' の正体
Pythonは実行時に、内部で特殊な変数 __name__ を自動で作ります。

直接実行したとき: __name__ の中身は '__main__' という文字列になります。
import されたとき: __name__ の中身は 'inter_02'（ファイル名）になります。

# --- 5. メイン実行部（ここが「本番」の順番） ---
if __name__ == '__main__':
    # ① まずハードウェアを準備する
 ..............................................    
    # ③ サーバーの設定と起動
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)　　　<- SSLの宣言
    base_path = os.path.dirname(os.path.abspath(__file__))
    ssl_ctx.load_cert_chain(os.path.join(base_path, 'localhost.pem'), os.path.join(base_path, 'localhost-key.pem'))　<- SSL　Fileの設定

    async def start_servers():
        global httpd
        async with websockets.serve(ws_handler, "0.0.0.0", 8700, ssl=ssl_ctx): <- WSS Serverの宣言
            # HTTPSサーバーをさらに別スレッドで走らせる
            httpd = HTTPServer(('', 443), RestrictedHTTPHandler)　<- HTTPS Serverの宣言
            httpd.socket = ssl_ctx.wrap_socket(httpd.socket, server_side=True)
            threading.Thread(target=httpd.serve_forever, daemon=True).start()  <- HTTPS Serverの宣言
            
            print("--- System Ready ---")
            while running: await asyncio.sleep(1)
　　　　　　　これ以降このループを実行するのみ。１秒に１回実行される
　　　　　　　　ー＞しかし裏ではHttpsとWssサーバは稼働している。２つのサーバを稼働させる為の空ループ
　　　　　　　終わりはrunning=Falseとなった時。これはプログラムで指定している。

    # 実行
    try:
        loop.run_until_complete(start_servers())
　　　　　start_servers()が終了するまで実行しなさいという関数
    except KeyboardInterrupt:
        print("\nStopping...")
        running = False
        if httpd:
            stop_thread = threading.Thread(target=httpd.shutdown)
            stop_thread.start()
            stop_thread.join()
    finally:
        if httpd: httpd.server_close()
        servo_ud.close()
        servo_rl.close()
        ssr.off()
        sys.exit(0)

try: except KeyboardInterrupt: finally:の関係

まずはtry:が実行される
　　ー＞loop.run_until_complete(start_servers())が実行
　　　　start_servers()が終了するまで実行しなさいという関数
　　　　start_servers()ではWss,Https Serverが起動。
　　　　　ー＞while running: await asyncio.sleep(1)　が鍵
except KeyboardInterrupt:
　　インデントがtry:と同じ位置にある事に注意
　　runningはCTRL+Cが押された時にFalseとなる。この時初めてstart_servers()が終了する
　　同じ位置にあるとstart_servers()が終了後CTRL+C受け入れとなりstart_servers()が終了しない
　　つまりexcept KeyboardInterrupt:はstart_servers()を終了させる事が出来る位置
　　try:と同じ位置にいる必要がる。
　　ここでの処理は、async withで処理されないthreadの後処理
finally:
　　ここで最終処理を行う。　　

HTTPS Server関係

 # --- 2. HTTPハンドラ（ブラウザに画面を出す役目） ---
class RestrictedHTTPHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        allowed_extensions = {'.html', '.css', '.js', '.ico'}
        _, ext = os.path.splitext(self.path)
        if self.path == "/" or ext.lower() in allowed_extensions:
            return super().do_GET()
        else:
            self.send_error(403, "Access Denied.")

os.path.splitext(self.path) の動き
この関数は、パスを「ファイル名」と「拡張子」の2つに分割して、タプルという形式で返します。
例えば self.path が "/intercom.js" だった場合：

左側（名前）: "/intercom"
右側（拡張子）: ".js"

コードにある _, ext という書き方は、「左側の名前はいらない（無視する）ので _ に捨てて、右側の拡張子だけを ext という変数に入れる」 という意味です。

if ext.lower() in allowed_extensions:
ext(拡張子の部分）を小文字に変換した文字列がallowed_extensionsに有るかチェック
有る：super().do_GET()　（標準のHTTP操作を行う関数）

WebSocketサーバー関係

# --- 4. WebSocket ハンドラー（スマホの命令をさばく） ---
async def ws_handler(websocket, path=None):
    global light_, rl_ang, rl_dir, ud_ang, ud_dir, running, is_homing
    try:
        async for message in websocket:
            data = json.loads(message)　　　＜ーデータを辞書形式に変換
            cmd = data.get("cmd")　　　　　＜ー{"cmd":, "値"} から"cmd"に対する値の取得
            if cmd == 'get_pos':　　　　　　＜ー位置データを送信
                pos_str = f"{int(-ud_ang)}/{int(-rl_ang)}"
                await websocket.send(json.dumps(pos_str))
            elif cmd in ['up', 'down', 'left', 'right', 'stop']:　＜ーその他のコマンド処理
                if cmd == 'up':      ud_dir = -1
                elif cmd == 'down':    ud_dir = 1
                elif cmd == 'right':   rl_dir = -1
                elif cmd == 'left':    rl_dir = 1
                elif cmd == 'stop':    ud_dir = rl_dir = 0
            elif cmd == 'light':
                light_ ^= 1
                ssr.on() if light_ else ssr.off()
            elif cmd == 'home':
                is_homing = True 
                rl_ang = ud_ang = rl_dir = ud_dir = 0
                servo_ud.angle = 0
                servo_rl.angle = 0
                time.sleep(1.5) 
                is_homing = False 
                pos_str = "0/0"
                await websocket.send(json.dumps(pos_str))                
    except Exception:
        pass

async for message in websocket:et, path=None):
待機: 「スマホからメッセージは届いているかな？」と確認します。
譲渡（重要！）: もし届いていなければ、**「届くまでこの処理は一時停止！その間、他の仕事（別のスマホの対応や、サーバーの維持など）をしていいよ！」**と、実行権を他に譲ります。
再開: スマホからデータが届いた瞬間、パッとこの場所に戻ってきて、次の行（json.loads など）を実行します。

`data = json.loads(message)`
役割：文字列を「辞書形式」に変換する
ブラウザからは `{"cmd": "up", "value": 10}` のような形式でデータが届きます。これは JSON（ジェイソン）という世界共通のデータ形式ですが、届いた瞬間は単なる「文字の羅列」です。

json.loads(): 「JSONの文字列（Load String）」を読み込んで、Python の 辞書（dict）型 に変換せよ、という命令です。
結果: これにより、data["cmd"] のようにキーを指定して値を取り出せるようになります。

`cmd = data.get("cmd")`
役割：辞書から「特定のデータ」を安全に取り出す
辞書になった `data` の中から、今回実行したい命令（`cmd` という名前のデータ）を取り出します。

なぜ data["cmd"] ではないのか？: もしブラウザから cmd というキーワードが含まれていない変なデータが届いた場合、data["cmd"] と書くとプログラムはエラーで強制終了してしまいます。
.get("cmd") のメリット: get を使うと、もし cmd が入っていなくてもエラーにならずに None（空っぽ）を返してくれます。プログラムを突然死させないための**「安全装置」**です。

except Exception: pass

構文の分解

except Exception: 「すべてのエラー（例外）」をキャッチするという意味です。計算ミス（ZeroDivisionError）や、ネットワークの切断、タイポなど、Pythonが投げるほぼすべてのエラーがここにかかります。
pass: 「何もしない」という命令です。エラーが起きたとき、本来なら画面に赤い文字でエラーメッセージが出ますが、それを一切出さずにスルーします。

なぜこのコードが使われるのか？
inter_02.py のようなハードウェア制御では、以下のような「一瞬の不具合」でプログラム全体を止めたくない場合に使われます。

通信のノイズ: WebSocketで一瞬変なデータが届いたが、無視して次のデータを待てばいい時。
ハードウェアの応答待ち: サーボモータが一瞬ビジーだったが、次のループでリトライすればいい時。

if name == ‘main’:

「このファイルが直接実行された時だけ、この下の処理を動かせ」**という決まり文句

なぜこれが必要なのか？
Pythonのファイルは、2つの使われ方をします。

直接実行される: python inter_02.py と打ち込んで動かす。
インポートされる: 他のプログラムから import inter_02 として、中にある関数だけを借りる。

もしこの if 文がないと、他のファイルから関数を借りようとしただけで、HTTPSサーバーやWSSサーバーが勝手に起動してしまいます。

__name__ と '__main__' の正体
Pythonは実行時に、内部で特殊な変数 __name__ を自動で作ります。

直接実行したとき: __name__ の中身は '__main__' という文字列になります。
import されたとき: __name__ の中身は 'inter_02'（ファイル名）になります。

つまり、if __name__ == '__main__': と書くことで、**「importされただけなら何もしない。直接実行された時だけ、サーバー起動などの重い処理を始めろ」**と指示しているのです。

用途	適したライブラリ
現在の日時を取得 (ローカルタイム)	datetime.datetime.now()
現在の時刻をエポック秒で取得	time.time()
時刻を特定のフォーマットで表示	time.localtime() + time.strftime()
タイムゾーンを考慮した時刻取得	datetime.now(pytz.timezone(…)) or arrow.now()
WebAPIや時刻処理を簡単に扱いたい	arrow.now()
時間計測 (コードの実行時間)	timeit.default_timer() or time.perf_counter()

処理名	概要
Popen()	プロセスの初期化、実行
Popen.wait()	プロセスの終了を待機　プロセス終了後、終了コード（リターンコード）を返す
Popen.poll()	非同期的にプロセスの出力を取得
Popen.communicate()	標準入力を渡し、標準出力と標準エラーを取得する
(stdout, stderr)	プロセスの標準出力や標準エラーを取得

文字	意味
`'r'`	読み込み用に開く (デフォルト)
`'w'`	書き込み用に開き、まずファイルを切り詰める
`'x'`	排他的な生成に開き、ファイルが存在する場合は失敗する
`'a'`	書き込み用に開き、ファイルが存在する場合には末尾に追記する
`'b'`	バイナリモード
`'t'`	テキストモード (デフォルト)
`'+'`	更新用に開く (読み込み・書き込み用)

ws／Http Server

if __name__ == ‘__main__’:

try: except KeyboardInterrupt: finally:の関係

HTTPS Server関係

WebSocketサーバー関係

cmd = data.get("cmd")役割：辞書から「特定のデータ」を安全に取り出す辞書になった data の中から、今回実行したい命令（cmd という名前のデータ）を取り出します。

except Exception: pass

if __name__ == ‘__main__’:

Thread

1. daemon=True の場合（おすすめ）

2. daemon=False（または指定なし）の場合

1. スレッドを「止める」方法

2. スレッドを「再開」する方法

3. 実務的な「負荷対策」の別解：待機時間を変える

4. それでもスレッドを完全にやめたい場合

文字列を特定の文字（区切り文字）で分割

演算子

ファイルの読み書き

読み

全文を一度に読み込む

1行ずつリストとして読み込む

末尾に改行コード（\n）を取り除く

Fileの終わり

書き

中身をすべて上書き

追記

リストをまとめて書き込む

よく使うオプション（モード）

注意点

バイナリモードの重要なルール

GPIO制御

複数行の文字列定義

３つのダブルまたはシングルコーテーション ”””…””” or ”’…”’ を使用

\（バックスラッシュ)を使用

()を使用

get()

os.path.splitext()

urlparse()

parse_qs(), iter(), next()

self.path.endswith

for

リスト,タプル

辞書

range関数

breakとcontinue

文字列を抽出

インデックス

スライス

f文字列

変数の値を文字列に挿入

式の評価

メソッドの呼び出し

辞書のアクセス

条件式（三項演算子）

フォーマット指定子の使用

左寄せ、中央寄せ、右寄せ

日付のフォーマット

2進数、8進数、16進数表現

デバッグ用の=指定子

時刻取得

1. datetime.datetime.now()

用途・シチュエーション:

注意点:

2. time.time()

用途・シチュエーション:

注意点:

3. time.localtime() + time.strftime()

用途・シチュエーション:

注意点:

4. datetime.now(pytz.timezone("Asia/Tokyo"))

用途・シチュエーション:

注意点:

5. arrow.now()

用途・シチュエーション:

注意点:

6. timeit.default_timer()

用途・シチュエーション:

注意点:

7. time.perf_counter()

用途・シチュエーション:

ws／Http　Server

if name == ‘main’:

`cmd = data.get("cmd")`
役割：辞書から「特定のデータ」を安全に取り出す
辞書になった `data` の中から、今回実行したい命令（`cmd` という名前のデータ）を取り出します。

if name == ‘main’:

1. `daemon=True` の場合（おすすめ）

2. `daemon=False`（または指定なし）の場合

３つのダブルまたはシングルコーテーション　”””…””” or ”’…”’　を使用

4. `datetime.now(pytz.timezone("Asia/Tokyo"))`

5. `arrow.now()`

6. `timeit.default_timer()`

7. `time.perf_counter()`