概要

海外ではエンジニアの面接においてコーディングテストというものが行われるらしく、多くの場合、特定の関数やクラスをお題に沿って実装するという物がメインである。

その対策としてLeetCodeなるサイトで対策を行うようだ。

早い話が本場でも行われているようなコーディングテストに耐えうるようなアルゴリズム力を鍛えるサイト。

せっかくだし人並みのアルゴリズム力くらいは持っておいた方がいいだろうということで不定期に問題を解いてその時に考えたやり方をメモ的に書いていこうかと思います。

Leetcode

ゼロから始めるLeetCode 目次

前回
ゼロから始めるLeetCode Day29「46. Permutations」

基本的にeasyのacceptanceが高い順から解いていこうかと思います。

Twitterやってます。

問題

234. Palindrome Linked List

単方向の連結リストが与えられるので、回文であるかをチェックしてください。

Example 1:

Input: 1->2
Output: false

Example 2:

Input: 1->2->2->1
Output: true

解法

最も簡単な方法としては、与えられた連結リストをひっくり返したものと一致すればTrueを、一致しなければFalseを返すというものです。

# Definition for singly-linked list.
# class ListNode:
#     def __init__(self, val=0, next=None):
#         self.val = val
#         self.next = next
class Solution:
    def isPalindrome(self, head: ListNode) -> bool:
        value = []
        while head:
            value.append(head.val)
            head = head.next
        return value == value[::-1]
# Runtime: 92 ms, faster than 15.28% of Python3 online submissions for Palindrome Linked List.
# Memory Usage: 24.1 MB, less than 11.54% of Python3 online submissions for Palindrome Linked List.

今回は少し時間がなかったため、他の人の解答をdiscussから持ってきます。

例えばO(1)の解答とかですね。

# Definition for singly-linked list.
# class ListNode:
#     def __init__(self, val=0, next=None):
#         self.val = val
#         self.next = next
class Solution:
    def isPalindrome(self, head):
        rev = None
        fast = head
        while fast and fast.next:
            fast = fast.next.next
            rev, rev.next, head = head, rev, head.next
        tail = head.next if fast else head
        isPali = True
        while rev:
            isPali = isPali and rev.val == tail.val
            head, head.next, rev = rev, head, rev.next
            tail = tail.next
        return isPali
# Runtime: 92 ms, faster than 15.28% of Python3 online submissions for Palindrome Linked List.
# Memory Usage: 24.1 MB, less than 11.54% of Python3 online submissions for Palindrome Linked List.

https://leetcode.com/problems/palindrome-linked-list/discuss/64500/11-lines-12-with-restore-O(n)-time-O(1)-space

わかりやすい解説としてはこのdiscussの内容がおすすめです。
視覚的に分かりやすいのすごい・・・

https://leetcode.com/problems/palindrome-linked-list/discuss/324358/O(n)-time-and-O(1)-space-with-explanation-(Python-and-C)

後、この本の連結リストの章にも複数の解法が載っていました。

世界で闘うプログラミング力を鍛える150問

時間がある人はチェックしてみると良いかもしれません。

以下、"AWS SDK for Python" に掲載されているAPIの中から、私的によく使う、boto3（Pythonライブラリー）関連のAPIをいくつか示します。
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/index.html

リソース名や各種設定値等は適宜読み替えてください~

Athena

# -*- coding: utf-8 -*-
import boto3
import time
athena = boto3.client("athena")

class QueryFailed(Exception):
    """
    Athenaのクエリ実行に失敗したときに呼ばれる例外クラス
    """
    pass

#クエリ実行（非同期処理）
start_query_response = athena.start_query_execution(
    QueryString = 'クエリ',
    QueryExecutionContext={
        "Database": "クエリするのに使うGlueDB名"
    },
    ResultConfiguration={
        "OutputLocation" : "s3://出力バケット名/キー"
    }
)
query_execution_id = start_query_response["QueryExecutionId"]
#クエリの実行状況を確認する
while True:
    query_status = athena.get_query_execution(
        QueryExecutionId = query_execution_id
    )
    query_execution_status = query_status['QueryExecution']['Status']['State']
    if query_execution_status == 'SUCCEEDED':
        break
    elif query_execution_status == "FAILED" or query_execution_status == "CANCELLED":
        raise QueryFailed(f"query_execution_status = {query_execution_status}")
        time.sleep(10)
#クエリ実行結果取得（成功した場合のみ）
query_results = athena.get_query_results(QueryExecutionId = query_execution_id)

DynamoDB

# -*- coding: utf-8 -*-
import boto3
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("table")

#項目１件取得
response = table.get_item(
    Key = {
        "id" : "1"
    }
)
#項目１件登録
response = table.put_item(
    Item = {
        "id" : "1",
        "key" : "value"
    }
)
#項目１件更新
response = table.update_item(
    Key = {
        "id" : "1"
    },
    UpdateExpression = "set key2 = :val",
    ExpressionAttributeValues = {
        ":val" : "value2"
    },
    ReturnValues = "UPDATED_NEW"
)
#項目１件削除
response = table.delete_item(
    Key = {
        "id" : "1"
    }
)


#全項目（全レコード）削除   truncate  or delete from table
# データ全件取得
delete_items = []
parameters   = {}
while True:
    response = table.scan(**parameters)
    delete_items.extend(response["Items"])
    if "LastEvaluatedKey" in response:
        parameters["ExclusiveStartKey"] = response["LastEvaluatedKey"]
    else:
        break
# キー抽出
key_names = [ x["AttributeName"] for x in table.key_schema ]
delete_keys = [ { k:v for k,v in x.items() if k in key_names } for x in delete_items ]
# データ削除
with table.batch_writer() as batch:
    for key in delete_keys:
        batch.delete_item(Key = key)



#テーブル削除
dynamodb_client = boto3.client('dynamodb')
autoscaling_client = boto3.client('application-autoscaling')
response = dynamodb_client.list_tables()
if 'TableNames' in response:
    for table_name in response['TableNames']:
        if table_name == "削除対象のテーブル名":
            dynamodb_client.delete_table(TableName = table_name)
            waiter = dynamodb_client.get_waiter("table_not_exists")
            waiter.wait(TableName = table_name)
            #Target TrackingタイプのAutoScalingをつけている場合、ScalingPolicyを削除することで、CloudWatchAlarmを同時に削除することができる。
            try:
                autoscaling_client.delete_scaling_policy(
                    PolicyName = f'{table_name}ReadCapacity',
                    ServiceNamespace = "dynamodb",
                    ResourceId = f"table/{table_name}",
                    ScalableDimension = "dynamodb:table:ReadCapacityUnits"
                )
            except autoscaling_client.exceptions.ObjectNotFoundException as e:
                print(f"type = {type(e)}, message = {e}")
            try:
                autoscaling_client.delete_scaling_policy(
                    PolicyName = f'{table_name}WriteCapacity',
                    ServiceNamespace = "dynamodb",
                    ResourceId = f"table/{table_name}",
                    ScalableDimension = "dynamodb:table:WriteCapacityUnits"
                )
            except autoscaling_client.exceptions.ObjectNotFoundException as e:
                print(f"type = {type(e)}, message = {e}")



#テーブル作成
table_name = "table"
dynamodb.create_table(
    TableName = table_name,
    KeySchema = [{
        "AttributeName" : "id",
        "KeyType" : "HASH"
    }],
    AttributeDefinitions = [{
        "AttributeName" : "id",
        "AttributeType" : "S"
    }],
    ProvisionedThroughput = {
        "ReadCapacityUnits" : 1,
        "WriteCapacityUnits" : 1
    }
)
waiter = dynamodb_client.get_waiter("table_exists")
waiter.wait(TableName = table_name)
# Target TrackingタイプのAutoScalingの設定
autoscaling_client.register_scalable_target(
    ServiceNamespace = "dynamodb",
    ResourceId = f"table/{table_name}",
    ScalableDimension = "dynamodb:table:ReadCapacityUnits",
    MinCapacity = 1,
    MaxCapacity = 10,
    RoleARN = "AutoScalingするためのIAM Role ARN"
)
autoscaling_client.register_scalable_target(
    ServiceNamespace = "dynamodb",
    ResourceId = f"table/{table_name}",
    ScalableDimension = "dynamodb:table:WriteCapacityUnits",
    MinCapacity = 1,
    MaxCapacity = 10,
    RoleARN = "AutoScalingするためのIAM Role ARN"
)
# スケーリングポリシーを設定
autoscaling_client.put_scaling_policy(
    ServiceNamespace='dynamodb',
    ResourceId = f"table/{table_name}",
    PolicyType = "TargetTrackingScaling",
    PolicyName = f"{table_name}ReadCapacity",
    ScalableDimension = "dynamodb:table:ReadCapacityUnits",
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue" : 70,
        "PredefinedMetricSpecification": {
            "PredefinedMetricType": "DynamoDBReadCapacityUtilization"
        },
       "ScaleOutCooldown" : 70,
       "ScaleInCooldown" : 70
   }
)
autoscaling_client.put_scaling_policy(
    ServiceNamespace='dynamodb',
    ResourceId = f"table/{table_name}",
    PolicyType = "TargetTrackingScaling",
    PolicyName = f"{table_name}WriteCapacity",
    ScalableDimension='dynamodb:table:WriteCapacityUnits',
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue" : 70,
        "PredefinedMetricSpecification": {
            "PredefinedMetricType": "DynamoDBWriteCapacityUtilization"
        },
       "ScaleOutCooldown" : 70,
       "ScaleInCooldown" : 70
   }
)



#テーブルスキーマ更新
response = dynamodb_client.update_table(
    AttributeDefinitions = [
        {
            'AttributeName': 'string',
            'AttributeType': 'S'|'N'|'B'
        },
    ],
    TableName = 'string',
    BillingMode = 'PROVISIONED'|'PAY_PER_REQUEST',
    ProvisionedThroughput = {
        'ReadCapacityUnits': 123,
        'WriteCapacityUnits': 123
    }
)


#バッチ処理
table = dynamodb.Table("table")
with table.batch_writer() as batch:
    for i in range(10 ** 6):
        batch.put_item(
            Item = {
                "id" : str(i + 1),
                "key" : f"key{i + 1}"
            }
        )

Lambda

