概要

pythonコード内でMySQLやPostgres、Seleniumなどのテストコンテナを作成することができる、testcontainersというライブラリがあるらしいというので試してみた結果です。
最初はDocker環境もいらないとかまさかないよね、と思いつつちょっと期待して試しましたが、予想通り、単にどこかにある既存のDockerホストに対して命令するだけだったのですが、それでも手動で上げ下げするよりは楽だと思われます。

対応範囲も広く、以下に対応しています。
- MySQL
- MariaDB
- PostgreSQL
- ElasticSearch
- Oracle DB
- MongoDB

MongoはなぜかReadmeに記載ないですが、以下のように対応しています。
https://testcontainers-python.readthedocs.io/en/latest/database.html#testcontainers.mongodb.MongoDbContainer

前提

自分のPCにDocker Toolboxを使ってコンテナ実行環境を作成しています。WindowsなのでDocker Desktop for Windowsを使いたいところですが、Hyper-Vを有効化できないため、インストールできません。またWindowsのバージョンも縛りがあり、WSL2も不可。
そこで少し古い方法ですが、Docker Toolboxというツールを使用しています。
https://docs.docker.com/toolbox/toolbox_install_windows/

Oracle Virtualboxという仮想化ソフトをHypervisorとして使用し、そのうえでLinuxを動かし、その上でDockerを起動する形になる。そのため、前提としてVirtualBoxをインストールする必要がある。(Toolboxをインストールする際に一緒にインストールすることもできる)

Host: Windows10(1803)
VirtualBox:6.0.12
Docker: 19.03.1

またDocker-Toolbox自体は起動された状態にしておく必要がある。(testcontainersからDocker-Toolboxの起動はできない。たぶん)

ちなみにコンテナ起動環境は、Windowsの場合、DOCKER_HOSTという環境変数を元に探しているとのこと。

echo %DOCKER_HOST%
tcp://192.168.99.100:2376

インストール

必要モジュールのインストールを実施。今回つかう範囲で必要だった上2つは追加で入れています。

pip install PyMySQL
pip install sqlalchemy
pip install testcontainers

公式にあるrequirementsファイルをpythonのバージョンに合わせて使用してもよいのですが、分量が多すぎたので個別にいれています。
https://github.com/testcontainers/testcontainers-python/tree/master/requirements

テスト

公式のテストコードそのままですが、以下のコードを記載したファイルを作成し、実行します。

app.py

import sqlalchemy
from testcontainers.mysql import MySqlContainer

with MySqlContainer('mysql:5.7.17') as mysql:
    engine = sqlalchemy.create_engine(mysql.get_connection_url())
    version, = engine.execute("select version()").fetchone()
    print(version)  # 5.7.17

イメージがない場合はpullしてきて起動してくれます。新規にコンテナを起動した場合は、コード実行後にコンテナも削除するようです。

テスト(MongoDBバージョン)

前提としてPyMongoをインストールしておきます。

pip install pymongo

以下のコードを実行。MySQL同様にバージョンを返しています。

app.py

from testcontainers.mongodb import MongoDbContainer

with MongoDbContainer("mongo:4.4.4") as mongo:
    db = mongo.get_connection_client()
    version = db.server_info()['version']
    print(version)  # 4.4.4

感想

今までは開発環境の操作とコーディング、テストは別々に行っていましたが、これをうまく使えば統合できそうだなと。ただまだそこまで至っていないので、まずはテストをしっかり書くところから取組みたいと思います。

SPA向けのクローラをpyppeteerで実装していたのですが、playwright-pythonに移行してみました。

基本的な使い心地はそのままに、細かいところで便利になっている印象だったので、pyppeteerを単純なクローラとして利用しているのでしたら、試してみてはいかがでしょうか。

■導入

普通にpip, pipenv, poetry等でplaywright-pythonを入れたあと、playwright installでバイナリを取得します。

$ poetry add playwright
$ poetry run playwright install

ただし、これだとchromium/firefox/webkitの全バイナリが入ってしまうので、クローラのように複数ブラウザで動かす必要がない場合は、明示的に指定するのがおすすめです。

$ poetry run playwright install webkit

■スクリプトの書き換え

pyppeteerとほとんど変わりませんが、asyncで書かなくても良い(もちろん書いても良い)ので、よりとっつきやすくなっています。

pyppeteer.py

import asyncio
from pyppeteer import launch

async def main():
    browser = await launch({    
        "args": ["--no-sandbox"],   
        "slowMo": 5,    
        "headless": True,
    })
    page = await browser.newPage()
    await page.goto('https://example.com')
    await page.screenshot({'path': 'example.png'})
    await browser.close()

asyncio.get_event_loop().run_until_complete(main())

playwright.py

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    iphone_11 = p.devices["iPhone 11 Pro"]
    browser = p.webkit.launch(headless=True)
    context = browser.new_context(**iphone_11)
    page = context.new_page()
    page.goto("https://example.com")
    page.screenshot(path="example.png")
    browser.close()

• 参考:
  • https://github.com/pyppeteer/pyppeteer#examples
  • https://github.com/microsoft/playwright-python#examples

page.waitFor()やslowMoが不要に

Auto-waitingが優秀で、画面描画に合わせてpage.waitFor()やslowMoで適宜waitを入れてやらなくても、いい感じに処理してくれます。

余計なwaitがないと実行時間も短くなるので、とりあえず消して試してみるのがおすすめです。

device

viewportとかその辺の設定がdeviceとしてビルトインで提供されていて楽ちんです。対応デバイス一覧は以下を参照。

任意のデバイスを定義して使う場合はこんな感じ。

• 設定値一覧
  • https://playwright.dev/python/docs/emulation/#devices

browser = playwright.webkit.launch(headless=True)
device = {
    'user_agent': 'foobar',
    'viewport': {
        'width': 1200,
        'height': 2048
    },
    'device_scale_factor': 1,
    'is_mobile': True,
    'has_touch': True,
    'default_browser_type': 'webkit'
}
context = browser.new_context(**device)

クローラとしてUAを明示したいだけなら、以下でOKです。

browser = playwright.webkit.launch(headless=True)
context = browser.new_context(user_agent="foobar")

HTML文字列取得

pyppeteerと同じなので、クローラがhtml文字列をlxmlやbs4に食わせるタイプとして実装されていれば、特に変更不要です。

python

page = context.new_page()
page.goto(url)

html = page.content()

■Dockernize

公式の https://github.com/microsoft/playwright-python/blob/master/Dockerfile を参考にしてください。

ライブラリが足りない場合、実行時に丁寧に表示されるので、とりあえず流してみて、エラーがあれば追加してあげるのが良さそうです。

shell

playwright._impl._api_types.Error: Host system is missing dependencies!

  Install missing packages with:
      sudo apt-get install libnss3\
          libnspr4\
          libasound2

参考に、以下に動作確認(2021/02/28: webkit用)したものを示します。

Dockerfile

FROM ubuntu:focal

# 1. Install latest Python
RUN apt-get update && apt-get install -y python3 python3-pip && \
    update-alternatives --install /usr/bin/pip pip /usr/bin/pip3 1 && \
    update-alternatives --install /usr/bin/python python /usr/bin/python3 1

# 2. Install WebKit dependencies
RUN apt-get update && DEBIAN_FRONTEND="noninteractive" apt-get install -y --no-install-recommends \
    libwoff1 \
    libopus0 \
    libwebp6 \
    libwebpdemux2 \
    libenchant1c2a \
    libgudev-1.0-0 \
    libsecret-1-0 \
    libhyphen0 \
    libgdk-pixbuf2.0-0 \
    libegl1 \
    libnotify4 \
    libxslt1.1 \
    libevent-2.1-7 \
    libgles2 \
    libxcomposite1 \
    libatk1.0-0 \
    libatk-bridge2.0-0 \
    libepoxy0 \
    libgtk-3-0 \
    libharfbuzz-icu0

# 3. Install gstreamer and plugins to support video playback in WebKit.
RUN apt-get update && apt-get install -y --no-install-recommends \
    libgstreamer-gl1.0-0 \
    libgstreamer-plugins-bad1.0-0 \
    gstreamer1.0-plugins-good \
    gstreamer1.0-libav \
    # add
    libvpx6\
    libgstreamer-plugins-base1.0-0\
    libgstreamer1.0-0\
    libopenjp2-7

RUN apt-get update \
    && pip install --upgrade pip \
    && pip install --no-cache-dir poetry && poetry config virtualenvs.create false && poetry install --no-dev \
    && playwright install webkit \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

なお、pyppeteerでDockerコンテナ経由の実行時に必要だった"args": ["--no-sandbox"]は不要になります。

(備考) tz選択が出て止まってしまう場合

Configuring tzdata
------------------

Please select the geographic area in which you live. Subsequent configuration
questions will narrow this down by presenting a list of cities, representing
the time zones in which they are located.

  1. Africa      4. Australia  7. Atlantic  10. Pacific  13. Etc
  2. America     5. Arctic     8. Europe    11. SystemV
  3. Antarctica  6. Asia       9. Indian    12. US
Geographic area:

DEBIAN_FRONTEND="noninteractive"を突っ込んでください。環境変数に設定してあげてもよいかと思います。

• 参考
  • Docker Ubuntu18.04でtzdataをinstallするときにtimezoneの選択をしないでinstallする - Qiita https://qiita.com/yagince/items/deba267f789604643bab

■移行で注意すべき点

pyppeteerで使われているChromiumのバージョンはかなり古い

pyppeteerで使われているChromiumのバージョンはかなり古い(REV:588429)です。
playwright-python(2021/02/28時点)だとREV:854489なので、そのあたりに起因する挙動の違いがないとも限りません。注意してください。

ちなみにMacOSXでplaywrightを動かした場合、バイナリは以下に配置されているようです。

shell

$ ls ~/Library/Caches/ms-playwright
chromium-854489 ffmpeg-1005     firefox-1234    webkit-1438

