ハッカソンの素材

README.md

ポケモン GO で試したスクリプト

ポケモン GO を自動化できないかと思って適当に書いたスクリプトです。大したことはやってません。

Walker.py

Android 内に立ち上がっている telnet サーバーと会話して位置情報を流し込み続けます。PokeVisioner.py を通して周辺のレアポケモンを探し、その方向に向かって歩きます。

Toucher.py

状況に応じて画面をタッチするスクリプトです。画面の画像を非同期に取得し続けるサーバーを PC 側に立ち上げておき、そのサーバーから画面情報を読み込みます。

「バトル画面」・「フィールド探索画面」・「ポケストップ画面」・「それ以外」を判別して、状況に応じた行動を取ります。

• バトル画面 - ボールを投げます。あまり上手ではありません。
• フィールド探索画面 - 自分の周辺をタッチしまくります。
• ポケストップ画面 - 回します。
• それ以外 - キャンセルボタンを連打します。

GPS 偽装

Android の GPS を外部から変更し続けるには MockGeoFix を使います。 Mock Location を使っていることがバレないように、Hide Mock Location を使います。 FusedLocation を殺しておくと良いという話は聞きます。

その他

• Android の input コマンドは実行するたびに VM を立ち上げるので非常に遅いです。これもサーバーを立ち上げておいて外部から流し込むのが良さそう。
• 1080x1920 の解像度だとキャプチャして読みこむだけで数秒かかるので、粗い画像をサクサク取得する方法があれば知りたいです。Vysor どうやってんだろう。
• 実は Xposed 使わなくても位置情報を書き換えるシステムアプリを作れば簡単にできそうな気がします。また時間があるときにやりたいです。

PokeVisioner.py

import json
import time
import urllib.request


def download_json(url):
    user_agent = "Mozilla/5.0 (iPhone; CPU iPhone OS 6_0 like Mac OS X) AppleWebKit/536.26 (KHTML, like Gecko) Version/6.0 Mobile/10A403 Safari/8536.25"
    headers = {'User-Agent': user_agent}
    req = urllib.request.Request(url, None, headers)
    with urllib.request.urlopen(req) as response:
        json_str = response.read().decode('utf-8')
        return json.loads(json_str)


def get_pokemon_json(lng, lat):
    url = "https://pokevision.com/map/scan/{lat}/{lng}".format(lng=lng, lat=lat)
    job_id = download_json(url)["jobId"]
    time.sleep(5)
    url = "https://pokevision.com/map/data/{lat}/{lng}/{job_id}".format(lng=lng, lat=lat, job_id=job_id)
    data = download_json(url)
    if not data["status"] or data["status"] != "success":
        return None
    return data["pokemon"]


POKEMON_DICT = {
    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: "ガーディ",
    59: "ウインディ",
    # 60: "ニョロモ",
    # 61: "ニョロゾ",
    62: "ニョロボン",
    63: "ケーシィ",
    64: "ユンゲラー",
    65: "フーディン",
    # 66: "ワンリキー",
    # 67: "ゴーリキー",
    # 68: "カイリキー",
    69: "マダツボミ",
    70: "ウツドン",
    71: "ウツボット",
    # 72: "メノクラゲ",
    # 73: "ドククラゲ",
    # 74: "イシツブテ",
    75: "ゴローン",
    76: "ゴローニャ",
    77: "ポニータ",
    78: "ギャロップ",
    79: "ヤドン",
    80: "ヤドラン",
    81: "コイル",
    82: "レアコイル",
    83: "カモネギ",
    # 84: "ドードー",
    # 85: "ドードリオ",
    86: "パウワウ",
    87: "ジュゴン",
    88: "ベトベター",
    89: "ベトベトン",
    90: "シェルダー",
    91: "パルシェン",
    92: "ゴース",
    93: "ゴースト",
    94: "ゲンガー",
    95: "イワーク",
    96: "スリープ",
    97: "スリーパー",
    # 98: "クラブ",
    99: "キングラー",
    100: "ビリリダマ",
    101: "マルマイン",
    # 102: "タマタマ",
    103: "ナッシー",
    104: "カラカラ",
    105: "ガラガラ",
    106: "サワムラー",
    107: "エビワラー",
    108: "ベロリンガ",
    109: "ガース",
    110: "マタドガス",
    111: "サイホーン",
    112: "サイドン",
    113: "ラッキー",
    114: "モンジャラ",
    115: "ガルーラ",
    116: "タッツー",
    117: "シードラ",
    # 118: "トサキント",
    119: "アズマオウ",
    # 120: "ヒトデマン",
    121: "スターミー",
    122: "バリヤード",
    123: "ストライク",
    124: "ルージュラ",
    125: "エレブー",
    126: "ブーバー",
    127: "カイロス",
    128: "ケンタロス",
    129: "コイキング",
    130: "ギャラドス",
    131: "ラプラス",
    132: "メタモン",
    133: "イーブイ",
    134: "シャワーズ",
    135: "サンダース",
    136: "ブースター",
    137: "ポリゴン",
    138: "オムナイト",
    139: "オムスター",
    140: "カブト",
    141: "カブトプス",
    142: "プテラ",
    143: "カビゴン",
    144: "フリーザー",
    145: "サンダー",
    146: "ファイヤー",
    147: "ミニリュウ",
    148: "ハクリュー",
    149: "カイリュー",
    150: "ミュウツー",
    151: "ミュウ"
}