# -*- coding: utf-8 -*-
import boto3
lambda = boto3.client('lambda ')

response = lambda.invoke(
    FunctionName = '処理対象のLambda',
    InvocationType = 'Event'|'RequestResponse'|'DryRun',
    LogType = 'None'|'Tail',
    Payload = b'bytes'|file
)

SageMaker

# -*- coding: utf-8 -*-
import boto3
sagemaker = boto3.client("sagemaker-runtime")

# sagemakerのエンドポイントにアクセスし予測結果を受け取る
response = sagemaker.invoke_endpoint(
    EndpointName = "SageMaker エンドポイント名",
    Body=b'bytes'|file,
    ContentType = 'text/csv', #The MIME type of the input data in the request body.
    Accept = 'application/json' #The desired MIME type of the inference in the response.
)

S3

# -*- coding: utf-8 -*-
import boto3
s3 = boto3.client('s3')
BUCKET_NAME = "処理対象のバケット名"

#１オブジェクト（ファイル）書き込み
s3.put_object(
    Bucket = BUCKET_NAME,
    Body = "データ内容。str型 or bytes型",
    Key = "S3のキー。s3:://バケット名/  以降のディレクトリおよびファイル名"
)
#１オブジェクト（ファイル）読み込み
s3.get_object(
    Bucket = BUCKET_NAME,
    Key = "S3のキー。s3:://バケット名/  以降のディレクトリおよびファイル名"
)
#１オブジェクト（ファイル）削除
s3.delete_object(
    Bucket = BUCKET_NAME,
    Key = "S3のキー。s3:://バケット名/  以降のディレクトリおよびファイル名"
)
#１オブジェクト（ファイル）を別の場所へコピー
s3.copy_object(
    Bucket = BUCKET_NAME,
    Key = "移行先S3キー。s3:://バケット名/  以降のディレクトリおよびファイル名",
    CopySource = {
        "Bucket" : "移行元バケット名",
        "Key" : "移行元S3キー"
    }
)


#指定プレフィックス以下 or バケット内全オブジェクト（ファイル）取得
BUCKET_NAME= "処理対象のバケット名"
contents = []
kwargs = {
    "Bucket" : BUCKET_NAME,
    "Prefix" : "検索対象のプレフィックス"
} 
while True:
    response = s3.list_objects_v2(**kwargs)
    if "Contents" in response:
        contents.extend(response["Contents"])
        if 'NextContinuationToken' in response:
            kwargs["ContinuationToken"] = response['NextContinuationToken']
            continue
        break
    else:
        break

SQS

# -*- coding: utf-8 -*-
import boto3
sqs = boto3.client('sqs')
QUEUE_URL= "SQSのキューURL。コンソールで確認できます。"

#SQSのキューからメッセージを全部取得。
while True:
    sqs_message = sqs.receive_message(
        QueueUrl = QUEUE_URL,
        MaxNumberOfMessages = 10
    )
    if "Messages" in sqs_message:
        for message in sqs_message["Messages"]:
            try:
                print(message)
                #取得できたメッセージは削除する。そうしないと重複取得される可能性あり
                sqs.delete_message(
                    QueueUrl = QUEUE_URL,
                    ReceiptHandle = message["ReceiptHandle"]
                )
            except Exception as e:
                print(f"type = {type(e)} , message = {e}")

以下、"AWS SDK for Python" に掲載されているAPIの中から、私的によく使う、boto3（Pythonライブラリー）関連のAPIをいくつか示します。
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/index.html

リソース名や各種設定値等は適宜読み替えてください~

Athena

# -*- coding: utf-8 -*-
import boto3
import time
athena = boto3.client("athena")

class QueryFailed(Exception):
    """
    Athenaのクエリ実行に失敗したときに呼ばれる例外クラス
    """
    pass

#クエリ実行（非同期処理）
start_query_response = athena.start_query_execution(
    QueryString = 'クエリ',
    QueryExecutionContext={
        "Database": "クエリするのに使うGlueDB名"
    },
    ResultConfiguration={
        "OutputLocation" : "s3://出力バケット名/キー"
    }
)
query_execution_id = start_query_response["QueryExecutionId"]
#クエリの実行状況を確認する
while True:
    query_status = athena.get_query_execution(
        QueryExecutionId = query_execution_id
    )
    query_execution_status = query_status['QueryExecution']['Status']['State']
    if query_execution_status == 'SUCCEEDED':
        break
    elif query_execution_status == "FAILED" or query_execution_status == "CANCELLED":
        raise QueryFailed(f"query_execution_status = {query_execution_status}")
        time.sleep(10)
#クエリ実行結果取得（成功した場合のみ）
query_results = athena.get_query_results(QueryExecutionId = query_execution_id)

DynamoDB

# -*- coding: utf-8 -*-
import boto3
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("table")
dynamodb_client = boto3.client('dynamodb')
autoscaling_client = boto3.client('application-autoscaling')

#項目１件取得
response = table.get_item(
    Key = {
        "id" : "1"
    }
)
#項目１件登録
response = table.put_item(
    Item = {
        "id" : "1",
        "key" : "value"
    }
)
#項目１件更新
response = table.update_item(
    Key = {
        "id" : "1"
    },
    UpdateExpression = "set key2 = :val",
    ExpressionAttributeValues = {
        ":val" : "value2"
    },
    ReturnValues = "UPDATED_NEW"
)
#項目１件削除
response = table.delete_item(
    Key = {
        "id" : "1"
    }
)


#全項目（全レコード）削除   truncate  or delete from table
# データ全件取得
delete_items = []
parameters   = {}
while True:
    response = table.scan(**parameters)
    delete_items.extend(response["Items"])
    if "LastEvaluatedKey" in response:
        parameters["ExclusiveStartKey"] = response["LastEvaluatedKey"]
    else:
        break
# キー抽出
key_names = [ x["AttributeName"] for x in table.key_schema ]
delete_keys = [ { k:v for k,v in x.items() if k in key_names } for x in delete_items ]
# データ削除
with table.batch_writer() as batch:
    for key in delete_keys:
        batch.delete_item(Key = key)



#テーブル削除
response = dynamodb_client.list_tables()
if 'TableNames' in response:
    for table_name in response['TableNames']:
        if table_name == "削除対象のテーブル名":
            dynamodb_client.delete_table(TableName = table_name)
            waiter = dynamodb_client.get_waiter("table_not_exists")
            waiter.wait(TableName = table_name)
            #Target TrackingタイプのAutoScalingをつけている場合、ScalingPolicyを削除することで、CloudWatchAlarmを同時に削除することができる。
            try:
                autoscaling_client.delete_scaling_policy(
                    PolicyName = f'{table_name}ReadCapacity',
                    ServiceNamespace = "dynamodb",
                    ResourceId = f"table/{table_name}",
                    ScalableDimension = "dynamodb:table:ReadCapacityUnits"
                )
            except autoscaling_client.exceptions.ObjectNotFoundException as e:
                print(f"type = {type(e)}, message = {e}")
            try:
                autoscaling_client.delete_scaling_policy(
                    PolicyName = f'{table_name}WriteCapacity',
                    ServiceNamespace = "dynamodb",
                    ResourceId = f"table/{table_name}",
                    ScalableDimension = "dynamodb:table:WriteCapacityUnits"
                )
            except autoscaling_client.exceptions.ObjectNotFoundException as e:
                print(f"type = {type(e)}, message = {e}")



#テーブル作成
table_name = "table"
dynamodb.create_table(
    TableName = table_name,
    KeySchema = [{
        "AttributeName" : "id",
        "KeyType" : "HASH"
    }],
    AttributeDefinitions = [{
        "AttributeName" : "id",
        "AttributeType" : "S"
    }],
    ProvisionedThroughput = {
        "ReadCapacityUnits" : 1,
        "WriteCapacityUnits" : 1
    }
)
waiter = dynamodb_client.get_waiter("table_exists")
waiter.wait(TableName = table_name)
# Target TrackingタイプのAutoScalingの設定
autoscaling_client.register_scalable_target(
    ServiceNamespace = "dynamodb",
    ResourceId = f"table/{table_name}",
    ScalableDimension = "dynamodb:table:ReadCapacityUnits",
    MinCapacity = 1,
    MaxCapacity = 10,
    RoleARN = "AutoScalingするためのIAM Role ARN"
)
autoscaling_client.register_scalable_target(
    ServiceNamespace = "dynamodb",
    ResourceId = f"table/{table_name}",
    ScalableDimension = "dynamodb:table:WriteCapacityUnits",
    MinCapacity = 1,
    MaxCapacity = 10,
    RoleARN = "AutoScalingするためのIAM Role ARN"
)
# スケーリングポリシーを設定
autoscaling_client.put_scaling_policy(
    ServiceNamespace='dynamodb',
    ResourceId = f"table/{table_name}",
    PolicyType = "TargetTrackingScaling",
    PolicyName = f"{table_name}ReadCapacity",
    ScalableDimension = "dynamodb:table:ReadCapacityUnits",
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue" : 70,
        "PredefinedMetricSpecification": {
            "PredefinedMetricType": "DynamoDBReadCapacityUtilization"
        },
       "ScaleOutCooldown" : 70,
       "ScaleInCooldown" : 70
   }
)
autoscaling_client.put_scaling_policy(
    ServiceNamespace='dynamodb',
    ResourceId = f"table/{table_name}",
    PolicyType = "TargetTrackingScaling",
    PolicyName = f"{table_name}WriteCapacity",
    ScalableDimension='dynamodb:table:WriteCapacityUnits',
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue" : 70,
        "PredefinedMetricSpecification": {
            "PredefinedMetricType": "DynamoDBWriteCapacityUtilization"
        },
       "ScaleOutCooldown" : 70,
       "ScaleInCooldown" : 70
   }
)



#テーブルスキーマ更新
response = dynamodb_client.update_table(
    AttributeDefinitions = [
        {
            'AttributeName': 'string',
            'AttributeType': 'S'|'N'|'B'
        },
    ],
    TableName = 'string',
    BillingMode = 'PROVISIONED'|'PAY_PER_REQUEST',
    ProvisionedThroughput = {
        'ReadCapacityUnits': 123,
        'WriteCapacityUnits': 123
    }
)