playwright-python公式のベースイメージはUbuntu Focal Fossa(20.04)

DockerHubのpythonイメージとかconda系を使ってる場合、playwrightに必要なライブラリがなくて困ったりするかもしれません。

私はpython:3.8.6-slimを使ってたんですがlibevent-2.1-7が見当たらなくてうーん…ってなりました。

頑張る気力がなかったので、playwrightに寄せる形でクローラのベースイメージをubuntu:focalにしてしまいましたが、クローラで使っている他のライブラリの依存関係が厳しめな場合、注意が必要です。

ちなみにubuntu:focalのpython3は3.8.5だったので、その辺も現在お使いのライブラリによっては、調整が必要になるかもしれません。

■最後に

「pyppeteerのスペルに気を使わなくて良くなる」、この一点だけでもplaywright-python、おすすめです！

分かること

・テスト駆動開発（TDD）はどの順序で開発すべきか？
・テスト駆動開発（TDD）はなぜ失敗をわざと行うのか？
・pytestはどう使うべきか？

目次

1. テスト駆動開発（TDD）について
2. Python, Pytestでテスト駆動開発（FizzBuzz問題）

1. テスト駆動開発（TDD）について

テスト駆動開発は以下3種のサイクルである。
（ToDoリストを作成し、簡単な処理から手を付けると良い）

１. RED（テスト失敗）
テストコードを作成してテストする。（実装コードを作成してないので失敗する）
2. GREEN（テスト成功）
成功するように必要最小限の実装コードを書く。（正しさや綺麗さは考えない）
3. リファクタリング
成功を維持したままコードを綺麗にする

(画像：50 分でわかるテスト駆動開発 より)

なぜ失敗を工程に入れるのか？

テストコードの正しさを確認（テストコードのテスト）する為。
誤りのある実装コードのテストが予想通りのエラーならば、正しい実装コードでもテストが正しく機能されると期待できる。

2. Pytestでテスト駆動開発（FizzBuzz問題）

FizzBuzz問題をテスト駆動開発で実装し、テスト駆動開発の工程を確認する。

FizzBuzz問題
1から100までの数をプリントするプログラムを書け。
ただし
3の倍数のときは数の代わりに｢Fizz｣とプリントし、
5の倍数のときは｢Buzz｣とプリントし、
3と5両方の倍数の場合には｢FizzBuzz｣とプリントすること。

0. [準備] ToDoリストを作成する

簡単な処理から優先順位を高くすると良い。

- [ ] 数を文字列にして返す

- [ ] ３の倍数のときは数の代わりに文字列'Fizz文字列'と返す
- [ ] ５の倍数のときは文字列'Buzz'と返す
- [ ] ３と５の倍数の場合には文字列'FizzBuzz'と返す

- [ ] 1から100までの数
- [ ] プリントする

1. ToDoリストから１つをピックアップしてサイクルを回す

- [ ] 数を文字列にして返す
    - [ ] 1を渡したら文字列"1"を返す  ← next
    - [ ] 2を渡したら文字列"2"を返す

必要ならToDoの深堀りをする。

1-1. RED（テスト失敗）

テストコードを作成してテストでわざと失敗する

1-1-1. テストコードを作成(test_fizz_buzz.py)

「1を渡したら文字列"1"を返す」テストを作成する

test_fizz_buzz.py

import pytest

class TestStringConvert:
    # 前準備
    @pytest.fixture
    def target(self):
        from fizz_buzz import string_convert        
        return string_convert

    # 検証
    def test_return_value(self, target):
        assert target(1) == '1'

この時点で既に、
・実装ファイルとテストコードのファイル名（fizz_buzz.py, test_fizz_buzz.py)
・数を文字列にして返す関数名
・数を文字列にして返す処理をテストするクラスと関数（TestStringConvert, test_return_value）
と考えることが多い為、ToDoは簡単な処理から始めるのがおすすめ。

"@pytest.fixture"はテストの前処理(実装コードで使う関数の呼び出し)を行うpytestの機能。

1-1-2. テストコードをテスト(test_fizz_buzz.py)

実装コードを作成してない（実装モジュールを作成してない）ので失敗する

pytest test_fizz_buzz.py
============================= test session starts =============================
-
collected 1 item

test_fizz_buzz.py E                                                      [100%]

=================================== ERRORS ====================================
____________ ERROR at setup of TestStringConvert.test_return_value ____________

self = <test_fizz_buzz.TestStringConvert object at 0x000001EF5648A190>

    @pytest.fixture
    def target(self):
>       from fizz_buzz import string_convert
E       ModuleNotFoundError: No module named 'fizz_buzz'

test_fizz_buzz.py:7: ModuleNotFoundError
=========================== short test summary info ===========================
ERROR test_fizz_buzz.py::TestStringConvert::test_return_value - ModuleNotFoun...
========================= 1 error in 0.18s =========================

予想（狙い）通り、fizz_buzzモジュールがないエラーなら成功。

E       ModuleNotFoundError: No module named 'fizz_buzz'

このエラーが出なければ、テストコードのどこかに誤りがあると分かる。

1-2. GREEN（テスト成功）

テストコードが成功することだけを優先して必要最小限の実装コードを書く。
（この時点で正しさや綺麗さは考えなくて良い）

1-2-1. 必要最小限の実装コードを書く(fizz_buzz.py)

fizz_buzz.py

def string_convert(x):
    return '1'

1-2-2. テストコードをテスト(test_fizz_buzz.py)

pytest test_fizz_buzz.py
============================= test session starts =============================
-
collected 1 item

test_fizz_buzz.py .                                                      [100%]
======================== 1 passed in 0.02s =========================

成功。

（1-3. リファクタリング）

今回はそのまま次のサイクルへ向かう。

2. 次のサイクルを回す

- [ ] 数を文字列にして返す　　　　　　　　 
    - [x] 1を渡したら文字列"1"を返す    
    - [ ] 2を渡したら文字列"2"を返す　  ← next

2-1. RED（テスト失敗）

2-1-1. テストコードを拡張(test_fizz_buzz.py)

「2を渡したら文字列"2"を返す」テストを追加する

test_fizz_buzz.py

 import pytest

 class TestStringConvert:
     # 前準備
     @pytest.fixture
     def target(self):
         from fizz_buzz import string_convert        
         return string_convert

+    check_value_data = [
+        (1, '1'),  # 1を渡したら文字列"1"を返す
+        (2, '2')   # 2を渡したら文字列"2"を返す
+    ]

     # 検証
-    def test_return_value(self, target):      
-       assert target(x) == '1'
+    @pytest.mark.parametrize('x, expected_return_value', check_value_data)
+    def test_return_value(self, target, x, expected_return_value):      
+        assert target(x) == expected_return_value

@pytest.mark.parametrizeで以下２パターン（計２回）のテストを定義する。

１．x = 1, expected_return_value= '1'
２．x = 2, expected_return_value= '2'

→１つのテストに複数のassert文を記述するのは避ける。

以下のような記述は避ける。
（assert1でエラーとなったとき、assert2のテスト結果が分からない為。）

    # 検証（避けたい記述方法）
    def test_return_value(self, target):      
        assert target(1) == '1'  # assert1 
        assert target(2) == '2'  # assert2（assert1がエラーだと実行されない）

2-1-2. テストコードをテスト(test_fizz_buzz.py)

現在の実装コード（fizz_buzz.py）はどんな値を入れても文字列1を返す為、エラーが出る。

pytest test_fizz_buzz.py
============================= test session starts =============================
-
collected 2 items

test_fizz_buzz.py .F                                                     [100%]

================================== FAILURES ===================================
__________________ TestStringConvert.test_return_value[2-2] ___________________

self = <test_fizz_buzz.TestStringConvert object at 0x0000013B7BC5C4F0>
target = <function string_convert at 0x0000013B7BC32DC0>, x = 2
expected_return_value = '2'

    @pytest.mark.parametrize('num, expected_return_value', check_value_data)
    def test_return_value(self, target, x, expected_return_value):
>       assert target(x) == expected_return_value
E       AssertionError: assert '1' == '2'
E         - 2
E         + 1

test_fizz_buzz.py:18: AssertionError
=========================== short test summary info ===========================
FAILED test_fizz_buzz.py::TestStringConvert::test_return_value[2-2] - Asserti...
=================== 1 failed, 1 passed in 0.19s ====================

2-2. GREEN（テスト成功）

テストの（狙い通りの）失敗を確認後、テストが成功するように実装コードを訂正する。

fizz_buzz.py

  def string_convert(x):
-     return '1'
+     return str(x)

PS C:\Users\spell\Documents\GIT\fizzbuzz> pytest test_fizz_buzz.py
============================= test session starts =============================
platform win32 -- Python 3.8.5, pytest-6.1.1, py-1.9.0, pluggy-0.13.1
rootdir: C:\Users\spell\Documents\GIT\fizzbuzz
collected 2 items

test_fizz_buzz.py ..                                                     [100%]
======================== 2 passed in 0.02s =========================

2つのテストが実行されたことも分かる。

collected 2 items

2-3. リファクタリング）

成功を維持したままコードを綺麗にする

test_fizz_buzz.py

 import pytest

 class TestStringConvert:
     @pytest.fixture
     def target(self):
         from fizz_buzz import string_convert        
         return string_convert

     check_value_data = [
         (1, '1'),
         (2, '2')
     ]

-    @pytest.mark.parametrize('x, expected_return_value', check_value_data)
-    def test_return_value(self, target, x, expected_return_value):      
-        assert target(x) == expected_return_value
+    @pytest.mark.parametrize('num, expected_return_value', check_value_data)
+    def test_return_value(self, target, num, expected_return_value):      
+        assert target(num) == expected_return_value

fizz_buzz.py

-  def string_convert(x):
-      return str(x)
+  def string_convert(num):
+      return str(num)

3. 次のサイクルを回す

