顔ランドマークデータセットまとめ (AFLW, LFPW, COFW, 300-W, WFLW)

仕事で顔ランドマーク推定について調査・検証などを行い始めているのですが、各手法の学習・評価で使われているデータセットについてかなり混乱しました。

• 元々は何を目的として (= 評価したくて) 作られたデータセットなのか？
• ランドマークの点数や位置、遮蔽されている場合のランドマークの有無など、どうなっているのか？
• (略字に "W" が多くて、区別して覚えられない...)

そこで、頻出する "in the wild" 系のデータセットを5つ取り上げて目的・収集方法・データサイズ・ランドマーク数などの観点で整理したいと思います。これら5つのデータセットは、コントロールされた撮影環境ではなく、いずれもインターネットの画像をソースとしており、様々な条件・シチュエーションで撮影された写真を集めたデータセットとなっています。そのためランドマーク推定でも難しい部類になります。

• AFLW
• LFPW
• COFW
• 300-W
• WFLW

主にこちらを参考にしました。

arxiv.org

AFLW (Annotated Facial Landmarks in the Wild)

AFLWは2011年発表のデータセットで、Flickrの画像をアノテーションすることで正面以外の多様な顔向きを含めることを実現しています。顔ランドマーク以外にも、顔向き (yaw, roll, pitch) ・表情・性別・年齢の推定などにも使えるデータセットとなっています。

• 公式サイト, paper (PDF)¹
  • 非営利の研究目的にのみ利用可能 (ダウンロード方法等は公式サイト参照)
• アノテーション有り画像数: 25993枚
  • Flickrがソース
  • 内21997枚が実際の写真
  • 概ねカラー画像で、同一画像に複数人を含む場合もある
• ランドマーク数: 21点
  • 見切れなどによって見えない場合はアノテーションされない (21点未満の画像も含む)
  • 正面だけではなく、広いレンジの顔向きを含む

サイトより抜粋

LFPW (Labeled Face Parts in the Wild)

CVPR2011で発表されたデータセットです。visibilityもラベルとして付与されています。

• 公式サイト, paper (PDF)²
  • アノテーションデータは公式サイトからダウンロード可能 (画像はURLで外部参照しているが、リンク切れも多い)
• アノテーション有り画像数: 1432枚
  • GoogleやFlickr、Yahooなどの検索で得られた画像から、商用の顔検出システムで抽出 (横顔が検出されずに除外されたものが多い)
• ランドマーク数: 29点
  • ソーシャルワーキングサービス (Amazon Mechanical Turk) で同じ画像を最大3人がアノテーション
  • 各ランドマークのvisibilityは4パターンで付けられてます (2や3でもx, yが付与されている)
    • 0: Visible
    • 1: obscured by hair/glasses/etc.
    • 2: hidden because of viewing angle
    • 3: hidden because of image crop

paper fig.2抜粋

COFW (Caltech Occluded Faces in the Wild)

RCPRという手法 (公式サイト参照) とともに2013年に提案された顔ランドマーク推定タスク用のデータセットです。大きく上下左右を向いていたり、手やサングラスなどで遮蔽されている画像が多く含まれます。数は少ないですが質は良さそうなので、特定タスクの評価用として使いやすいかもです。

• 公式サイト, paper (PDF)³
  • 公式サイトからダウロード可能
• アノテーション有り画像数: 1007枚
  • 4人のCV専門家が複数のデータソースからサンプリング
• ランドマーク数: 29点
  • LFPWと同じ
  • 全体のランドマークの内23%が遮蔽されています

paper抜粋

300-W

ICCV2013と併せて開催されたベンチマークチャレンジで使われたデータセットです。複数のデータセットを、半教師あり学習で再アノテーションしたデータセットとなっています。

• 公式サイト, paper(PDF)⁴
  • 非商用の研究・学術用途のみ
• 学習セットは、4つのデータセット (LFPW, AFW, HELEN, XM2VTS) を半教師あり学習⁵で再アノテーション + IBUGのデータセット (135枚)
  • このIBUGの135枚を指して "300-W" と呼ばれることもあります
• テストセットは、新たに集められた600枚 (屋内300枚 + 屋外300枚)

paper table 1.抜粋

paper table.2抜粋

• ランドマーク数: 68点
  • Multi-PIEデータセットと揃えている

公式サイトFigure 1抜粋

WFLW (Wider Facial Landmarks in-the-wild)

CVPR2018にて発表されたデータセット。WIDER FACE⁶ (顔検出タスクのデータセット) をベースにアノテーションしてます。

• 公式サイト, paper (PDF)⁷
  • 画像・アノテーションの両方を公式サイトからダウンロード可能