#バッチ処理
table = dynamodb.Table("table")
with table.batch_writer() as batch:
    for i in range(10 ** 6):
        batch.put_item(
            Item = {
                "id" : str(i + 1),
                "key" : f"key{i + 1}"
            }
        )

Lambda

# -*- coding: utf-8 -*-
import boto3
lambda = boto3.client('lambda ')

response = lambda.invoke(
    FunctionName = '処理対象のLambda',
    InvocationType = 'Event'|'RequestResponse'|'DryRun',
    LogType = 'None'|'Tail',
    Payload = b'bytes'|file
)

SageMaker

# -*- coding: utf-8 -*-
import boto3
sagemaker = boto3.client("sagemaker-runtime")

# sagemakerのエンドポイントにアクセスし予測結果を受け取る
response = sagemaker.invoke_endpoint(
    EndpointName = "SageMaker エンドポイント名",
    Body=b'bytes'|file,
    ContentType = 'text/csv', #The MIME type of the input data in the request body.
    Accept = 'application/json' #The desired MIME type of the inference in the response.
)

SQS

# -*- coding: utf-8 -*-
import boto3
sqs = boto3.client('sqs')
QUEUE_URL= "SQSのキューURL。コンソールで確認できます。"

#SQSのキューからメッセージを全部取得。
while True:
    sqs_message = sqs.receive_message(
        QueueUrl = QUEUE_URL,
        MaxNumberOfMessages = 10
    )
    if "Messages" in sqs_message:
        for message in sqs_message["Messages"]:
            try:
                print(message)
                #取得できたメッセージは削除する。そうしないと重複取得される可能性あり
                sqs.delete_message(
                    QueueUrl = QUEUE_URL,
                    ReceiptHandle = message["ReceiptHandle"]
                )
            except Exception as e:
                print(f"type = {type(e)} , message = {e}")

S3

# -*- coding: utf-8 -*-
import boto3
s3 = boto3.client('s3')
BUCKET_NAME = "処理対象のバケット名"

#１オブジェクト（ファイル）書き込み
s3.put_object(
    Bucket = BUCKET_NAME,
    Body = "データ内容。str型 or bytes型",
    Key = "S3のキー。s3:://バケット名/  以降のディレクトリおよびファイル名"
)
#１オブジェクト（ファイル）読み込み
s3.get_object(
    Bucket = BUCKET_NAME,
    Key = "S3のキー。s3:://バケット名/  以降のディレクトリおよびファイル名"
)
#１オブジェクト（ファイル）削除
s3.delete_object(
    Bucket = BUCKET_NAME,
    Key = "S3のキー。s3:://バケット名/  以降のディレクトリおよびファイル名"
)
#１オブジェクト（ファイル）を別の場所へコピー
s3.copy_object(
    Bucket = BUCKET_NAME,
    Key = "移行先S3キー。s3:://バケット名/  以降のディレクトリおよびファイル名",
    CopySource = {
        "Bucket" : "移行元バケット名",
        "Key" : "移行元S3キー"
    }
)


#指定プレフィックス以下 or バケット内全オブジェクト（ファイル）取得
BUCKET_NAME= "処理対象のバケット名"
contents = []
kwargs = {
    "Bucket" : BUCKET_NAME,
    "Prefix" : "検索対象のプレフィックス"
} 
while True:
    response = s3.list_objects_v2(**kwargs)
    if "Contents" in response:
        contents.extend(response["Contents"])
        if 'NextContinuationToken' in response:
            kwargs["ContinuationToken"] = response['NextContinuationToken']
            continue
    break

自然言語処理における深層学習

自然言語処理において深層学習は下記のような様々なタスクで使われています。

文章を異なる言語に翻訳する機械翻訳
重要な情報だけを抽出する自動要約
文書と質問を元に回答する機械読解
画像に関する質問に応答するシステム
etc

これらはいずれも世界中の研究者・有名IT企業が精力的に取り組んでいる分野です。
これらに用いられる最新の手法にはほぼ必ずと言っていいほど深層学習が用いられています。
よって、自然言語処理関連の研究・ビジネスをしようとしている人にとって
深層学習・ニューラルネットワークを学ぶことはほぼ必須と言えるでしょう。

では、なぜ自然言語処理で深層学習がこれほど使われているのか
単語をコンピュータで扱うためにはどうしても数値に変換する必要があります。
その古典的な方法として、

One hot vector
TFIDF

などが挙げられます。

実際これらは手軽に行えるので手始めに自然言語処理で何かしたい時には最適な手法でしょう。

しかし、これらのベクトルには

①スパース性
ベクトルの次元が数万〜数十万になり、しばしばメモリが足りなくなる。

②意味を扱えない
全ての単語が出現頻度という特徴でしか差別化されず、個性が失われる。

などの問題があります。

そこでニューラルネットワークのモデルを使うと
誤差逆伝播法によって単語のベクトルを学習できるため

Embedding(各単語にわずか数百次元のベクトルを割り当てること)だけでよく

さらに文脈を考慮しながら単語のベクトルを学習できるので
単語の意味が似ているものは似たようなベクトルが得られるなど
TFIDFなどに比べて単語の意味を扱えると言った利点があります。

Embedding

Embeddingとは日本語で「埋め込み」という意味です。

深層学習による自然言語処理では、単語という記号を
d次元ベクトル（dは100〜300程度）に埋め込む、Embeddingという処理を行います。

単語を扱うニューラルネットワークのモデルを構築する際、一番始めに行うのがこのEmbeddingです。

kerasではEmbedding Layerというものが用意されており、以下のように使えます。

from keras.models import Sequential
from keras.layers.embeddings import Embedding

vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length = 20 # 文の長さ

model = Sequential()
model.add(Embedding(input_dim=vocab_size, output_dim=embedding_dim, input_length=seq_length))

ここで必ず指定しなければならない値が、input_dimとoutput_dim、input_lengthです。

input_dim: 語彙数、単語の種類の数
output_dim: 単語ベクトルの次元
input_length: 各文の長さ

全単語の単語ベクトルを結合した

Embedding Matrixと呼ばれるものの例を図に示しました。

前提として各単語に特有のIDを割り振り
そのIDがEmbedding Matrixの何行目になるのかを指します。
（ID=0が0行目、ID=iがi行目です。）

そして各単語のベクトルがEmbedding Matrixの行に相当します。
Embedding Matrixの横の長さdは単語ベクトルの次元に相当します。

各セルの値はkerasが自動でランダムな値を格納してくれます。
また、図のように

多くの場合0行目にはunk、すなわちUnknown（未知語）を割り当てます。

unkを使う理由は、扱う語彙を限定してその他のマイナーな単語は全て
Unknownとすることでメモリを節約するためです。
語彙の制限の仕方は
扱うコーパス（文書）に出現する頻度が一定以上のものだけを扱うなどが一般的です。

Embeddingへの入力は単語に割り当てたIDからなる行列で
サイズは（バッチサイズ、文の長さ）である必要があります。

ここでバッチサイズとは、一度に並列して計算を行うデータ（文の）数を表します。

文の長さは全てのデータで統一する必要があります。
ここで問題になるのが、文の長さは一定ではないということです。
そこで恣意的に文の長さをDとし、

長さD以下の文は長さがDになるよう0を追加する
長さD以上の文は長さがDになるよう末尾から単語を削る

これを一般的にpaddingと呼びます

使用例はこちらです。

import numpy as np
from keras.models import Sequential
from keras.layers.embeddings import Embedding


batch_size = 32 # バッチサイズ
vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length = 20 # 文の長さ

# 本来は単語をIDに変換する必要がありますが、今回は簡単に入力データを準備しました。
input_data = np.arange(batch_size * seq_length).reshape(batch_size, seq_length)

# modelにEmbeddingを追加してください。
model = Sequential()
model.add(Embedding(input_dim=vocab_size, output_dim=embedding_dim, input_length=seq_length))

# input_dataのshapeがどのように変わるのか確認してください。
output = model.predict(input_data)
print(output.shape)

RNN

この節ではRNNについて学びます。

RNNとはRecurrent Neural Networkの略称で
日本語で「再帰ニューラルネット」などと訳されます。

可変長系列、すなわち任意の長さの入力列を扱うことに優れており
自然言語処理において頻繁に使われる重要な機構です。

言語以外でも、時系列データであれば使えるので、音声認識など活用の幅は広いです。

Recurrentは「繰り返しの」という意味があります。

つまりRNNとは「繰り返しの」ニューラルネットワークという意味です。

もっとも簡単なRNNは以下の式で表されます。

ここでtをよく時刻などと表現します。
xtは時刻tの入力、htは時刻tの隠れ状態ベクトル、ytは時刻tの出力で
これら3つは全てベクトルです。

h0には零ベクトルを使うことが多いです。

f,gは活性化関数でシグモイド関数などが使われます。

Wh,Wy,Uとbh,byはそれぞれ学習可能な行列とバイアスです。

簡単な図解をすると図のようになります。

RNNには上で紹介した以外の亜種が多くありますが
基本の構造は図のようになります。

また、kerasを使うときにこれらの厳密な定義を覚えておく必要はありません。
入力列を順番に入力していき、各時刻で隠れ状態ベクトルと出力が得られるということを覚えておきます。

またRNNにも他のニューラルネットと同様に複数の層を重ねることができます。

LSTM

LSTMとは

LSTMとはRNNの一種で、 LSTMはRNNに特有な欠点を補う機構を持っています。

そのRNN特有の欠点について説明します。
RNNは時間方向に深いニューラルネットですので、初期の方に入力した値を考慮して
パラメータを学習させるのが難しい、すなわち長期記憶(long-term memory)が苦手だと言われています。

感覚的に言うと、RNNは初めの方に入力された要素を「忘れて」しまうのです。

これを防ぐための機構として有名なのがLSTMです。

LSTMとはLong Short-Term Memoryの略称で
その名の通り長期記憶と短期記憶を可能にするものです。

世界中の研究者に幅広く使われています。

LSTMはRNNに「ゲート」という概念を導入したもので
ゲート付きRNNの一種です。 このゲートによって長期記憶と短期記憶が可能になります。

LSTMの概要は図の通りです。

LSTMの内部は以下の式で記述されます。 （覚える必要はありません。）

