たけのこの里が好きな G くんのために、きのこの山を分別する装置を作ってあげた。

git clone https://github.com/aws-samples/aws-ml-jp

# 撮影した画像のディレクトリ指定
TRAIN_RAWIMAGE_DIR = './train_raw_images/'
# S3 にアップロードする先の Prefix の指定
BASE_PREFIX = 'takenoko_kinoko_gt'
# ユニークなラベリングジョブ名の生成
GT_JOB_NAME = f'{BASE_PREFIX}-{uuid.uuid4()}'.replace('_','-') 
# 撮影した画像を S3 にアップロードし、その URI を取得する
rawimage_s3_uri = sagemaker.session.Session().upload_data(TRAIN_RAWIMAGE_DIR,key_prefix=BASE_PREFIX)
# ラベリングジョブ名とラベリング画像を配置した URI 、ラベリングに使う IAM ロールを取得
print(f'GroundTruth job name : {GT_JOB_NAME}')
print(f'GroundTruth Target : {rawimage_s3_uri}/')
print(f'GroundTruth Role : {sagemaker.get_execution_role()}')

# クロップした画像にきのこの山やたけのこの里が映っている場合、
# クロップした後のきのこの山やたけのこの里が1/4以下かどうかを判定するヘルパー関数

def fix_bbox(l,t,r,b,w,h):
    # 判定結果、NG なら False にする
    judge = True
    # ラベリング結果のクロップ補正後の値が負の値ならば 0 に、イメージサイズより大きければイメージサイズに補正する
    fix_left = 0 if l < 0 else l
    fix_top = 0 if t < 0 else t
    fix_right = w if r > w else r
    fix_bottom = h if b > h else b
    # 領域外ならラベリング無しとする
    if l > w or t > h or r < 0 or b <0:
        judge=False
    # 基の面積の1/4以下ならアノテーション無しとする
    elif (r-l)*(b-t)/4 > (fix_right-fix_left)*(fix_bottom-fix_top):
        judge=False
    
    return judge,(fix_left,fix_top,fix_right,fix_bottom)
    
# ラベリング結果をテキストとして読み込む
with open('manifest/output.manifest','r') as f:
    manifest_line_list = f.readlines()

# クロップした結果のきのこの山やたけのこの里の位置情報を格納する辞書 
annotation_dict = {
    'images':[],
    'annotations':[]
}

# クロップサイズの定数
IMAGE_SIZE_TUPLE=(512,512)

# クロップした画像のファイル名に使う一意なシーケンス番号
IMAGE_ID = 0

# クロップした画像の保存先
OUTPUT_DIR = './train_random_crop_images/'

# (re-run用の削除コマンド)
!rm -rf {OUTPUT_DIR}*.png

# ラベリング結果の行数分ループする
# ラベリング結果は 1 行につき 1 画像格納される
for manifest_line in manifest_line_list:
    # 画像のラベリング結果の読み込み
    manifest_dict = json.loads(manifest_line)
    # 画像のファイル名取得(ラベリング結果に格納されている)
    filename = manifest_dict['source-ref'].split('/')[-1]
    # 元画像のサイズを取得(ラベリング結果に格納されている)
    image_size_tuple=(manifest_dict['kinoko-takenoko-aug']['image_size'][0]['width'],manifest_dict['kinoko-takenoko-aug']['image_size'][0]['height'])
    # PIL で画像を開く
    raw_img = Image.open(os.path.join(TRAIN_RAWIMAGE_DIR,filename))
    # 20 回クロップする
    for i in range(20):
        # ループするかどうかのフラグ(画像にきのこの山やたけのこの里が 2 枚未満だったらクロップをやりなおし)
        loop = True
        while loop:
            # クロップを行う左上の座標を設定
            rand_x = np.random.randint(0,image_size_tuple[0]-IMAGE_SIZE_TUPLE[0])
            rand_y = np.random.randint(0,image_size_tuple[1]-IMAGE_SIZE_TUPLE[1])
            # クロップする
            crop_img = raw_img.crop((
                rand_x,
                rand_y,
                rand_x + IMAGE_SIZE_TUPLE[0],
                rand_y + IMAGE_SIZE_TUPLE[1]
            ))
            # クロップ後のきのこの山やたけのこの里の位置を格納するリスト
            annotation_list = []
            # 元画像のラベリング結果をループ
            for annotation in manifest_dict['kinoko-takenoko-aug']['annotations']:
                # クロップした後のきのこの山やたけのこの里の座標に補正
                left = annotation['left'] - rand_x
                top = annotation['top'] - rand_y
                right = annotation['left'] + annotation['width'] - rand_x
                bottom = annotation['top'] + annotation['height'] - rand_y
                # きのこの山やたけのこの里があるかどうかを判定
                judge,(left,top,right,bottom) = fix_bbox(left,top,right,bottom,IMAGE_SIZE_TUPLE[0],IMAGE_SIZE_TUPLE[1])
                if judge:
                    # きのこの山やたけのこの里があったら位置とラベルを追加
                    annotation_list.append(
                        {
                            'bbox':[left,top,right,bottom],
                            'category_id':annotation['class_id']
                        }
                    )
            # きのこの山やたけのこの里と数が2未満だったらクロップやり直し
            if len(annotation_list) > 1:
                loop = False
        
        # クロップしたら画像を保存する
        save_file_name = f'{str(IMAGE_ID).zfill(5)}_{str(i).zfill(5)}_{filename}'.replace('jpg','png')
        crop_img.save(os.path.join(OUTPUT_DIR,save_file_name))
        
        # 補正済ラベリング結果を出力用辞書に格納
        annotation_dict['images'].append(
            {
                'file_name' : save_file_name,
                'height' : IMAGE_SIZE_TUPLE[1],
                'width' : IMAGE_SIZE_TUPLE[0],
                'id' : IMAGE_ID
            }
        )
        for annotation in annotation_list:                  
            annotation_dict['annotations'].append(
                {
                    'image_id': IMAGE_ID,
                    'bbox':annotation['bbox'],
                    'category_id':annotation['category_id']
                }
            )
        IMAGE_ID += 1