- [X] 数を文字列にして返す                          ← finish
    - [X] 1を渡したら文字列'1'を返す
    - [X] 2を渡したら文字列'2'を返す

- [ ] 3の倍数のときは数の代わりに文字列'Fizz'と返す   
    - [ ] 3を渡したら文字列'Fizz'を返す             ← next
- [ ] 5の倍数のときは文字列'Buzz'と返す
- [ ] 3と5の倍数の場合には文字列'FizzBuzz'と返す

- [ ] 1から100までの数
- [ ] プリントする

3-1. RED（テスト失敗）

3-1-1. テストコードを拡張(test_fizz_buzz.py)

「3を渡したら文字列"Fizz"を返す」テストを追加する

test_fizz_buzz.py

import pytest

class TestStringConvert:
    @pytest.fixture
    def target(self):
        from fizz_buzz import string_convert        
        return string_convert

    check_value_data = [
        (1, '1'),
        (2, '2'),
        (3, 'Fizz')
    ]

    @pytest.mark.parametrize('num, expected_return_value', check_value_data)
    def test_return_value(self, target, num, expected_return_value):      
        assert target(num) == expected_return_value

3-2. GREEN（テスト成功）

3-2-1. 成功するように必要最小限のコードを書く（倍数関係なく3ならFizzを返す）

fizz_buzz.py

 def string_convert(num):
+    if num == 3:
+        return 'Fizz'

     return str(num)

3-2-2. 成功したら正しい処理（3の倍数ならFizzを返す）をするよう訂正

fizz_buzz.py

 def string_convert(num):
-    if num == 3:
+    if num % 3 == 0:
         return 'Fizz'

     return str(num)

3-3. リファクタリング

・テストコードから仕様が読み取れない（個別の具体的な動きしかない）為、テストコードだけで仕様を読み取れるようにする。
・また、テストのパターン数に対称性を持たせる。
（メンテナンスを減らす為にも場合分けは減らして対称性を持たせる）

test_fizz_buzz.py

 import pytest

 class TestStringConvert:
     @pytest.fixture
     def target(self):
         from fizz_buzz import string_convert        
         return string_convert

     check_value_data = [
-        (1, '1'),
-        (2, '2'),
+        # 3の倍数のとき
         (3, 'Fizz'),
+        # その他の数のとき
+        (1, '1')
     ]

     @pytest.mark.parametrize('num, expected_return_value', check_value_data)
     def test_return_value(self, target, num, expected_return_value):      
         assert target(num) == expected_return_value

今回はコメントアウトで説明したが、以下の手法でも効果的。

・テスト関数名を長くして仕様を説明する。
・パターンごとにテストクラス分けを行う。

以上を繰り返すのがテスト駆動開発（TDD）である。

参考動画：50 分でわかるテスト駆動開発（@t_wadaさんによるライブコーディング）

C - Modulo Summation

O(N)
最小公倍数を計算。
その後、最小公倍数-1の値で余りを取得し加算していきます。

python

import math
import heapq
import itertools
from functools import reduce

def my_lcm_base(x, y):
    return (x * y) // math.gcd(x, y)

# main
def main():
    N = int(input())
    A = list(map(int, input().split()))

    g = reduce(my_lcm_base, A, 1)
    g = g - 1
    res=0
    for i in range(0, len(A)):
        res += g % A[i]

    print(res)

# エントリポイント
if __name__ == '__main__':
    main()

D - Flipping Signs

O(N)
i, i+1に-1をかけていくのを繰り返す。
最終的に奇数なら−符号が一つ存在、偶数なら全て＋符号になります。

偶数なら全てを正整数に変更して全ての整数の合計を出力。

奇数なら全てを正整数に変更して最小値を取得。
全ての整数-(最小値*2)を計算。
正整数にした際の最小値を負数に変更して、全ての整数の合計を出力。

python

import math
import heapq
import itertools
from functools import reduce

def my_lcm_base(x, y):
    return (x * y) // math.gcd(x, y)

# main
def main():
    N = int(input())
    A = list(map(int, input().split()))

    cnt=0
    for i in range(0, len(A)):
        if A[i]<0:
            cnt+=1

    min_v=10000000000
    for i in range(0, len(A)):
        if A[i]<0:
            A[i] = A[i] * -1
        min_v = min(min_v, A[i])

    if cnt%2==0:
        print(sum(A))
    else:
        print(sum(A)-(min_v*2))

# エントリポイント
if __name__ == '__main__':
    main()

Gaussian-Bernouili RBMって

RBMと言えば可視層と隠れ層がBernouili分布に従いますよね。
本題のGaussian-Bernouili RBM（GBRBM）は可視層がGaussian分布で隠れ層がBernouili分布に従うRBMなわけです。
GBRBM学習って分散の学習が上手くいかない〜、とか、分散を1に固定して学習しない記事や実装コードばかり目にするので、
GBRBMを使おうとしてた時に相当困りました・・・（４年前）。
ということでコードを混ぜてGBRBMについて！！！

分散の学習について

悩んでる人は当時の僕だけじゃないはずですが、
一応前置きすると、可視特徴量の分散をきちんと再現できるわけでは無いです。
あくまで「途中でパラメータが発散してしまう！」などの、学習が破綻するようなレベルの問題をクリアするためのお話です。

結論から言うと、”学習時にCD法を行う際、きちんと隠れ層をサンプリングしてから可視層の分布を推定する”ってことです。
多くの記事やコードでは
$v_0$ ⇒ $p(h_0)$ ⇒ $p(v_1)$ ⇒ $p(h_1)$
のように、CD法を行ってます。
これを
$v_0$ ⇒ $p(h_0)$ ⇒ $h_0$ ⇒ $p(v_1)$ ⇒ $v_1$ ⇒ $p(h_1)$
のように、ちゃんと$h_0$をサンプリングしてから$p(v_1)$を算出すれば学習もちゃんと進みます。

「RBMの隠れ層の値は 1 or 0 程度の情報量しか持ってない」という前提で可視層が予測されるので
学習時はちゃんとサンプリングした方が良いそうです。

ただし、学習が終わってしまえば、デコード時は隠れ層の期待値からデコードして大丈夫（というかそうすべき？）です。

初期化

import numpy as np

class GBRBM(object):
  def __init__(self, I, J):
    self.W = np.random.randn(I, J)
    self.b = np.zeros(I)
    self.c = np.zeros(J)
    self.z = np.zeros(I)

Wは接続重み、bとcはそれぞれ可視層のバイアスと隠れ層のバイアスです。
分散$\sigma^2_i$の学習ですが、非負制約を満たさなくてはいけないので
$$
z_i = \log\sigma^2_i
$$
として、$\sigma^2_i$の代わりに$z_i$を学習します。

ネットワーク定義

  def propup(self, v):
    v_ = v / np.exp(self.z)
    tmp = np.dot(v_, self.W) + self.c
    ph = self.sigmoid(tmp)
    return ph

  def propdown(self, h):
    v_mean = np.dot(h, self.W.T) + self.b
    return v_mean

  def contrastive_divergence(self, v, k=1):
    sigma = np.sqrt(np.exp(self.z))
    ph0 = self.propup(v) # p(h|v)
    phk = ph0
    for step in range(k):
      hk = np.random.binomial(size=phk.shape, n=1, p=phk) # sampling h 
      pvk = self.propdown(hk) # p(v|h)
      vk = np.random.randn(pvk.shape[0], pvk.shape[1])*sigma + pvk 
      phk = self.propup(vk) # p(h|v)    

    return ph0, vk, phk

propupは$p(h|v)$を算出するブロックで、propdownは$p(v|h)$を算出するブロックです。
contrastive_divergenceはCD法によるサンプリングブロックで、
さっきも書きましたが$p(v|h)$を算出する際にphk（$p(h|v)$のこと）を使うのではなく
ちゃんとサンプリングしたhkを使います。
ただ、勾配計算にはどうせ期待値を使うのでサンプリングした値ではなく$p(h|v)$を返します。

勾配計算

  def get_grad(self, v):
    N = v.shape[0]

    #--- CD_1 ---#
    v0 = v
    h0, v1, h1 = self.contrastive_divergence(v0)

    var = np.exp(self.params['z'])
    v0_, v1_ = v0/var, v1/var

    W, b = self.W, self.b
    #--- Calculate gradient params ---#
    vw0 = np.dot(v0_.T, h0)
    vw1 = np.dot(v1_.T, h1)
    dW = (vw0 - vw1) / N
    db = np.mean(v0_ - v1_, axis=0) 

    vz1 = 0.5*((v0 - b)**2) - np.dot(h0, W.T)*v0
    vz2 = 0.5*((v1 - b)**2) - np.dot(h1, W.T)*v1
    dz = np.mean((vz1 - vz2)/var, axis=0)

    return dW, db, dz

ここはどの記事でもコードでもほぼ一緒だと思います。

全体コード

import numpy as np