※ ⊙は要素ごとの掛け算、アダマール積を表します

ここでiを入力ゲート、fを忘却ゲート、oを出力ゲートと呼びます。

これらのゲートの活性化関数にsigmoid関数が使われていることに注意してください。 sigmoid関数とは

のことです。よってこれらのゲートの出力は0から1の間の数なのです。

入力ゲートのiは¯htに要素ごとにかけられており
時刻tにおける入力からの情報をどれだけ伝えるかを調整していることが分かります。
ゲートとはその名の通り「門」のイメージからきています。

忘却ゲートのfはct−1に要素ごとにかけられており
時刻t-1までの情報を時刻tにどれだけ伝えるか（＝過去の情報をどれだけ忘れるか）を調整しています。

そして出力ゲートのoは時刻1からtまでの情報を含むctから
どれだけの情報を隠れ状態ベクトルht出力として使うかを調整しています。

これらが短期記憶と長期記憶を実現させるために考案されたLSTMの全容です。

LSTMの実装

早速kerasでLSTMを実装していきます。
kerasにはLSTMを簡単に使うことができるモジュールが用意されており
説明したような数式を意識することなくLSTMを使うことができます。

具体的には以下のように使います。

from keras.layers.recurrent import LSTM

units = 200

lstm = LSTM(units, return_sequences=True)

ここでunitsとはLSTMの隠れ状態ベクトルの次元数です。
大抵100から300程度の値にします。

一般に学習すべきパラメータの数は多いほど複雑な現象をモデル化できますが
その分学習させるのが大変です（消費メモリが増える、学習時間が長い）。

学習環境に合わせて調節すると良いでしょう。

そして今回は

return_sequencesという引数を指定します。

これはLSTMの出力の形式をどのようにするかを決めるための引数で、

return_sequencesがTrueなら
LSTMは全ての入力系列に対応する出力系列（隠れ状態ベクトル h1h1 〜 hThT ）を出力します。

return_sequencesがFalseなら
LSTMは最後の時刻Tにおける隠れ状態ベクトルのみを出力します。

後の章で全ての出力系列を使うことになるので、ここではreturn_sequencesをTrueにしておきます。

モデルの構築方法はEmbeddingと繋げると以下のようです。

from keras.models import Sequential
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM

vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length = 20 # 文の長さ
lstm_units = 200 # LSTMの隠れ状態ベクトルの次元数

model = Sequential()
model.add(Embedding(input_dim=vocab_size, output_dim=embedding_dim))
model.add(LSTM(lstm_units, return_sequences=True))

使用例はこちら

import numpy as np
from keras.models import Sequential
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM


batch_size = 32 # バッチサイズ
vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length = 20 # 文の長さ
lstm_units = 200 # LSTMの隠れ状態ベクトルの次元数

# 今回も簡単に入力データを準備しました。
input_data = np.arange(batch_size * seq_length).reshape(batch_size, seq_length)

# modelにLSTMを追加
model = Sequential()
model.add(Embedding(input_dim=vocab_size, output_dim=embedding_dim, input_length=seq_length))
model.add(LSTM(lstm_units,return_sequences=True))

# input_dataのshapeがどのように変わるのか確認してください。
output = model.predict(input_data)
print(output.shape)

BiLSTM

これまで、LSTM含めRNNに入力系列 x={x1,...,xT}をx1から
xTにかけて先頭から順に入力していました。

これまでとは逆にxTからx1にかけて後ろから順に入力して行くこともできます。

さらに2つの入力させる向きを組み合わせた

双方向再帰ニューラルネット（bi-directional recurrent neural network）がよく用いられます。

これの利点は、各時刻において先頭から伝播してきた情報と後ろから伝播してきた情報
すなわち入力系列全体の情報を考慮できることです。

そして2方向のLSTMを繋げたものを

Bidirectional LSTM、通称BiLSTMと言います。

概要は図の通りです。

2つの向きのRNNを繋げる方法はいくつかあるのですが、説明するより先に実装する方法を紹介します。

from keras.layers.recurrent import LSTM
from keras.layers.wrappers import Bidirectional

bilstm = Bidirectional(LSTM(units, return_sequences=True), merge_mode='sum')

このようにLSTMを引数にすることで簡単に実装できます。

もう一つの引数であるmerge_modeが2方向のLSTMをどう繋げるかを指定するためのもので
基本的に{'sum', 'mul', 'concat', 'ave'}の中から選びます。

sum: 要素和
mul: 要素積
concat: 結合
ave: 平均
None: 結合せずにlistを返す

例はこちら

import numpy as np
from keras.models import Sequential
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM
from keras.layers.wrappers import Bidirectional


batch_size = 32 # バッチサイズ
vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length = 20 # 文の長さ
lstm_units = 200 # LSTMの隠れ状態ベクトルの次元数

# 今回も簡単に入力データを準備しました。
input_data = np.arange(batch_size * seq_length).reshape(batch_size, seq_length)

# modelにBiLSTMを追加。
model = Sequential()
model.add(Embedding(input_dim=vocab_size, output_dim=embedding_dim, input_length=seq_length))
model.add(Bidirectional(LSTM(lstm_units, return_sequences=True), merge_mode='concat'))

# input_dataのshapeがどのように変わるのか確認してください。
output = model.predict(input_data)
print(output.shape)

Softmax関数

一旦話は逸れますが、ここではSoftmax関数の扱いに慣れてもらいます。

Softmax関数は活性化関数の一種で
クラス分類を行う際にニューラルネットの一番出力に近い層で使われます。

Softmax関数の定義は、入力列 x={x1,...,xn}（各要素は実数）を

をi番目の要素に持つような出力列 y={y1,...,yn}に変換することで、
この定義から分かるように出力列yは

を満たします。
これは入力列が実数値の場合でした。

実際にkerasで実装するときは

from keras.layers.core import Activation

# xのサイズ: [バッチサイズ、クラス数]

y = Activation('softmax')(x)

# sum(y[0]) = 1, sum(y[1]) = 1, ...

のように、バッチごとにsoftmax関数を適用して使います。

これはActivation('softmax')のデフォルトの設定が
入力xのサイズの最後の要素、クラス数の軸方向にsoftmaxを作用させるというものだからです。

つまり、xのサイズが[バッチサイズ、d、クラス数]のように3次元であっても
Activation('softmax')を適用可能です。

余談ですが、keras.models.Sequentialを使わずにこのようにモデルを記述する方法を

Functional APIと言います。

使用例はこちら

import numpy as np
from keras.layers import Input
from keras.layers.core import Activation
from keras.models import Model


x = Input(shape=(20, 5))
# xにsoftmaxを作用
y = Activation('softmax')(x)

model = Model(inputs=x, outputs=y)

sample_input = np.ones((12, 20, 5))
sample_output = model.predict(sample_input)

print(np.sum(sample_output, axis=2))

Attention

Attentionとは

今2つの文章 s={s1,...,sN},t={t1,...,tL} があるとします。

ここでは仮にsを質問文とし、tをそれに対する回答文の候補だとします。

この時、機械に自動でtがsに対する回答文として妥当かどうか判断させるにはどのようにしたら良いでしょうか？

tだけをいくら眺めても、tが妥当かどうかは分かりません。
sの情報を参照しつつ、tが妥当かどうか判断する必要があります。

そこでAttention Mechanism（注意機構）という機構が役に立ちます。

これまでの章で文をRNNによって隠れ状態ベクトルに変換できることを学びました。

具体的には2つの別々のRNNを用意し、一方のRNNによって
sを隠れ状態ベクトル

に変換し、もう一方のRNNによってtを隠れ状態ベクトル

に変換することができます。

そこでsの情報を考慮してtの情報を使うために、以下のようにtの各時刻において
sの各時刻の隠れ状態ベクトルを考慮した特徴を計算します。

このように参照元の系列sのどこに”注意”するかを対象の系列tの各時刻において計算することで
臨機応変に参照元の系列の情報を考慮しながら対象の系列の情報を抽出することができます。

概要は図の通りです。

図には単方向のRNNの場合を示しましたが、双方向のRNNであってもAttentionは適用可能です。
また
参照元と対象のRNNの最大時刻（隠れ状態ベクトルの総数）が異なる場合でもAttentionを適用可能です。

このAttentionという機構は深層学習による自然言語処理では当たり前のように使われる重要な技術で、

機械翻訳や、自動要約、対話の論文で頻繁に登場します。

歴史的には機械翻訳に初めて使われて以来その有用性が広く認知されるようになりました。

また、今回はsの隠れ状態ベクトルの重み付き平均を用いる

Soft Attentionを紹介しましたが

ランダムに1つの隠れ状態ベクトルを選択する

Hard Attentionも存在します。

さらにそこから派生して、画像認識の分野でも使われることがあります。

中でもGoogleが発表した

Attention is all you needという論文はとても有名です。

Attentionの実装

kerasでAttentionを実装するためには

Mergeレイヤーを使う必要があります。

kerasのバージョン2.0以降では前の章まで使っていたSequential ModelをMergeすることができないので、

ここではkerasのFunctional APIを使います。Functional APIの簡単な使い方は以下のようです。

Sequential Modelではただ新しいLayerをaddしていくだけでしたが、

Functional APIを使うことでもっと自在に複雑なモデルを組むことができます。

from keras.layers import Input, Dense
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM
from keras.layers.wrappers import Bidirectional
from keras.layers.merge import dot, concatenate
from keras.layers.core import Activation
from keras.models import Model

batch_size = 32 # バッチサイズ
vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length1 = 20 # 文1の長さ
seq_length2 = 30 # 文2の長さ
lstm_units = 200 # LSTMの隠れ状態ベクトルの次元数
hidden_dim = 200 # 最終出力のベクトルの次元数

input1 = Input(shape=(seq_length1,))
embed1 = Embedding(input_dim=vocab_size, output_dim=embedding_dim, input_length=seq_length1)(input1)
bilstm1 = Bidirectional(LSTM(lstm_units, return_sequences=True), merge_mode='concat')(embed1)

input2 = Input(shape=(seq_length2,))
embed2 = Embedding(input_dim=vocab_size, output_dim=embedding_dim, input_length=seq_length2)(input2)
bilstm2 = Bidirectional(LSTM(lstm_units, return_sequences=True), merge_mode='concat')(embed2)