• アノテーション有り画像数: 10000枚
  • ポーズ、表情、イルミネーション、メイク、遮蔽、ピンぼけなどのメタデータを含む
  • 場合分けして評価したい場合に使いやすい

公式サイト抜粋

• ランドマーク数: 98点
  • 遮蔽・見切れに対してもアノテーションされている

公式サイト抜粋

まとめ

今回は顔ランドマーク検出で用いられるデータセット (AFLW, LFPW, COFW, 300-W, WFLW) を5つを紹介しました。

---

1. Annotated Facial Landmarks in the Wild: A Large-scale, Real-world Database for Facial Landmark Localization. Martin Koestinger, Paul Wohlhart, Peter M. Roth and Horst Bischof. In Proc. First IEEE International Workshop on Benchmarking Facial Image Analysis Technologies, 2011↩

2. Localizing Parts of Faces Using a Consensus of Exemplars. Peter N. Belhumeur, David W. Jacobs, David J. Kriegman, Neeraj Kumar. Proceedings of the 24th IEEE Conference on Computer Vision and Pattern Recognition (CVPR), June 2011.↩

3. Robust face landmark estimation under occlusion. 
X. P. Burgos-Artizzu, P. Perona and P. Dollár. 
ICCV 2013, Sydney, Australia, December 2013.↩

4. Sagonas, C., Tzimiropoulos, G., Zafeiriou, S., Pantic, M.: 300 faces in-the-wild challenge: The first facial landmark localization challenge. In: IEEE International Conference on Computer Vision, 300 Faces in-the-Wild Challenge (300-W). Sydney, Australia (2013)↩

5. C. Sagonas, G. Tzimiropoulos, S. Zafeiriou, and M. Pantic. A semi-automatic methodology for facial landmark annotation. In Computer Vision and Pattern Recognition Workshops (CVPRW), 2013 IEEE Conference on, pages 896–903. IEEE, 2013.↩

6. S. Yang, P. Luo, C. C. Loy, and X. Tang. Wider face: A face detection benchmark. In CVPR, 2016.↩

7. Wu, Wayne and Qian, Chen and Yang, Shuo and Wang, Quan and Cai, Yici and Zhou, Qiang. Look at Boundary: A Boundary-Aware Face Alignment Algorithm. CVPR, 2018.↩

yacsで実験パラメータを設定する

今回はyacsを用いたパラメータ管理について整理します。

yacs

yacsはPythonコード + YAMLファイルで実験条件などのパラメータを管理できるようにするPythonライブラリです。

Pythonでyamlの設定ファイルを読み込むようなライブラリとしてはPyYAMLやruamel.yamlなどがよく使われます。これらのプリミティブにyamlを読み書きするライブラリと異なり、yacsは実験管理にフォーカスして抽象化されており、主にニューラルネットワークのハイパパラメータ設定などで用いられます。

• yacsはPyYAMLを使って実装されています
• ハイパパラメータには、バッチサイズや学習率といった学習時のパラメータに加えて、畳み込み層の入力サイズや深さなどのネットワークのパラメータも含みます

github.com

yacsを使った設定

yacsを使ってパラメータの設定を行っていきます。大まかな流れは4段階です。

1. デフォルトパラメータのロード
2. 実験ごとのパラメータで上書き
3. パラメータを変更できないように固める
4. パラメータにアクセス

詳細は後述しますが、こういった使い方になります。

from config import get_cfg_defaults

# デフォルトの設定をロード
cfg = get_cfg_defaults()
# 実験ごとのパラメータで上書き
cfg.merge_from_file("experiments/EXP_A.yaml")
# パラメータを変更できないように固める
cfg.freeze()
# パラメータにアクセス
print(cfg.OUTPUT_DIR)

ファイル構成はこんな感じです。

お約束のpip install yacsから始めます。

$ python --version
Python 3.7.4

$ pip install yacs

$ pip list | grep yacs
yacs       0.1.6 

デフォルトパラメータのロード

最初にデフォルト値を生成する処理をPythonで記述します。ファイル名にはconfig.pyやdefaults.pyが推奨されています (以下ではconfig.pyに実装してます) 。

キーとなるのはCfgNodeで、ここに設定値を持たせることができます。
またCfgNodeは入れ子にすることもできますので、階層構造 (MODELのINPUT_SIZE...など) を表現できます。

最後にデフォルト値を設定し終わったCfgNodeのルート (_C) をcloneメソッドで返す関数 get_cfg_defaults を用意しています。

from yacs.config import CfgNode as CN

_C = CN()