# ランダムクロップ補正後のラベリング結果を出力
with open('annotations.json','wt') as f:
    f.write(json.dumps(annotation_dict))       

# 出力したディレクトリを prefix として使う
prefix = OUTPUT_DIR[2:-1]
# re-run 用の削除コマンド
!aws s3 rm s3://{bucket}/{prefix} --recursive
# ランダムクロップした画像をアップロード    
image_s3_uri = sagemaker.session.Session().upload_data(OUTPUT_DIR,key_prefix=f'{prefix}/images')
# ラベリング結果をアップロード
annotatione_s3_uri = sagemaker.session.Session().upload_data('./annotations.json',key_prefix=prefix)
# Fine-Tune で使う URI を出力
paste_str = image_s3_uri.replace('/images','')
print(f"paste string to S3 bucket address：{paste_str}")

er-runtime クライアントの生成
smr_client = boto3.client('sagemaker-runtime')
# エンドポイントの名前
ENDPOINT_NAME='jumpstart-ftc-kinoko-detection-endpoint'
# 推論する画像の場所
TEST_IMAGE_FILE = 'test_raw_images/lattice.jpg'

# 推論対象の画像を開いて変数に格納
with open(TEST_IMAGE_FILE, 'rb') as f:
    img_bin = f.read()

# 推論を実行
response = smr_client.invoke_endpoint(EndpointName=ENDPOINT_NAME, ContentType='application/x-image', Body=img_bin)

# 推論結果を読み込む
model_predictions = json.loads(response['Body'].read())

# 結果を可視化
# テスト画像を PIL を通して numpy array として開く
image_np = np.array(Image.open(TEST_IMAGE_FILE))
# matplotlibで描画する
fig = plt.figure(figsize=(20,20))
ax = plt.axes()
ax.imshow(image_np)
# 推論結果を変数に展開
bboxes, classes, confidences = model_predictions['normalized_boxes'], model_predictions['classes'], model_predictions['scores']
# 物体検出結果を検出した分だけループする
for idx in range(len(bboxes)):
    # 信頼度スコアが 0.5 以上のみ可視化する
    if confidences[idx]>0.5:
        # 検出した座標（左上を(0,0),右下を(1,1)とした相対座標)を取得
        left, bot, right, top = bboxes[idx]
        # 相対座標を絶対座標に変換する
        x, w = [val * image_np.shape[1] for val in [left, right - left]]
        y, h = [val * image_np.shape[0] for val in [bot, top - bot]]
        # 検出した物体の ID を take/kino に読み替える
        class_name = 'take' if int(classes[idx])==0 else 'kino'
        # take/kinoに対して矩形で描画するための色を設定する
        color = 'blue' if class_name == 'take' else 'red'
        # matplotlib に検出した物体に矩形を描画する
        rect = patches.Rectangle((x, y), w, h, linewidth=3, edgecolor=color, facecolor='none')
        ax.add_patch(rect)
        # 左上に検出結果と信頼度スコアを描画する
        ax.text(x, y, "{} {:.0f}%".format(class_name, confidences[idx]*100), bbox=dict(facecolor='white', alpha=0.5))