# 要素ごとの積を計算する
product = dot([bilstm2, bilstm1], axes=2) # productのサイズ：[バッチサイズ、文2の長さ、文1の長さ]

a = Activation('softmax')(product)
c = dot([a, bilstm1], axes=[2, 1])
c_bilstm2 = concatenate([c, bilstm2], axis=2)
h = Dense(hidden_dim, activation='tanh')(c_bilstm2)

model = Model(inputs=[input1, input2], outputs=h)

このように各Layerを関数のように使うのでFunctional APIと呼ばれています。

また新しく登場したInputレイヤーで指定するshapeにはbatchサイズを入れないよう注意しましょう。

そしてModelを定義するときは引数にinputsとoutputsを指定する必要がありますが、

入力や出力が複数ある場合はリストに入れて渡せば大丈夫です。

そして新しく登場したdot([u, v], axes=2)は、uとvのバッチごとの行列積を計算します。

指定したaxesにおける次元数はuとvで等しくなければなりません。

また、dot([u, v], axes=[1,2])とするとuの1次元とvの2次元
のように別々の次元を指定することもできます。

それでは以下の数式を元にFunctional APIを使ってAttentionを実装してみます。

使用例はこちら

import numpy as np
from keras.layers import Input, Dense
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM
from keras.layers.wrappers import Bidirectional
from keras.layers.merge import dot, concatenate
from keras.layers.core import Activation
from keras.models import Model

batch_size = 32 # バッチサイズ
vocab_size = 1000 # 扱う語彙の数
embedding_dim = 100 # 単語ベクトルの次元
seq_length1 = 20 # 文1の長さ
seq_length2 = 30 # 文2の長さ
lstm_units = 200 # LSTMの隠れ状態ベクトルの次元数
hidden_dim = 200 # 最終出力のベクトルの次元数

# 2つのLSTMに共通のEmbeddingLayerを使うため、はじめにEmbeddingLayerを定義します。
embedding = Embedding(input_dim=vocab_size, output_dim=embedding_dim)

input1 = Input(shape=(seq_length1,))
embed1 = embedding(input1)
bilstm1 = Bidirectional(LSTM(lstm_units, return_sequences=True), merge_mode='concat')(embed1)

input2 = Input(shape=(seq_length2,))
embed2 = embedding(input2)
bilstm2 = Bidirectional(LSTM(lstm_units, return_sequences=True), merge_mode='concat')(embed2)

# 要素ごとの積を計算する
product = dot([bilstm2, bilstm1], axes=2) # サイズ：[バッチサイズ、文2の長さ、文1の長さ]

# ここにAttention mechanismを実装
a = Activation('softmax')(product)
c = dot([a, bilstm1], axes=[2, 1])
c_bilstm2 = concatenate([c, bilstm2], axis=2)
h = Dense(hidden_dim, activation='tanh')(c_bilstm2)


model = Model(inputs=[input1, input2], outputs=h)

sample_input1 = np.arange(batch_size * seq_length1).reshape(batch_size, seq_length1)
sample_input2 = np.arange(batch_size * seq_length2).reshape(batch_size, seq_length2)

sample_output = model.predict([sample_input1, sample_input2])
print(sample_output.shape)

Dropout

Dropoutとは訓練時に変数の一部をランダムに0に設定することによって
汎化性能を上げ、過学習を防ぐための手法です。

①過学習とは

ニューラルネットなどのモデルで教師あり学習を行う場合
しばしば訓練データに適合しすぎて
評価データでのパフォーマンスが訓練データに比べて著しく下がる「過学習」を起こしてしまいます。

②汎化性能とは

訓練データで過学習することなく、訓練データと評価データに関わらず
一般的に高いパフォーマンスができることを「汎化性能」が高いと言います。

実際に使うときは、変数のうち0に設定する割合を0から1の間の値で設定して
Dropoutレイヤーを追加します。

# Sequentialモデルを使う場合
from keras.models import Sequential
from keras.layers import Dropout


model = Sequential()

...

model.add(Dropout(0.3))

# Functional APIを使う場合
from keras.layers import Dropout

y = Dropout(0.3)(x)

使用例はこちら

import numpy as np
from keras.layers import Input, Dropout
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM
from keras.layers.wrappers import Bidirectional
from keras.models import Model

batch_size = 32  # バッチサイズ
vocab_size = 1000  # 扱う語彙の数
embedding_dim = 100  # 単語ベクトルの次元
seq_length = 20  # 文1の長さ
lstm_units = 200  # LSTMの隠れ状態ベクトルの次元数

input = Input(shape=(seq_length,))

embed = Embedding(input_dim=vocab_size, output_dim=embedding_dim,
                  input_length=seq_length)(input)

bilstm = Bidirectional(
    LSTM(lstm_units, return_sequences=True), merge_mode='concat')(embed)

output = Dropout(0.3)(bilstm)

model = Model(inputs=input, outputs=output)

sample_input = np.arange(
    batch_size * seq_length).reshape(batch_size, seq_length)

sample_output = model.predict(sample_input)

print(sample_output.shape)

はじめに

巷で流行っている自然言語100本ノックにPythonで挑戦してみます。
解き方はいろいろあると思うので、いろいろな解き方を示したいと思います。
全ての解き方を網羅することは当然ながらできません。
「こんな解き方があるよ」や「ここ間違ってるよ」という点があれば、ぜひコメントでお教えいただければと思います。

言語処理100本ノック 2020とは

言語処理100本ノック 2020は、東北大学の乾・鈴木研究室のプログラミング基礎研勉強会で使われている教材です。
詳しくは、言語処理100本ノックについてを参照して下さい。

Qiitaの記事へのリンク

	1問目	2問目	3問目	4問目	5問目	6問目	7問目	8問目	9問目	10問目
第1章	00	01	02	03	04	05	06	07	08	09
第2章	10	11	12	13	14	15	16	17	18	19
第3章	20	21	22	23	24	25	26	27	28	29
第4章	30	31	32	33	34	35	36	37	38	39
第5章	40	41	42	43	44	45	46	47	48	49
第6章	50	51	52	53	54	55	56	57	58	59
第7章	60	61	62	63	64	65	66	67	68	69
第8章	70	71	72	73	74	75	76	77	78	79
第9章	80	81	82	83	84	85	86	87	88	89
第10章	90	91	92	93	94	95	96	97	98	99

01. 「パタトクカシーー」

問題

01. 「パタトクカシーー」の問題は、下記の内容です。

「パタトクカシーー」という文字列の1,3,5,7文字目を取り出して連結した文字列を得よ．

解答

問題を読むと、パタトクカシーーの1,3,5,7文字目なので、答えはパトカーです。
以下に示すどの解答でも、最後の式を評価するとパトカーと返ってきます。

シンプル

素直に文字列の末尾に追加していく方法があります。
enumerate(s)は、n=len(s)として(0,s[0]),...,(n-1,s[n-1])を走査するためのイテレータを返します。

# シンプル
s = "パタトクカシーー"

result = ""
for n,c in enumerate(s):
    if n%2==0:
        result += c
result #=> 'パトカー'

関数型的

関数型由来のmapやfilterを使って書くこともできます。
この例ではあまりメリットを感じられませんね…。

# 関数型的
s = "パタトクカシーー"

"".join(
    map(
        lambda x:x[1],
        filter(
            lambda x:x[0]%2==0,
            enumerate(s)
        )
    )
) #=> 'パトカー'

ラムダ式で定義した関数に名前を付けたほうが若干ましでしょうか。

# 関数型的
s = "パタトクカシーー"

second_elem = lambda x:x[1]
is_first_elem_even = lambda x:x[0]%2==0

"".join(map(
        second_elem,
        filter(
            is_first_elem_even,
            enumerate(s)
        )
    )
) #=> 'パトカー'

名前を付けすぎるとやりすぎに感じますね。

# 関数型的
s = "パタトクカシーー"

first_elem = lambda x:x[0]
second_elem = lambda x:x[1]
is_first_elem_even = lambda x:first_elem(x)%2==0

"".join(map(
        second_elem,
        filter(
            is_first_elem_even,
            enumerate(s)
        )
    )
) #=> 'パトカー'

これくらいの関数は準備されていてほしいですね。
なお、ラムダ式による関数定義は、下記の通常の関数定義と一緒です。

def first_elem(x):
    return x[0]

def second_elem(x):
    return x[1]

def is_first_elem_even(x):
    return first_elem(x)%2==0

おとなしく変数に入れたほうがいいかもしれません。
filterやmapが返してくるのはイテレータ(ジェネレータ)なので、変数に入れるたびにリストが生成されたりはしません。
JavaScriptみたいに、コールバックが後ろ側のほうがよかった気がしてなりません。

# 関数型的
s = "パタトクカシーー"

first_elem = lambda x:x[0]
second_elem = lambda x:x[1]
is_first_elem_even = lambda x:first_elem(x)%2==0

filtered = filter(is_first_elem_even, enumerate(s))
mapped = map(second_elem, filtered)
"".join(mapped) #=> 'パトカー'

内包表記

mapやfilterにこだわらずに、内包表記で書いたほうが簡単です。
[x for x in l]と書くとリスト内包表記ですし、(x for x in l)と書くとジェネレータ内包表記です。
f((x for x in l))は、f(x for x in l)とかっこを一つ減らして書くことができます。
複数の変数に値を同時に代入すれば、xで受け取ってx[0]やx[1]で取り出すよりもわかりやすくなります。

# 内包表記
s = "パタトクカシーー"

"".join(c for n,c in enumerate(s) if n%2==0) #=> 'パトカー'

Pythonic

Python的であることを Pythonic といいますが、このコードが一番Pythonicでしょうか。
リストや文字列に対して、s[start:end:step]でアクセスする方法をスライスと呼びます。
stepに2を指定すると1こ飛ばしになります。

# Pythonic
s = "パタトクカシーー"

s[::2] #=> 'パトカー'

その他の解答

インデックスで回す

enumerateなんて使わずに、自分でインデックスを用意して回す方法もあります。

# 自力で回す
s = "パタトクカシーー"

result = ""
n = 0
for c in s:
    if n%2==0:
        result += c
    n += 1
result #=> 'パトカー'

リストに追加して後で結合する

