カテゴリー: LINE Messaging API

LINE notifyのモジュール化

LINE notifyのモジュール化

別のメソッドでも使えるようにモジュール化する
また

1
message = 'ファイルパス自動取得テスト'

の部分は
他のプログラムで
生成された文字列を受け取って実行するようにコードを変更する

1
vim line_notify.py

で

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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)

とりあえずこれを使えるかテストする

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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}個')

の中で呼び出すようにする

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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)

これを

1
python count_inventory_terminal.py data_bak/Baskulin4.jpg

で実行すると

1
2
3
4
5
6
7
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"}

となり画像つきメッセージが送信される

次は在庫の数が１以下のものをリストにして送信するようにする

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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.")

これで今度は

1
python count_inventory_terminal.py data_bak/potato_starch1.jpg

として検出されない時には

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

LINE Messasging API　でメッセージ送信

LINE Messasging API

LINE Messasging APIでメッセージ送信を試してみた

を参考に実践

なお後で調べたら
Messaging APIを始めよう

にドキュメントによれば
QRコードでのログインもできるらしい

まずスマホの LINEアプリで
メルアドとパスワードを設定

ログイン許可にチェックを入れる

https://developers.line.biz/console/
へアクセスしログイン

ユーザ名
メールアドレスを設定

I have read and agreed to the
LINE Developers Agreement
（規約に同意します）
にチェックをいれ

Create my account をクリック

これでアカウントの作成ができるので

次にプロバイダーの作成

下へスクロールし
Create a new provider をクリック

任意の名前を入力
ただしすでに他の人が使っていないものであること

Create をクリック

次にチャンネルの作成

使用するのは
Messaging API なので

Create a Messaging API channel
をクリック

Channel type
と
Provider
については
デフォルトのままでOK

Channel name には任意のチャンネル名

Channel description には
チャンネルの説明

Category には業種

Subcategory には業種の詳しいカテゴリ

Email address にはメルアド

I have read and agree to the LINE Official Account Terms of Use
LINE公式アカウント利用規約 の内容に同意します

I have read and agree to the LINE Official Account API Terms of Use
LINE公式アカウントAPI利用規約 の内容に同意します

にチェックをいれて

Create をクリック

これでチャンネルが作成されるので

Messaging API をクリックし
QRコードで自分のLINEアプリで友達登録しておく

次にメッセージ送信のテスト

アクセストークンが必要なので
Channel access token(long-lived) の
issue をクリック

これでアクセストークンが発行される

アクセストークンの種類については
チャネルアクセストークン

の公式ドキュメントを参考に

次にIDの確認
これは自分のIDを使うので
Basic settings タブをクリック

Your user ID の部分を確認

ここまでできたら
curl コマンドでメッセージ送信

1
2
3
4
5
6
7
8
9
10
11
12
curl -v -X POST https://api.line.me/v2/bot/message/push \
-H 'Content-Type:application/json' \
-H 'Authorization: Bearer { アクセストークン }' \
-d '{
    "to": "ユーザID",
    "messages":[
        {
            "type":"text",
            "text":"Hello, world"
        }
    ]
}'

これを実行すると

LINEアプリに
hello world
と書かれたメッセージが届く

なおデフォルトでは自動応答メッセージがあり
これを無効にするには
Messaging API をクリック

Auto-reply messages の Edit をクリック

応答メッセージをオフに設定

これで自動応答メッセージがでなくなる

今後の課題としては
LINEトークルームを作成
Google Assistant SDK をラズパイに入れる
google-home-notifier　設定
とやっていけば

【ベビテクDIY】LINEやスマートスピーカーでらくらく家事育児効率化 その２

のように
LINEのやり取りを音声のみでできるいう予定