LINE notifyのモジュール化

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カメラで行うのか、それとも
ラズパイゼロなどで撮影した画像を使うのか、それを考えてからまた改良していこうと思います