IoTデバイスの開発にクリーンアーキテクチャを適用してみた話

はじめに

クリーンアーキテクチャは、アンクル・ボブが提唱しているアーキテクチャです。この考えは特定の分野に限定されず、Web系の開発でよく使われることが多いと思います。

今回、IoTデバイスの開発でも活用してみて、開発効率が向上した実感を得たので紹介します。

開発言語はPythonを使用していますが、他の言語でも同様の考え方で適用できると思います。

IoTデバイスの開発

クラウド側では、システムの保守性や拡張性を高めるため、クリーンアーキテクチャを採用していました。導入までが大変でしたが、軌道に乗ってくるとテストプログラムが書きやすいことが実感できました。デバイス側の開発にもこのメリットを生かしたいと考え、デバイス側の開発でもクリーンアーキテクチャを採用することにしました。

適用方法

クリーンアーキテクチャの詳細は書籍やQiitaの記事などで説明されているので、ここではIoTデバイスの開発にクリーンアーキテクチャを適用する際のポイントを紹介します。

入力→ビジネスロジック→出力に分離

デバイスが持つビジネスロジックを明確にし、それ以外の部分と分離します。IoT機器の開発では、入力→ビジネスロジック→出力の3つに分けることが多いと思います。入力の部分はセンサーから値を読み取る処理、出力の部分は接続されている機器を動作させる処理になると思います。クラウドとの通信が入力や出力になる場合もあると思います。

ハードウェアに依存しないUnitTest

分離することでテストが行いやすくなります。入力部分と出力部分をモックに置き換えることで、ビジネスロジックのテストが行いやすくなります。特にIoT機器では、入出力の部分はハードウェアに依存するので、実機がないとテストできないことがあります。この置き換えにより、ハードウェアがなくてもビジネスロジックのテストが行えるようになります。これが最も大きなメリットだと思います。

入力、出力部分を個別開発

入力→ビジネスロジック→出力この一連の処理を一つのプログラムで記述すると、入力の処理と出力の処理が密に結合しています。開発を行う場合にも、入力部分と出力部分を個別に開発することが難しく、一つのブラックボックス化してしまうことも多いと思います。

ビジネスロジックを境界にして入力、出力を分離することで、入力部分と出力部分を独立したモジュールにできます。これにより、入力部分と出力部分の開発を別々に行うことが可能になります。

ビジネスロジックがほとんどない例もあると思います。例えば、センサーから取得した値をそのままクラウドに送信するような場合です。このような場合もあえてこの形式に当てはめることで、入力の処理と出力の処理を分離することができます。

見通しがよくなる

第三者がプログラムを見た時に、どの部分がビジネスロジックで、どの部分が入力、出力処理なのかが明確になります。これにより、プログラムの見通しがよくなり、保守性が向上します。

一つ例を挙げてみます。センサーからの入力値に応じて照明を点灯させるようなIoT機器があったとします。この機器に照明が点灯した時にクラウドにも通知する機能を追加するとします。

【ビジネスロジックが分離されていない場合】

たぶん次のような作業になると思います。

1. センサーの値を取得している処理を見つける。
2. 順に処理の流れを追って、照明の点灯を行っている所を特定する。
3. 照明の点灯を行っている処理にクラウドに通知する処理を追加する。

このような調査を行いやすいように、ドキュメントを整備したり、コードのコメントを充実させることも大切ですが、ビジネスロジックが一定の規則で作成されていれば、修正箇所を特定するのは容易になります。

【ビジネスロジックが分離されている場合】

1. クラウドに通知する出力処理を作成する。
2. 該当するビジネスロジックにクラウドへの通知を追加する。

照明の点灯を判定するビジネスロジック部分にクラウドへの通知を追加すれば良いので、修正部分が明確です。

サンプルプログラム

クリーンアーキテクチャを適用したIoTデバイスの開発のサンプルプログラムを説明用に作成してみました。入出力を実際にどのように書けばよいのかの参考になればと思います。LightingLogicの__init__の引数に、light_controllerとcloud_notifierを渡していることが重要です。これにより、ビジネスロジックは固定した特定のクラスではなく、パラメータとして渡された入出力処理に依存するようになっています。

