Python, クラス分割, 型ヒント, クリーンアーキテクチャ

お題

トイドローン Tello のプログラミング記事です。

複雑化したソースコード main.py の改善を行います。

問題

まずはなぜ「複雑化している」という印象を受けるのか考えてみます。 現在のソースコードを改めて眺めてみると、 いろんな機能が1つの main.py の中に混然一体となっています。この「いろんな機能」と「混然一体」の掛け算が「複雑」という印象を与えるようです。

  • 現在の main.py の中に存在する機能:
    • データの格納場所 (Infoクラス)
    • 画面表示(メインループ)
    • ドローンへのコマンド送信処理 (send_command関数)
    • ドローンの状態受信処理 (receive_state関数)
    • ドローンのビデオ受信処理 (receive_video関数)
    • スレッド起動 (threading.Thread … start)

確かに機能は多いです。これらがすべて main.py という同じファイルの中にあるので複雑に見えるのは仕方ないといえそうです。

解決策

この問題の解決策としてはシンプルに「機能ごとにソースコードを格納するファイルを分ける」がよさそうです。また同時にクラス化にも対応してみたいと思います。現在はほぼ「1機能=1関数」なのでクラス化のメリットはさほどないのですが、将来的にひとつの機能を複数の関数で実装したくなった場合には、クラスによって関数などの機能実装がグループ化されている方がソースコードを追っかけやすくなるので。

クラス分割の方針

main.py で実装されている機能を図にすると以下のような役割分担になっています。

現在の機能

こうして図にしてみると機能ごとの役割分担的にはそれぞれの箱の単位で綺麗にまとまっているように見えます。また（機能を表す）箱と箱の間の依存関係も一方向（処理→データ）で疎になっている（各処理の間には依存関係がない）ので、とりあえずこの単位をそのままクラスにしても大丈夫そうです。

  • クラス分割案:
    • データの格納場所 - Info クラス
    • 画面表示 - Dashboard クラス
    • ドローンへのコマンド送信処理 - DroneCommandRequester クラス
    • ドローンの状態受信処理 - DroneStateReceiver クラス
    • ドローンのビデオ受信処理 - DroneVideoReceiver クラス
    • 起動処理 - Runner クラス

クラス分割後

(1) データ格納場所の実装

まずデータの格納場所としてInfo クラスを作成します。

これは main.py の中ですでにクラスになっていたので、単純にそのまま info.py というファイルに切り出しただけのものです。

(2) 各処理の実装

次に画面表示機能やドローンへの受送信機能を切り出します。

以下のクラスに分割できました。

各クラスの記述内容は main.py に書かれていた内容をそのまま転記しています。 違いは「クラスの中にあるメソッドとして処理を実装する」という点のみです。

ちなみに今回実装したメソッドの基本構造は全部同じで以下の形になっています。

  • メソッドの内容:
    1. 起動したら (メソッドが呼び出されたら) まず初期処理を行う。
    2. ループを回してひたすら処理を続ける。

とっても単純ですね…。具体的には以下のような実装になります。

class Dashboard:
    # 処理実行開始
    def run(self, info):
        # 画面レイアウトなどの初期処理を実行
        (...)

        while True:
          # 画面からの入力や画面表示を行う
          (...)
class DroneCommandRequester:
    # 処理実行開始
    def start(self, info):
        # ソケットの作成などの初期処理を実行
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        (...)

        while info.is_active():
          # コマンドの送受信を実行
          (...)
class DroneStateReceiver:
    # 処理実行開始
    def start(self, info):
        # ソケットの作成などの初期処理を実行
        state_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        (...)

        while info.is_active():
          # 状態受信を実行
          (...)
class DroneVideoReceiver:
    # 処理実行開始
    def start(self, info):
        # VideoCaptureでの画像受信準備を行う
        cap = cv2.VideoCapture("udp://0.0.0.0:11111?overrun_nonfatal=1")

        while info.is_active():
          # 画像受信処理を実行
          (...)

メソッドは以下のルールで名前付けを行いました。

  • メソッド名のルール
    • run: メインスレッドで動作する処理
    • start: メインスレッド以外のスレッド上で動作する処理

各処理の役割はクラス名 (例: DroneCommandRequester, DroneVideoReceiver) で表現できているため、メソッド名にあえてバラバラな名前を付ける理由はありません。 そこで上記のようなルールにして統一してみました。

同じ名前にすると以下のようなメリットもあります。

  • メソッド名を統一するメリット
    • そのほうが 覚えやすい
    • すべての処理を「同じもの」として扱えるため、 便利

後者の「便利」の具体的な例は次項で説明する「起動処理」です。

(3) 起動処理

各クラスの起動処理は Runner クラス で行っています。 具体的には「各クラスの start メソッドを専用のスレッド上で起動し、最後にメインスレッドで run メソッドを呼び出す」です。

図にするとこんなイメージになります。

クラス分割後

main.py 上ではこんなコードでした。

state_receive_thread = threading.Thread(target=receive_state)
state_receive_thread.start()

(...)

video_receive_thread = threading.Thread(target=receive_video)
video_receive_thread.start()

(...)

command_send_thread = threading.Thread(target=send_command)
command_send_thread.start()

(...)

while True:
  (...)

このようなほとんど同じ記述を何度も繰り返すコードは「コピペコード」として忌み嫌われますが、前述のようにメソッドの名前に関するルールを決めておけば以下のように重複のないコードが書けます。