class GBRBM(object):
  def __init__(self, I, J):
    self.W = np.random.randn(I, J)
    self.b = np.zeros(I)
    self.c = np.zeros(J)
    self.z = np.zeros(I)

  def propup(self, v):
    v_ = v / np.exp(self.z)
    tmp = np.dot(v_, self.W) + self.c
    ph = self.sigmoid(tmp)
    return ph

  def propdown(self, h):
    v_mean = np.dot(h, self.W.T) + self.b
    return v_mean

  def contrastive_divergence(self, v, k=1):
    sigma = np.sqrt(np.exp(self.z))
    ph0 = self.propup(v) # p(h|v)
    phk = ph0
    for step in range(k):
      hk = np.random.binomial(size=phk.shape, n=1, p=phk) # sampling h 
      pvk = self.propdown(hk) # p(v|h)
      vk = np.random.randn(pvk.shape[0], pvk.shape[1])*sigma + pvk 
      phk = self.propup(vk) # p(h|v)    

    return ph0, vk, phk

  def get_grad(self, v):
    N = v.shape[0]

    #--- CD_1 ---#
    v0 = v
    h0, v1, h1 = self.contrastive_divergence(v0)

    var = np.exp(self.params['z'])
    v0_, v1_ = v0/var, v1/var

    W, b = self.W, self.b
    #--- Calculate gradient params ---#
    vw0 = np.dot(v0_.T, h0)
    vw1 = np.dot(v1_.T, h1)
    dW = (vw0 - vw1) / N
    db = np.mean(v0_ - v1_, axis=0) 

    vz1 = 0.5*((v0 - b)**2) - np.dot(h0, W.T)*v0
    vz2 = 0.5*((v1 - b)**2) - np.dot(h1, W.T)*v1
    dz = np.mean((vz1 - vz2)/var, axis=0)

    return dW, db, dz

  def train(self, v, nbatch=256, nepoch=100, lr=0.01): 
    N = v.shape[0]

    for epoch in range(n_epoch):
      perm = np.random.permutation(N)

      for i in range(0, N-nbatch, nbatch):
        v_batch = np.asarray(v[perm[i:i+n_batch]])
        dW, db, dc, dz = self.get_grads(v_batch)

        self.W -= lr * dW
        self.b -= lr * db
        self.c -= lr * dc
        self.z -= lr * dz

  def sigmoid(self, x):
    return 1./(1. + np.exp(x))

初めに

ganやらcnnやらで二次元の画像のデータセットを作りたくなったので
タグを指定してpixivから画像をダウンロードするスクリプトを作った。

pythonのpixivpyを使ってpixivから特定のユーザーの作品をすべて一括でダウンロードする（うごイラ含む）
これをベースに作りました。

注意

pixivpyはおそらく中国の有志が作ったライブラリであり、公式のものではありません。
そのためログイン時のユーザーid、passwordは流出してもいいように捨て垢を使うなどの方法を使うことをお勧めします。
また乱用厳禁です。
スクレイピングのルールとして1秒当たりリクエリは1回までという暗黙のルールがあります。
これがないとdos攻撃となってしまい、法的にアウトです。
sleepをコメントアウトする際は自己責任でお願いします。
また何か問題が起きても自己責任でお願いします。

ソースコード

tag_downloader.py

from pixivpy3 import *
import json
import os
from time import sleep



#タグによるフィルター　書き方 target_tag = ["Fate/GrandOrder","FGO","FateGO","Fate/staynight"]
target_tag = [] #target_tag内に複数書くと、そのうち少なくとも一つあればダウンロード
target_tag2 = []#さらにtarget_tag2に書くとtarget_tagを満たし、かつtarget_tag2を満たすものだけダウンロード
extag = ["R-18"]#extag内のタグが一つでも入っていればダウンロードしない
#ブックマーク数によるフィルター、最小値を設定、0ならすべて
score=0
#閲覧数によるフィルター、最小値を設定、0ならすべて
view=1000

#画像を保存するディレクトリ
saving_direcory_path = "./img/"
#検索タグ
serch_tag = "初音ミク"
#何ページ分取得するか（1ページ30作品）
max_page = 10



#def search_illust(self, word, search_target='partial_match_for_tags', sort='date_desc', duration=None):
# 搜索 (Search) 
# search_target - 検索タイプ
#   partial_match_for_tags  - 部分一致
#   exact_match_for_tags    - 完全一致
#   title_and_caption       - 説明文？
# sort: [date_desc, date_asc, popular_desc] - popular_desc：プレミアム会員のみ
# duration: [within_last_day, within_last_week, within_last_month]
# start_date, end_date: '2020-07-01'


#2021/2/21方法変更
api = AppPixivAPI()
api.auth(refresh_token=REFRESH_TOKEN)
aapi = AppPixivAPI()
aapi.auth(refresh_token=REFRESH_TOKEN)

if not os.path.exists(saving_direcory_path):
    os.mkdir(saving_direcory_path)
separator = "------------------------------------------------------------"

def making_illust_list(serch_tag):
    ids = []
    urls = []
    for p in range(1, max_page+1):
        sleep(1)
        try:
            print("p" + str(p))
            if p == 1:
                a = aapi.search_illust(serch_tag, search_target='partial_match_for_tags', sort='popular_desc')
            else:
                a = aapi.search_illust(**nxt)
            illusts = a.illusts
            for illust in illusts:
                ids.append(illust.id)
            nxt = aapi.parse_qs(a.next_url)
            if nxt == None:
                break
        except:
            print("erro")
    return ids




def download(ids):
    sleep(1)         
    for idnum in ids:
        illust =  api.works(idnum).response[0]

        #フィルター

        #タグによるフィルター
        if len(list(set(target_tag)&set(illust.tags))) == 0 and target_tag != []:
            continue
        if len(list(set(target_tag2)&set(illust.tags))) == 0 and target_tag2 != []:
            continue
        #exタグが一つでも入っていたらスキップ
        if len(list(set(extag)&set(illust.tags))) > 0 :
            continue
        #score以上の作品だけダウンロード
        if illust.stats.favorited_count.private + illust.stats.favorited_count.public < score :
            continue
        #view以上の作品だけダウンロード
        if illust.stats.views_count < view :
            continue

        print("Title: {}".format(illust.title))
        print("score: {}".format(illust.stats.favorited_count.private + illust.stats.favorited_count.public))
        print("view: {}".format(illust.stats.views_count))

        #if the illustration has already downloaded, skip downloading it
        if os.path.exists(saving_direcory_path+str(illust.id)+"_p0.png") or os.path.exists(saving_direcory_path+str(illust.id)+"_p0.jpg") or os.path.exists(saving_direcory_path+str(illust.id)+'_ugoira') or os.path.exists(saving_direcory_path+str(illust.id)+"_p0.gif"):
            print("Title:"+str(illust.title)+" has already downloaded.")
            print(separator)
            continue

        sleep(1)
        # illustrations with only one picture
        #複数の絵を投稿しているといろいろなタグが混ざるので一枚絵のみダウンロードすることにした。
        if illust.type == "illustration":
            aapi.download(illust.image_urls.large, saving_direcory_path)
            #print("Download complete!　Thanks to {0}{1}!!".format(illust.user.id, illust.user.name))
        print(separator)



ids = making_illust_list(serch_tag)
download(ids)

使い方

pixivpyでログインできなくなったので、新しいログイン方法の紹介
これでログインしてください。

firt.py

#タグによるフィルター　書き方 target_tag = ["Fate/GrandOrder","FGO","FateGO","Fate/staynight"]
target_tag = [] #target_tag内に複数書くと、そのうち少なくとも一つあればダウンロード
target_tag2 = []#さらにtarget_tag2に書くとtarget_tagを満たし、かつtarget_tag2を満たすものだけダウンロード
extag = ["R-18"]#extag内のタグが一つでも入っていればダウンロードしない
#ブックマーク数によるフィルター、最小値を設定、0ならすべて
score=0
#閲覧数によるフィルター、最小値を設定、0ならすべて
view=1000

#画像を保存するディレクトリ
saving_direcory_path = "./img/"
#検索タグ
serch_tag = "初音ミク"
#何ページ分取得するか（1ページ30作品）
max_page = 10

ここの数値をいじって実行してください。
target_tagあたりは今回はあまり関係ないですが、元のコードを流用したため残っています。
詳しくは
pythonのpixivpyを使ってpixivから特定のユーザーの作品をすべて一括でダウンロードする（うごイラ含む）
を見てください。

• pytesseractの概要と使用方法についてメモする。

pytesseract 概要

• OCRツールTesseractのPythonラッパー。

• PillowやNumPyなどの形式で解析対象データを受け取ることが可能。

• コマンド呼び出しで実行。

インストール

• pipコマンドを使用し、インストールする。

pip install pytesseract

※必要に応じて解析データの読み込み用途でPillowなどもインストールする。

※tesseract本体も必要。Linux環境などではapt-get -y install tesseract-ocr tesseract-ocr-jpnでインストールしておく。

使用方法

• Pythonコード

from pytesseract import pytesseract
from PIL import Image

