LINE notifyのモジュール化
別のメソッドでも使えるようにモジュール化する
また
message = 'ファイルパス自動取得テスト'
の部分は
他のプログラムで
生成された文字列を受け取って実行するようにコードを変更する
vim line_notify.py
で
import requests
import os
from PIL import Image
from io import BytesIO
from utils import load_config, get_latest_directory, get_image_files
def resize_image_if_needed(image_data, max_size=3 * 1024 * 1024):
if len(image_data) > max_size:
image = Image.open(BytesIO(image_data))
new_size = (image.width // 2, image.height // 2)
image = image.resize(new_size, Image.LANCZOS)
output = BytesIO()
image_format = image.format if image.format else 'JPEG'
image.save(output, format=image_format)
return output.getvalue()
return image_data
def send_line_notify(message, config_path='config.json'):
# 設定ファイルを読み込む
config = load_config(config_path)
# 設定ファイルからトークンとディレクトリパスを取得
token = config['token']
base_path = config['image_file_path']
# 最新のpredictディレクトリを取得
latest_dir = get_latest_directory(base_path)
image_files = get_image_files(latest_dir)
url = 'https://notify-api.line.me/api/notify'
headers = {'Authorization': f"Bearer {token}"}
params = {'message': message}
# 最新のpredictディレクトリ内の全ての画像ファイルに対してLINE Notify APIにリクエストを送信
for image_file_path in image_files:
with open(image_file_path, 'rb') as img_file:
img_data = img_file.read()
img_data = resize_image_if_needed(img_data)
# ファイルデータをバイトデータとして用意
files = {'imageFile': BytesIO(img_data)}
files['imageFile'].name = os.path.basename(image_file_path)
# LINE Notify APIにリクエストを送信
res = requests.post(url, headers=headers, params=params, files=files)
# レスポンスを出力
print(f"File: {image_file_path}")
print(res.status_code)
print(res.text)
とりあえずこれを使えるかテストする
import argparse
import json
import cv2
from ultralytics import YOLO
from collections import defaultdict
# コマンドライン引数の解析
parser = argparse.ArgumentParser(description="YOLOv8 Object Detection")
parser.add_argument('image_path', type=str, help='Path to the input image file')
args = parser.parse_args()
# ラベルマッピングファイルのパス
label_mapping_path = 'label_mapping.json'
# JSONファイルからクラスラベルのマッピングを読み込み
with open(label_mapping_path, 'r', encoding='utf-8') as f:
label_mapping = json.load(f)
# YOLOv8モデルのロード
model = YOLO('inventory_model/best.pt') # ここで適切なモデルを選択
# 画像のロード
image = cv2.imread(args.image_path)
# 画像の検出
results = model(image, save=True, conf=0.2, iou=0.5)
# 検出結果の取得
detections = results[0] # 最初の結果を取得
classes = detections.boxes.cls
# 検出物体のカウント
object_counts = defaultdict(int)
for cls in classes:
class_label = model.names[int(cls)]
if class_label in label_mapping:
label = label_mapping[class_label]
else:
label = class_label
object_counts[label] += 1
# 検出結果の表示
for label, count in object_counts.items():
print(f'{label}: {count}個')
の中で呼び出すようにする
import argparse
import json
import cv2
from ultralytics import YOLO
from collections import defaultdict
from line_notify import send_line_notify # インポートを追加
# コマンドライン引数の解析
parser = argparse.ArgumentParser(description="YOLOv8 Object Detection")
parser.add_argument('image_path', type=str, help='Path to the input image file')
args = parser.parse_args()
# ラベルマッピングファイルのパス
label_mapping_path = 'label_mapping.json'
# JSONファイルからクラスラベルのマッピングを読み込み
with open(label_mapping_path, 'r', encoding='utf-8') as f:
label_mapping = json.load(f)
# YOLOv8モデルのロード
model = YOLO('inventory_model/best.pt') # ここで適切なモデルを選択
# 画像のロード
image = cv2.imread(args.image_path)
# 画像の検出
results = model(image, save=True, conf=0.2, iou=0.5)
# 検出結果の取得
detections = results[0] # 最初の結果を取得
classes = detections.boxes.cls
# 検出物体のカウント
object_counts = defaultdict(int)
for cls in classes:
class_label = model.names[int(cls)]
if class_label in label_mapping:
label = label_mapping[class_label]
else:
label = class_label
object_counts[label] += 1
# 検出結果のメッセージ生成
message_lines = [f'{label}: {count}個' for label, count in object_counts.items()]
message = '\n'.join(message_lines)
# 検出結果の表示
for line in message_lines:
print(line)
# LINE Notifyにメッセージを送信
send_line_notify(message)
これを
python count_inventory_terminal.py data_bak/Baskulin4.jpg
で実行すると
0: 640x512 1 baskulin, 125.4ms
Speed: 7.7ms preprocess, 125.4ms inference, 7.6ms postprocess per image at shape (1, 3, 640, 512)
Results saved to runs/detect/predict4
バスクリン: 1個
File: runs/detect/predict4/image0.jpg
200
{"status":200,"message":"ok"}
となり画像つきメッセージが送信される
次は在庫の数が１以下のものをリストにして送信するようにする
import argparse
import json
import cv2
from ultralytics import YOLO
from collections import defaultdict
from line_notify import send_line_notify # インポートを追加
# コマンドライン引数の解析
parser = argparse.ArgumentParser(description="YOLOv8 Object Detection")
parser.add_argument('image_path', type=str, help='Path to the input image file')
args = parser.parse_args()
# ラベルマッピングファイルのパス
label_mapping_path = 'label_mapping.json'
# JSONファイルからクラスラベルのマッピングを読み込み
with open(label_mapping_path, 'r', encoding='utf-8') as f:
label_mapping = json.load(f)
# YOLOv8モデルのロード
model = YOLO('inventory_model/best.pt') # ここで適切なモデルを選択
# 画像のロード
image = cv2.imread(args.image_path)
# 画像の検出
results = model(image, save=True, conf=0.2, iou=0.5)
# 検出結果の取得
detections = results[0] # 最初の結果を取得
classes = detections.boxes.cls
# 検出物体のカウント
object_counts = defaultdict(int)
for cls in classes:
class_label = model.names[int(cls)]
if class_label in label_mapping:
label = label_mapping[class_label]
else:
label = class_label
object_counts[label] += 1
# 検出結果のフィルタリング（1以下のもの）
filtered_object_counts = {label: count for label, count in object_counts.items() if count <= 1}
# フィルタリングされた検出結果のメッセージ生成
message_lines = [f'{label}: {count}個' for label, count in filtered_object_counts.items()]
message = '\n'.join(message_lines)
# 検出結果の表示
for line in message_lines:
print(line)
# LINE Notifyにメッセージを送信（フィルタリングされた結果のみ）
if message:
send_line_notify(message)
else:
print("No objects with counts of 1 or less detected.")
これで今度は
python count_inventory_terminal.py data_bak/potato_starch1.jpg
として検出されない時には
0: 640x512 (no detections), 123.0ms Speed: 5.7ms preprocess, 123.0ms inference, 5.4ms postprocess per image at shape (1, 3, 640, 512) Results saved to runs/detect/predict6 No objects with counts of 1 or less detected.
となって
LINE送信はされなくなる
今回の画像はモデルの学習不足のためか
片栗粉の検出ができなかったので
それを認識できない場合のテストに使った
しかし、画像読み取りエラーなどを考慮し
今後何らかのアクションを取るようにした方が良いかもしれない
エラーログ以外のものを考えるようにする
また、送信するタイミングは、在庫数が１以下になった時に送るようにしました。
この場合、画像が検出できなかったりした時に判定ができないため
今後の課題とします
解決方法としては、検出結果をDBへ格納しておき
実行したタイムスタンプも記録、検出結果が０の時にはアラートを飛ばすなどがありそうです
とりあえず、ターミナル実行のみの状態なので
今後はどこから画像を撮ってくるのか、またwebカメラで行うのか、それとも
ラズパイゼロなどで撮影した画像を使うのか、それを考えてからまた改良していこうと思います