言語処理100本ノック 2020を解く(01. 文字列の逆順)と同じようにリストに追加して後で結合することもできます。

# シンプル
s = "パタトクカシーー"

l = []
for n, c in enumerate(s):
    if n%2==0:
        l.append(c)
"".join(l) #=> 'パトカー'

振り返り

「文字列を結合する」のと「リストに追加する」のとどっちがいいか悩むことがあります。
文字ではなく一般のオブジェクトのリスト扱うなら必然的にリストに追加することになりますが、今回のようにゴールが文字列の生成ならどちらでもいいんじゃないでしょうか。
サイズが大きくなって遅くなってきたら立ち止まって考えましょう。

# シンプル
s = "パタトクカシーー"

result = ""
for n,c in enumerate(s):
    if n%2==0:
        result += c
result #=> 'パトカー'

# シンプル
s = "パタトクカシーー"

l = []
for n, c in enumerate(s):
    if n%2==0:
        l.append(c)
"".join(l) #=> 'パトカー'

TL;DR

PyPiに上げられているC言語依存のpythonモジュールは、alpine標準のmuslには対応してないから毎回手元でコンパイルされるよ。
docker:alpineでどうしても使いたい場合は必要モジュールをビルドしたイメージを個別に用意しておくと良いよ。

alpineさんちのpip事情

pythonのモジュール管理行うpip。そのpipで採用されており快適にモジュールの導入を可能にしているwheel形式だが、alpineに対応した.whlがPyPi上に存在しないモジュールがある。

wheel

先に少しだけwheelの話。
wheelは元々Eggという形式の後継で作られたもので、Built Distributionと呼ばれるインストール形式であるそう。
Built Distributionはここの用語集を見る限りでは以下の通り。

Distribution 形式のうち、中身のファイルとメタデータをターゲットシステムの正しい場所へ移動するだけでインストールができるもの。

Egg や Wheel の実体は zip/tar などの圧縮形式の拡張なので、pip install時はダウンロードして展開後にpipで管理している場所にファイルを移動しているだけとなる。なので速い。

コンパイル済み拡張モジュールを含んだパッケージ¹も同様で取得・展開・移動するだけなので、これによりストレスフリーに使うことができている訳だ。
...ただしPyPi上に使用しているOS・アーキテクチャに対応したwheelファイルが存在する場合のみである。²

alineはmusl-libcで動く

alpineはmuslを採用している。
muslは軽量・高速・シンプルを目標に標準Cライブラリの実装を１から行っており、glibcなどの非標準な拡張ライブラリにも対応している。alpineにぴったりだ。
ただし、完全な互換はまだ実現できていない模様で稀に使いたい関数が存在しなかったりする。

で、今回のここで書く内容もalpineがmuslを使用している事が原因で掲題通りの事象が発生している。

ちなみに読み方はマッスルらしい。つよそう。

alpineはglibcを使っていない

今日、多くのLinuxディストリビューションはglibcを採用しているし、バイナリも共有ライブラリとして動的リンクさせてるのも少なくない。
numpyなどC言語を使用しているモジュールも例に漏れず³、PyPiにアップロードされているwheelファイルはglibc環境下で動くことを想定したビルドがなされている。つまり musl lib上で動く事を想定していない。⁴
そのためalpine上でpip install numpyをするとサポートされた.whlが存在しないため.zipやら.tar.gzが降ってくる。ビルド前のソースコードを丸々落としてきているのだ。

Collecting numpy
  Downloading numpy-1.18.3.zip (5.4 MB)
     |████████████████████████████████| 5.4 MB 1.3 MB/s 
  Installing build dependencies ... done

ダウンロード後はalpine用に一からコンパイルが実行されwheelファイルを作成する。そのあと出来上がったwheelファイルをpipは取り込み直すことでpython上でimportできるようにしている。
これがC言語依存モジュールをpip installすると時間がかかる原因である。

※ 他言語やOS・アーキテクチャに依存しないモジュールや、pureなpythonで書かれたモジュールだとalpineでもwheel形式で落ちてくるため時間はかからない。

浮かび上がる諸問題

C言語依存しているものでも数秒でコンパイルできるものもあるが、numpyを入れようとすると数分程度かかってしまうしnumpy依存のscipyなどを使おうとすると更に数十分コンパイルに時間がかかってしまうこともある。

一回限りのビルドで今後全ての開発環境を賄えるローカル環境であればそれでも良いかもしれないが、
製品やサービスをCI/CDを含めた環境構築することを考えていくとなると、膨大なコンパイル時間はとてつもないほど大きな障害となる。
git のブランチをフックにしてunitテストが走るCI環境、サーバーなどにサービスを安全にデリバリーするCD環境、etc、etc...
毎回毎回長いコンパイルが走ってしまうと細かい修正の確認ですら一時間単位で浪費してしまうことになるし、他の作業が滞ってしまう原因にもなる。
そしてなによりtwitterをする時間が減ってしまう。大問題だ。
実際に業務でpandasをalpineに突っ込んでしまった時は睡眠時間まで減ってしまったのだから、冗談抜きで死活問題である。

回避策

これを防ぐためには以下のものが考えられる。

1. alpineを諦める
2. コンパイル済みdocker imageで対処する
   1. ベースのディストリビューションとして使う
   2. multi stage ビルドを利用する

alpineを諦める

最も手間が少なく考える時間をかける必要がない有効な方法である。
ubuntuやcentosなどコンパイル済みで手段も確立しているディストリビューションに乗り換えてしまうのだ。
ただし、既に作り込んでしまっていて容易に乗り換えられないケースもあるのでは無かろうか？
業務で詰まった時は次の方法を取った。

コンパイル済みdocker imageで対処する

wheelのコンパイルが済んだ状態のdocker imageを自由に使える場所に置いてしまい、使いたい時にpullするというもの。

base-image

FROM python:3.8-alpine3.11
COPY require/requirements.txt /tmp
RUN apk update && \
  apk add --no-cache hoge-dev && \
  pip install --upgrade --no-cache-dir pip setuptools wheel && \
  pip install -r --no-cache-dir /tmp/requirements.txt

先ずはこんな感じでimageを作っておき、docker pushでdocker hubやawsのecrなどに保存しておく。

docker push hoge/huga:latest

2パターンあるがビルド完了しているものを使うという意味では同じ。

ベースのイメージとして使う

公式イメージなどと同じようにFROMを使ってベースイメージとして使う。
簡単だが、コンパイル時のみに必要なライブラリを消し忘れたりするとこちらのイメージサイズも大きくなってしまうなど、base-imagedockerfileへの依存度が高くなりメンテナンス性が下がってしまう。

builder-pattern

FROM hoge/huga:latest

multi stage ビルドとして使う

Docker17.05以上なら使える機能で、成果物だけイメージから取り出すことができる方法。
wheelコンパイルに関連するものはhoge/huga:latestに封じ込められるので後々気が楽になるから個人的におすすめ。

builder-pattern

FROM hoge/huga:latest as pip_build
FROM python:3.8-alpine3.11

COPY --from=pip_build /usr/local/lib/ /usr/local/lib/
COPY --from=pip_build /usr/local/bin/ /usr/local/bin/
COPY --from=pip_build /usr/local/include/ /usr/local/include/

上記だとディレクトリを直接張り付けてしまっているので、コンパイル済みの*.whlをhoge/huga:latestから持ってきてpip installする方が安全かも。

欠点

解決策としなかったのは上記の方法だと、複数のdockerイメージのメンテナンスを避けられないためだ。
作成した実行環境よりもソースコードで使用するライブラリや書式の方が最新になってしまった場合が起こるたび、長大なコンパイルを実行しpushし直す必要が出てくる...
安眠できる日々は長くは続かなさそうだ。

参考文献

• Using Alpine can make Python Docker builds 50× slower : 滅茶苦茶参考にしました。
• Python Packaging User Guide:用語集
• Introduction to musl
• お前のDockerイメージはまだ重い???
• マルチステージビルドの利用

---

1. 用語集曰くBinary Distributionと呼ぶらしく、拡張モジュールごと移動させ使用している。 ↩

2. PyPiのモジュールのページ左側[Navigation]->[Download files]から確認できる。 ↩

3. ダウンロードしたwheelをunzipで展開してやると*.soが見える。pipが正しく動いている以上、OS・アーキテクチャに最適化されたものが存在しているはずだ。 ↩

4. muslはアプリケーションを単一のポータブルなバイナリファイルとして配布できるように静的リンクに最適化している。(wikipediaから引用) ld-musl-x86_64.so.1しか存在しないのでld-linux-x86-64.so.2にダイナミックリンクしているものは軒並み落ちる。 ↩

はじめに

人工知能の記事を見ていると「学習回数」という言葉が多く出てきます。

どの記事も難しい言葉を使って説明されているので、分かりにくい！と思った方も多いと思います。

今回は「学習回数」を分かりやすく説明します。

書いてあること

1. 学習回数とは？
2. バッチサイズとは？
3. まとめ

1. 学習回数とは？

下の画像を見てください。これが学習回数です。

1つずつ説明していきます。

例えば、今あなたは、イヌとネコの画像を仕分ける人工知能(AI)を作ろうとしています。
イヌの画像を3,000枚、ネコの画像を3,000枚、合計6,000枚用意しました。

この6,000枚の画像をAIに見せて、イヌネコの特徴を覚えさせていきます。しかし、AIくんは、「6,000枚を一度見ただけ」で賢くなることはできません。これは、先生が授業で言った内容を一度で覚えられないのと同じです。

なので、6,000枚を何回も何回も、繰り返し見ることで、覚えようとします。簡単に言えば、この「何回見たか(覚えたか)」が学習回数になります。

6,000枚の画像を3回見たら、学習回数3回。
(合計18,000枚)
6,000枚の画像を5回見たら、学習回数5回。
(合計30,000枚)
6,000枚の画像を100回見たら、学習回数100回。
(合計6,000,000枚)

学習回数が多くなれば、AIはその分賢くなっていきます。1回目(最初の6,000枚)は「イヌかネコか分からない…」となっていた画像も、2回、3回と繰り返し見せていくと、「さっきは分からなかったけど、今ならわかるぞ！」という感じで賢くなっていきます。