_C.OUTPUT_DIR = "output"
_C.LOG_DIR = "log"

_C.MODEL = CN()
_C.MODEL.NAME = "CNN"
_C.MODEL.INPUT_SIZE = [256, 256]

_C.TRAIN = CN()
_C.TRAIN.DEVICE = "cuda:0"
_C.TRAIN.BATCH_SIZE = 16
_C.TRAIN.LR = 0.0001

_C.TEST = CN()
_C.TEST.DEVICE = "cpu"
_C.TEST.BATCH_SIZE = 1

def get_cfg_defaults():
  return _C.clone()

このデフォルト設定を読み込む実装 (main.py) が以下です。

from config import get_cfg_defaults

cfg = get_cfg_defaults()

print(type(cfg))  # <class 'yacs.config.CfgNode'>
print(cfg)
# LOG_DIR: log
# MODEL:
#   INPUT_SIZE: [256, 256]
#   NAME: CNN
# OUTPUT_DIR: output
# TEST:
#   BATCH_SIZE: 1
#   DEVICE: cpu
# TRAIN:
#   BATCH_SIZE: 16
#   DEVICE: cuda:0
#   LR: 0.0001

### 実験ごとにチューニングされたパラメータで上書き 次にこのデフォルト設定を、実験ごとのパラメータで上書きします。この上書きパラメータはYAMLで定義します。ここでは experiments/EXP_A.yaml というファイルに以下のパラメータが記述されているものとします。

TEST:
  DEVICE: "cuda:0"
  BATCH_SIZE: 16

ロードしたCfgNodeオブジェクトのmerge_from_fileメソッドをコールすることで、引数に渡した設定ファイルで自身のパラメータが上書きされます。

• TESTのBATCH_SIZEとDEVICEが変更されていることに注目です

cfg.merge_from_file("experiments/EXP_A.yaml")

print(cfg)
# LOG_DIR: log
# MODEL:
#   INPUT_SIZE: [256, 256]
#   NAME: CNN
# OUTPUT_DIR: output
# TEST:
#   BATCH_SIZE: 16
#   DEVICE: cuda:0
# TRAIN:
#   BATCH_SIZE: 16
#   DEVICE: cuda:0
#   LR: 0.0001

ちなみにデフォルトに無いキーのパラメータを含む設定ファイルをマージしようとするとエラーになります。想定外のパラメータをアプリケーションでロードできないように保護しています。

NON_EXISTS: "hoge"

cfg.merge_from_file("experiments/EXP_B.yaml")
# -> KeyError: 'Non-existent config key: NON_EXISTS'

パラメータを変更できないように固める

最後にパラメータを変更できないようにfreezeメソッドでイミュータブルにします。freeze後にパラメータの値を変更しようとするとエラーになります。これによって設定ファイル以外からのパラメータの変更を阻止しています。

cfg.freeze()

cfg.LOG_DIR = "hoge"
# -> AttributeError: Attempted to set LOG_DIR to hoge, but CfgNode is immutable

パラメータにアクセス

ロードしたパラメータにアクセスする際は、アトリビュート形式でアクセスします。

print(cfg.OUTPUT_DIR)  # output
print(cfg.MODEL.INPUT_SIZE)  # [256, 256]

まとめ

今回はディープラーニングプロジェクトのパラメータ管理でよく用いられるyacsについて紹介しました。

実際の利用にあたっては デフォルトパラメータのロード -> 実験ごとのパラメータ (YAML) で上書き -> 固定 の手順を間違えやすそうですので、もう1層ラップした関数があると良いかと思います。

def load_config(config_path):
    cfg = get_cfg_defaults()
    cfg.merge_from_file(config_path)
    cfg.freeze()
    return cfg

cfg = load_config("experiments/EXP_A.yaml")

Python: motoでS3・DynamoDB・SQSのモックを作る

AWSで稼働するアプリケーションのユニットテストを作るときに厄介なのが、依存しているマネージドサービスのモックをどうやって整えるかです。無しというわけにはいかないですが、unittest.Mockなどで自作するのもかなり大変です。

こうしたモックを簡単に作れるようにアシストするのがmotoです。
今回は、Webやバッチなどのサービスアプリケーションから、AWSのAPIを通じてアクセスすることが多いS3・DynamoDB・SQSに絞ってmotoの使い方を紹介していきます。

moto

motoはAWSのマネージドサービスのモックを提供します。

github.com

汎用的なモックライブラリでは、テスト作成者が、テスト対象のモジュール内で呼び出しているboto3コードのクラスや返り値などを自前で書き換えることが多いかと思います。