def get_rare_pokemon(lng, lat):
    data = get_pokemon_json(lng, lat)
    if not data:
        return lng, lat
    for r in data:
        if r["pokemonId"] in POKEMON_DICT and r["is_alive"]:
            print(POKEMON_DICT[r["pokemonId"]])
            return r["longitude"], r["latitude"]
    return lng, lat

Toucher.py

import argparse
import io
import subprocess
import threading
import time
import urllib.request

import numpy
from PIL import Image
from numpy import random

adb_path = "/usr/bin/adb"


def adb_exec(commands):
    commands = [str(c) for c in commands]
    commands.insert(0, adb_path)
    commands.insert(1, "shell")
    commands.insert(2, "input")
    subprocess.check_output(commands)


def throw_ball():
    height = random.randint(100, 1100)
    adb_exec(["swipe", 540, 1400, 540, height, 250])


def random_touch():
    radius = 250
    center = (540, 1150)

    rad = random.randint(0, 1000) / 1000.0 * numpy.pi * 2
    r = random.randint(0, 1000) / 1000.0 * radius
    y = numpy.sin(rad) * r
    x = numpy.cos(rad) * r
    cx, cy = center
    x += cx
    y += cy
    adb_exec(["touchscreen", "tap", x, y])


def swipe_poke_stop():
    adb_exec(["swipe", 200, 800, 600, 800, 200])


def send():
    adb_exec(["touchscreen", "tap", 540, 540])
    time.sleep(1)
    adb_exec(["swipe", 800, 1700, 800, 100, 1200])
    time.sleep(1)
    adb_exec(["touchscreen", "tap", "700", "1500"])
    adb_exec(["touchscreen", "tap", "540", "1000"])


def power_up():
    adb_exec(["touchscreen", "tap", "700", "1500"])
    adb_exec(["touchscreen", "tap", "540", "1000"])


def tap_cancel_button():
    adb_exec(["touchscreen", "tap", 540, 1700])


def crop(img, center, edge):
    x, y = center
    x -= edge / 2
    y -= edge / 2
    c = img.crop((x, y, x + edge, y + edge))
    return c


def crop_ball(img):
    return crop(img=img, center=(540, 1625), edge=150)


def convert_to_array(img):
    return numpy.asarray(img)


def match(img1, img2):
    a1 = convert_to_array(img1)
    a2 = convert_to_array(img2)
    cnt = 0
    dif = 0
    for i in range(0, len(a1)):
        for j in range(0, len(a1[i])):
            cnt += 255
            x = int(a1[i][j]) - int(a2[i][j])
            dif += abs(x)
    return (cnt - dif) / float(cnt)


def crop_away(img):
    return crop(img=img, center=(121, 150), edge=144)


def crop_camera(img):
    return crop(img=img, center=(939, 1428), edge=182)


def crop_nap(img):
    return crop(img=img, center=(939, 1652), edge=182)


def crop_stop(img):
    return crop(img=img, center=(50, 1700), edge=100)


def is_in_field(img):
    return match(crop_ball(img), Image.open("ball.png")) > 0.9


def is_in_battle(img):
    c = match(crop_camera(img), Image.open("camera.png"))
    n = match(crop_nap(img), Image.open("nap.png"))
    a = match(crop_away(img), Image.open("away.png"))

    return max(c, n, a) > 0.9


def is_poke_stop(img):
    return match(crop_stop(img), Image.open("stop.png")) > 0.95


def init_q():
    q = []
    for i in range(0, 500):
        radius = 250
        center = (540, 1150)

        rad = random.randint(0, 1000) / 1000.0 * numpy.pi * 2
        r = random.randint(0, 1000) / 1000.0 * radius
        y = numpy.sin(rad) * r
        x = numpy.cos(rad) * r
        cx, cy = center
        x += cx
        y += cy
        q.append((x, y))
    return q