基本はこんな感じですが、実際は少し違った方法で学習しています。数年前にどこかの偉い人が、「もっと効率よく賢くなる方法」を考えました。それが、「バッチサイズ」を使用する方法です。

2.バッチサイズとは？

もっと効率よく覚えるために、バッチサイズという考え方を使います。例えば、バッチサイズ=600の場合は、下のようになります。

6,000枚の画像を600枚ずつのグループに分けます。この時の「600」というのがバッチサイズになります。6,000枚を600枚ずつに分けるので、全部で10個のグループができることになります。AIはこのグループごとに学習していきます。

ここで注意すべき点があります。グループ分けされる画像は、最初の画像から順番に選ばれていくのではなく、6,000枚から「ランダム」に選ばれていきます。

6,000枚から適当に600枚を選ぶ
→グループ1
6,000枚から適当に600枚を選ぶ
→グループ2
6,000枚から適当に600枚を選ぶ
→グループ3

のようにして10個のグループを作ります。この時、イヌが何枚ずつで、ネコが何枚ずつ、という風に決める必要はありません。めちゃくちゃ適当に600枚ずつのグループを作ります。

「そーやってランダムに選んでいくと、2.3回選ばれる画像とか、1回も選ばれない画像が出てくるんじゃね？」と、疑問に思った方もいると思います。

その通りです。
このように「適当に選ぶ」ことによって、柔軟なAIが作れるそうです。(へーそうなんだーくらいに思ってもらえれば大丈夫です)

また、6,000枚をバッチサイズ=600で分けると、10個のグループができますが、この「10」のことを「イテレーション」と呼びます。

イテレーションは日本語で「繰り返し」という意味になります。「1回、2回、3回」のように、「イテレーション1、イテレーション2、イテレーション3」という雰囲気で使います。

つまり今回の例でいくと、
6,000枚のデータセットをバッチサイズ=600で、イテレーション=10で学習させた。という風になります。

ちなみにプログラマー達は、学習回数のことを「エポック数」と呼びます。
学習回数3回はエポック3、
学習回数10回はエポック10、
学習回数100回はエポック100、
のようになります。

3.まとめ

エポック、バッチサイズ、イテレーションをまとめた図が、下の図になります。

この図が理解できれば、人工知能の記事や論文を見たときに「学習回数」という単語が出てきたときでも理解できるはずです！

人工知能/AI/機械学習をもっと詳しく

「ねこアレルギーのAI」
https://t.co/4ltE8gzBVv?amp=1
YouTubeで機械学習について発信しています。
お時間ある方は覗いていただけると喜びます。

created by NekoAllergy

PythonでぷよぷよAIを作ったので記事にさせていただきます。
詳しくない方でもイメージしやすいように基本的なところから説明していくので、既に知っている方はどんどん読み飛ばしてください。

アルゴリズム

まず、コンピュータは次置く場所を「次こう置いて、その次はこう置いて...」というように先読みをして決めていきます。
この先読みは、もちろん深ければ深いほどいいのですが、見えてる手は今フィールドに落ちてきてるぷよ、NEXTぷよ、NEXTNEXTぷよの3つしかないので、見えない未来のツモは補完しながら探索していきます。
今回は10手先まで探索するようにしています。
ぷよを置く行動は22パターンあり、10手先まで考えると全ての置けるパターンは22^10となります。
これを計算するにはとても長い時間がかかってしまうので、評価の低い手をそれ以上探索しないようにするためにビームサーチを使います。今回はビーム幅は50としました。評価関数(コンピュータが手の良さを評価する基準)はこちらの記事を参考に「各列にそれぞれの色のぷよを2つ落とした時に起きる最大連鎖数×10+1列目、6列目の高さ×2+2列目、5列目の高さ+各ぷよの連結数の合計^2」としました。

コード

コードを実行すると組んでる様子のGIFと30手後の盤面の画像が作られます。(ぷよの画像をあらかじめimgフォルダに入れておいてください)
copy.deepcopyが遅いので、cPickleモジュールを使って高速化しました。これでかなり早くなりました。copy.deepcopyでやると15手組むのに2時間くらいかかります。あとリスト内包表記も早くなってる気がしました。

puyoAI.py

import pprint
import random
import _pickle as cPickle
import time
from PIL import Image

#x列の高さを数える
def search_height(field,x):
    return len([y[x] for y in field if y[x] != 0])

def next_create():
    color_type = ["G","Y","B","R"]
    return [random.choice(color_type) for i in range(2)]

def possible_moves(field,next):
    all_possible_moves = []
    #ぷよを横にしておいた時の全ての置き方
    for x in range(6):
        if x+1 < 6:
            #14段目まで埋まってないとき
            if search_height(field,x) < 14 and search_height(field,x+1) < 14:
                copy_field = cPickle.loads(cPickle.dumps(field, -1))
                #ぷよぷよの仕様で14段目のぷよは消える
                if search_height(field,x)+1 != 14:
                    copy_field[-(search_height(field,x)+1)][x] = next[0]
                #ぷよぷよの仕様で14段目のぷよは消える
                if search_height(field,x+1)+1 != 14:
                    copy_field[-(search_height(field,x+1)+1)][x+1] = next[1]
                all_possible_moves.append(copy_field)
    #2つとも同じ色のぷよの時は置いた後のフィールドが被る置き方があるからそこをカット
    if next[0] != next[1]:
        for x in range(6):
            if x+1 < 6:
                #14段目まで埋まってないとき
                if search_height(field,x) < 14 and search_height(field,x+1) < 14:
                    copy_field = cPickle.loads(cPickle.dumps(field, -1))
                    #ぷよぷよの仕様で14段目のぷよは消える
                    if search_height(field,x)+1 != 14:
                        copy_field[-(search_height(field,x)+1)][x] = next[1]
                    #ぷよぷよの仕様で14段目のぷよは消える
                    if search_height(field,x+1)+1 != 14:
                        copy_field[-(search_height(field,x+1)+1)][x+1] = next[0]
                    all_possible_moves.append(copy_field)
    #ぷよを縦にしておいた時の全ての置き方
    for x in range(6):
        if search_height(field,x) <= 12:
            copy_field = cPickle.loads(cPickle.dumps(field, -1))
            copy_field[-(search_height(field,x)+1)][x] = next[0]
            #ぷよぷよの仕様で14段目のぷよは消える
            if search_height(field,x)+2 != 14:
                copy_field[-(search_height(field,x)+2)][x] = next[1]
            all_possible_moves.append(copy_field)
    #2つとも同じ色のぷよの時は置いた後のフィールドが被る置き方があるからそこをカット
    if next[0] != next[1]:
        for x in range(6):
            if search_height(field,x) <= 12:
                copy_field = cPickle.loads(cPickle.dumps(field, -1))
                copy_field[-(search_height(field,x)+1)][x] = next[1]
                #ぷよぷよの仕様で14段目のぷよは消える
                if search_height(field,x)+2 != 14:
                    copy_field[-(search_height(field,x)+2)][x] = next[0]
                all_possible_moves.append(copy_field)
    return all_possible_moves

def count(field,y,x):
    global n
    c = field[y][x]
    field[y][x] = 1
    n +=1

    if x+1 < 6 and field[y][x+1] == c and c != 1:
        count(field,y,x+1)
    if y+1 < 14 and field[y+1][x] == c and c != 1:
        count(field,y+1,x)
    if x-1 >= 0 and field[y][x-1] == c and c != 1:
        count(field,y,x-1)
    if y-1 >= 0 and field[y-1][x] == c and c != 1:
        count(field,y-1,x)
    return n

def drop(field,y,x):
    while y >= 0:
        if y > 0:
            field[y][x] = field[y-1][x]
            y -= 1
        #14段目は空白
        else:
            field[y][x] = 0
            y -= 1
    return field

def chain(field,chain_count):
    global n
    copy_field = cPickle.loads(cPickle.dumps(field, -1))
    #4つ以上繋がっているところにフラグを立てる
    for y in range(14):
        for x in range(6):
            n = 0
            if field[y][x] != 0 and count(cPickle.loads(cPickle.dumps(field, -1)),y,x) >= 4:
                copy_field[y][x] = 1
    #フラグが一つもなかったら終了
    flag_count = 0
    for y in copy_field:
        flag_count += y.count(1)
    if flag_count == 0:
        return copy_field,chain_count
    #浮いてるぷよを落とす
    for y in range(14):
        for x in range(6):
            if copy_field[y][x] == 1:
                drop(copy_field,y,x)
    chain_count +=1
    return chain(copy_field,chain_count)

def height_evaluation(field):
    point = 0
    for x in range(6):
        if x == 0 or x == 5:
            point += len([y[x] for y in field if y[x] != 0])*2
        if x == 1 or x == 4:
            point += len([y[x] for y in field if y[x] != 0])
    return point

#各列にそれぞれの色のぷよを2つ落とした時の最大連鎖数
def feature_chain_evaluation(field):
    chain_counts = []
    color_type = ["G","Y","B","R"]
    for x in range(6):
        for color in color_type:
            copy_field = cPickle.loads(cPickle.dumps(field, -1))
            if [y[x] for y in field].count(0) > 2:
                copy_field[-(search_height(copy_field,x)+1)][x] = color
                copy_field[-(search_height(copy_field,x)+2)][x] = color
                chain_counts.append(chain(copy_field,0)[1])
            else:
                chain_counts.append(0)
    return max(chain_counts)

def count_evaluation(field):
    global n
    point = 0
    for y in range(14):
        for x in range(6):
            if field[y][x] != 0:
                n = 0
                point += count(cPickle.loads(cPickle.dumps(field, -1)),y,x)**2
    return point

def beam_search(depth0s,next,depth):
    print("現在の探索深さ:{}".format(depth))
    if depth > 10:
        return depth0s
    evaluation_results = []
    for depth0 in depth0s:
        for depth1 in possible_moves(depth0[1],next):
            #ぷよが消える置き方をしない,3列目の12段目に置かない
            if chain(depth1,0)[1] == 0 and depth1[2][2] == 0:
                evaluation_results.append([depth0[0],depth1,feature_chain_evaluation(depth1)*10 + height_evaluation(depth1) + count_evaluation(depth1)])
    return beam_search(sorted(evaluation_results, key=lambda x:x[2], reverse=True)[:50],next_create(),depth+1)