project/
│-- main.py
│-- input/         
│   └── sensor_input.py    # 入力層: センサーデータ取得  
│-- use_case/       
│   └── lighting_logic.py  # ビジネスロジック  
│-- output/        
│   ├── light_controller.py # 出力層: 照明の制御  
│   └── cloud_notifier.py   # 出力層: クラウド通知
└── test/
    └── test_logic.py      # ユニットテスト

sensor_input.py

class SensorInput:
    def __init__(self, threshold: float):
        self.threshold = threshold

    def get_sensor_value(self) -> float:
        # 実際のセンサー入力処理 (モックでランダム値を返す)
        import random
        return random.uniform(0, 100)

light_controller.py

class LightController:
    def turn_on(self):
        print("Light: ON")

    def turn_off(self):
        print("Light: OFF")

cloud_notifier.py

class CloudNotifier:
    def notify(self, message: str):
        # クラウド通知処理 (ここでは標準出力)
        print(f"Cloud Notification: {message}")

lighting_logic.py

class LightingLogic:
    def __init__(self, light_controller, cloud_notifier):
        self.light_controller = light_controller
        self.cloud_notifier = cloud_notifier

    def process_sensor_data(self, sensor_value: float, threshold: float):
        if sensor_value >= threshold:
            self.light_controller.turn_on()
            self.cloud_notifier.notify(f"Light turned ON. Sensor value: {sensor_value}")
        else:
            self.light_controller.turn_off()
            self.cloud_notifier.notify(f"Light turned OFF. Sensor value: {sensor_value}")

main.py

from input.sensor_input import SensorInput
from output.light_controller import LightController
from output.cloud_notifier import CloudNotifier
from use_case.lighting_logic import LightingLogic

def main():
    # センサー値の閾値設定
    THRESHOLD = 50.0

    # クリーンアーキテクチャの各層のインスタンス生成
    sensor_input = SensorInput(THRESHOLD)
    light_controller = LightController()
    cloud_notifier = CloudNotifier()

    # ビジネスロジックの初期化
    lighting_logic = LightingLogic(light_controller, cloud_notifier)

    # センサー値取得とビジネスロジック処理
    sensor_value = sensor_input.get_sensor_value()
    print(f"Sensor Value: {sensor_value}")
    lighting_logic.process_sensor_data(sensor_value, THRESHOLD)

if __name__ == "__main__":
    main()

test_logic.py

import pytest
from unittest.mock import Mock
from use_case.lighting_logic import LightingLogic

@pytest.fixture
def setup_logic():
    """Mockを使用してビジネスロジックのセットアップ"""
    light_controller = Mock()
    cloud_notifier = Mock()
    lighting_logic = LightingLogic(light_controller, cloud_notifier)
    return lighting_logic, light_controller, cloud_notifier

def test_process_sensor_data_on(setup_logic):
    lighting_logic, light_controller, cloud_notifier = setup_logic

    # センサー値が閾値以上の場合
    lighting_logic.process_sensor_data(sensor_value=60, threshold=50)

    # 照明がONになったか確認
    light_controller.turn_on.assert_called_once()
    light_controller.turn_off.assert_not_called()

    # クラウド通知が正しいメッセージで呼ばれたか確認
    cloud_notifier.notify.assert_called_once_with("Light turned ON. Sensor value: 60")

def test_process_sensor_data_off(setup_logic):
    lighting_logic, light_controller, cloud_notifier = setup_logic

    # センサー値が閾値未満の場合
    lighting_logic.process_sensor_data(sensor_value=40, threshold=50)

    # 照明がOFFになったか確認
    light_controller.turn_off.assert_called_once()
    light_controller.turn_on.assert_not_called()

    # クラウド通知が正しいメッセージで呼ばれたか確認
    cloud_notifier.notify.assert_called_once_with("Light turned OFF. Sensor value: 40")

クリーンコードでは、依存性逆転の原則 (Dependency Inversion Principle, DIP)というのがあり、ビジネスロジックが直接出力層を呼び出さないようにする事が望ましいですが、ここでは簡単のために省略しています。この部分の配慮については、また別の機会に紹介したいと思います。

まとめ

IoTデバイスの開発でも、クリーンアーキテクチャを適用する事で得られたメリット

• ビジネスロジックの独立性が高まる
• テストが行いやすくなる
• 入力、出力の処理を個別に開発できる