class Runner:

    # startables 引数は start メソッドを持つオブジェクトの配列
    def run(self, startables, runnable):
        # Info を初期化する
        info = Info()

        # 引数で渡された startables の全要素に対してスレッド起動処理を呼び出す
        for startable in startables:
            threading.Thread(target=startable.start, args=(info,)).start()

        # メインスレッドで起動する処理 (run メソッド) を呼び出す
        runnable.run(info)

便利ですね。

(4) main.py

最後に main.py です。ほとんどの処理は別のクラスに移動しましたが、2つの役割を残しました。

  • main.py の役割:
    • 各クラスの初期化を行う
    • Runner クラスの run を呼び出す
from runner import Runner
from drone_command_requester import DroneCommandRequester
from drone_state_receiver import DroneStateReceiver
from drone_video_receiver import DroneVideoReceiver
from dashboard import Dashboard

# メインスレッドで起動する Dashboard の初期化
runnable = Dashboard()

# 別スレッドで起動するクラスの初期化
startables = [DroneCommandRequester(), DroneStateReceiver(), DroneVideoReceiver()]

# Runner の初期化
runner = Runner()

# Runner の run を呼び出す
runner.run(startables, runnable)

型ヒントの導入

ここまでの実装で、当初の目的どおりクラス分割ができて正常に動作するものにもなっているのですが、今後の拡張に備えて型ヒントも導入してみたいと思います。型ヒントを導入すれば Visual Studio Code等のエディタがコードの入力補完や誤りの指摘をしてくれるようになって開発がはかどるので。

例えば Info クラスでは、以下のように各変数やメソッドの引数・戻り値に型情報を記述します。

class Info:
    __state: dict[str, float]
    __is_active: bool
    __image: Mat
    __command: str
    __sent_command: str
    __result: str

    def __init__(self) -> None:
    (...)

Runner の run メソッドに渡す2つの引数には該当する型がないため、新たに以下の型を定義します。

  • Runnable: メインスレッドで動作する処理 (runメソッドを持つクラス)
  • Startable: メインスレッド以外のスレッド上で動作する処理 (startメソッドを持つクラス)

これを用いて Runner の run メソッドに型ヒントを追加すると以下の形になります。

class Runner:
    def run(self, startables: list[Startable], runnable: Runnable) -> None:

ちなみに Startable および Runnable はあっさりめの実装にしてます。

class Startable:
    def start(self, info: Info) -> None:
        pass

class Runnable:
    def run(self, info: Info) -> None:
        pass

最後に DroneCommandRequester, DroneStateReceiver, DroneVideoReceiver は Startable を親クラスとし、 Dashboard は Runnable を親クラスとする形に書き換えます。

class DroneCommandRequester(Startable):
    def start(self, info: Info) -> None:

...

class DroneStateReceiver(Startable):
    def start(self, info: Info) -> None:

...

class DroneVideoReceiver(Startable):
    def start(self, info: Info) -> None:

...

class Dashboard(Runnable):
    def run(self, info: Info) -> None:

クリーンアーキテクチャ

現在のクラス構成はクリーンアーキテクチャに沿った形になっています。

クリーンアーキテクチャ

  • インターフェースや実装の変更がシステム全体の挙動を破壊する可能性が高いもの (方針) を内側の層に配置し、変更がシステム全体の挙動に影響を及ぼしにくいもの (詳細) を外側の層に配置しています。
  • 依存の方向が「外側 → 内側」の一方通行でその逆はありません。具体的に言うと、クラスを実装するコードは内側の層に属するクラスを import しますが、内側の層に属するクラスが外側の層に属するクラスを import することはありません。つまり、外側の層の変更は内側の層に影響を及ぼしません (依存性のルールに従う)。
  • ユースケース層の Runner は run メソッドの中でインターフェースアダプタ層の DroneCommandRequester クラス等の run メソッドを呼び出していますが、具象クラスを名指しせずに抽象クラス Startable を利用して呼び出しています (DIP: 依存関係逆転の原則に従う)。これにより内側の層が外側に依存することを回避しています。
  • 一番外側にある (究極的な詳細である) main.py がその他のクラスを直接作成するファクトリの役割と実行を開始するドライバの役割を担っています。

クリーンアーキテクチャを採用するメリット

この構成にすると、今回の開発ではものすごくうれしいことがあります。

それは「テスト（試行錯誤）が気軽できるようになること」です。

例えば「コマンド文字列入力ではなく『指で示した方向に移動させる』というようなハンドジェスチャーでドローン操作ができるようにしたい」という要求があるとします。これに対応するためには、ハンドジェスチャーを画像認識してドローンへのコマンドに変換する必要があります。この「画像認識」にはかなりの試行錯誤が必要になるはずです。試行錯誤のたびに毎回ドローンを飛ばすとなると、いくらバッテリーの替えがあったとしてもなかなかつらい状況になりそうです。

このとき「ドローンから受信した動画の代わりに PC のWebカメラから受信した動画を使う」ということができれば、試行錯誤のハードルがかなり下がるはずです。

今回のようにクリーンアーキテクチャを意識したクラス構成にした場合は「DroneVideoReceiver の Web カメラ版実装を作成して main.py で差し替える」という最小限の変更を行うだけで、他のクラスには全く手を加えなくても安全にこれに対応できます。

次回

ということで次回からは実際に以下のテーマで進めてみたいと思います。

Tello 関連記事

かわかみしんいち。島根県津和野町在住のフリーランスエンジニア。複合現実(Mixed Reality)と3DUXでおもちゃを作るのが趣味。 https://github.com/ototadana