class TouchThread(threading.Thread):
    def __init__(self):
        super().__init__()
        self.state = ""
        self.q = init_q()
        self.cnt = 0

    def set_state(self, s):
        self.state = s
        print(s)

    def random_touch(self):
        x, y = self.q[self.cnt]
        self.cnt += 1
        if self.cnt >= len(self.q):
            self.cnt = 0
        adb_exec(["touchscreen", "tap", x, y])

    def run(self):
        while True:
            if self.state == "battle":
                throw_ball()
            elif self.state == "field":
                self.random_touch()
            elif self.state == "stop":
                swipe_poke_stop()
                time.sleep(2)
                tap_cancel_button()
            else:
                tap_cancel_button()


def main(args):
    t = TouchThread()
    t.start()
    state = ""
    cnt = 0
    while True:
        url = args.u
        path = io.BytesIO(urllib.request.urlopen(url).read())
        img_color = Image.open(path)
        img = img_color.convert("L")

        if is_in_field(img):
            state = "field"
        elif is_in_battle(img):
            state = "battle"
        elif is_poke_stop(img):
            state = "stop"
        else:
            if state == "field":
                state = "stop"
            else:
                state = "unknown"

        t.set_state(state)
        cnt += 1


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("-u", default="http://127.0.0.1:53516/screenshot.jpg?password=27023775", help="URL")
    main(parser.parse_args())

Walker.py

import argparse
import os
import select
import socket
import subprocess
import threading
import time

from geopy.distance import vincenty

from PokeVisioner import get_rare_pokemon

# 調整用パラメータ
delta = 0.000020
sleep_sec = 0.7

running = False


def kill_pokemon_go():
    from Toucher import adb_path
    commands = [adb_path, "shell", "am", "force-stop", "com.nianticlabs.pokemongo"]
    subprocess.check_output(commands)


def move(s, lng, lat):
    """
    GPS の座標を (lng, lat) に変更します
    :param s: telnet 用の socket
    :param lng: 移動先の longitude
    :param lat: 移動先の latitude
    :return: 移動に成功したら True を返します。失敗したら False
    """
    global running
    try:
        while True:
            r_list, w_list, _ = select.select([s], [s], [])
            if s in r_list:
                x = s.recv(1024)
                if b"KO: password required" in x:
                    s.close()
                    print(
                        "Password protection is enabled MockGeoFix settings. This is not supported.")
                    return False
                if x == '':
                    s.close()
                    print("Connection closed.")
                    return False
                if b'OK' in x:
                    running = True
                elif running:
                    return False
                time.sleep(sleep_sec)
                return True
            if s in w_list:
                s.send(b"geo fix %f %f\r\n" % (lng, lat))
            time.sleep(sleep_sec)
    except:
        return False


def start_to_tuple(start):
    a = start.split(",")
    lng = a[0]
    lat = a[1]
    return float(lng), float(lat)


class WalkThread(threading.Thread):
    def __init__(self, ip, port, start):
        super().__init__()
        self.ip = ip
        self.port = port
        self.lng, self.lat = start_to_tuple(start)
        self.g_lng, self.g_lat = self.lng, self.lat

    def set_goal(self, lng, lat):
        self.g_lng = lng
        self.g_lat = lat

    def get_geo(self):
        return self.lng, self.lat

    def run(self):
        s = socket.socket()
        s.connect((self.ip, self.port))

        ids = [0] * 100
        ids.extend([1] * 100)
        ids.extend([2] * 100)
        ids.extend([3] * 100)
        i = 0

        dx = [0, 0, 1, -1]
        dy = [1, -1, 0, 0]

        while True:
            if vincenty((self.lng, self.lat), (self.g_lng, self.g_lat)).meters > 1000:
                if abs(self.g_lat - self.lat) > abs(self.g_lng - self.lng):
                    if self.g_lat > self.lat:
                        self.lat += delta
                    else:
                        self.lat -= delta
                else:
                    if self.g_lng > self.lng:
                        self.lng += delta
                    else:
                        self.lng -= delta
            else:
                i %= len(ids)
                x = ids[i]
                self.lng += dx[x] * delta
                self.lat += dy[x] * delta
                i += 1

            if not move(s, self.lng, self.lat):
                break
            message = "({lng}, {lat})".format(lng=self.lng, lat=self.lat)
            os.system("clear")
            print(message)
        s.close()


def run(ip, port, start):
    try:
        t = WalkThread(ip, port, start)
        t.start()

        while True:
            lng, lat = t.get_geo()
            lng, lat = get_rare_pokemon(lng, lat)
            t.set_goal(lng, lat)
            time.sleep(60 * 5)
        kill_pokemon_go()
    except:
        kill_pokemon_go()


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", default="192.168.128.100",
                        help="IP address of telnet server")
    parser.add_argument("-p", type=int, default=5554, help="port of telnet server")
    parser.add_argument("-s", type=str, default="139.54496900012708,35.72819600000127",
                        help="start longitude & latitude")
    args = parser.parse_args()
    run(ip=args.a, port=args.p, start=args.s)