motoでは同じboto3のインタフェースを使ってマネージドサービスをオブジェクトとして作成して、アノテーションを使ってテストコードに差し込むことで実現します。いわば仮想的なAWS環境を作成して、その上で稼働するアプリケーションとしてテストするのと同等になります。
これによって、テスト対象コードの内部実装への依存が切れます。つまり、テスト対象コードが呼び出すboto3のメソッドが変わっても、テストコードを変える必要がなくなります。モックのための実装も減り、テストコードも簡潔になります。 さらに、Flaskサーバとして独立させる機能 (Standalone Server Mode) も持っており、Pythonに限らず他の言語で書かれたアプリケーションのテストにも利用できます。

一方で、motoの欠点はサポートしているAWSマネージドサービスとAPIが限られていることです (以下で一覧化されています) 。また、マネージドサービスの作成をboto3のインタフェースで行う必要がありますので、テスト作成時にその仕様を調べる必要もあります。

https://github.com/spulec/moto/blob/master/IMPLEMENTATION_COVERAGE.md

motoを使ったユニットテスト

実際にmotoを使ったユニットテストを見ていきます。

いずれも大きく2ステップです。

1. モックするマネージドサービスをテストメソッド (またはクラス) にアノテーションする
2. テスト対象コードが依存しているマネージドサービスをboto3で作成する (必要に応じてデータも投入)
3. テストコード内でテスト対象メソッドを呼び出して出力などをassert

motoとboto3をインストールします。pytestはオプションです。

$ python --version
Python 3.7.4

$ pip install pytest boto3 moto

S3

最初にS3です。テスト対象となる関数 (main.py) を以下に示します。

• upload_to_bucket: moto-exampleバケットの"data/"配下に所定パスのファイルをアップロード
• download_from_bucket: moto-exampleバケットの"data/"配下のファイルを所定パスへダウンロード

import boto3
import datetime

def upload_to_bucket(file_path: str, file_name: str) -> bool:
    s3_client = boto3.client("s3")

    _ = s3_client.upload_file(file_path, "moto-example", "data/" + file_name)

    return True

def download_from_bucket(file_name: str, file_path: str) -> bool:
    s3_client = boto3.client("s3")

    _ = s3_client.download_file("moto-example", "data/" + file_name, file_path)

    return True

次にテストコード (test_main.py) です。

@mock_s3アノテーションに着目してください。これによってテストクラス内でのboto3を使ったS3アクセスは、実際のS3サービスにアクセスすることなく、モックオブジェクトがリクエスト・レスポンスを代わりに返します。このモッククラスはサービス (リソース) ごとに定義されています。
また、テストコード内でboto3を介してバケットを生成しています。モックされているのでテストに必要なデータもboto3オペレーションで完結できます。

• 例外時もboto3のインタフェース同様、botocoreの例外オブジェクト (ここではClientError) がスローされます

import pytest
import boto3
from botocore.exceptions import ClientError
from moto import mock_s3

# テスト対象となる関数
from .main import (
    upload_to_bucket, download_from_bucket,
)

@mock_s3
class TestS3Methods:
    bucket = "moto-example"

    def test_upload_succeed(self):
        # バケットの生成
        s3 = boto3.resource("s3")
        s3.create_bucket(Bucket=TestS3Methods.bucket)

        assert upload_to_bucket("./data/example.txt", "example.txt")

        # アップロードされたファイルをGet
        body = s3.Object(TestS3Methods.bucket, "data/example.txt").get()["Body"].read().decode("utf-8")

        assert body == "Hello, world!"

    def test_download_failed(self):
        s3 = boto3.resource("s3")
        s3.create_bucket(Bucket=TestS3Methods.bucket)

        # botocoreの例外クラスがスローされる
        with pytest.raises(ClientError):
            download_from_bucket("nonexist.txt", "output/example.txt")

例えば "data/" を忘れて間違ったパスへアップロードされるようになっていたとしても...

def upload_to_bucket(file_path: str, file_name: str) -> bool:
    s3_client = boto3.client("s3")

    # _ = s3_client.upload_file(file_path, "moto-example", "data/" + file_name)
   _ = s3_client.upload_file(file_path, "moto-example", file_name)

    return True

テストで気づくことができます。

FAILED test_main.py::TestS3Methods::test_upload_succeed - botocore.errorfactory.NoSuchKey: An error occurred (NoSuchKey) when calling the GetObject operation: The specified key does not exist.

DynamoDB

DynamoDBも2つの関数をテストします。moto-exampleテーブルはuser_idが数値型ハッシュキーとなります。

