Flutter × Edge AI × IoT で「クラウド非接続」在庫管理システムを構築する #2
はじめに
前回の記事では、
・ Flutter × Edge AI × IoT を組み合わせて、クラウド非接続でも動作する
・在庫管理（カウント系）システムを構築するための全体アーキテクチャ
を、現場 DX を想定した構成例として紹介しました。
今回はその続きとして、
・ USB カメラ → Edge AI（TensorFlow Lite）→ Flutter アプリ
・という最小ループを「完全ローカル」で動かすところまで
を、実装イメージ付きで整理します。
実運用で使うコードは案件ごとに変わりますが、
- どのレイヤで何を担当させるか
- どこからどこまでを「HTTP API」として切り出すか
- Flutter アプリ側はどのように結果を受け取るか
といった設計の方針は共通化できます。
この記事は、以下のような方を想定しています。
- 現場向けの画像認識・在庫カウントを PoC したいエンジニア
- 「クラウドは使えないが、AI でカウントや検品をしたい」現場 DX 担当者
- Flutter を既に業務アプリで使っており、Edge AI との連携を検討している方
全体像のおさらい：今回フォーカスする部分
前回記事では、次のようなレイヤ構成を紹介しました。
・入力レイヤ
- USB カメラ／産業用カメラ
- 距離センサ・QR コードリーダ等（必要に応じて）
・推論レイヤ（Edge AI） - Raspberry Pi / x86 PC / NPU 搭載ボード
- TensorFlow Lite / ONNX Runtime
・プリケーションレイヤ（UI） - Flutter (Windows / Web / Linux など)
- 結果表示・在庫数管理・設定画面
・データレイヤ - ローカル DB（SQLite 等）
- CSV / JSON エクスポート
・ネットワーク - ローカル LAN or スタンドアロン
- インターネット非接続前提
今回はこのうち、もっとも小さな単位である
・「USB カメラから撮像」→「Edge 側で推論」→「Flutter から HTTP で呼び出し」
という部分にフォーカスします。
在庫 DB や複数ライン展開、ログ設計などは後続の記事で扱い、
まずは「動く最小構成」をクリアすることを目的にします。
開発環境と前提条件
この記事では、具体的な環境の一例として次のような構成を想定します。
- OS：Windows 10 / 11（Linux でもほぼ同様）
- 言語 / ランタイム
- Python 3.x
- TensorFlow Lite Runtime
- Flutter 3.x（stable） - ハードウェア
- UVC 対応 USB カメラ 1 台
- 開発用 PC（Edge デバイスを兼ねる）
ディレクトリ構成はシンプルに、次のように分けておきます。
project-root/
edge/ # Python + TFLite の推論サーバ
app/ # Flutter アプリ
前提として、以下は既にできる読者を想定します。
・Flutter プロジェクトの作成や、基本的な画面開発
・Python スクリプトの実行
・pip などによる必要パッケージのインストール
Step 1: 学習済みモデルを TFLite に用意する
本番環境では、学習データの収集やモデル設計・学習が大きなテーマになりますが、
この記事では焦点を「連携方法」に絞るため、次の前提を置きます。
・既に画像認識モデルが学習済みである
・例：棚の箱をカウントする物体検出モデル
・例：OK / NG を判定する画像分類モデル
・そのモデルを TensorFlow Lite 形式（.tflite）に変換済みである
TFLite 形式のファイルが 1 つあれば、
Edge デバイス側では次のような形で配置しておきます。
edge/
model/
inventory_model.tflite
server.py
camera_utils.py
inference.py
変換手順自体（SavedModel / Keras → TFLite Converter）は別記事で整理する前提とし、
ここでは「model/inventory_model.tflite が存在する」状態から進めます。
Step 2: Python + TFLite で「ローカル推論サーバ」を立てる
ここからは Edge 側の役割です。
・USB カメラから画像を 1 枚取得
・前処理を行いモデルに入力
・推論結果（個数等）を JSON にまとめ、HTTP 経由で返す
という流れを、Python スクリプトとして実装します。
2-1. USB カメラからの画像取得
まずは USB カメラから 1 フレーム取得するヘルパー関数を用意します。
わかりやすさ重視で、ここでは OpenCV を利用します。
edge/camera_utils.py のイメージ：
import cv2
def capture_frame(device_index: int = 0):
cap = cv2.VideoCapture(device_index)
if not cap.isOpened():
raise RuntimeError("Camera device not opened")
try:
# 必要なら解像度設定
# cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
# cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
ret, frame = cap.read()
if not ret:
raise RuntimeError("Failed to capture frame")
return frame
finally:
cap.release()
実運用では、毎回オープン／クローズだとオーバーヘッドが大きいので、
カメラを常時オープンしておく実装にすることが多いです。
ここでは「仕組み理解」を優先し、シンプルな形にしています。
2-2. 前処理と TFLite 推論
続いて、TFLite モデルを使って画像から結果を出す関数を用意します。
edge/inference.py のイメージ：
import numpy as np
import cv2
import tflite_runtime.interpreter as tflite
class InventoryModel:
def __init__(self, model_path: str):
self.interpreter = tflite.Interpreter(model_path=model_path)
self.interpreter.allocate_tensors()
self.input_details = self.interpreter.get_input_details()
self.output_details = self.interpreter.get_output_details()
def preprocess(self, frame):
# 入力サイズにリサイズ
h, w = self.input_details[0]["shape"][1:3]
img = cv2.resize(frame, (w, h))
# モデルに応じて正規化等を実施
img = img.astype("float32") / 255.0
img = np.expand_dims(img, axis=0)
return img
def infer(self, frame):
input_data = self.preprocess(frame)
self.interpreter.set_tensor(
self.input_details[0]["index"], input_data
)
self.interpreter.invoke()
output_data = self.interpreter.get_tensor(
self.output_details[0]["index"]
)
# ここでは「個数」を返す想定とし、実際のモデルに合わせて変換する
count = int(output_data[0][0])
return {"count": count}
実際には、物体検出モデルならバウンディングボックスやスコアを返すなど、
後処理はモデルごとに変わります。
記事としては、処理の流れ（前処理→推論→後処理→辞書化）が分かれば十分です。
2-3. HTTP API で結果を提供（FastAPI など）
最後に、推論処理を HTTP API として公開します。
ここでは FastAPI を例にしますが、Flask でも同じ考え方です。
edge/server.py のイメージ：
from datetime import datetime
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import uvicorn
from camera_utils import capture_frame
from inference import InventoryModel
app = FastAPI()
model = InventoryModel("model/inventory_model.tflite")
@app.get("/health")
def health():
return {"status": "ok", "time": datetime.now().isoformat()}
@app.get("/predict")
def predict():
try:
frame = capture_frame()
result = model.infer(frame)
return {
"count": result["count"],
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
# 本番ではログ出力なども行う
return JSONResponse(
status_code=500,
content={"error": str(e)},
)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)
このサーバを起動すると、
GET http://127.0.0.1:8000/health
GET http://127.0.0.1:8000/predict
といったエンドポイントがローカル PC 上で利用できるようになります。
ここまでで、
Edge 側が「USB カメラ付きのローカル推論サーバ」として振る舞う
状態が整いました。
Step 3: Flutter からローカル HTTP にアクセスする
次は、Flutter アプリ側からこのローカル API を呼び出し、
結果を UI 上に表示する部分です。
3-1. HTTP クライアントの用意
Flutter プロジェクト（app/）を作成し、
pubspec.yaml に HTTP クライアントライブラリを追加します。
ependencies:
flutter:
sdk: flutter
http: ^1.2.0 # バージョンは適宜
シンプルな API クライアントを 1 ファイルにまとめておきます。
lib/api_client.dart のイメージ：
import 'dart:convert';
import 'package:http/http.dart' as http;
class EdgeApiClient {
EdgeApiClient({this.baseUrl = 'http://127.0.0.1:8000'});
final String baseUrl;
Future<Map<String, dynamic>> health() async {
final url = Uri.parse('$baseUrl/health');
final res = await http.get(url).timeout(const Duration(seconds: 3));
if (res.statusCode != 200) {
throw Exception('Health check failed: ${res.statusCode}');
}
return jsonDecode(res.body) as Map<String, dynamic>;
}
Future<Map<String, dynamic>> predict() async {
final url = Uri.parse('$baseUrl/predict');
final res = await http.get(url).timeout(const Duration(seconds: 5));
if (res.statusCode != 200) {
throw Exception('Predict failed: ${res.statusCode}');
}
return jsonDecode(res.body) as Map<String, dynamic>;
}
}
3-2. シンプルな UI で結果を表示
最小の画面では、次の 3 要素があれば十分です。
・「カメラで撮影して推論」ボタン
・推論結果（個数）
・最終更新時刻
lib/main.dart のイメージ：
import 'package:flutter/material.dart';
import 'api_client.dart';
void main() {
runApp(const EdgeDemoApp());
}
class EdgeDemoApp extends StatelessWidget {
const EdgeDemoApp({super.key});
@override
Widget build(BuildContext context) {
return MaterialApp(
title: 'Edge AI Inventory Demo',
theme: ThemeData.dark(),
home: const InventoryPage(),
);
}
}
class InventoryPage extends StatefulWidget {
const InventoryPage({super.key});
@override
State<InventoryPage> createState() => _InventoryPageState();
}
class _InventoryPageState extends State<InventoryPage> {
final _api = EdgeApiClient();
int? _count;
String? _lastUpdated;
String? _error;
bool _isLoading = false;
Future<void> _runInference() async {
setState(() {
_isLoading = true;
_error = null;
});
try {
final result = await _api.predict();
setState(() {
_count = result['count'] as int?;
_lastUpdated = result['timestamp'] as String?;
});
} catch (e) {
setState(() {
_error = e.toString();
});
} finally {
setState(() {
_isLoading = false;
});
}
}
@override
Widget build(BuildContext context) {
return Scaffold(
appBar: AppBar(
title: const Text('Edge AI Inventory Demo'),
),
body: Padding(
padding: const EdgeInsets.all(16),
child: Column(
crossAxisAlignment: CrossAxisAlignment.start,
children: [
ElevatedButton(
onPressed: _isLoading ? null : _runInference,
child: _isLoading
? const CircularProgressIndicator()
: const Text('カメラで撮影して推論'),
),
const SizedBox(height: 24),
Text('推論結果（個数）: ${_count ?? '-'}'),
const SizedBox(height: 8),
Text('最終更新時刻: ${_lastUpdated ?? '-'}'),
const SizedBox(height: 16),
if (_error != null) ...[
const Divider(),
const Text('エラー', style: TextStyle(color: Colors.red)),
Text(
_error!,
style: const TextStyle(fontSize: 12),
),
],
],
),
),
);
}
}
ここまで実装すると、
１．Edge 側で python server.py を起動
２．Flutter アプリを起動
３．ボタンを押すと、カメラで 1 枚撮影 → 推論 → 個数表示
という流れが、クラウドに依存せずローカルだけで完結します。
3-3. 「完全ローカル」であることの確認ポイント
ローカル完結を確認するため、次のようなチェックを行うと安心です。
・PC のネットワークケーブルを抜いた状態でも動作するか
・baseUrl が http://127.0.0.1:8000 以外になっていないか
・OS のファイアウォール設定でローカルアクセスが許可されているか
このあたりを一度確認しておくと、
「外部に情報が出ていかない Edge 構成」であることを説明しやすくなります。
Step 4: 最低限のエラーハンドリングとデバッグ
実際に現場で使い始めると、まず直面するのは「動いていないときに何が悪いか分かりにくい」という問題です。
最低限、次のような観点でエラーハンドリング・ログ出力を入れておくと運用しやすくなります。
・Edge 側
・カメラが認識されていない場合に、ログと明確なエラーメッセージを出す
・モデルファイルが見つからない場合に即時に起動失敗する
・health API で簡易チェックができるようにする
・Flutter 側
・タイムアウト・HTTP 500 などを捕捉し、ユーザー向けメッセージを表示
・「Edge 側が起動していません」のようなメッセージ化
・デバッグ用にレスポンスをコンソールログにも出力
PoC 段階では、UI にすべてを詰め込む必要はありませんが、
「どこまで到達しているか」を把握できるログ行を数行入れておくだけでも、
トラブルシュートの速度が大きく変わります。
発展パターン：ストリーミングやパフォーマンス改善
今回の構成は「ボタンを押すたびに 1 フレーム推論する」最小構成です。
実案件では、次のような発展パターンを検討します。
・ストリーミング（WebSocket 等）
・カメラを常時回し続け、一定間隔で推論結果をストリームとして送る
・Flutter 側では StreamBuilder などでリアルタイム更新
・推論頻度の間引き
・毎フレーム推論せず、0.5 秒ごと・1 秒ごとに推論
・カウント用途であれば多くの場合十分
・モデル・ハードウェアのチューニング
・モデルの軽量化（量子化・入力サイズ削減）
・NPU 搭載ボードの活用
・ローカル DB との統合
・推論結果を SQLite に蓄積
・日次・月次の在庫推移レポートを生成
これらは別記事で掘り下げる予定ですが、
最初の PoC 段階では「単発推論で OK」と割り切って始めると進めやすくなります。
まとめと次回予告
本記事では、第 2 回として
・学習済みモデルを TFLite として Edge デバイスに配置
・Python + FastAPI で USB カメラ付きのローカル推論サーバを構築
・Flutter からローカル HTTP API を叩き、結果を UI に表示
という、「最小の実装ループ」を具体的なコードイメージ付きで整理しました。
この構成まで動けば、
「クラウドに依存せず、1 台の PC 上で完結する Edge AI × Flutter アプリ」
のベースが出来上がったことになります。
次回以降は、
・WebSocket ベースでのストリーミング UI
・SQLite などローカル DB への記録とグラフ表示
・実際の現場導入を見据えた運用設計（ログ・更新・監視）
といったトピックも含めて、
より「業務システムとしての完成度」を高めるための方法を整理していく予定です。
Discussion