def next_move(field,next):
    evaluation_results = []
    for depth1 in possible_moves(field,next[0]):
        #ぷよが消える置き方をしない,3列目の12段目に置かない
        if chain(depth1,0)[1] == 0 and depth1[2][2] == 0:
            for depth2 in possible_moves(depth1,next[1]):
                #ぷよが消える置き方をしない,3列目の12段目に置かない
                if chain(depth2,0)[1] == 0 and depth2[2][2] == 0:
                    evaluation_results.append([depth1,depth2,feature_chain_evaluation(depth2)*10 + height_evaluation(depth2) + count_evaluation(depth2)])
    #評価値の合計が一番高い手を採用
    #dic = {フィールド:評価値}
    dic = {}
    beam_search_result = beam_search(sorted(evaluation_results, key=lambda x:x[2], reverse=True)[:50],next[2],3)
    for i in beam_search_result:
        #フィールドを文字列にして辞書のkeyにする
        #初期値を入れる(0)
        dic["".join([str(x) for y in i[0] for x in y])] = 0
    for i in beam_search_result:
        #フィールドに評価値を加算
        dic["".join([str(x) for y in i[0] for x in y])] += i[2]
    #評価値の合計が最も高いフィールド(文字列)
    final_choice = sorted(dic.items(), key=lambda x:x[1], reverse=True)[0][0]
    #文字列から二次元配列に直してreturn
    return [[n if n != "0" else 0 for n in final_choice[i:i+6]] for i in range(0,len(final_choice),6)]

def field_to_img(field):
    green = Image.open("img/green.png")
    yellow = Image.open("img/yellow.png")
    blue = Image.open("img/blue.png")
    red = Image.open("img/red.png")
    blank = Image.open("img/blank.png")
    imgs = [green,yellow,blue,red,blank]
    color_type = ["G","Y","B","R",0]
    field_img = Image.new("RGB", (green.width*6, green.height*14))
    start_y = 0
    for y in field:
        field_x_img = Image.new("RGB", (green.width*6, green.height))
        start_x = 0
        for x in y:
            for img,color in zip(imgs,color_type):
                if x == color:
                    field_x_img.paste(img, (start_x, 0))
                    start_x += img.width
        field_img.paste(field_x_img, (0, start_y))
        start_y += field_x_img.height
    return field_img

def main():
    start = time.time()
    imgs = []
    field = initial_field
    next = [next_create(),next_create(),next_create()]
    for i in range(30):
        field = next_move(field,next)
        pprint.pprint(field)
        imgs.append(field_to_img(field))
        next.pop(0)
        next.append(next_create())
    imgs[0].save("result.gif",save_all=True, append_images=imgs[1:], optimize=False, duration=500, loop=0)
    imgs[-1].save("result.png")
    print("処理時間:{}秒".format(time.time()-start))

if __name__ == "__main__":
    initial_field = [[0]*6 for i in range(14)]
    main()

結果

30手番シミュレーションした結果

組んでる様子

消えてる様子(10連鎖)

よくないところ

・一手考えるのがとても遅い
・30手組むのに30分ぐらいかかる

改善策

・c++とかもっと早い言語でやろう
・ビットボードを使おう
・盤面をハッシュ化して同じ盤面に対して何度も評価する必要を無くそう

はじめに

Qiitaの記事を書くのが初めてなので、間違えているところなどあると思います。
また、Pythonに関してもKerasなどの機械学習ライブラリで遊ぶ程度の初心者なのでコードをもっと良く書ける部分があると思います。
「ここをこうした方がいいよ」などコメントでアドバイスを頂けると嬉しいです。

今回作成したもの

タイトルにもある通りPythonからMusic(iTunes)をコントロールできるプログラムを作成しました。

(わかりにくくてごめんなさい。)

手の認識とジェスチャー

手の認識はmetalwhaleさんのGithubのリポジトリを参考に、wolterlwさんのプログラムを使わせていただきました。
説明を見たところ、認識にはGoogleのmediapipeをPythonで扱えるようにした物らしいです。
（手の検出器は顔などと違い全然ないことにびっくりしました...）

使用するには以下のモジュールとhand_tracker.pyが必要なのでインストールをしてください。

numpy
opencv
tensorflow

ジェスチャーはTheJLifeXさんがC言語で作成した手のランドマークの位置でジェスチャーを判断するプログラムを参考に両手と上下の判断をできるようにpythonで書きました。

landmark.py

def finger(landmark):
    x = 0
    y = 1
    thumbFinger = False
    firstFinger = False
    secondFinger = False
    thirdFinger = False
    fourthFinger = False

    if landmark[9][y] < landmark[0][y]:
        Hand_direction_y = 'up'
    else:
        Hand_direction_y = 'down'

    landmark_point = landmark[2][x]
    if landmark[5][x] < landmark[17][x]:
        if landmark[3][x] < landmark_point and landmark[4][x] < landmark_point:
            thumbFinger = True
        Hand_direction_x = 'right'
    else:
        if landmark[3][x] > landmark_point and landmark[4][x] > landmark_point:
            thumbFinger = True
        Hand_direction_x = 'left'

    landmark_point = landmark[6][y]
    if landmark[7][y] < landmark_point and landmark[8][y] < landmark_point:
        firstFinger = True

    landmark_point = landmark[10][y]
    if landmark[11][y] < landmark_point and landmark[12][y] < landmark_point:
        secondFinger = True

    landmark_point = landmark[14][y]
    if landmark[15][y] < landmark_point and landmark[16][y] < landmark_point:
        thirdFinger = True

    landmark_point = landmark[18][y]
    if landmark[19][y] < landmark_point and landmark[20][y] < landmark_point:
        fourthFinger = True

    if thumbFinger and firstFinger and secondFinger and thirdFinger and fourthFinger:
        hand = 'five'
    elif not thumbFinger and firstFinger and secondFinger and thirdFinger and fourthFinger:
        hand = 'four'
    elif not thumbFinger and firstFinger and secondFinger and thirdFinger and not fourthFinger:
        hand = 'tree'
    elif not thumbFinger and firstFinger and secondFinger and not thirdFinger and not fourthFinger:
        hand = 'two'
    elif not thumbFinger and firstFinger and not secondFinger and not thirdFinger and not fourthFinger:
        hand = 'one'
    elif not thumbFinger and not firstFinger and not secondFinger and not thirdFinger and not fourthFinger:
        hand = 'zero'
    elif thumbFinger and not firstFinger and not secondFinger and not thirdFinger and fourthFinger:
        hand = 'aloha'
    elif not thumbFinger and firstFinger and not secondFinger and not thirdFinger and fourthFinger:
        hand = 'fox'
    elif thumbFinger and firstFinger and not secondFinger and not thirdFinger and not fourthFinger:
        hand = 'up'
    elif thumbFinger and firstFinger and not secondFinger and not thirdFinger and fourthFinger:
        hand = 'RankaLee'
    else:
        hand = None

    return hand, Hand_direction_x, Hand_direction_y

Musicの制御

Musicの制御には、自分が作成したControl-itunes-from-pythonを使用しました。
このプログラムはsubprocessからOsascriptを実行しているだけのプログラムです。

(GithubのReadmeにもある通りmacOS 10.15 Catalina以上でしか実行できません。）

作成したコード

今回自分が作成したコードです。
ここでは、映像を取得後、手の認識やジェスシャーなどを行いその結果でMusicの操作を行うコード実行しています。

main.py

# coding=utf-8
import cv2
import numpy as np
from hand_tracker import HandTracker
from control import Control,Get_data
from landmark import finger
import subprocess
import shlex

hand_detector = HandTracker("./files/palm_detection_without_custom_op.tflite","./files/hand_landmark.tflite","./files/anchors.csv")
cap = cv2.VideoCapture(0)
before = None

def music(hand, Hand_direction_x, Hand_direction_y, before):
    if hand is not before:
        if hand is 'two' and Hand_direction_x == 'right' and Hand_direction_y == 'up':
            Control.next_track()
            return 'Next track'
        elif hand is 'two' and Hand_direction_x == 'left' and Hand_direction_y == 'up':
            Control.back_track()
            return 'Back track'
        elif hand is 'RankaLee' and Hand_direction_x == 'left' and Hand_direction_y == 'up':
            Control.previous_track()
            return 'Previous track'
        elif hand is 'aloha' and Hand_direction_x == 'right' and Hand_direction_y == 'up':
            Control.playpause()
            return 'Playpause'
        elif hand is 'fox' and Hand_direction_x == 'right' and Hand_direction_y == 'up':
            display = f'osascript -e \'display notification \"{Get_data.current_track_artist()}\" with title \"{Get_data.current_track()}\"\''
            subprocess.Popen(shlex.split(display),stdout=subprocess.PIPE)
            return 'display'

while True:
    ret, frame = cap.read()
    if ret:
        frame = cv2.flip(frame, 1)
        landmark, _ = hand_detector(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        if landmark is not None:
            hand, Hand_direction_x, Hand_direction_y = finger(landmark)
            control = music(hand, Hand_direction_x, Hand_direction_y, before)
            before = hand
            for landmark_point in landmark:
                x, y = landmark_point
                cv2.circle(frame, (int(x), int(y)), 2, (255, 0, 0), 4)
        cv2.imshow('Hand', frame)
        if cv2.waitKey(10) == 27:
            cap.release()
            cv2.destroyAllWindows()
            break

10行目あたりのhand_detector = HandTrackerは手の認識に必要なファイルを設定しています。
今回はmetalwhaleさんのhand_trackingリポジトリの中にあるmodelsを使いました。

14行目の def music ではジェスチャーで判断したデータを元にMusicの制御をするプログラムです。
ここを書き換えれば他の物を制御できたりするようになります。

さいごに

予想より、手の認識やランドマークが正確でびっくりしました。
今回は音楽の制御に使いましたが、別のことにも活用できると思うので、色々と面白いことをしてコメントなどに書いていただけると嬉しいです！！

今回作成したコードはGithubの方にあげたので、よければ使ってください（hand_tracker.pyとmodelsのダウンロードだけお願いします）

初めてのQittaの記事で全然説明などできていなかったり文字がおかしいところがあると思います。
もしあった場合は編集リクエストやコメントなどで教えてください。