• put_to_dynamo: user_id, access_count, last_accessed_atのアイテムをmoto-exampleテーブルへputする
• get_from_dynamo: user_idをキーとしてmoto-exampleテーブルのアイテムをgetして辞書型で返す

import boto3
import datetime

def put_to_dynamo(user_id: int, access_count: int, last_accessed_at: datetime.datetime):
    dynamo_client = boto3.client("dynamodb")

    item = {
        "user_id": {"N": str(user_id)},
        "access_count": {"N": str(access_count)},
        "last_accessed_at": {"S": last_accessed_at.isoformat()},
    }

    dynamo_client.put_item(TableName="moto-example", Item=item)

def get_from_dynamo(user_id: int) -> dict:
    dynamo_client = boto3.client("dynamodb")

    key = {"user_id": {"N": str(user_id)}}

    item = dynamo_client.get_item(TableName="moto-example", Key=key)["Item"]

    return {
        "user_id": int(item["user_id"]["N"]),
        "access_count": int(item["access_count"]["N"]),
        "last_accessed_at": datetime.datetime.fromisoformat(item["last_accessed_at"]["S"]),
    }

テストコードは以下です。putとgetをまとめて1メソッドでテストしており (実用上はテストデータを作って1関数ずつテストするほうが望ましいです) 、setup_methodでテーブルを生成しています。

注意すべきはmock_dynamodb2を使っている点です。mock_dynmodbもありますが、最近のboto3では2の方を使うべきです (イシュー) 。

import datetime
import pytest
import boto3
from moto import mock_dynamodb2

# テスト対象
from .main import put_to_dynamo, get_from_dynamo

@mock_dynamodb2
class TestDynamoMethods:
    def setup_method(self, method):
        dynamodb = boto3.resource("dynamodb", region_name="ap-northeast-1")
        dynamodb.create_table(
            TableName="moto-example",
            KeySchema=[{"AttributeName": "user_id", "KeyType": "HASH"}],
            AttributeDefinitions=[{"AttributeName": "user_id", "AttributeType": "N"}],
        )

    def test_put_get_succeed(self):
        put_to_dynamo(user_id=33, access_count=10, last_accessed_at=datetime.datetime(2020, 3, 21, 10, 30, 15))

        item = get_from_dynamo(user_id=33)
        assert item["user_id"] == 33
        assert item["access_count"] == 10
        assert item["last_accessed_at"] == datetime.datetime(2020, 3, 21, 10, 30, 15)

SQS

最後にSQSです。

• send_to_sqs: SQSへ本文がbodyとなるメッセージを1通送信する
• receive_from_sqs: SQSからメッセージを1通受信して、本文とreceipt_handleのタプルを返す
  • メッセージが無い場合は、いずれの値もNone

import boto3

def send_to_sqs(queue_url: str, body: str):
    sqs_client = boto3.client("sqs")

    response = sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody=body
    )

    return response["MessageId"]

def receive_from_sqs(queue_url: str) -> (str, str):
    sqs_client = boto3.client("sqs")

    response = sqs_client.receive_message(QueueUrl=queue_url)

    if "Messages" not in response or len(response["Messages"]) == 0:
        return None, None

    message = response["Messages"][0]

    return message["Body"], message["ReceiptHandle"]

SQSではアカウントIDを含むURLを引数に持たせる必要があります。そのため、create_queueから返されるQueueUrlを保存し、テストしたい関数に渡しています。(通常、こうしたURLは引数で直接渡されるのではなく、設定ファイルや環境変数などでインスタンス生成時に渡ってくるものかと思います。)

import pytest
import boto3
from mock_sqs

# テスト対象
from .main import send_to_sqs, receive_from_sqs

@mock_sqs
class TestSqsMethods:
    def setup_method(self, method):
        sqs = boto3.client('sqs')
        response = sqs.create_queue(QueueName="moto-example")
        self.queue_url = response["QueueUrl"]

    def test_send_receive_succeed(self):
        assert send_to_sqs(queue_url=self.queue_url, body="Hello, world!")

        body, receipt_handle = receive_from_sqs(queue_url=self.queue_url)
        assert body == "Hello, world!"
        assert receipt_handle

    def test_receive_empty(self):
        body, receipt_handle = receive_from_sqs(queue_url=self.queue_url)
        assert body is None
        assert receipt_handle is None

まとめ

今回はmotoを使ってS3・DynamoDB・SQSに依存するアプリケーションのテストを実装してみました。

複雑になりがちなモックの作成を省力化できる点もそうですが、個人的にはテスト前にあるべき状態 (想定している状態) というのがboto3でコード化されているので、仕様の明瞭性が大きく上がるのも、motoの大きなメリットだと思います。積極的に使っていきましょう。