# 読み込み対象ファイルの指定
img = Image.open("./test.png, "r")
# tesseractコマンドのインストールパス
pytesseract.tesseract_cmd = "/usr/bin/tesseract"
# 文字列として出力できる。
ocr_result = pytesseract.image_to_string(img, lang="eng+jpn")

参考情報

• pytesseract

はじめに

はじめまして。
tit_BTCQASH と申します。

この記事では上位20%以上の性能が出る予測値をNumeraiにサブミットする方法を解説します。

Numeraiにステークしたいけど、毎週プログラミングなんてめんどくさい・プログラミングなんてできない、という人向けです。

Numeraiって何？という方は、私の過去記事( 日本語版 /英語版)をご参照ください。

データのダウンロードから予測の提出まで

本項ではNumeraiに予測値を提出する方法を説明します。
使用する計算モデルは公式 がGithubで頒布しているexample_model.pyです。
公式が配布しているサンプルモデルですが、下手な独自モデルよりも性能が良いです。また、Numeraiに提出する予測値を格納したCSVファイルも公式が毎週配布しています。

それでは、提出用ファイルのダウンロードと提出方法について説明いたします。

公式サイトの左側にあるDownload Dataを押し、Zipファイルを解凍すると、example_predictions.csvというファイルがあります。

example_predictions.csvを公式サイト左側にあるUpload Predictions（Download Dataの下）からアップロードすれば完了です。
（ね？簡単でしょう？）
本ファイルは毎週土曜日18:00(UTC)にダウンロードできるようになります。本ファイルを月曜日の14:30(UTC)までに提出すれば、その週から開催されるNumeraiトーナメントに参加できます。

example_model.pyの性能

example_model.pyから計算した予測値は毎週公式が提出してくれています。本モデルの性能について下図に示します。

本モデルの性能で一番見るべきところはCorr Reptationです。Corr Reptationとは20ラウンドのCorr平均値を表しており、Corr*100の値が週当たりの平均リターン[%]を表します。example_model.pyは週当たり2.83%の平均リターンがあり、年利換算で約150%とかなり高いです。
一方mmcはマイナスの値なので、他のモデルに対する優位性はありません。そのため、ステーキングをする場合はCorrのみがよいでしょう。

また、ランキングを見ると過去20ラウンドでは200-700位の間に位置することが分かります。現在4861モデルがNumeraiに存在するので、example_model.pyは上位4~15%程度の性能を持つことがわかります。
下手に独自モデルを作って予測値を提出するより、example_model.pyを出した方が高いリターンを得られるかもしれません。

さいごに

NMRのトークン価値は変動しますので、NMRトーナメントで優秀な成績を修め、NMRの枚数を増やしても、JPY建てでは損失が発生する可能性があります。

また、NMRを一度ステークすると、トーナメント期間中は引き出せないことも注意が必要です。

個人的には、下手に年利が高いDefiよりもリスクが低いと感じておりますが、リスク感度は個人個人で異なると思いますので、ご自身の判断でご参加ください。

チップ用
NMR:0x0000000000000000000000000000000000021d96

はじめに

よくプログラムの定期実行の方法として、Windowsではタスクの登録、Linuxではcronでの定期実行が紹介されるが、ローカル環境でこれらを行う場合は、もちろんPCを起動しておく必要がある。しかし、たかが1プログラムのために電気代を消費するのも勿体ないし、サーバーを借りて実行するほどでもない、といった需要に一番マッチするのがGithub Actionsである。
※この記事はGitに関しての記事ではないため、コミットやプッシュなどは理解しているものとして話を進める。

GitHub Actionsとは?

Github Actionsはその名の通りGitHubから提供されているワークフローを自動化するためのツールである。
具体的には、リポジトリの.github/workflowsの配下にYAML形式の実行制御ファイルを配置することによって利用することができる。Linuxで利用されるcronによる定期実行やレポジトリへのプッシュを検知して、ワークフローを実行することもできる。

公式ドキュメント -> https://docs.github.com/ja/actions/reference/workflow-syntax-for-github-actions

実行するトリガーとなるイベントを検知

リポジトリのmainブランチにプッシュされたときに実行

on:
  push:
    branches:
      - main

時間を設定して定期実行

on:
  schedule:
    - cron: '0 3 * * *'

※時間はUTCであることに注意が必要。上記の例では、1日に1回日本時間の12:00に実行。
cronの時間設定については↓の記事を参照
クーロン(cron)をさわってみるお

実行する環境の設定

jobs:
  build:
    runs-on: ubuntu-latest

他にもWindows ServerやMacOSなども使用できる。利用する場合は公式ドキュメントを参照されたい。

ワークフローの実行

ここからは主にLinuxで実行したいコマンドを書いていくことになる。(Linuxのコマンドの内容については解説しない)
今回は例としてPythonの環境を構築して、レポジトリ内のmain.pyを実行し、実行した結果をレポジトリに自動でプッシュするように設定する。

.github/workflows/actionstest.yaml

name: ActionsTest # GitHub Actionsにつける名前。任意で良い。

on:
  schedule:
    - cron: '0 3 * * *' # cronで定期実行

jobs:
  build:
    runs-on: ubuntu-latest # 最新のubuntu環境で下記のジョブを実行
    steps:
      - uses: actions/checkout@v2 # おまじない

      - name: Setup python # ワークフローのセクションごとに設定する名前。特に設定する必要はないが、どこでエラーが起きているかを把握する為にも設定しておいた方が良い。
        uses: actions/setup-python@v2 # Pythonのセットアップ
        with:
          python-version: "3.x" # Pythonのバージョン指定

      - name: Install dependencies # Pythonの依存環境のインストール
        run: | #このような書き方で複数行を一気に実行することができる。
          python -m pip install --upgrade pip
          pip install pandas
          pip install pytz
          pip install matplotlib
          pip install numpy

      - name: Run main.py # Pythonファイルの実行
        run: |
          python main.py

      - name: Commit and Push # 実行した結果をプッシュして変更をレポジトリに反映
        run: |
          git config user.name "YOUR NAME"
          git config user.email "YOUR MAIL"
          git add .
          git commit -m "Commit Message"
          git pull
          git push origin master

※最後のGitにコミットする前の名前とEmailは必須だそうだ。(この行を省いた状態でテストしていない...)

これだけで簡単にスクリプトを実行し、レポジトリに反映するスクリプトが出来る。ただし、月当たりの利用時間の制限があるので、サーバーとして運用するのは厳しいかと思われる。（無料枠では2000分/月）

その他の機能

上記で紹介したもののほかにも、環境変数を利用したり、特定のシェルを利用することもできる。これらの機能の詳細については、下記のドキュメントを参考にされたい。

公式ドキュメント -> https://docs.github.com/ja/actions/reference/workflow-syntax-for-github-actions

書いてること

ラズパイでラジコンを作ったので、その方法をここに残します。
専用アプリ不要で手持ちのスマホがコントローラーになります。

完成品はこちら ↓

サーボモータをラズパイで制御して、Webサイト経由で操作するのできた！ pic.twitter.com/7USpyYXTpx

— suo-takefumi (@zsipparu) February 28, 2021

構成図

全体の構成図です。

• ラズパイで4つのサーボモーターを制御
  • server.pyから制御信号を出す
  • ※Python3スクリプト
• index.htmlとserver.pyはSocket通信
• http://{ラズパイ}:8000/index.htmlを開いたスマホがコントローラーになる
• ＜制御の流れ概要＞
  • index.htmlから5パターンの信号(前進/後進/左回り/右回り/停止)を発信
  • server.pyがindex.htmlからの信号をもとにサーボモーターを制御

サーボモーターの配線はこちら

Raspberry Piの情報

$ lsb_release -a
No LSB modules are available.
Distributor ID: Raspbian
Description:    Raspbian GNU/Linux 10 (buster)
Release:        10
Codename:       buster

$ python3 -V
Python 3.7.3

スクリプト

スクリプトは全てラズパイの/home/pi/data/web配下に格納します。
ディレクトリ構成は以下のようになります。
ソケット通信については こちらのサイト を参考にさせて頂きました。

pi@raspi:~/data/web $ tree
.
├── images
│   ├── back.png
│   ├── go.png
│   ├── left.png
│   ├── right.png
│   └── stop.png
├── index.html
└── server.py

コントローラーになるindex.htmlのソースです。
index.htmlはSocket通信ではクライアント側になり、Socket通信のサーバー側となるserver.pyに対し5種類の信号(GO/BACK/LEFT/RIGHT/STOP)を送ります。

index.html

<!DOCTYPE html>
<html>/home/pi/data/web
<head>
  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
  <title>コントローラー</title>
  </head>
<body>
  <img onclick="sendMessage('GO')"    src="images/go.png" />
  <img onclick="sendMessage('BACK')"  src="images/back.png" />
  <img onclick="sendMessage('LEFT')"  src="images/left.png" />
  <img onclick="sendMessage('RIGHT')" src="images/right.png" />
  <img onclick="sendMessage('STOP')"  src="images/stop.png" />
  <br />
  <!-- 出力 area -->
  <textarea id="messageTextArea" rows="10" cols="50"></textarea>
  <script type="text/javascript">
    var webSocket = new WebSocket("ws://192.168.3.19:9998"); // ウェブサーバを接続
    var messageTextArea = document.getElementById("messageTextArea"); // ウェブサーバから受信したデータを出力するオブジェクトを取得
    // ソケット接続すれば呼び出す関数。
    webSocket.onopen = function(message){
      messageTextArea.value += "Server connect...\n";
    };
    // ソケット接続が切ると呼び出す関数。
    webSocket.onclose = function(message){
      messageTextArea.value += "Server Disconnect...\n";
    };
    // ソケット通信中でエラーが発生すれば呼び出す関数。
    webSocket.onerror = function(message){
      messageTextArea.value += "error...\n";
    };
    // ソケットサーバからメッセージが受信すれば呼び出す関数。
    webSocket.onmessage = function(message){
      // 出力areaにメッセージを表示する。
      messageTextArea.value += "Recieve From Server => "+message.data+"\n";
    };

    function sendMessage(argVal){
      messageTextArea.value += "Send to Server => "+ argVal +"\n";
      webSocket.send(argVal); // WebSocketでtextMessageのオブジェクトの値をサーバに送信
    }

    // 通信を切断する。
    function disconnect(){
      webSocket.close();
    }
  </script>
</body>
</html>

サーバー側でサーボモーターを制御するserver.pyのソースです。
server.pyはindex.htmlから受信した信号(GO/BACK/LEFT/RIGHT/STOP)に応じモーターを制御します。

server.py

import asyncio
import websockets
import RPi.GPIO as GPIO
import time

GPIO.setmode(GPIO.BCM)

wheelPin = {
    "FrontRight": "2",
    "FrontLeft" : "3",
    "RearRight" : "17",
    "RearLeft"  : "4"
  }

wheelPwm = {
    "FrontRight": "",
    "FrontLeft" : "",
    "RearRight" : "",
    "RearLeft"  : ""
}

## 角度をラジアンに変換
def exchDegreeToRadian(argDegree):
    return 2.5 + (12.0 - 2.5) / 180 * (argDegree + 90)

## サーボモータ制御
def wheelControl(argFlg, argTerm):
    global wheelPwm
    act01 = exchDegreeToRadian( -90 )
    act02 = exchDegreeToRadian(  90 )
    act03 = exchDegreeToRadian(   0 )
    if(argFlg == "GO"):
        wheelPwm["FrontRight"].ChangeDutyCycle( act01 )
        wheelPwm["FrontLeft" ].ChangeDutyCycle( act01 )
        wheelPwm["RearRight" ].ChangeDutyCycle( act01 )
        wheelPwm["RearLeft"  ].ChangeDutyCycle( act01 )
    elif(argFlg == "BACK"):
        wheelPwm["FrontRight"].ChangeDutyCycle( act02 )
        wheelPwm["FrontLeft" ].ChangeDutyCycle( act02 )
        wheelPwm["RearRight" ].ChangeDutyCycle( act02 )
        wheelPwm["RearLeft"  ].ChangeDutyCycle( act02 )
    elif(argFlg == "STOP"):
        wheelPwm["FrontRight"].ChangeDutyCycle( act03 )
        wheelPwm["FrontLeft" ].ChangeDutyCycle( act03 )
        wheelPwm["RearRight" ].ChangeDutyCycle( act03 )
        wheelPwm["RearLeft"  ].ChangeDutyCycle( act03 )
    elif(argFlg == "LEFT"):
        wheelPwm["FrontRight"].ChangeDutyCycle( act01 )
        wheelPwm["FrontLeft" ].ChangeDutyCycle( act02 )
        wheelPwm["RearRight" ].ChangeDutyCycle( act01 )
        wheelPwm["RearLeft"  ].ChangeDutyCycle( act02 )
    elif(argFlg == "RIGHT"):
        wheelPwm["FrontRight"].ChangeDutyCycle( act02 )
        wheelPwm["FrontLeft" ].ChangeDutyCycle( act01 )
        wheelPwm["RearRight" ].ChangeDutyCycle( act02 )
        wheelPwm["RearLeft"  ].ChangeDutyCycle( act01 )

    time.sleep(argTerm)

#####################################################
# ここから処理を開始

## サーボモータの初期設定
for wheel in wheelPin:
    print("wheel={0} , pin={1}".format( wheel, wheelPin[wheel] ) )
    PinNo = int(wheelPin[wheel]) # GPIO
    GPIO.setup(PinNo, GPIO.OUT)
    wheelPwm[wheel] = GPIO.PWM(PinNo, 50)
    wheelPwm[wheel].start(0.0)

##----------------------------------------##

# クライアント接続すると呼び出す
async def accept(websocket, path):
  while True:
    # クライアントからメッセージを待機
    data = await websocket.recv()
    print("receive : " + data)
    # クライアントでechoを付けて再送信
    #await websocket.send("echo : " + data)
    if data == "GO":
      wheelControl("GO", 0.1)
    elif data == "BACK":
      wheelControl("BACK", 0.1)
    elif data == "STOP":
      wheelControl("STOP", 0.1)
    elif data == "LEFT":
      wheelControl("LEFT", 0.1)
    elif data == "RIGHT":
      wheelControl("RIGHT", 0.1)

# WebSocketサーバー生成
start_server = websockets.serve(accept, "raspi203.local", 9998)

# 非同期でサーバを待機する。
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

以上で全ての準備が完了しました。

実行方法

2つコンソールを起動し、コンソール①ではWebサーバーを起動し、コンソール②ではserver.pyを実行します。

コンソール①

$ cd /home/pi/data/web
$ python3 -m http.server 8000

コンソール②

$ cd /home/pi/data/web
$ python3 server.py

実際の実行画面はこんな感じ

これでサービスが起動状態になりました。スマホなどでhttp://{ラズパイIP}:8000/index.htmlを開くとそのデバイスがコントローラーになります。

おわりに

今回は、とりあえずラズパイをラジコンにするまでを紹介しました。今後は、コントローラのUIをもっとリッチにするとか、ラジコンにカメラを搭載しコントローラ側に表示するとか、機能を盛り込んでいきます。

こんにちは！バイオインフォマティクス系オタク修士学生のroadricefieldです！AtCoder ABC193参加しました！

A問題

何円安くなっているかを引き算で計算してそれが定価の何％に当たるか計算して出力すればいいですね．「割合」は小学5年生の算数で習います．

（私の解答）

A, B = map(int, input().split())
print(((A-B)/A)*100)

B問題

まず問題設定より，A_iがX_i以上である店iでは，高橋くんがその店にたどり着いたときにはスヌケマシンは売り切れてしまっています．このことに気をつけてすべての店についてスヌケマシンを買えるのか，買えるのであれば価格はいくらなのかを調べて最安値を出力すればいいです．

（私の解答）

N = int(input())

ans = 10000000000

for _ in range(N):
  A, P, X = map(int, input().split())
  if A >= X: continue
  if P < ans: ans = P

if ans != 10000000000: print(ans)
else: print(-1)

私の解答ではどこの店でもスヌケマシンが買えない場合はansが最後に10000000000のままとなっていますのでそのようなときは-1を出力するように書いています．

C問題

変数ansをNとして，a^bと表すことのできる数を列挙していってNよりも小さいかどうかを判定していき，小さいならば順次ansから1を引いていって，a^bがNよりも大きくなってしまったところで処理を終了してその時のansを出力するという方針で解きました．

問題は「a^bと表すことのできる数を列挙」をどのように実現するかです．a, bそれぞれ2からスタートして1ずつ増やしながら列挙していけばよいでしょう．しかし，そのままやると途中で重複が起きてしまいます．たとえばa = 2, b = 4 のときa^b = 16ですがa = 4, b = 2のときもa^b = 16です．これに気をつけなければなりません．と，いうわけで一度出現した数を記録しておいて一度現れた数は飛ばしながら処理をすすめることにしましょう．aに代入する数がすでに出現している場合はbがどんな数でもa^bはすでに出現していることを利用するなどして以下のように書きました．

（私の解答）

N = int(input())

ans = N
done = set()

for i in range(2,10**5+5):
  if i**2 > N: break
  if i in done: continue
  j = 2
  while 1:
    if i**j > N: break
    else:
      done.add(i**j)
      ans -= 1
      j += 1

print(ans)

a^bで表せる数がさほど多くないので普通に間に合います．

D問題

これは高校の数学Aの確率ですね完全に．高橋くんと青木くんの裏向きのカードの組み合わせとしてあり得るすべての場合の数，(9*K*-8)(9*K*-9)で高橋くんが勝つ組み合わせの数を割ればいいです．ここで数字が同じ2枚のカードは異なるものとして扱わなければなりませんね．

まず最初に1~9のカードがそれぞれ何枚が表向きに場に出ていないカードとして残っているかを計算しておいて，それに気をつけながら高橋くん，青木くんの裏側のカードの数字のすべての組み合わせ（9 × 9 = 81通り）についてそれぞれ検討していきます．気をつけることとしては例えば高橋くんと青木くんの裏向きになっているカードの数が同じである場合を検討するとき，そもそもその数字のカードが表向きでないものとして2枚以上残っていなければそのような場合は存在しないことになります．高橋くんが勝つ数字の組み合わせとなる場合の数を計算していってその総和を(9*K*-8)(9*K*-9)で割ったものが答えになりますね．

（私の解答）

K = int(input())
S = input()
S = list(map(int, list(S[:4])))
T = input()
T = list(map(int, list(T[:4])))
ST = S + T

left_card = [K]*9

for i in range(1,10): left_card[i-1] -= ST.count(i)

bumbo = (9*K - 8)*(9*K - 9)
bunshi = 0

for i in range(1,10):
  if left_card[i-1] == 0: continue
  for j in range(1,10):
    if left_card[j-1] == 0: continue
    S.append(i)
    T.append(j)
    Takahashi = 0
    Aoki = 0
    for k in range(1,10):
      Takahashi += k*(10**S.count(k))
      Aoki += k*(10**T.count(k))
    if Takahashi > Aoki:
      if i == j and left_card[i-1] == 1:
        pass
      elif i == j and left_card[i-1] != 1:
        bunshi += left_card[i-1]*(left_card[i-1]-1)
      else:
        bunshi += left_card[i-1]*left_card[j-1]
    S = S[:4]
    T = T[:4]

print(bunshi/bumbo)

今回はアルゴリズムの力というか思い通りの処理を実装する力が問われるような問題でしたね（D問題までは）．て，いうかAtCoderってすごく難しくてやっとD問題まで安定して解けるようになってきた～と思ったら次に待ち受けるのはDよりはるかに難しいE問題ですからね！？時間足りひん！！Eを解けるようになるにはA~Dを解き切る時間を超大幅に短縮する鍛錬を積むか，解けるはずのC,D問題を後回しにしてEに果敢に立ち向かうかしかないですね......

目次

• 1.背景
• 2.構成/構築手順
  • 2-1.Lambdaの構築
    • 2-1-1.IAMロールの作成
    • 2-1-2.Lambda関数の作成
    • 2-1-3.ソースコードのデプロイ
    • 2-1-4.レイヤー作成
    • 2-1-5.タイムアウト値の延長
  • 2-2.APIによる収益情報取得
  • 2-3.EventBridgeによるトリガー定義
  • 2-4.S3バケットへの書き出し・読み込み
  • 2-5.Amazon SNSによるメール通知
• 3.実行結果
• 4.終わりに
• 5.更新履歴

1. 背景

　日々メインPCとは別にマイニングリグを稼働させて、NiceHashでマイニングをしているのですが、残高確認する際にNiceHashのダッシュボードを都度確認するのは面倒なので日次ジョブとして通知したいと思ってました。
　業務ではコーディングや、AWSに触れる機会が一切ないので、勉強がてらAWS上でスマホへメール通知するシステムを構築してみました。
　最重要なマイニングリグのヘルスチェックエラー通知はMySettingsから設定できる ¹ 仕様なので、今回は収益情報のみをメールの通知対象としました。

2. 構成/構築手順

システム構成は、AWS Lambdaを中心とした基本的なサーバレスアーキテクチャです。

処理の流れ
　1. EventBridge(CroudWatch Event)の日次実行cronがトリガーとなり、Lambda関数をキック
　2. Lambdaでは、外部APIからマイニング収益情報を取得
　3. S3バケットへ残高情報を書き込み、前日の残高情報を取得、残高の増減を算出
　4. Amazon SNSへ値をpublishしてスマホへメールを通知

2-1.Lambdaの構築

2-1-1.IAMロールの作成

AWSサービス間を連携するために新規IAMロールを作成し必要なポリシーをアタッチする
・IAMを起動し、ユースケースLambdaを選択し「次のステップ」をクリック

・S3バケットへ残高情報を読み書きするためにAmazonS3FullAccess、収益情報をメール通知するためにAmazonSNSFullAccessポリシーをロールにアタッチし「次のステップ」をクリック

・タグの追加は不要なので何も記入せず「次のステップ」をクリック
・ロール名は適当にNiceHash-Nortificationとして「ロールの作成」をクリック

2-1-2.Lambda関数の作成

呼び出されるLambda関数本体を作成する
・サービスからLambdaを起動し、以下のように入力し「関数の作成」をクリック

　関数名：「NiceHash-Nortification-Mail」
　ランタイム：「Python 3.6」#Python3系ならたぶんＯＫ
　アクセス権限：「NiceHash-Nortification」#作成したIAMロール

2-1-3.ソースコードのデプロイ

Lambdaで実行するプログラムをデプロイする
・以下4つのpythonファイルを新規に作成してコードをデプロイ

NiceHash-Nortification-Mail

NiceHash-Nortification-Mail/
├ lambda_function.py
├ nicehash.py
├ marketrate.py
└ s3inout.py

Lambdaで呼び出されるメインプログラム

lambda_function.py

import json
import datetime
import boto3

import nicehash
import marketrate
import s3inout
#Function kicked by AWS Lambda
def lambda_handler(event, context):
    client = boto3.client('sns')
    #Amazon SNS
    TOPIC_ARN = 'arn:aws:sns:xx-xxxx-x:xxxxxxxxxx:NiceHashSNS' # SNSのARNを指定
    msg = create_message()
    subject = "NiceHash-Mining 日次収益通知"
    #Send a notification message to Amazon SNS
    response = client.publish(
        TopicArn = TOPIC_ARN,
        Message = msg,
        Subject = subject
    )
#Function to get a nortification message
def create_message():
    #NiceHash API    
    host = 'https://api2.nicehash.com'
    organisation_id = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx' # hogehoge
    key = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx' # API Key Code
    secret = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx' # API Secret Key Code
    market='BTC'

    #S3 bucket
    bucket_name = '[bucket_name]'#hogehoge
    key_name = '[balance_filename]'#hogehoge

    #Get mining information from NiceHash API
    PrivateApi = nicehash.private_api(host, organisation_id, key, secret)
    accounts_info = PrivateApi.get_accounts_for_currency(market)
    balance_row = float(accounts_info['totalBalance'])

    #Get currency_to_JPY_rate from CoinGecko API
    TradeTable = marketrate.trade_table(market)
    rate = TradeTable.get_rate()
    balance_jpy = int(balance_row*rate)

    #S3 dealer
    S3dealer = s3inout.s3_dealer(bucket = bucket_name, key = key_name)
    pre_balance = int(S3dealer.read_from_s3_bucket())
    diff = balance_jpy - pre_balance
    S3dealer.write_to_s3_bucket(str(balance_jpy))

    #Nortification message
    time_text = "時刻: " + str(datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))))[:19]
    market_text = "仮想通貨： " + market
    rate_text = "単位仮想通貨価値: " + str(rate) + "円"
    balance_text = "現在の残高: " + str(balance_jpy) + "円"
    pre_balance_text = "昨日の残高: " + str(pre_balance) + "円"
    symbol = "+" if diff > 0 else ""
    diff_txt = "【日次収益: " + str(symbol) + str(diff) + "円】"
    mon_revenue = "推定月次収益: " + str(diff*30) + "円"
    ann_revenue = "推定年次収益: " + str(diff*365) + "円"
    msg = '\n'.join([time_text,market_text,rate_text,balance_text,pre_balance_text,diff_txt,mon_revenue,ann_revenue])
    return msg