# 画像を切り出すためのコード
# 開始地点設定
x ,y = 0,512
# 切り出すサイズ設定
CROP_SIZE=(512,512)
# 切り出す対象の画像を PIL で開く
img = Image.open('./test_raw_images/takenoko.jpg')
# 切り出した画像を保存するディレクトリ
CROP_DIR = './test_crop_images/'
# re-run 用の削除コマンド
!rm -rf {CROP_DIR}/*.png
# 1pxずらしてループ
for i in range(img.size[0]-CROP_SIZE[0]):
    # 画像の切り出し
    crop_img = img.crop((i,y,i+CROP_SIZE[0],y+CROP_SIZE[1]))
    # 切り出した画像を保存
    file_name = f'{CROP_DIR}{str(i).zfill(5)}.png'
    crop_img.save(file_name)

# 検出結果を保存するディレクトリを設定
DETECT_DIR='./test_detect_images/'
# re-run 用の削除コマンド
!rm -rf {DETECT_DIR}/*.png

# 切り出した画像分だけループ
for img_file_path in sorted(glob(f'{CROP_DIR}*.png')):
    # 切り出した画像を開く
    with open(img_file_path,'rb') as f:
        img_bin = f.read()
    # 推論エンドポイントに画像を投げる
    response = smr_client.invoke_endpoint(EndpointName=ENDPOINT_NAME, ContentType='application/x-image', Body=img_bin)
    # 推論結果を読み込む
    pred=json.loads(response['Body'].read())
    # 推論結果を展開
    bboxes, classes, confidences = pred['normalized_boxes'], pred['classes'], pred['scores']
    # 切り出した画像を PIL で開く
    img = Image.open(img_file_path)
    # 矩形やテキストを描くために draw インスタンスを生成
    draw = ImageDraw.Draw(img)
    # 検出したkino/take分ループ
    for i in range(len(bboxes)):
        # 信頼度スコアが0.8以上のみ描画する
        if confidences[i]>0.8:
            # 矩形の相対座標を取得
            left, top, right, bottom = bboxes[i]
            # 矩形の相対座標を絶対座標に変換
            left = img.size[0] * left
            top = img.size[1] * top
            right = img.size[0] * right
            bottom = img.size[1] * bottom
            # 検出した物体の ID を take/kino に読み替える
            text = 'take' if int(classes[i])==0 else 'kino'
            # take/kinoに対して矩形で描画するための色を設定する
            color = 'blue' if text == 'take' else 'red'
            # 矩形の左上に表示する文字の大きさを設定、きのこの山なら大きくする
            TEXTSIZE=14 if classes[i]=='0' else 18
            # 矩形の先の太さを設定、きのこの山なら太くする
            LINEWIDTH=4 if classes[i]=='0' else 6
            # 矩形を描画する
            draw.rectangle([(left,top),(right,bottom)], outline=color, width=LINEWIDTH)
            # 矩形の左上に描画する信頼度スコアの取得
            text += f' {str(round(confidences[i],3))}'
            # テキストを描画する場所を取得
            txpos = (left, top-TEXTSIZE-LINEWIDTH//2)
            # フォントの設定
            font = ImageFont.truetype("/usr/share/fonts/truetype/noto/NotoMono-Regular.ttf", size=TEXTSIZE)
            # 描画するテキストのサイズを取得
            txw, txh = draw.textsize(text, font=font)
            # テキストの背景用の矩形を描画
            draw.rectangle([txpos, (left+txw, top)], outline=color, fill=color, width=LINEWIDTH)
            # テキストを描画
            draw.text(txpos, text, fill='white',font=font)
    # 画像をファイルに書き出す
    img.save(img_file_path.replace(CROP_DIR,DETECT_DIR))

# 書き出した画像を連結して動画にする    
fourcc = cv2.VideoWriter_fourcc('m', 'p', '4', 'v')
video = cv2.VideoWriter('./video.mp4',fourcc, 120.0, CROP_SIZE)
for img_file_path in sorted(glob(f'{DETECT_DIR}*.png')):
    img = cv2.imread(img_file_path)
    video.write(img)
video.release()

import boto3
sm_client = boto3.client('sagemaker')
# Endpoint の詳細から EndpointConfigName を取得
endpoint_config_name = sm_client.describe_endpoint(EndpointName='jumpstart-ftc-kinoko-detection-endpoint')['EndpointConfigName']
# EndpointConfig の詳細から ModelName を取得
model_name = sm_client.describe_endpoint_config(EndpointConfigName=endpoint_config_name)['ProductionVariants'][0]['ModelName']
# Model の詳細を取得
model_detail = sm_client.describe_model(ModelName=model_name)
# Model の詳細からクラウド推論に使っているコンテナイメージを取得
model_container_image = model_detail['PrimaryContainer']['Image']
# Model の詳細からクラウド推論に使っているモデルの URI を取得
model_s3_uri = model_detail['PrimaryContainer']['ModelDataUrl']
# コンテナイメージと URI を取得
print(model_container_image)
print(model_s3_uri)

763104351884.dkr.ecr.{REGION}.amazonaws.com/mxnet-inference:1.7.0-cpu-py3
s3://sagemaker-{REGION}-{ACCOUNT_ID}/mxnet-od-ssd-512-mobilenet1-0-coco-YYYYMMDD-HHMISS/model.tar.gz

# re-run 用の削除コマンド
!rm -rf model/
# modelダウンロードディレクトリ作成
!mkdir -p model/
# モデルを S3 からダウンロード
!aws s3 cp {model_s3_uri} ./model/
# モデルディレクトリへ移動
%cd model/
# 解凍
!tar zxvf model.tar.gz

!ls -l
total 432644
drwxr-xr-x 4 root root      6144 Feb  2 05:38 code/
-rw-r--r-- 1 root root  46233667 Jan 31 08:26 finetuned_model.params
-rw-r--r-- 1 root root        18 Jan 31 08:26 labels_info.json
-rw-r--r-- 1 root root 304524052 Feb  2 03:17 model.tar.gz
-rw-r--r-- 1 root root        62 Jan 31 08:26 model_info.json
-rw-r--r-- 1 root root         0 Jan 31 08:26 ssd_512_mobilenet1.0_coco-da9756fa.lock
-rw-r--r-- 1 root root  92247566 Jan 31 08:26 ssd_512_mobilenet1.0_coco-da9756fa.params

%cd code
!ls -l
total 2124
-rw-rw-r-- 1 root root       0 Jan 22 01:04 __init__.py
drwxr-xr-x 2 root root    6144 Jan 22 02:18 constants/
-rw-rw-r-- 1 root root    9240 Jan 22 01:04 inference.py
drwxr-xr-x 3 root root    6144 Jan 22 02:18 lib/
-rw-rw-r-- 1 root root       0 Jan 22 01:04 requirements.txt
-rw-rw-r-- 1 root root       5 Jan 22 01:04 version

# 依存しているライブラリをインストール
!pip install sagemaker-inference

import json, inference # 推論コードを読み込む
from PIL import Image, ImageDraw, ImageFont

# モデルのロード
model = inference.model_fn('../')
# 推論に使う画像ファイル
IMAGE_FILE = '../../test_raw_images/lattice.jpg'
# 画像ファイルをバイナリとして開く
with open(IMAGE_FILE,'rb') as f:
    img_bin = f.read()
# 推論
pred = json.loads(inference.transform_fn(model,img_bin,'application/x-image','application/json'))
# 推論結果の展開
bboxes, classes, confidences = pred['normalized_boxes'], pred['classes'], pred['scores']
# 推論結果の描画
img = Image.open(IMAGE_FILE)
draw = ImageDraw.Draw(img)
for i in range(len(bboxes)):
    # 信頼度スコアが0.8以上のみ描画する
    if confidences[i]>0.8:
        # 矩形の相対座標を取得
        left, top, right, bottom = bboxes[i]
        # 矩形の相対座標を絶対座標に変換
        left = img.size[0] * left
        top = img.size[1] * top
        right = img.size[0] * right
        bottom = img.size[1] * bottom
        # 検出した物体の ID を take/kino に読み替える
        text = 'take' if int(classes[i])==0 else 'kino'
        # take/kinoに対して矩形で描画するための色を設定する
        color = 'blue' if text == 'take' else 'red'
        # 矩形の左上に表示する文字の大きさを設定、きのこの山なら大きくする
        TEXTSIZE=14 if classes[i]=='0' else 18
        # 矩形の先の太さを設定、きのこの山なら太くする
        LINEWIDTH=4 if classes[i]=='0' else 6
        # 矩形を描画する
        draw.rectangle([(left,top),(right,bottom)], outline=color, width=LINEWIDTH)
        # 矩形の左上に描画する信頼度スコアの取得
        text += f' {str(round(confidences[i],3))}'
        # テキストを描画する場所を取得
        txpos = (left, top-TEXTSIZE-LINEWIDTH//2)
        # フォントの設定
        font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSerif.ttf", size=TEXTSIZE)
        # 描画するテキストのサイズを取得
        txw, txh = draw.textsize(text, font=font)
        # テキストの背景用の矩形を描画
        draw.rectangle([txpos, (left+txw, top)], outline=color, fill=color, width=LINEWIDTH)
        # テキストを描画
        draw.text(txpos, text, fill='white',font=font)
# 画像をファイルに書き出す
img.save('pred_lattice.png')
# jupyter に画像を表示
img