NiceHash API （2-2で説明）

nicehash.py

from datetime import datetime
from time import mktime
import uuid
import hmac
import requests
import json
from hashlib import sha256
import optparse
import sys
class private_api:
  def __init__(self, host, organisation_id, key, secret, verbose=False):
      self.key = key
      self.secret = secret
      self.organisation_id = organisation_id
      self.host = host
      self.verbose = verbose
  def request(self, method, path, query, body):
      xtime = self.get_epoch_ms_from_now()
      xnonce = str(uuid.uuid4())
      message = bytearray(self.key, 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray(str(xtime), 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray(xnonce, 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray(self.organisation_id, 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray(method, 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray(path, 'utf-8')
      message += bytearray('\x00', 'utf-8')
      message += bytearray(query, 'utf-8')
      if body:
          body_json = json.dumps(body)
          message += bytearray('\x00', 'utf-8')
          message += bytearray(body_json, 'utf-8')
      digest = hmac.new(bytearray(self.secret, 'utf-8'), message, sha256).hexdigest()
      xauth = self.key + ":" + digest
      headers = {
          'X-Time': str(xtime),
          'X-Nonce': xnonce,
          'X-Auth': xauth,
          'Content-Type': 'application/json',
          'X-Organization-Id': self.organisation_id,
          'X-Request-Id': str(uuid.uuid4())
      }
      s = requests.Session()
      s.headers = headers
      url = self.host + path
      if query:
          url += '?' + query
      if self.verbose:
          print(method, url)
      if body:
          response = s.request(method, url, data=body_json)
      else:
          response = s.request(method, url)
      if response.status_code == 200:
          return response.json()
      elif response.content:
          raise Exception(str(response.status_code) + ": " + response.reason + ": " + str(response.content))
      else:
          raise Exception(str(response.status_code) + ": " + response.reason)

  def get_epoch_ms_from_now(self):
      now = datetime.now()
      now_ec_since_epoch = mktime(now.timetuple()) + now.microsecond / 1000000.0
      return int(now_ec_since_epoch * 1000)

  def algo_settings_from_response(self, algorithm, algo_response):
      algo_setting = None

      for item in algo_response['miningAlgorithms']:
          if item['algorithm'] == algorithm:
              algo_setting = item
      if algo_setting is None:
          raise Exception('Settings for algorithm not found in algo_response parameter')
      return algo_setting

  def get_accounts(self):
      return self.request('GET', '/main/api/v2/accounting/accounts2/', '', None)

  def get_accounts_for_currency(self, currency):
      return self.request('GET', '/main/api/v2/accounting/account2/' + currency, '', None)

  def get_withdrawal_addresses(self, currency, size, page):
      params = "currency={}&size={}&page={}".format(currency, size, page)
      return self.request('GET', '/main/api/v2/accounting/withdrawalAddresses/', params, None)

  def get_withdrawal_types(self):
      return self.request('GET', '/main/api/v2/accounting/withdrawalAddresses/types/', '', None)

  def withdraw_request(self, address_id, amount, currency):
      withdraw_data = {
          "withdrawalAddressId": address_id,
          "amount": amount,
          "currency": currency
      }
      return self.request('POST', '/main/api/v2/accounting/withdrawal/', '', withdraw_data)

  def get_my_active_orders(self, algorithm, market, limit):
      ts = self.get_epoch_ms_from_now()
      params = "algorithm={}&market={}&ts={}&limit={}&op=LT".format(algorithm, market, ts, limit)
      return self.request('GET', '/main/api/v2/hashpower/myOrders', params, None)

  def create_pool(self, name, algorithm, pool_host, pool_port, username, password):
      pool_data = {
          "name": name,
          "algorithm": algorithm,
          "stratumHostname": pool_host,
          "stratumPort": pool_port,
          "username": username,
          "password": password
      }
      return self.request('POST', '/main/api/v2/pool/', '', pool_data)

  def delete_pool(self, pool_id):
      return self.request('DELETE', '/main/api/v2/pool/' + pool_id, '', None)

  def get_my_pools(self, page, size):
      return self.request('GET', '/main/api/v2/pools/', '', None)

  def get_hashpower_orderbook(self, algorithm):
      return self.request('GET', '/main/api/v2/hashpower/orderBook/', 'algorithm=' + algorithm, None )

  def create_hashpower_order(self, market, type, algorithm, price, limit, amount, pool_id, algo_response):
      algo_setting = self.algo_settings_from_response(algorithm, algo_response)
      order_data = {
          "market": market,
          "algorithm": algorithm,
          "amount": amount,
          "price": price,
          "limit": limit,
          "poolId": pool_id,
          "type": type,
          "marketFactor": algo_setting['marketFactor'],
          "displayMarketFactor": algo_setting['displayMarketFactor']
      }
      return self.request('POST', '/main/api/v2/hashpower/order/', '', order_data)

  def cancel_hashpower_order(self, order_id):
      return self.request('DELETE', '/main/api/v2/hashpower/order/' + order_id, '', None)

  def refill_hashpower_order(self, order_id, amount):
      refill_data = {
          "amount": amount
      }
      return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/refill/', '', refill_data)

  def set_price_hashpower_order(self, order_id, price, algorithm, algo_response):
      algo_setting = self.algo_settings_from_response(algorithm, algo_response)
      price_data = {
          "price": price,
          "marketFactor": algo_setting['marketFactor'],
          "displayMarketFactor": algo_setting['displayMarketFactor']
      }
      return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/updatePriceAndLimit/', '',
                          price_data)

  def set_limit_hashpower_order(self, order_id, limit, algorithm, algo_response):
      algo_setting = self.algo_settings_from_response(algorithm, algo_response)
      limit_data = {
          "limit": limit,
          "marketFactor": algo_setting['marketFactor'],
          "displayMarketFactor": algo_setting['displayMarketFactor']
      }
      return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/updatePriceAndLimit/', '',
                          limit_data)

  def set_price_and_limit_hashpower_order(self, order_id, price, limit, algorithm, algo_response):
      algo_setting = self.algo_settings_from_response(algorithm, algo_response)
      price_data = {
          "price": price,
          "limit": limit,
          "marketFactor": algo_setting['marketFactor'],
          "displayMarketFactor": algo_setting['displayMarketFactor']
      }
      return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/updatePriceAndLimit/', '',
                          price_data)

  def get_my_exchange_orders(self, market):
      return self.request('GET', '/exchange/api/v2/myOrders', 'market=' + market, None)

  def get_my_exchange_trades(self, market):
      return self.request('GET','/exchange/api/v2/myTrades', 'market=' + market, None)

  def create_exchange_limit_order(self, market, side, quantity, price):
      query = "market={}&side={}&type=limit&quantity={}&price={}".format(market, side, quantity, price)
      return self.request('POST', '/exchange/api/v2/order', query, None)

  def create_exchange_buy_market_order(self, market, quantity):
      query = "market={}&side=buy&type=market&secQuantity={}".format(market, quantity)
      return self.request('POST', '/exchange/api/v2/order', query, None)

  def create_exchange_sell_market_order(self, market, quantity):
      query = "market={}&side=sell&type=market&quantity={}".format(market, quantity)
      return self.request('POST', '/exchange/api/v2/order', query, None)

  def cancel_exchange_order(self, market, order_id):
      query = "market={}&orderId={}".format(market, order_id)
      return self.request('DELETE', '/exchange/api/v2/order', query, None)

if __name__ == "__main__":
 parser = optparse.OptionParser()
 parser.add_option('-b', '--base_url', dest="base", help="Api base url", default="https://api2.nicehash.com")
 parser.add_option('-o', '--organization_id', dest="org", help="Organization id")
 parser.add_option('-k', '--key', dest="key", help="Api key")
 parser.add_option('-s', '--secret', dest="secret", help="Secret for api key")
 parser.add_option('-m', '--method', dest="method", help="Method for request", default="GET")
 parser.add_option('-p', '--path', dest="path", help="Path for request", default="/")
 parser.add_option('-q', '--params', dest="params", help="Parameters for request")
 parser.add_option('-d', '--body', dest="body", help="Body for request")
 options, args = parser.parse_args()
 private_api = private_api(options.base, options.org, options.key, options.secret)
 params = ''
 if options.params is not None:
     params = options.params
 try:
     response = private_api.request(options.method, options.path, params, options.body)
 except Exception as ex:
     print("Unexpected error:", ex)
     exit(1)
 print(response)
 exit(0)

CoinGecko API （2-2で説明）

marketrate.py

import requests
import json
class trade_table:
    def __init__(self, market="BTC"):
        #currency-name conversion table
        self.currency_rename_table = {'BTC':'Bitcoin','ETH':'Ethereum','LTC':'Litecoin',
                                      'XRP':'XRP','RVN':'Ravencoin','MATIC':'Polygon',
                                      'BCH':'Bitcoin Cash','XLM':'Stellar','XMR':'Monero','DASH':'Dash'}
        self.market = self.currency_rename_table[market]

    def get_rate(self):
        body = requests.get('https://api.coingecko.com/api/v3/coins/markets?vs_currency=jpy')
        coingecko = json.loads(body.text)
        idx = 0
        while coingecko[idx]['name'] != self.market:
            idx += 1
            #Escape of illegal market_currency name
            if idx > 100:
                return "trade_table_err"
        #market-currency_to_JPY_rate
        else:
            return int(coingecko[idx]['current_price'])

S3バケットへの収益情報の読み込み・書き出し（2-3で説明）

s3inout.py

import boto3
class s3_dealer:
  def __init__(self, bucket = 'nice-hash-balance', key = 'balance_latest.txt'):
    self.bucket = bucket
    self.key = key

  #Get balance of the previous day
  def read_from_s3_bucket(self):
    S3 = boto3.client('s3')
    res = S3.get_object(Bucket=self.bucket, Key=self.key)
    body = res['Body'].read()
    return body.decode('utf-8')

  #Export balance
  def write_to_s3_bucket(self, balance):
    S3 = boto3.resource('s3')
    obj = S3.Object(self.bucket, self.key)
    obj.put(Body=balance)

2-1-4.レイヤー作成

必要なモジュールをLambdaのレイヤーに取り込む
・2-1-3 記載のソースをデプロイしただけで実行するとrequestsモジュールが読み込めず以下エラーが発生してしまうため、外部モジュールをLayersへ定義する

{
  "errorMessage": "Unable to import module 'lambda_function': No module named 'requests'",
  "errorType": "Runtime.ImportModuleError"
}

・レイヤーファイルを作成するために、EC2でAmazon Linux AMIから新規インスタンスを作成する
・2-1-1の手順で、EC2のロールに対してS3のアクセスポリシーをアタッチ
　※インターネット環境に接続されたWSLやUbuntu等のUNIXマシンであれば何でもOK
・EC2インスタンスへコンソール接続し、以下CLIコマンドを打鍵してレイヤーファイルを作成する

ec2-user

[ec2-user@ip-xxx-xx-xx-xxx ~]$ su -
[root@ip-xxx-xx-xx-xxx ~]# mkdir layer/
[root@ip-xxx-xx-xx-xxx ~]# cd layer
[root@ip-xxx-xx-xx-xxx ~]# yum -y install gcc gcc-c++ kernel-devel python-devel libxslt-devel libffi-devel openssl-devel
[root@ip-xxx-xx-xx-xxx ~]# yum -y install python-pip
[root@ip-xxx-xx-xx-xxx ~]# pip install -t ./ requests
[root@ip-xxx-xx-xx-xxx ~]# cd ../
[root@ip-xxx-xx-xx-xxx ~]# zip -r Layer.zip layer/

・レイヤーファイルを、S3バケットへアップロード

ec2-user

[root@ip-xxx-xx-xx-xxx ~]# chmod 777 Layer.zip
[root@ip-xxx-xx-xx-xxx ~]# aws s3 cp Layer.zip s3://layerzip-s3

・S3でEC2からアップロードしたレイヤーファイルのオブジェクトURLを取得

・LambdaでS3のオブジェクトURLから名前を適当にImportRequestsとしてレイヤーを作成

・Lambdaで「レイヤーの追加」をクリック

・カスタムレイヤーから作成したImportRequestsを読み込む

2-1-5.タイムアウト値の延長

タイムアウトエラーを回避するためにタイムアウト値を変更する
・Lambdaはデフォルトだと、メモリ：128MB、タイムアウト：3秒になっているため、タイムアウトのみ「3秒 ⇒ 5秒」へ変更する

2-2.APIによる収益情報取得

外部APIからマイニング収益情報を取得する
・LambdaとNiceHash APIを連携するために、NiceHashへログインしてMySettingsからAPI Keysを発行する

・NiceHash API（nicehash.py）で収益情報を取得するために、lambda_function.pyの対象箇所に発行したAPI Keys、組織IDを入力する

lambda_function.py

#NiceHash API    
host = 'https://api2.nicehash.com'
organisation_id = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx' # hogehoge
key = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx' # API Key Code
secret = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx' # API Secret Key Code

・NiceHash APIのみでは、円相場の情報は取得できないため、別の外部API CoinGecko API（marketrate.py）を呼び出して、各仮想通貨市場の日本円相場を取得する。

2-3.EventBridgeによるトリガー定義

日次ジョブとしてLambdaをキックするためのトリガーを定義する
・Lambdaから「トリガーの追加」をクリック
・EventBridgeを選択し、以下のように入力し「追加」をクリック
　※AWSでcronを定義する際は、crontabとの違いや時差を考慮する必要があるため注意する ²

ルール：「新規ルールの作成」
ルール名：DailyTrigger
ルールタイプ：スケジュール式
スケジュール式：cron(0 15 * * ? *) # 毎日0:00に実行するcron

2-4.S3バケットへの書き出し・読み込み

前日の収益との比較を行うため、S3バケットのファイルに対して書き出し・読み込みを行う
・S3バケットbucket_nameを作成して、前日の残高(円)を整数で記載したダミーファイルbalance_filename.csvを予め格納しておく

bucket_name/balance_filename.csv

28583

・LambdaとS3のサービス間で連携するために、lambda_function.pyの対象箇所を編集する

lambda_function.py

#NiceHash API    
#S3 bucket
bucket_name = '[bucket_name]' # S3バケット名
key_name = '[balance_filename.csv]' # 残高情報が記載されたファイル名

2-5.Amazon SNSによるメール通知

取得した収益情報をメール通知するために、「SNSとメール」「LambdaとSNS」の連携を行う
・Amazon SNSを起動し、適当にNiceHashSNSとしてトピックを作成する
・作成したトピックNiceHashSNSのARNをコピーする

・「SNSとメール」を連携するために、作成したトピックNiceHashSNSを開き、「サブスクリプションの作成」をクリック

・プロトコル：Eメール、エンドポイント：通知したいメールアドレスを指定し、「サブスクリプションの作成」をクリック
・サブスクリプションを作成すると、指定したメールアドレスに通知メールが届くので、挿入されたリンクへアクセスしサブスクリプションのアクティベーションを行う
・「LambdaとSNS」を連携するために、lambda_function.pyの対象箇所を編集する

lambda_function.py

#Amazon SNS
TOPIC_ARN = 'arn:aws:sns:xx-xxxx-x:xxxxxxxxxx:NiceHashSNS' # コピーしたSNSのARNを記載

3. 実行結果

・毎日0:00になるとEventBridgeがLambdaをキックして、日次の通知メールが来るようになりました。

4. 終わりに

・APIで情報取得している割に、通知はメールというのは構成的にビミョかったので、この後にLINE APIを活用してLINE通知する構成に変更しました。次回は、LINE通知する構成について投稿予定です。

5. 更新履歴

ver. 1.0 初版投稿 2021/02/28

---

1. 【2018年2月】マイニングするならNicehashの通知設定をしておこう！ ↩

2. Amazon CloudWatch Events で cron 式を使う場合は時差に気をつける ↩