Blockly Custom Blocks 前回（Blockly Python Code Generator(2)）に続き、今回は具体的にカスタマイズブロックを作っていきます。googleのガイドを参照して作成します。 Custom Blocks 目標 Pythonを使い、SeleniumでEdge起動し任意のURLを開くまでを作ります。 将来的にはテスト自動駆動をBlocklyで作ることができれば良いかな、と。 準備 EdgeをSeleniumで起動するための環境は先に作っておきます。 インストール Pythonの環境も設定します。pipでSeleniumを有効にしておいてください。 pip install selenium webdriver_managerも使用します。これは自動でwebdriverをインストールするためのモジュールです。 pip install webdriver_manager ただし、現状ではログ処理部分にバグがあるため、そのままでは動作しません。 Pullリクエストは出ているので、将来的には修正されると思われます。 それまでは、自分でパッチをあてることになります。 Fixes edge and IE driver initialization issue テスト 動作するかどうかをスクリプトで試してみます。 edge_test.py from selenium import webdriver from webdriver_manager.microsoft import EdgeChromiumDriverManager executable_path = EdgeChromiumDriverManager().install() driver = webdriver.Edge(executable_path) driver.get("http://www.google.co.jp/") ブラウザが開けることを確認します。 Blockly Custom Blocks 作成 実際にカスタマイズBlockを作っていきます。 Blockの定義 Blocklyでブロックを作ることができます。 Blockly Developer Toolsか、ローカルのBlockly DemoのBlockly Developer Toolsを開きます。 開始ブロック Seleniumを実行するためにはimportが必要です。 これを実装するためのブロックを作成します。 nameを「selenium_start」に変更します。 入力は無いため、inputsにはdummy inputを設定し、text「開始」を接続します。 接続は下のみのため、「bottom connection」にします。 右側に定義用のソースが作成されているのが見えます。 次のブロック作成前にSaveをクリックして保存するのを忘れないようにして下さい。 Edge開始ブロック Edgeを開始するためのブロックを作成します。 nameを「selenium_edge_start」に変更します。 入力は無いため、inputsにはdummy inputを設定し、text「Edge開始」を接続します。 接続は上下必要なため、「top + ottom connection」にします。 開くブロック URLを開くためのブロックを作成します。 nameを「selenium_get」に変更します。 URLを文字列で受け取るため、inputsにはvalue inputを設定します。 text「開く」を接続します。 接続できるタイプを文字列に限定するため、typeはanyからStringに変更します。 これも接続は上下必要なため、「top + ottom connection」にします。 一通り作成したら、Download Block Librayで保存しておいて下さい。 Block Exporter 作成したらBlock Exporterタブに切り替えます。 左側に作成したブロックがありますので、すべて選択してください。 中央のExport Settingsにそれぞれファイル名を設定し、Exportをクリックして下さい。 定義ファイルが出力されます。 Workspace Factory Workspace Factoryに切り替えます。 Toolboxに作成したブロックを並べ、表示確認をします。 Exportをクリックすると、参考ソースが出力されます。 コード追加 CustomBlockPython.jsはコード出力ための定義ですが、現状ではダミーコードが入っていますので、それを実際に出力するコードへと変更します。 CustomBlockPython.js Blockly.Python['selenium_start'] = function(block) { var code = 'from selenium import webdriver\n'; code += 'from webdriver_manager.microsoft import EdgeChromiumDriverManager\n' return code; }; Blockly.Python['selenium_edge_start'] = function(block) { var code = 'executable_path = EdgeChromiumDriverManager().install()\n'; code += 'driver = webdriver.Edge(executable_path)\n' return code; }; Blockly.Python['selenium_get'] = function(block) { var value_url = Blockly.Python.valueToCode(block, 'url', Blockly.Python.ORDER_ATOMIC); var code = 'driver.get(' + value_url +')\n'; return code; }; 前回（Blockly Python Code Generator(2)）のソースを元にカスタマイズブロックを表示できるように改造していきます。 ブロック定義ファイル 作成したCustomBlockDefine.jsとCustomBlockPython.jsは前回のindex.htmlと同じディレクトリに配置します。 ToolBox カスタマイズブロックをメニューに追加します。 ntoolbox.xml(追加分) <category name="Custom"> <block type="selenium_start"></block> <block type="selenium_edge_start"></block> <block type="selenium_get"></block> </category> HTML カスタマイズデータを読み込む処理をHTMLに追加します。 index.html(追加分) <script src="./CustomBlockJson.js"></script> <script src="./CustomBlockPython.js"></script> 実行 実際に動作させてみます。 出力されたPythonコードを実際に実行した結果、Edgeが起動しました。

備忘録用のコードです。 test_perspective.py # -*- coding: utf-8 -*- """ Created on Fri Jan 28 21:37:16 2022 @author: komin """ # モジュール import cv2 import numpy as np # マウスイベント def mouse_click(event, x, y, flags, param): if (event == cv2.EVENT_LBUTTONDOWN) and (len(pts) < 4): origen = len(pts) pts.append((x,y)) cv2.drawMarker(img,(x,y), (0, 255, 0), thickness=5) # 画像サイズ取得 def img_size(img): h, w, c = np.shape(img) size_array = [(0, 0), (w, 0), (w, h), (0, h)] return size_array # 本文 if __name__ == '__main__': pts = [] img = cv2.imread("lenna.jpg") cv2.namedWindow("test") cv2.setMouseCallback("test", mouse_click) while True: cv2.imshow("test", img) k = cv2.waitKey(1) & 0xff if k == ord("q"): break elif (k == ord("w")) and ((len(pts) > 3)): # 変換前後の対応点を設定 set_points = np.float32(pts) tra_points = np.float32(img_size(img)) # 変換マトリクスと射影変換 trans_matrix = cv2.getPerspectiveTransform(set_points, tra_points) image_origen = cv2.imread("lenna.jpg") ih,iw,ic = np.shape(image_origen) i_trans = cv2.warpPerspective(image_origen, trans_matrix, (iw, ih)) img = cv2.hconcat([img,i_trans]) cv2.destroyAllWindows()

これは何？ 先日まで Mid 2015 の 15 inch MacBook Pro (Core i7 クアッド / 2.2 GHz) を使っていた。 先日 MacBook Pro 14 inch (M1 非Max) を手に入れたんだけど、あんまり速くないなと思うことがあったので、今日も楽しいマイクロベンチマーク。 計算内容 ruby で書くと短くていいね。 ruby N=10000 r=(1..N).max_by{ |x| ((N-x)**x/7) % 6074001001 } p r こういう内容。なんの意味もない。 出力は 8663 となれば正解。 これを、go, java, c++ with boost (clang, gcc), ruby, python3, node で試した。 以降、グラフで出てくる "m1", "rosetta", "amd64" の意味は下表のとおり。 記号 実行ハードウェア バイナリ m1 MacBook Pro 14 inch (M1 非Max) arm64 rosetta MacBook Pro 14 inch (M1 非Max) x86_64 amd64 MacBook Pro (Core i7, Mid 2015) x86_64 コンパイルするチーム go, java, c++ with boost (clang, g++-11)。 各コンパイラは下記の通り go version go1.17.5 darwin/arm64 openjdk 17.0.1 2021-10-19 LTS Apple clang version 13.0.0 (clang-1300.0.29.30) g++-11 (Homebrew GCC 11.2.0_3) 11.2.0 java と clang の rosetta はサボった。 結果は下記の通り。 time コマンドの real の値を出しているので棒が短いほど速い。 ちなみに、 real なのは並列実行を優遇するため。実際、 Java は user が real の 1.5倍ぐらいある。 結果は下図。 目盛りを見ると分かる通り、 go が速い。意外と clang が M1 を使いこなせてない感じ。 全体的にはまあそうだよねという結果だと思う。 コンパイルしないチーム 続いて、 ruby, python3, node。 各環境は下記の通り ruby 3.1.0p0 (2021-12-25 revision fb4df44d16) [arm64-darwin21] / for m1 ruby 3.0.3p157 (2021-11-24 revision 3fb7d2cadc) [x86_64-darwin21] / for rosetta ruby 3.1.0p0 (2021-12-25 revision fb4df44d16) [x86_64-darwin21] / for amd64 Python 3.9.10 (main, Jan 15 2022, 11:40:53) / for m1 Python 3.9.10 (main, Jan 15 2022, 11:48:04) / for rosetta Python 3.9.10 (main, Jan 15 2022, 11:48:04) / for amd64 node v17.3.0 / for m1 and rosetta node v17.3.1 / for amd64 なんか ruby と node のバージョンが合ってないけど気にしない。 結果は下図。 こちらはわりと思いがけなかった。 node はまあまあそうだよねという内容。m1 と amd64 の差はもうちょっとあってもいいかなと思うけど。 python3 は、三者ほぼ同タイム。 そして ruby は m1 が一番遅いという意外な展開。よく見てみると、m1 が遅いというより、amd64 が速すぎる。amd64 の中では go と並んでほぼ最速。m1 が遅いと書いたもののそれは ruby 内の比較の話。m1 内での比較だと Java と同等、clang より速い。node には負けるけど。 ruby や python で m1 がふるわないのは、おそらく、x86_64 バイナリは SSEとかをたっぷり使っていて、ARM64 バイナリは NEONとかを使いこなせてないんだろうと想像する。調べてないので想像するだけ。 まとめ M1 ネイティブでも rosetta 2 より遅いこともあるよ。 とはいえ。ほとんどの場合は M1 ネイティブは速いし、今遅いものもそのうち早くなるんじゃないかと思うよ（思うだけ）。

import pandas as pd df = pd.DataFrame([[0,1,2,1,1], [1,2,4,6,7], [2,3,6,8], [3,4,8,3]]) df.to_excel('test.xlsx', sheet_name='Sheet1', index=False, header=False) import openpyxl from openpyxl.chart import LineChart, Reference wb = openpyxl.load_workbook('test.xlsx') for sheet_name in wb.sheetnames: sheet = wb[sheet_name] # 秒数の列に変換 last_row = sheet.max_row last_col = sheet.max_column # 各データで何行目まで格納されているか calced_last_rows = [] for col in sheet.iter_cols(): for i, cell in enumerate(col): # print(si) if cell.value is None: calced_last_rows.append(i) break if i == last_row-1: calced_last_rows.append(i+1) for i in range(last_row): sheet.cell(row=i+1, column=1).value *= 0.1 sheet.cell(row=last_row+2, column=1).value = "最終距離" for i in range(last_col): if(i != 0): sheet.cell(row=last_row+2, column=i+1).value = sheet.cell(row=calced_last_rows[i], column=i+1).value sheet.cell(row=last_row+3, column=1).value = "実行時間" for i in range(last_col): if(i != 0): sheet.cell(row=last_row+3, column=i+1).value = sheet.cell(row=calced_last_rows[i], column=1).value sheet.cell(row=last_row+4, column=1).value = "単位m時間" sheet.cell(row=last_row+4, column=2).value = "" #グラフの設定 values = Reference(sheet, min_col=2, min_row=1, max_row=last_row, max_col=last_col) categories = Reference(sheet, min_col=1, min_row=1, max_col=1, max_row=last_row) chart = LineChart() chart.add_data(values) chart.set_categories(categories) sheet.add_chart(chart, 'A' + str(last_row + 5)) wb.save('test.xlsx') '''

本記事はProjectEulerの「100番以下の問題の説明は記載可能」という規定に基づいて回答のヒントが書かれていますので、自分である程度考えてみてから読まれることをお勧めします。 問題 68. 魔法5角リング 原文 Problem 68: Magic 5-gon ring 問題の要約 図 １のような図形のマルに1~10の数字を入れで以下の条件を満たすもので答えの16桁が最大になるものを求めよ 直線の3個のマルの数字の合計はすべて同じ 答えは、それぞれの3個の数字を外側から並べ、それをさらに外側の数の一番小さい数のものから時計回りに列挙したもので答えを表す。例えば図 2の3角魔法リングでは432:621:513となり、これが最大となる。 【図 １】 【図 2】 問題の考え方 なかなか、ややこしい問題なのでステップを踏んで考えて行きたいと思います。 5角魔法リングでは二けたの10が含まれ、これが内側のリングにあると答えに2回現れるので17桁になってします。したがって10は外側にある。 答えの16桁の最も大きい桁の数は外のリングの最小値、これを最大にするためには10以外の大きい数字「6,7,8,9」が外のリングに来て6が16桁の最初の数になる。 その6からなる3つの数の2番目が答えの16桁の2桁目になるので、内のリングの最大値5を選びます。 外のリングが「6,7,8,9,10」、内のリングが「1,2,3,4,5」とすると16桁の合計は内のリングは2回登場するので(6+7+8+9+10)+2*(1+2+3+4+5)=70。したがって3つの数の和は14になる。 6からなる3つの数の3つ目は14-6-5=3。すると内のリングの残りの3つは「1,3,4」 内のリングの各辺の和は{4,5,6,7,8}にならなければならないので、「5,3,1,4,2」が来まり、外のリング「6,10,9,8,7」も来まる。 これを答えの16桁にすると653:1031:914:842:725になる。 というように解いていくとプログラムを書かなくても紙と鉛筆で答えが出せてしまいました。

本記事はProjectEulerの「100番以下の問題の説明は記載可能」という規定に基づいて回答のヒントが書かれていますので、自分である程度考えてみてから読まれることをお勧めします。 問題 67. 経路の合計の最大値 （その２） 原文 Problem 67: Maximum path sum II 問題の要約：triangle.txtの100行のピラミッド型の数の上から下まで数字をたどった経路の合計の最大値を求めよ これはProblem 18: 経路の合計の最大値 （その１）の拡大版と言えます。アルゴリズムはProblem 18と同じで良いので。データを同様な配列に入れればプロクラムにはそのままでOKです。 for y in range(len(triarr)-2,-1,-1): for x in range(y+1): triarr[y][x] += max(triarr[y+1][x],triarr[y+1][x+1]) print(f"Answer: {triarr[0][0]}") ファイルから読んで配列に入れるプロクラム。 # show upload dialog from google.colab import files uploaded = files.upload() #---- read numbers from file f = open("p067_triangle.txt") triNums = f.readlines() f.close() triarr = [list(map(int,row.split())) for row in triNums] print(triarr[:20]) （開発環境：Google Colab）

はじめに 去年の9月ごろ、社内にて営業や人事といった技術以外の部署の人間を巻き込んで、 リテラシー向上のためのIT講習会を行いました。 その際の事前質問で、今流行りの言語を教えて欲しいと言われたので、その時の内容です。 眠らせておくのも勿体無かったのでQiitaにも載せておきたいと思います。 ※この記事には、私の独断と偏見が結構含まれています。 資料 30分という枠の中で、Zoomの画面共有で資料のスライドを進めながら質問や解説を入れていたので、 少し説明が足りない＆分かりづらいところもあるかもしれません。 今流行りのプログラミング言語とは なにをもって流行りと判断するか？ ソースコードのバージョン管理システム GitHubの使用ランキング(2021)によると 1位.JavaScript 2位.Python 3位.Java 4位.TypeScript 5位.C# 6位.PHP 7位.C++ 8位.C 9位.Shell 10位.Ruby これらは業務での使用率が高い言語と言えるが、良い意味で枯れている(比較的安定した)言語も多い。 枯れている言語が果たして流行っていると言えるのだろうか？ 広辞苑によると、「流行」とは、 ①流れ行くこと。 ②急にある現象が世間一般にひろくゆきわたること、特に、衣服・化粧・思想などの様式が一時的にひろく行われること、はやり。 ③(芭蕉の用語)不易 流行(ふえきりゅうこう) と定義されている。 Qiitaのプログラミング記事の 月間Post数(2021/9)によると 1位.Python 879post 2位.JavaScript 503post 3位.Ruby (Rails含む) 440post 4位.Swift 246post 今、話題になっている言語はこれらと言える。 JavaScriptに関しては、Node.jsの存在もありフロントエンド、バックエンド共に記事がpostされており、 Pythonに関しては機械学習やデータ分析系の記事が多い。 今回は記事のPost数が一番多い、Pythonを流行っている言語として紹介する。 どんな特徴があるのか？ Python言語の特徴として 1.シンプルな構文で汎用性が高い。 2.フレームワークやライブラリといったOSSも豊富。 3.人工知能を始めとした様々な分野で使われている。 1.シンプルな構文で汎用性が高い ・「テキスト読み取り処理」を他言語と比較すると C言語の場合 #include #define BUFF_SIZE 256 int main(void) { FILE *fp; char line[BUFF_SIZE]; fp = fopen(“sample.txt”, “rb”); if (fp == NULL) { return -1; } while (fgets(line, sizeof(line), fp) != NULL) { printf(“%s\n”, line); } fclose(fp); return 0; } Pythonの場合 with open(“sample.txt”, “r”) as f: for line in f: print(line) またPythonは、ブロックをインデントで指定する「Off-side Rule（オフサイドルール）」を採用しているので、他人が書いたコードも比較的読みやすいと言われている。 2.フレームワークやライブラリといったOSSも豊富 ・Django（ジャンゴ）/ FW Webシステム開発用のフレームワークで、ユーザ認証やRSSフィードといった基本的なWebの機能が備わっている。 ・Flask（フラスク）、bottle（ボトル）/ FW WSGI（Web Server Gateway Interface）であり、Webサーバへのリクエストを複数のWebアプリケーションを利用して処理するシステム。 ・Anaconda（アナコンダ）/ ライブラリパッケージ 大規模データ処理や予測分析などで使用される各種データサイエンスライブラリをインストール出来る。 補足 WSGIは、PythonでWebアプリケーションとWebサーバーを接続する際に考案されたインターフェース定義。 その昔、PythonのWebフレームワークが急増したことにより、 フレームワークごとにサーバーと接続するためのインターフェースが異なっていたため、同一のアプリケーションであっても、接続できるサーバーが制限される（あるサーバーには接続できるのに、別のサーバーには接続できない）という事態が生じてしまった。 そのため、インターフェースを統一して、どのフレームワークでも同じように各種サーバーに接続できるようにしよう、ということで生まれたのがWSGIである。 WSGIアプリケーションの定義として、 WSGIアプリケーションは、呼び出し可能なオブジェクトとして定義する。 このオブジェクトが呼び出される際、第一引数に環境変数が渡され、第二引数にステータスコードとレスポンスヘッダを受け取る呼び出し可能なオブジェクトが渡される。 第二引数に渡されたオブジェクトを呼び出して、ステータスコードとレスポンスヘッダ情報を渡す。 戻り値として、バイト文字列をyieldするiterableなオブジェクトを返す。 といったものがある。 3.人工知能を始めとした様々な分野で使われている ・実は、 Pythonといったら機械学習やAIといった人工知能をイメージするが、 実は1990年から存在する言語なので、もともとAIのために作られた言語ではない。 Webシステム等、汎用的に様々なシステムを作れる。 機械学習の際に、非常に重要なビックデータ処理やデータ分析のフレームワークやライブラリがPythonに多い。シンプルに記述出来るためOSS作成の際にエンジニアに選ばれる。コミュニティも活発。 データ分析には主に3段階 ・クローリング（クローラーがWeb上を巡回して情報を集める工程） ・スクレイピング（Web上で抽出したデータを分析、格納可能なデータに変換） ・データ前処理（収集したデータを扱いやすいように変換） それぞれの工程でライブラリが用意されており、Pythonでのデータ処理、分析がやりやすい。 では、実際にPythonはどんなシステムに使用されているのか？ 一部抜粋 サーバサイドのデータ読み込みや動画処理等に使用されている。 YouTubuの開発者によると、「Pythonは私たちのサイトにとって十分に高速であり、最小限の開発者で、記録的な速さで保守可能な機能を作成することができます」といっている。 GmailやGoogle MapsのサーバサイドでPythonが使用されている。 instagram, spotify, Pepper　etc... まとめ Pythonは今、世界で最も人気の高いプログラミング言語の一つであり、 日本においてもPythonエンジニアの需要は年々高まってきている。 これからプログラミングを学び始める人はもちろん、他言語を扱う現役エンジニアも新たに勉強しておいて損はない言語といえるだろう。 近年、Pythonをメイン言語として選定する会社や、サービスが増加傾向にあるので、今後もさらなる人気の向上が見込まれる言語であることは間違いないだろう。 おわり

前回まででFlaskの基本的な使い方を学べたものとして、もう少し実践的なアプリケーションを実装します。 新たに空のプロジェクトを作成します。 前提条件 DBはMySQLを使用するものとし、既に以下の設定を行っているものとします。 127.0.0.1:3306 で接続可能な状態 スキーマmy_schemaを作成済み ユーザーはroot、パスワードもroot テーブルには何もない状態 PythonでどうDBを扱うのか いきなり話が逸れますが、ORMという技術を使わない実装方法について説明します。 まずはDBに対応したDB Driverを探しましょう。 DB　 Driver MySQL PyMySQL PostgreSQL Psycopg SQLServer pyodbc SQLite sqlite3 DBに合わせたDriverを使い、SQLを実行します。 以下はMySQLとSQLServerの比較です。 MySQLの場合 import pymysql.cursors # Connect to the database connection = pymysql.connect(host='localhost', user='user', password='passwd', database='db', cursorclass=pymysql.cursors.DictCursor) with connection: with connection.cursor() as cursor: # Create a new record sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)" cursor.execute(sql, ('webmaster@python.org', 'very-secret')) # connection is not autocommit by default. So you must commit to save # your changes. connection.commit() SQLServerの場合 import pyodbc # Some other example server values are # server = 'localhost\sqlexpress' # for a named instance # server = 'myserver,port' # to specify an alternate port server = 'tcp:myserver.database.windows.net' database = 'mydb' username = 'myusername' password = 'mypassword' cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER='+server+';DATABASE='+database+';UID='+username+';PWD='+ password) cursor = cnxn.cursor() #Sample insert query count = cursor.execute(""" INSERT INTO SalesLT.Product (Name, ProductNumber, StandardCost, ListPrice, SellStartDate) VALUES (?,?,?,?,?)""", 'SQL Server Express New 20', 'SQLEXPRESS New 20', 0, 0, CURRENT_TIMESTAMP).rowcount cnxn.commit() DBに接続してINSERTするだけのコードですがかなり実装に差があるのがわかると思います。 特定のDBしか使用しないならいいでしょうが、そうとは限らないでしょう。 「最初はシンプルにSQLiteで実装して、後から別のDBに変えよう」 「SQLServerで実装したが、Json形式のサポートが弱いからPostgreSQLに変えよう」 ということはままあります。 今回はそういった煩わしさをなくしたいので、ORM、Migrationといった技術を使い、SQLを書くのではなくDBをインスタンスのように扱います。 パッケージインストール pip install Flask Flask-Migrate flask-restx Flask-SQLAlchemy mysqlclient Flask-SQLAlchemyがORM、Flask-MigrateがMigrationのためのパッケージです。 SQLAlchemy というパッケージもあります。Flask-SQLAlchemyはFlask拡張パッケージとして実装されており、FlaskでORMを利用する際に便利です。 app.py の作成 まずはメインファイルを作成しましょう。 app.py from flask import Flask from flask_migrate import Migrate from flask_sqlalchemy import SQLAlchemy from flask_restx import Api app = Flask(__name__) # flask-restxの設定 app.config['RESTX_MASK_SWAGGER'] = False api = Api(app, version='1.0', title='Sample API', description='A sample API') # Flask-SQLAlchemyの設定 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@127.0.0.1/my_schema?charset=utf8mb4' db = SQLAlchemy(app) # Flask-Migrateの設定 migrate = Migrate(app, db) # 別ファイルをインポートする(後で作成) from routes.user import user_module app.register_blueprint(user_module) Userモデルの作成 models/user from datetime import datetime from app import db, api from sqlalchemy import Column, Integer, String, Boolean, Date, DateTime, Float, Text from flask_restx import fields # DB固有の型はこう使う # from sqlalchemy.dialects.mysql import TINYINT, TIMESTAMP class User(db.Model): __tablename__ = 'user' id = Column(Integer, index=True, primary_key=True, autoincrement=True) username = Column(String(80), nullable=False) email = Column(String(40), nullable=False) UserDto = api.model('UserDto', { 'id': fields.Integer(min=0, description='primary_key'), 'username': fields.String(min_length=0, max_length=80), 'email': fields.String(min_length=0, max_length=40), }) UserInsertDto = api.model('UserInsertDto', { 'username': fields.String(required=True, min_length=0, max_length=80), 'email': fields.String(required=True, min_length=0, max_length=40), }) UserUpdateDto = api.model('UserUpdateDto', { 'username': fields.String(min_length=0, max_length=80), 'email': fields.String(min_length=0, max_length=40), }) UserクラスがMySQL上のuserテーブルと対応します。 OpenAPI定義のため以下の３つのDtoも定義してあります。 - UserDto: Select時のレスポンス - UserInsertDto: Insert時のリクエスト - UserUpdateDto :Update時のリクエスト Userルートの定義 routes/user.py from flask import request, Blueprint from flask_restx import Resource from models.user import User, UserDto, UserInsertDto, UserUpdateDto from app import db, api user_module = Blueprint('user_module', __name__) @api.route('/user') class Route(Resource): @api.marshal_list_with(UserDto) def get(self): """ 全レコードを取得します """ users = db.session.query(User).all() return users @api.expect(UserInsertDto) @api.marshal_with(UserDto) def post(self): """ レコードを１件追加します """ json_body = request.json new_user = User(**json_body) db.session.add(new_user) db.session.commit() return new_user @api.route('/user/<int:id>') class RouteWithId(Resource): @api.marshal_with(UserDto) def get(self, id: int): """ IDを指定してレコードを１件取得します """ user = db.session.query(User).filter(User.id == id).one() return user @api.expect(UserUpdateDto) @api.marshal_with(UserDto) def patch(self, id: int): """ IDを指定してレコードを１件更新します """ json_body = request.json update_user = db.session.query(User).filter(User.id == id).one() for k, v in json_body.items(): setattr(update_user, k, v) db.session.commit() return update_user @api.marshal_with(UserDto) def delete(self, id: int): """ IDを指定してレコードを１件削除します """ delete_user = db.session.query(User).filter(User.id == id).one() db.session.delete(delete_user) db.session.commit() return delete_user userテーブルに対するCRUDを行えるようGET(Select), POST(Insert), PATCH(Update), DELETE(Delete)を定義しています。 GET /userで全レコード取得、GET /user/{id} で１件取得します。 複数レコード取得する場合はCRUD + L(LはList)というらしいです。 @api.expect(UserInsertDto)、@api.marshal_with(UserDto)の部分が重要で @api.expect(UserInsertDto)がサーバへ送られてくるデータの形の指定。 @api.marshal_with(UserDto)がサーバが返すべきデータの形の指定だけではなく、Userクラスからjson形式への変換も行ってくれます。 試しに @api.marshal_with(UserDto)を削除すると正しくレスポンスを返せません。 -> TypeError: Object of type User is not JSON serializableになります また、@api.marshal_list_with(UserDto)はUserクラスを複数(配列で)返す場合の書き方で、@api.marshal_with(UserDto, as_list=True)と書いても同じです。 DBの作成 実装はできましたがDBはまだ作成されていません。作成しましょう。 マイグレーションリポジトリの作成 初回のみ必要。やり直す場合はmigrationsフォルダを削除してから行う。 flask db init PS D:\Python\flask_qiita> flask db init Creating directory D:\Python\flask_qiita\migrations ... done Creating directory D:\Python\flask_qiita\migrations\versions ... done Generating D:\Python\flask_qiita\migrations\alembic.ini ... done Generating D:\Python\flask_qiita\migrations\env.py ... done Generating D:\Python\flask_qiita\migrations\README ... done Generating D:\Python\flask_qiita\migrations\script.py.mako ... done Please edit configuration/connection/logging settings in 'D:\\Python\\flask_qiita\\migrations\\alembic.ini' before proceeding. migrationsフォルダが作成されます。この時点では初期設定のみです。 新規作成後はmigrations/env.pyのcontext.configure()のkwargsに compare_type=Trueとcompare_server_default=Trueを追加した方がいいかもしれません。 追加しないと「型」と「デフォルト値」の変更が検知されません。 def run_migrations_online(): # ... with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata, compare_type=True, # <- 追加 compare_server_default=True, # <- 追加 process_revision_directives=process_revision_directives, **current_app.extensions['migrate'].configure_args ) マイグレーションファイルの作成 flask db migrate PS D:\Python\flask_qiita> flask db migrate INFO [alembic.runtime.migration] Context impl MySQLImpl. INFO [alembic.runtime.migration] Will assume non-transactional DDL. INFO [alembic.autogenerate.compare] Detected added table 'user' INFO [alembic.autogenerate.compare] Detected added index 'ix_user_id' on '['id']' Generating D:\Python\flask_qiita\migrations\versions\5ebb528e8a5e_.py ... done 現在のDBの状態とFlaskのModelとの差を検出できました。 この時点ではalenmbic_versionというテーブルが作成されたのみで、userテーブルは作成されていません。 alenmbic_versionテーブルはDBのバージョン管理のためのテーブルです。 マイグレーションの適用 flask db upgrade PS D:\Python\flask_qiita> flask db upgrade INFO [alembic.runtime.migration] Context impl MySQLImpl. INFO [alembic.runtime.migration] Will assume non-transactional DDL. INFO [alembic.runtime.migration] Running upgrade -> 5ebb528e8a5e, empty message テーブル作成ができました。flask runでサーバを起動すればSwaggerUIからDBの更新が行えます。 DB移行を行う DBへの列の追加を試みます。 Userクラスの変更だけでDBの移行自体は行えますがOpenAPIとの整合性を維持するため ...Dtoも一緒に変更します。 models/user.py from datetime import datetime from app import db, api from sqlalchemy import Column, Integer, String, Boolean, Date, DateTime, Float, Text from flask_restx import fields # DB固有の型はこう使う # from sqlalchemy.dialects.mysql import TINYINT, TIMESTAMP class User(db.Model): __tablename__ = 'user' id = Column(Integer, index=True, primary_key=True, autoincrement=True) username = Column(String(80), nullable=False) email = Column(String(40), nullable=False) birth_day = Column(Date, nullable=False, default=datetime.now) height = Column(Float, nullable=False, default=0.0) memo = Column(Text, nullable=False, default='') created_at = Column(DateTime(timezone=True), nullable=False, default=datetime.now) updated_at = Column(DateTime(timezone=True), nullable=False, default=datetime.now, onupdate=datetime.now) is_active = Column(Boolean, nullable=False, default=True) UserDto = api.model('UserDto', { 'id': fields.Integer(min=0, description='primary_key'), 'username': fields.String(min_length=0, max_length=80), 'email': fields.String(min_length=0, max_length=40), 'birth_day': fields.Date(example=datetime.now().strftime('%Y-%m-%d')), 'height': fields.Float(example=0.0), 'memo': fields.String(example='memo'), 'created_at': fields.DateTime(example=datetime.now().strftime('%Y-%m-%dT%H:%M:%S')), 'updated_at': fields.DateTime(example=datetime.now().strftime('%Y-%m-%dT%H:%M:%S')), 'is_active': fields.Boolean, }) UserInsertDto = api.model('UserInsertDto', { 'username': fields.String(required=True, min_length=0, max_length=80), 'email': fields.String(required=True, min_length=0, max_length=40), 'birth_day': fields.Date(default=datetime.now().strftime('%Y-%m-%d')), 'height': fields.Float(default=0.0), 'memo': fields.String(default=''), 'is_active': fields.Boolean(default=True), }) UserUpdateDto = api.model('UserUpdateDto', { 'username': fields.String(min_length=0, max_length=80), 'email': fields.String(min_length=0, max_length=40), 'birth_day': fields.Date(), 'height': fields.Float(), 'memo': fields.String(), 'is_active': fields.Boolean(), }) マイグレーションファイルの作成 & 適用 flask db migrate flask db upgrade PS D:\Python\flask_qiita> flask db migrate Creating directory D:\Python\flask_qiita\migrations ... done Creating directory D:\Python\flask_qiita\migrations\versions ... done Generating D:\Python\flask_qiita\migrations\alembic.ini ... done Generating D:\Python\flask_qiita\migrations\env.py ... done Generating D:\Python\flask_qiita\migrations\README ... done INFO [alembic.autogenerate.compare] Detected added column 'user.memo' INFO [alembic.autogenerate.compare] Detected added column 'user.created_at' INFO [alembic.autogenerate.compare] Detected added column 'user.updated_at' INFO [alembic.autogenerate.compare] Detected added column 'user.is_active' Generating D:\Python\flask_qiita\migrations\versions\afc20f79324a_.py ... done PS D:\Python\flask_qiita> flask db upgrade INFO [alembic.runtime.migration] Context impl MySQLImpl. INFO [alembic.runtime.migration] Will assume non-transactional DDL. INFO [alembic.runtime.migration] Running upgrade 5ebb528e8a5e -> afc20f79324a, empty message Modelの変更と同じようにDBが変更され、移行できました。 補足など ORMを使えばDBの扱いが楽になりますし、Migrationがあれば多少の変更なら問題なしです。 とはいえMigrationによるDB移行は万能ではありません。大きな変更はうまく移行できない場合もありますので過信しすぎないようにしてください。 1. Flask 環境構築、簡単なバックエンドサーバー実装 2. FlaskでOpenAPI定義の自動出力、Swagger UIページの作成 3. ORM + MigrationでSQLを一切書かずにDBのCRUDや移行を行う 4. AWSにデプロイ (未作成)

 初めに 全部で4本の記事に分けて書いています！ 第1回 株価予測_可視化編 第2回 株価予測_LSTM ver.1 第3回 株価予測_LSTM ver.2 第4回 株価予測_アプリ化編 今回は株価予測のアプリ化を行なっていきます。Flaskの特徴は、1つのファイルでアプリケーションを作成できることです。これはmicroframeworkと呼ばれています。それに加えて、大きなアプリケーションを作るための拡張機能も備えている優れものです。 このシリーズで行っている株価のチャート分析は意思決定のためのテクニカル指標に過ぎません。なので、予測に絶対は無いことと、株やFXで損する・得することの判断は自己責任でお願いします。ちなみに私は過去にFX(レバレッジ20倍)でテクニカル指標を過信し過ぎて。。。。なので皆さんも実際に取引する時はファンダメンタルズ分析もしっかり取り入れつつ、楽しく取引しましょう！！ 今回の章立て 今回のファイル構成 ngrok 最後に 参考文献  今回のファイル構成 flask（ファイル） ┣ main.py ┣ predict.py ┣ static ┃ ┣ images ━ xxx.png ┃ ┣ predict_file ━ xxx.png ┃ ┗ train_files ━ xxx.pth ┃ ┗ xxx.pkl ┗ templates ┣ get.html # 株価入力 ┗ post.html # 株価予測結果 main.pyでwebアプリを起動します。 predict.pyは前回の株価予測を少し変えた内容が入っているファイルです。 staticにはWebページに表示するための画像ファイルや予測のためのpklファイルが入っています。 templatesにはwebページに記載するためのhtmlファイルが入っています。 今回もcolab上でモジュールのインポートします。 !pip3 install torch==1.6.0+cu101 !pip3 install torchvision==0.7.0+cu101 !pip3 install numpy==1.18.5 !pip3 install matplotlib==3.2.2 !pip3 install scikit-learn==0.23.1 !pip3 install seaborn==0.11.0 !pip3 install flask==1.1.2 !pip install flask-ngrok 次にflask.zipとしたFlaskをWeb上で起動するためのフォルダへ移動し、ファイルを確認します。 from google.colab import files uploaded = files.upload() !unzip flask.zip %cd flask !ls 次にngrokをインストールします。 !pip install pyngrok !ngrok authtoken xxxxx #Your Authtoken 最後にmain.pyを起動します。 !python3 main.py 実行すると以下のようなものがRunningとして表示されるので、~~~~ngrok.ioをクリックしてwebアプリを起動します。 Running on http://xxx.x.x.x:xxxx/ (Press CTRL+C to quit) Running on http://xxxx-xx-xxx-xx-xxx.ngrok.io Traffic stats available on http://xxx.x.x.x:xxxx ngrokについて まず、webアプリをGoogle colab上で立ち上げるためには、tokenの取得とコードの追記が必要です。まずtokenの取得をしましょう。 ngrokにアクセスします。 Sign upします。 メールが届いているので、承認します。 Your Authtokenからトークン確認します。 以下のコマンドで実行します。 !pip install pyngrok !ngrok authtoken xxxxx #Your Authtoken 最後に Flaskに関しては入門書をやってみた感じですが、先駆者の皆様のサイトをみながら、書籍を参考にさせていただきながら、一応完成はしました！ 参考文献 Flask Tutorial ゼロからFlaskがよくわかる本: Pythonで作るWebアプリケーション開発入門 動かしながら学ぶ PyTorchプログラミング入門 ngrok：詳しくは以下を参照 GoogleColabでFlaskアプリを実行する方法(ngrok token必要になった模様) Quickly share ML WebApps from Google Colab using ngrok for Free ローカルで実行したい場合はこちら➡︎【Macユーザー向け】ngrokでローカル環境のサイトを外部へ公開

目次 プログラマーのための統計学 - 目次 概要 共分散と共分散行列(または分散共分散行列)について書く。 pythonのmultivariate_normalとかで使うやつ。 import numpy as np # 期待値 mean = np.array([3, 5]) # 分散共分散行列（covariance matrix） cov = np.array([[4, -1.2], [-1.2, 1]]) # 多変量正規分布に従う乱数を生成 np.random.multivariate_normal(mean, cov, size=200) 前提条件 この記事を読むには、以下を理解していることが前提となります。 共分散とは？何の役に立つの？ 国語の点数が高い人は、数学の点数も高いのか？といった傾向を見るときに役に立つ。 Xが国語の点数、Yが数学の点数としたとき、 「Xの偏差 × Yの偏差」の平均 を出すことで求められる。 以下の生徒の共分散を算出してみる。(100点満点) 生徒 国語 数学 A 50 50 B 50 70 C 80 60 D 70 90 E 90 100 偏差の出し方については以下を参照 偏差とは、平均値と各データの差を全て足したものの平均のことです。 https://qiita.com/Esfahan/items/2e70635552d7d661bc6b#%E5%88%86%E6%95%A3 国語の平均点 68 = \dfrac{1}{5}(50+50+80+70+90) 数学の平均点 74 = \dfrac{1}{5}(50+70+60+90+100) それぞれの「国語の偏差 × 数学の偏差」は以下のようになる。 \begin{align} (A) \, 432 &= (50−68)(50−74) \\ (B) \, 72 &= (50-68)(70-74) \\ (C) \, −168 &= (80-68)(60-74) \\ (D) \, 32 &= (70-68)(90-74)) \\ (E) \, 572 &= (90-68)(100-74) \end{align} 共分散は， 「Xの偏差 × Yの偏差」の平均なので， 188 = \dfrac{1}{5}(432+72-168+32+572) ​ となる。 この値が正の値であれば、国語の点数が高い人は数学も高い、(国語の点数が低い人は数学も低い、も成り立つ) 負の値であれば、国語の点数が高い人は数学の点数が低い、 0であれば（に近ければ）2つの科目に関係性は薄い、ということになる。 上記の例では188という大きな値が出ているので、国語の点数が高い人は数学も高い、という結果が得られたことになる。 共分散は以下の式で表すことができる。 Sxy = \frac{1}{n}(x-\vec{x})^{'}(y-\vec{y}) 共分散の問題点 上記の例では100点満点同士の科目の比較をしたが、仮に片方が100点満点、もう片方が10点満点のように、単位が異なる場合共分散の数値が大きく変わってしまうので、共分散の値だけを見て一概に判断はできないという問題がある。 この問題を解消するには、相関係数というものを用いる。（今回は説明しない） https://manabitimes.jp/math/854 分散共分散行列 分散共分散行列の求め方 上記の例と同じデータの分散共分散行列を求めてみる。 生徒 国語 数学 A 50 50 B 50 70 C 80 60 D 70 90 E 90 100 国語の平均点 E[X] = 68 = \dfrac{1}{5}(50+50+80+70+90) 数学の平均点 E[Y] = 74 = \dfrac{1}{5}(50+70+60+90+100) 偏差ベクトル（平均からの差） 生徒 国語 数学 A (50−68) (50−74) B (50-68) (70-74) C (80-68) (60-74) D (70-68) (90-74) E (90-68) (100-74) 国語の分散 σ^2_{x} = 256 = \dfrac{1280}{5} = \dfrac{1}{5}((50−68)^2+(50-68)^2+(80-68)^2+(70-68)^2+(90-68)^2) 数学の分散 σ^2_{y} = 344 = \dfrac{1720}{5} = \dfrac{1}{5}((50−74)^2+(70-74)^2+(60-74)^2+(90-74)^2+(100-74)^2) 国語と数学の共分散(前項で算出してるので過程は省略) \sigma_{XY} = 188 = \dfrac{1}{5}(432+72-168+32+572) 分散共分散行列は以下となる。 Σ = \begin{pmatrix} 256 & 188 \\ 188 & 344 \end{pmatrix} 共分散行列のまとめ 共分散行列は以下のように表すことができる。 \begin{bmatrix} S_{xx} & S_{xy} \\ S_{xy} & S_{yy} \end{bmatrix} 式の形から、「xの分散」は「xとxの共分散」と同じなのでSxxと表す。(前述の例では国語の分散=256) 同様に、「yの分散」は「yとyの共分散」と同じなので Syyで表す。(前述の例では数学の分散=344) xの分散 = (x - 平均値)^2 = xとxの共分散 = (x - 平均値)(x - 平均値)) \\ yの分散 = (y - 平均値)^2 = yとyの共分散 = (y - 平均値)(y - 平均値)) 参考

概要 REST API を利用して、リソースグループ内にあるデプロイ名毎にデプロイ情報・履歴を取得する Python プログラムです。 デプロイ履歴（情報）を取得するプログラムフローは以下となっています。 Azureリソースに REST API アクセスするためのアクセストークンの取得 REST API を利用したデプロイ履歴情報の取得 取得した履歴情報をデプロイ名単位でローカルファイルへ保存 REST API を利用したデプロイ操作情報の取得 取得した操作情報をデプロイ名単位でローカルファイルへ保存 実行環境 macOS Monterey 12.1 python 3.8.3 Azure CLI 2.28.0 前提条件 Azure環境がすでに用意されていること（テナント/サブスクリプション） ローカル環境に「azure cli」がインストールされていること リソースグループ（rg_ituru_deploy_test）配下にvNetリソース（vnet_ituru_deploy_test）のAzure環境が構築されていること 実行プログラム GetDepeloymentHistory.py import os import time import json import argparse import requests import sys from datetime import datetime from azure.identity import AzureCliCredential, DefaultAzureCredential from azure.mgmt.resource import ResourceManagementClient # Azure Info TENANT_ID = os.environ['AZURE_TENANT_ID'] CLIENT_ID = os.environ['AZURE_CLIENT_ID'] CLIENT_KEY = os.environ['AZURE_CLIENT_SECRET'] # Retrieve subscription ID from environment variable. SUBSCRIPTION_ID = os.environ["ARM_SUBSCRIPTION_ID"] API_VERSION = "2021-04-01" # リソースの管理オブジェクトの取得 def GetResourceManagementClient(): resource_client = ResourceManagementClient( # credential=DefaultAzureCredential() credential=AzureCliCredential(), subscription_id=SUBSCRIPTION_ID ) return resource_client # サブスクリプションのリソースグループ一覧の取得 def GetResourceGroup(): # Obtain the management object for resources. resource_client = GetResourceManagementClient() # Retrieve the list of resource groups group_list = resource_client.resource_groups.list() # Show the groups in formatted output column_width = 50 print("Resource Group".ljust(column_width) + "Location") print("-" * (column_width * 2)) for group in list(group_list): print(f"{group.name:<{column_width}}{group.location}") # デプロイ情報・履歴データをローカルファイルへ保存の取得 def DeploymentToLocalFile(item, filename): # 保存するファイル名の生成 print(filename) # json.dump関数でファイルに書き込む fw = open(filename,'w') json.dump(item,fw,indent=2) fw.close() # 指定したリソースグループ内のデプロイ名毎のデプロイ操作情報の取得 def GetDepeloymentOperation(access_token, rg_name, dp_name, now): # access_token を含めたヘッダ情報 headers = { 'Authorization': 'Bearer %s' % access_token } # 取得URL DeplumentOperation_URL = 'https://management.azure.com/subscriptions/' + \ SUBSCRIPTION_ID + '/resourcegroups/' + rg_name + \ '/deployments/' + dp_name + '/operations?api-version=' + API_VERSION # print(DeplumentOperation_URL) # 情報取得のGetリクエスト try : res2 = requests.get( DeplumentOperation_URL, headers=headers ) except Exception as err: print("Something Fatal Happened") print (err) sys.exit() # requrest処理をクローズする res2.close # res2をjsonファイルに整形しユーザ情報の取得 item = res2.json()['value'] if len(item) > 0: # print(json.dumps(item, indent = 2)) # 保存するファイル名の生成 filename = './output/Deployment_Operation__' + rg_name + '_' + dp_name + '_' + now.strftime('%Y%m%d_%H%M%S') + '.json' # デプロイ履歴データをローカルファイルへ DeploymentToLocalFile(item, filename) # 指定したリソースグループ内のデプロイ名毎のデプロイ履歴情報の取得 def GetDepeloymentInfo(access_token, rg_name): # access_token を含めたヘッダ情報 headers = { 'Authorization': 'Bearer %s' % access_token } # 取得URL DeplumentInfo_URL = 'https://management.azure.com/subscriptions/' + \ SUBSCRIPTION_ID + '/resourcegroups/' + rg_name + \ '/providers/Microsoft.Resources/deployments/?api-version=' + API_VERSION # print(DeplumentInfo_URL) # 情報取得のGetリクエスト try : res1 = requests.get( DeplumentInfo_URL, headers=headers ) except Exception as err: print("Something Fatal Happened") print (err) sys.exit() # requrest処理をクローズする res1.close # デプロイ情報の取得有無判斷 if len(res1.json()['value']) == 0: print("\n AzureCLI もしくは Terraform 等で構築されたリソースグループの場合、取得できません、、、、 \n") return 0 # res1をjsonファイルに整形しデプロイ履歴情報の取得 for num, item in enumerate(res1.json()['value']): print("\n■ ■ ■ {}".format(item["name"])) # print(json.dumps(item, indent = 2)) # 保存するファイル名の生成 now = datetime.now() filename = './output/Deployment__' + rg_name + '_' + item["name"] + '_' + now.strftime('%Y%m%d_%H%M%S') + '.json' # デプロイ情報データをローカルファイルへ DeploymentToLocalFile(item, filename) # デプロイ操作データの取得 GetDepeloymentOperation(access_token, rg_name, item["name"], now) return num+1 # Azureにアクセスするためのトークンの取得（Access_Token） def get_azure_access_token() -> str: # access_token を取得するためのヘッダ情報 headers = { 'Accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded' } payload = { 'client_id': CLIENT_ID, 'scope': 'https://management.azure.com/.default', 'grant_type': 'client_credentials', 'client_secret': CLIENT_KEY } # access_token を取得するためのURLを生成 TokenGet_URL = "https://login.microsoftonline.com/" + \ TENANT_ID + "/oauth2/v2.0/token" # アクセストークンの取得 try : response = requests.get( TokenGet_URL, headers=headers, data=payload ) except Exception as err: print("Something Fatal Happened") print (err) sys.exit() # requrest処理のクローズ response.close jsonObj = json.loads(response.text) return jsonObj["access_token"] # メイン if __name__ == '__main__': parser = argparse.ArgumentParser(description='引数なし：リソースグループ一覧の取得、　　引数あり：指定したリソースグループ内のリソース毎のデプロイ情報・履歴の取得') parser.add_argument('-g', '--rg', type=str, help='リソースグループ名') args = parser.parse_args() start = time.time() if args.rg == None : GetResourceGroup() else : access_token = get_azure_access_token() print(f"取得アクセストークン :\n{access_token}\n") cnt = GetDepeloymentInfo(access_token, args.rg) print("\n Deplyment_name 数 : " + str(cnt)) generate_time = time.time() - start print("\n 取得時間:{0}".format(generate_time) + " [sec] \n") プログラムのHELP表示 ## HELPの表示 $ python GetDepeloymentHistory.py -h usage: GetDepeloymentHistory.py [-h] [-g RG] 引数なし：リソースグループ一覧の取得、　　引数あり：指定したリソースグループ内のリソース毎のデプロイ履歴・操作情報の取得 optional arguments: -h, --help show this help message and exit -g RG, --rg RG リソースグループ名 プログラムの実行 まずは最初に、リソースグループ一覧の取得 $ python GetDepeloymentHistory.py Resource Group Location ---------------------------------------------------------------------------------------------------- rg_AzureStackHCI eastus rg_NetworkWatcher japaneast ： 省略 ： rg_ituru_vm02 japaneast rg_ituru_deploy_test japaneast 取得時間:0.4982631206512451 [sec] 上記のリソースグループ一覧から、指定したリソースグループのデプロイ情報・履歴を取得し、ローカルファイルへ保存 $ python GetDepeloymentHistory.py -g rg_ituru_deploy_test 取得アクセストークン : eyJ0eXAiOiJ・・・省略・・・VjCV6lrbWc8Dw ■ ■ ■ Microsoft.VirtualNetwork-20220201112413 ./output/Deployment__rg_ituru_deploy_test_Microsoft.VirtualNetwork-20220201112413_20220201_160456.json ./output/Deployment_Operation__rg_ituru_deploy_test_Microsoft.VirtualNetwork-20220201112413_20220201_160456.json Deplyment_name 数 : 1 取得時間:0.49964094161987305 [sec] Outputファイルの確認 $ ls -l ./output total 32 drwxr-xr-x 6 ituru staff 192 2 1 16:08 ./ drwxr-xr-x 12 ituru staff 384 2 1 16:05 ../ -rw-r--r-- 1 ituru staff 1483 2 1 16:04 Deployment_Operation__rg_ituru_deploy_test_Microsoft.VirtualNetwork-20220201112413_20220201_160456.json -rw-r--r-- 1 ituru staff 2511 2 1 16:04 Deployment__rg_ituru_deploy_test_Microsoft.VirtualNetwork-20220201112413_20220201_160456.json デプロイ履歴情報（Deployment__rg_xxxxxxxxxxx.json） { "id": "/subscriptions/xxxxxxxx-1717-dada-9779-zzzzzzzzzzzz/resourceGroups/rg_ituru_deploy_test/providers/Microsoft.Resources/deployments/Microsoft.VirtualNetwork-20220201112413", "name": "Microsoft.VirtualNetwork-20220201112413", "type": "Microsoft.Resources/deployments", "tags": { "primaryResourceId": "/subscriptions/xxxxxxxx-1717-dada-9779-zzzzzzzzzzzz/resourceGroups/rg_ituru_deploy_test/providers/Microsoft.Network/virtualNetworks/vnet_ituru_deploy_test", "marketplaceItemId": "Microsoft.VirtualNetwork-ARM" }, "properties": { "templateHash": "15981414355539578904", "parameters": { "location": { "type": "String", "value": "japaneast" }, "extendedLocation": { "type": "Object", "value": {} }, "virtualNetworkName": { "type": "String", "value": "vnet_ituru_deploy_test" }, "resourceGroup": { "type": "String", "value": "rg_ituru_deploy_test" }, "addressSpaces": { "type": "Array", "value": [ "10.6.0.0/16" ] }, "ipv6Enabled": { "type": "Bool", "value": false }, "subnetCount": { "type": "Int", "value": 1 }, "subnet0_name": { "type": "String", "value": "default" }, "subnet0_addressRange": { "type": "String", "value": "10.6.0.0/24" }, "ddosProtectionPlanEnabled": { "type": "Bool", "value": false }, "firewallEnabled": { "type": "Bool", "value": false }, "bastionEnabled": { "type": "Bool", "value": false } }, "mode": "Incremental", "debugSetting": { "detailLevel": "None" }, "provisioningState": "Succeeded", "timestamp": "2022-02-01T02:24:55.4064206Z", "duration": "PT4.7454974S", "correlationId": "a891f3f3-4477-4545-acac-d748ffdd2857", "providers": [ { "namespace": "Microsoft.Network", "resourceTypes": [ { "resourceType": "VirtualNetworks", "locations": [ "japaneast" ] } ] } ], "dependencies": [], "outputResources": [ { "id": "/subscriptions/xxxxxxxx-1717-dada-9779-zzzzzzzzzzzz/resourceGroups/rg_ituru_deploy_test/providers/Microsoft.Network/VirtualNetworks/vnet_ituru_deploy_test" } ], "validationLevel": "Template" } } デプロイ操作情報（Deployment_Operation__rg_xxxxxxxxxxx.json） [ { "id": "/subscriptions/xxxxxxxx-1717-dada-9779-zzzzzzzzzzzz/resourceGroups/rg_ituru_deploy_test/providers/Microsoft.Resources/deployments/Microsoft.VirtualNetwork-20220201112413/operations/557B59888B6CD293", "operationId": "557B59888B6CD293", "properties": { "provisioningOperation": "Create", "provisioningState": "Succeeded", "timestamp": "2022-02-01T02:24:55.2896236Z", "duration": "PT2.9130169S", "trackingId": "3838ccbb-f42d-4c26-92be-b46005570e40", "serviceRequestId": "f597efef-c06c-4ee4-8683-083f3a862760", "statusCode": "OK", "targetResource": { "id": "/subscriptions/xxxxxxxx-1717-dada-9779-zzzzzzzzzzzz/resourceGroups/rg_ituru_deploy_test/providers/Microsoft.Network/VirtualNetworks/vnet_ituru_deploy_test", "resourceType": "Microsoft.Network/VirtualNetworks", "resourceName": "vnet_ituru_deploy_test" } } }, { "id": "/subscriptions/xxxxxxxx-1717-dada-9779-zzzzzzzzzzzz/resourceGroups/rg_ituru_deploy_test/providers/Microsoft.Resources/deployments/Microsoft.VirtualNetwork-20220201112413/operations/00585579225949072200", "operationId": "00585579225949072200", "properties": { "provisioningOperation": "EvaluateDeploymentOutput", "provisioningState": "Succeeded", "timestamp": "2022-02-01T02:24:55.3801548Z", "duration": "PT3.0035481S", "trackingId": "5252d103-4555-4c4c-821c-0c5ef88ccc34", "statusCode": "OK" } } ] まとめ このプログラムで、指定するリソースグループ内にあるリソースのデプロイ履歴・操作情報をサクッと取得し、その内容を json形式 でローカルファイルに保存できました。 この２つのデプロイ履歴情報とデプロイ操作情報の json ファイルは、Azure Portalから「リソースグループ」ー「設定：デプロイ」からデプロイ名を選択し、「展開の詳細」ー「ダウンロード」から入手できる「deployment.json（履歴情報）」「deployment_operations.json（操作情報）」の２つのファイルと同等となっています。 ただ、この２つのデプロイ情報は、AzurePortalからリソースを作成した場合のみ取得可能(?)であり、AzureCLI / Terraform 等でリソースを作成した場合には取得不可となります。なんでだろ、、、、、 AzureCLI / Terraform 等でリソースを作成した場合、デプロイ情報の取得は「テンプレートのエクスポート」しかないんでしょうか？、、、、、、、

概要 　PyTorch様公式のTransformer実装が複雑怪奇だったため，より簡素な実装をしていきます．公式とは逆にbatch firstな点を注意してください． 　※今回は自己回帰型のエンコーダのみの実装になります． 準備 　python = "3.6.8" 　pytorch = "1.6.0" ソースコード インポート類 import copy import math import numpy as np import torch torch.manual_seed(41) import torch.optim as optim import torch.nn as nn from torch.nn import functional as F Softmaxを使ったMultiHeadのSelfAttention Input: QKV [batch,head,length,head_dim], mask [batch,len,len] Output: X [batch,head,length,head_dim] class SoftmaxAttention(nn.Module): def __init__(self, head_dim): super().__init__() self.head_dim = head_dim def forward(self, Q, K, V, mask=None): logit = torch.einsum("bhld,bhmd->bhlm",Q,K)/math.sqrt(self.head_dim) if mask!=None: logit = logit + mask[:,None,:,:] attention_weight = F.softmax(logit, dim=-1) X = torch.einsum("bhlm,bhmd->bhld",attention_weight,V) return X MultiHead化を含むSelfAttention層全体 Input: X [batch,length,dim], mask [batch,len,len] Output: X [batch,length,dim] class SelfAttention(nn.Module): def __init__(self, dim, head_dim, num_head): super().__init__() self.dim = dim self.head_dim = head_dim self.num_head = num_head assert dim == head_dim*num_head,print("ASSERT #dim = head_dim * num_head") self.W_q = nn.Linear(self.dim, self.dim) self.W_k = nn.Linear(self.dim, self.dim) self.W_v = nn.Linear(self.dim, self.dim) self.attn = SoftmaxAttention(head_dim) def forward(self, X, mask): Q = self.split_heads(self.W_q(X)) K = self.split_heads(self.W_k(X)) V = self.split_heads(self.W_v(X)) attn_out = self.attn(Q.float(), K.float(), V.float(), mask.float()) attn_out = self.combine_heads(attn_out) return attn_out #[batch,head,len,head_dim]->[batch,len,dim] def combine_heads(self, X): X = X.transpose(1, 2) X = X.reshape(X.size(0), X.size(1), self.num_head * self.head_dim) return X #[batch,len,dim]->[batch,head,len,head_dim] def split_heads(self, X): X = X.reshape(X.size(0), X.size(1), self.num_head, self.head_dim) X = X.transpose(1, 2) return X 位置エンコーディング（公式の実装はbatch firstではない事に注意） class PositionalEncoding(nn.Module): def __init__(self, d_model, dropout=0.1, max_len=5000): super().__init__() self.dropout = nn.Dropout(p=dropout) position = torch.arange(max_len).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2) * (-math.log(10000.0) / d_model)) pe = torch.zeros(max_len, 1, d_model) pe[:, 0, 0::2] = torch.sin(position * div_term) pe[:, 0, 1::2] = torch.cos(position * div_term) self.register_buffer('pe', pe.transpose(0,1)) #[batch,len,dim]->[batch,len,dim] def forward(self, x): x = x + self.pe[:,:x.size(1),:] return self.dropout(x) Transformerエンコーダ層 Input: x [batch,length,dim], mask [batch,len,len] Output: x [batch,length,dim] class TransformerEncoderLayer(nn.Module): def __init__(self, attn_layer, ff_layer, norm_layer, drop_layer): super().__init__() self.attn_layer = attn_layer self.ff_layer = ff_layer self.norm_layer = norm_layer self.drop_layer = drop_layer def forward(self, x, mask): x = self.drop_layer(self.attn_layer(x,mask)) + x x = self.norm_layer(x) x = self.drop_layer(self.ff_layer(x)) + x x = self.norm_layer(x) return x Transformerエンコーダ全体 　コメントアウト部分はキーパディング部分に対するmask処理です（あまり効果が無いので入れていません） Input: x [batch,length] Output: logits [batch,length,vocab_size] class TransformerEncoder(nn.Module): def __init__(self, num_layer, dim, ff_dim, head_dim, num_head, vocab_size=None, drop_p=0.1): super().__init__() self.vocab_size = vocab_size self.emb = nn.Embedding(vocab_size,dim) if vocab_size!=None else nn.Identity() self.pos_encoder = PositionalEncoding(dim) self.output = nn.Linear(dim,vocab_size) if vocab_size!=None else nn.Identity() self.encoders = ModuleList([TransformerEncoderLayer(SelfAttention(dim, head_dim, num_head),\ nn.Sequential(nn.Linear(dim,ff_dim),nn.ReLU(),nn.Linear(ff_dim,dim)),\ LayerNorm(dim, eps=1e-5),\ nn.Dropout(drop_p)) for i in range(num_layer)]) def generate_square_subsequent_mask(self, sz): mask = (torch.triu(torch.ones(sz, sz)) == 1).transpose(0, 1) mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0)) return mask # def generate_key_padding_mask(self, src, pad_id=0): # f = torch.full_like(src,False).bool().to() # t = torch.full_like(src,True).bool() # return torch.where(src==pad_id,t,f) def forward(self, x, key_mask=None, sq_mask=None): mask = torch.zeros(x.size(0),x.size(1),x.size(1)).bool().to(x.device) # if self.vocab_size != None: # key_mask = self.generate_key_padding_mask(x).to(x.device) sq_mask = self.generate_square_subsequent_mask(x.size(-1)).to(x.device) # if key_mask != None: # mask = mask.bool().to(x.device) + torch.cat([key_mask.unsqueeze(1)]*x.size(1),dim=1).bool() + torch.cat([key_mask.unsqueeze(2)]*x.size(1),dim=2).bool() # mask = torch.zeros_like(mask).masked_fill_(mask,float("-inf")).to(x.device) if sq_mask!=None: mask = mask + sq_mask[None,:,:] mask = mask.float().to(x.device) x = self.emb(x) x = self.pos_encoder(x) for layer in self.encoders: x = layer(x,mask) return self.output(x) モデルの宣言 model = TransformerEncoder(vocab_size=VOCAB_SIZE,num_layer=6,dim=512,ff_dim=1024,head_dim=64,num_head=8) 試しにテキストを自己回帰で学習してみましょう． pip install transformers from torch.utils.data import DataLoader, Dataset TRAIN_BATCH = 40 VAL_BATCH = 20 LEARNING_RATE = 1e-4 with open("text.txt","r",encoding="utf-8") as r: #適当なテキストデータ lines = [line.strip() for line in r.readlines()] import transformers from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("cl-tohoku/bert-base-japanese-char") # tokenizer = AutoTokenizer.from_pretrained("cl-tohoku/bert-base-japanese") PAD_IDX = tokenizer.pad_token_id CLS_IDX = tokenizer.cls_token_id EOS_IDX = tokenizer.eos_token_id VOCAB_SIZE = tokenizer.vocab_size class MyDataset(Dataset): def __init__(self,lines,_tokenizer): self.text = lines self.tokenizer = _tokenizer def __len__(self): return len(self.text) def __getitem__(self, idx): text = self.text[idx] encode = self.tokenizer(text) return torch.tensor(encode["input_ids"]) def collate_fn(batch): x = torch.nn.utils.rnn.pad_sequence(batch, batch_first=True, padding_value=PAD_IDX) return x dataset = MyDataset(lines,tokenizer) train_length = int(len(dataset)*0.9) val_length = len(dataset) - train_length train,val = torch.utils.data.random_split(dataset,[train_length,val_length]) train_loader = DataLoader(train,batch_size=TRAIN_BATCH,shuffle=True,collate_fn=collate_fn) val_loader = DataLoader(val,batch_size=VAL_BATCH,shuffle=False,collate_fn=collate_fn) optim = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = model.to(device) def cut_tensor(x,max_len=512): if x.size(-1)>max_len: return x[:,:max_len] else: return x epoch = 20 for i in range(epoch): model.train() step = 0 train_epoch_loss = 0 for batch in train_loader: step += 1 src = batch # src = cut_tensor(src,max_len=SEQUENCE_LENGTH+1)#max_length_cut src = src.to(device) output = model(src[:,:-1]) optim.zero_grad() loss = F.cross_entropy(output.reshape(-1, output.size(-1)), src[:,1:].reshape(-1), ignore_index = PAD_IDX) loss.backward() optim.step() train_epoch_loss += loss.item() train_epoch_loss /= step model.eval() step = 0 val_epoch_loss = 0 for batch in val_loader: step += 1 src = batch # src = cut_tensor(src,max_len=SEQUENCE_LENGTH+1)#max_length_cut src = src.to(device) with torch.no_grad(): output = model(src[:,:-1]) loss = F.cross_entropy(output.reshape(-1, output.size(-1)), src[:,1:].reshape(-1), ignore_index = PAD_IDX) val_epoch_loss += loss.item() val_epoch_loss /= step print("\rSTEP:{}\tTRAINLOSS:{}\tVALLOSS:{}".format(i,train_epoch_loss,val_epoch_loss)) 　トークナイザ―は東北大学様のBERTの事前学習モデルよりお借りしました． 　PyTorchの公式実装と同じか，より低い損失値になったので良い感じです． まとめ 公式実装がややこしかったので，簡潔な自己回帰のTransformerエンコーダのコードを書きました． 備考 layer(x, mask) -> layer(x)　とすれば自己回帰ではなくなります． キーパディングのmask処理を入れるには，TransformerEncoderのコメントアウトを全て外してください．（GPU処理じゃないとエラーが出る場合があるそうです） 最後に 　間違っている点などございましたら，コメント等で優しく指摘して頂けると助かります．（気付かなければ申し訳ありません）

1.よく使うライブラリ テストコードを書く際に自分が高頻度で使うライブラリを記載します。 1-1.Seleniumがサポートしているライブラリ ライブラリ名 説明 import文 webdriver Webdriverを読み込む from selenium import webdriver keys キーボード入力を行う from selenium.webdriver.common.keys import Keys ~Exception 例外を定義する from selenium.common.exceptions import NoSuchElementException By 要素の抽出条件等を指定する from selenium.webdriver.common.by import By Options Webdriverの実行オプションを設定する from selenium.webdriver.chrome.options import Options chromedriver_binary chromedriver-binaryを読み込む import chromedriver_binary WebDriverWait 明示的待機を定義する from selenium.webdriver.support.ui import WebDriverWait expected_conditions 明示的待機の条件を指定する from selenium.webdriver.support import expected_conditions as EC 1-2.Python標準ライブラリ ライブラリ名 説明 time 強制待機(time.sleep(n))を設定 date 日付を取得 os パスの取得やディレクトリの作成 logging ロガーを設定 requests 結果連携用のSlackAPIの実行 2.導入 2-1.ディレクトリ構成 workdir workdir/ ∟ ph_main.py ∟ test_pattern.csv ∟ chromedriver ∟ each_ph_pkg/ 2-2.test_pattern.csvファイルの作成 パターン名 メールアドレス パスワード 項目1 項目2 基本購入シナリオ1 hogehuga001@example.com Passw0rd 0 0 基本購入シナリオ2 hogehuga002@example.com Passw0rd 1 0 test_pattern.csv パターン名,メールアドレス,パスワード,項目1,項目2 基本購入シナリオ1,hogehuga001@example.com,Passw0rd,0,0 基本購入シナリオ2,hogehuga002@example.com,Passw0rd,1,0 2-3.メインスクリプト(ph_main.py)の作成 test.py from selenium import webdriver from selenium.webdriver.common.keys import Keys from selenium.webdriver.chrome.options import Options from datetime import date import os import time import requests driver = webdriver.Chrome(executable_path='./chromedriver') #カレントディレクトリにあるchromedriverを指定 driver.implicitly_wait(10) #暗黙的な待機(要素が見つかるまで最大10秒待つ) #ファイル命名用の現在日付の取得 current_time = datetime.now() datetime_str = current_time.strftime("%Y%m%d%H%M%S") #エビデンスキャプチャ格納用フォルダ作成 evidence_folder = os.path.join(os.getcwd(), "evidence") if os.path.exsists(evidence_folder): pass else: os.makedirs(evidence_folder) #試験結果を書き込むログファイル用のフォルダを準備 result_folder = ps.path.join(os.getcwd(), "result") if os.path.exsists(result_folder): pass else: os.makedirs(result_folder) result_list = [] #test_pattern.csvファイルの読み込み/試験実行 with open("test_pattern.csv", "r") as t: row = csv.reader(t) test_patterns = [row for row in reader] for n in range(1, len(test_patterns)+1): pattern_name = test_patterns[n][1] email_address = test_patterns[n][2] password = test_patterns[n][3] item1 = test_patterns[n][4] item2 = test_patterns[n][5] for _ in range(2): #失敗時1回のみ再実行 retry_count = 0 try: #処理を記述 driver.get("https://example.com") retry_count = 0 except: #失敗時の処理を記述 retry_count += 1 else: break else: #リトライも失敗した際の処理を記述 evidence_file_name = pattern_name + "_" + datetime_str driver.save_screenshot(evidence_folder) if retry_count = 0: result_row = pattern_name + ": OK" else: result_row = pattern_name + ": NG" result_list.append(result_row) #試験結果の書き込み with open("./result/result_{}.csv".format(datetime_str), "w") as r: writer = csv.writer(r, lineterminater="¥n") for results in result_list: writer.writerow(results) #Slackのチャンネルに結果を連携 class SendResultToSlack: def __init__(self, token): self.token = token def send_result(self, message, channel): params = {"channel": channel, "text": message} r = requests.post('https://slack.com/api/chat.postMessage', headers={'Authorization': 'Bearer ' + self.token}, params=params) print("return ", r.json()) with open("./result/result_{}.csv".format(datetime_str), "r") as r: texts = r.read() test_result = "本日のリグレッションテストの結果です。\n" + texts token = '<TOKEN>' slack = SendResultToSlack(token) slack.send_result(test_result, "<CHANNEL_NAME>") driver.quit()

背景：損失関数をコードを通して理解したい 先日は「$y=x^2$」を使って簡単な勾配降下法の可視化を行いました（勾配降下法を可視化したい件）。1変数の勾配降下法についての可視化は出来たので、2変数や多変数についての勾配降下法についても同じように実装したいと思います。 目的：勾配降下法を一般化していきながら理解したい 2変数や多変数についての勾配降下法の実装を通して、誤差逆伝播の流れを実感します。 方法：2変数関数で勾配降下法を行う $z=x^2+y^2$ を使って可視化します。前回は微分した時の傾きが0の所で勾配降下法を終えましたが、今回は任意の回数行う事にします。 結果：3次元空間の勾配降下法の変化を描写 前回アニメーションが出来たので今回も作ろうと思いましたが、エラーが出て出来ませんでした。結局、出力の変位をプロットするだけで終わってしまいました。というか、参考にしたコードそのままです。これはこれで初めの方が学習による変化の大きさが見れました。それが以下の図になります。 コードはこちらです。 勾配降下法.ipynb 勾配降下法（$z=x^2+y^2$）にあります。 課題：一般化した勾配降下法の実装 損失関数や評価関数、重みの更新についての理解をコードを通して深めたい。 3次元のアニメーションを動画に残す方法が分かりませんでした。先日、やった方法は視点を変える動画で保存できましたが、今回は、点が動くものです。位置情報はリストに入れているのですが、3次元でプロットする際にイテレーション出来ないと怒られた気がします。出来る方がいらっしゃったらアドバイスして頂ければ嬉しいです！ 関連 勾配降下法を可視化したい件

いよいよ本題に挑戦！ 前回は、今回の検証で使用するデータ生成ツーールを紹介させて頂きましたが、今回はいよいよ以前のスクリプトをベースに雪だるま方式の作り込みを行い、幾つかのSQLクエリを一定時間毎に定期的ワークフロー処理してみる事にします。 因みにどうやってスケジュールを定義するのか・・・ この一連の検証を企画した際に、どうやってスケジュールを設定するのか？が大きな壁だと覚悟していましたが、実際作業を進めていくと・・ schedule = IntervalSchedule(interval=timedelta(seconds=Offset_Sec)) と書くだけでOK!だと解り、以前の叩き台を with Flow(FLOW_NAME, schedule) as flow: とすれば良いと知り・・・・ある意味で愕然としました・・・ また、彼らのホームページやGitHubには沢山のサンプルが有りますので、是非一読されることをお勧め致します。 今回使用するワークフローはこんな感じです・・・ 今回は、以前のモノをベースに必要そうな処理を雪だるま式に追加し、なんとかそれっぽい系のモノに仕上げてみました。毎度おなじみのNDA（ノン・ダメ出し・アグリーメント）ベースで公開しますので、適宜に改造等を行って自由に使ってください。 # # Prefectの検証：少し重たいSQLを振り回せ！版 # # お約束の初期設定 import prefect import datetime from prefect import task, Flow from datetime import timedelta from prefect.schedules import IntervalSchedule import pymysql.cursors # # 広域変数設定 Yes = 1 No = 0 # 各種設定 Offset_Sec = 30 # 処理間隔の設定（この例では30秒間隔） Column_Number = 20 # 今回の参照テーブルのカラム数（ここも実情に併せて変更する） Data_Lines　= 2 　　　　　　　　　　　# 一部出力の際のデータ行数（ここも実情に併せて変更する） Console_Out_ALL = Yes # コンソールに出力するパターンを設定(Yes:全ての情報 No:Data_Lines行だけ) Aggregate = No # 処理を集計型で行うかRawデータ表示を行うかを選択（ここも実情に併せて変更する） # # SigleStoreとの接続情報（適宜変更） SS_Host = "zzz.zzz.zzz.zzz" SS_Port = 3306 SS_User = "zzzzz" SS_Pass = "zzzzz" SS_DB = "zzzzz" SS_Char = "utf8" # FLOW_NAME = "Prefect_Demo_FLOW"　　　　　# 処理フローの名前 TABLE_NAME = "Prefect_Demo_TABLE"　# SingleStoreに作成するテーブルの名前 # ログ出力用メッセージ LOG_MESSAGE1 = "SingleStoreと接続開始" LOG_MESSAGE2 = "SingleStoreと接続完了" LOG_MESSAGE3 = "SingleStore上のテーブル名をクエリします" LOG_MESSAGE4 = "現在登録されているカラム情報は以下の通りです（全部表示）" LOG_MESSAGE5 = "現在登録されているカラム情報は以下の通りです（一部抜粋）" LOG_MESSAGE6 = "SingleStoreとの接続を終了" LOG_MESSAGE7 = "SingleStore連携処理の終了" # SQL関連メッセージ SQL_MESSAGE1 = "カテゴリ別の期間売上" SQL_MESSAGE2 = "カテゴリ別の総売上" SQL_MESSAGE3 = "カテゴリ別の期間出荷" SQL_MESSAGE4 = "カテゴリ別の総出荷" SQL_MESSAGE5 = "配送センター別の期間出荷" SQL_MESSAGE6 = "配送センター別の総出荷" # # SQLクエリで使うSQL文(全データ対象:固定) # # 全てのカラム情報を全部クエリする SQL00 = "SELECT * FROM " + TABLE_NAME + " ORDER BY id" # カテゴリ別の総売上クエリ SQL01 = "SELECT SUM(Payment) as PP, Category FROM " + TABLE_NAME + " GROUP BY Category ORDER BY PP DESC" # カテゴリ別の総出荷数クエリ SQL02 = "SELECT SUM(Units) as UU, Category FROM " + TABLE_NAME + " GROUP BY Category ORDER BY UU DESC" # 配送センター別の総取扱数クエリ SQL03 = "SELECT SUM(Units) as LL, Logistics FROM " + TABLE_NAME + " GROUP BY Logistics ORDER BY LL DESC" # # SingleStoreに接続して処理に必要なポインタを返す # def Open_DB(): # ターゲットのデータベースに接続してポインタを取得 db = pymysql.connect( host = SS_Host, port = SS_Port, user = SS_User, password = SS_Pass, db = SS_DB, charset = SS_Char, cursorclass = pymysql.cursors.DictCursor) return(db)　# データベース処理に必要な情報を戻す # # コンソールに1行単位で全カラム情報を表示する def Print_Query_Data(Data, i, Type): if i % Column_Number == 0: print(Data) if Type == Yes: print("=====================================") else: print("/////////////////////////////////////") return("") # # 指示されたSQLを実行して結果を戻す def Print_SQL_Result(cursor, db, SQL_Data): Tmp_Data = [] cursor.execute(SQL_Data) db.commit() # クエリ結果を取得して利用可能な状態にする for Query_Data in cursor.fetchall(): for item in Query_Data.values(): Tmp_Data.append(item) return(Tmp_Data) # # Prefectで使用するタスクを作成 # @task def Prefect_Batch_Task(): logger = prefect.context.get("logger") # ログ出力の準備 dt_now = datetime.datetime.now() # 現在の時刻を取得 dt_old = dt_now - datetime.timedelta(seconds=Offset_Sec) # 現在時間から設定時間分戻った時刻を生成 # 全てのカラム情報を期間設定してクエリする SQL10 = "SELECT * FROM " + TABLE_NAME + " WHERE ts_SS BETWEEN '" + str(dt_old) + "' AND '" + str(dt_now) + "' ORDER BY id" # カテゴリ別の売上を期間設定集計する SQL11 = "SELECT SUM(Payment) as PP, Category FROM " + TABLE_NAME + " WHERE ts_SS BETWEEN '" + str(dt_old) + "' AND '" + str(dt_now) + "' GROUP BY Category ORDER BY PP DESC" # カテゴリ別の出荷数を期間設定集計する SQL12 = "SELECT SUM(Units) as UU, Category FROM " + TABLE_NAME + " WHERE ts_SS BETWEEN '" + str(dt_old) + "' AND '" + str(dt_now) + "' GROUP BY Category ORDER BY UU DESC" # 配送センター別の出荷数を期間設定集計する SQL13 = "SELECT SUM(Units) as UU, Logistics FROM " + TABLE_NAME + " WHERE ts_SS BETWEEN '" + str(dt_old) + "' AND '" + str(dt_now) + "' GROUP BY Logistics ORDER BY UU DESC" logger.info(LOG_MESSAGE1) # 処理の過程をログで出力 db = Open_DB() # SingleStoreと接続 logger.info(LOG_MESSAGE2) # 処理の過程をログで出力 with db.cursor() as cursor: logger.info(LOG_MESSAGE3) # 処理の過程をログで出力 if Aggregate == Yes: # 集計系のSQL形式で出力 logger.info(SQL_MESSAGE1) logger.info(Print_SQL_Result(cursor, db, SQL11)) logger.info(SQL_MESSAGE2) logger.info(Print_SQL_Result(cursor, db, SQL01)) logger.info(SQL_MESSAGE3) logger.info(Print_SQL_Result(cursor, db, SQL12)) logger.info(SQL_MESSAGE4) logger.info(Print_SQL_Result(cursor, db, SQL02)) logger.info(SQL_MESSAGE5) logger.info(Print_SQL_Result(cursor, db, SQL13)) logger.info(SQL_MESSAGE6) logger.info(Print_SQL_Result(cursor, db, SQL03)) else: # Raw形式で出力 i = 1 DB_Data = "" Tmp_Data = Print_SQL_Result(cursor, db, SQL10) # クエリ処理を行い結果を取得 if Console_Out_ALL == Yes: # 全部を表示 logger.info(LOG_MESSAGE4) for loop in Tmp_Data: DB_Data = DB_Data + " " + str(Tmp_Data[i-1]) if i % Column_Number == 0: DB_Data = Print_Query_Data(DB_Data, i, Yes) i = i + 1 else: # 指定された行数を表示 logger.info(LOG_MESSAGE5) for loop in range(Column_Number * Data_Lines): DB_Data = DB_Data + " " + str(Tmp_Data[i-1]) if i % Column_Number == 0: DB_Data = Print_Query_Data(DB_Data, i, No) i = i + 1 logger.info(LOG_MESSAGE6) # 処理の過程をログで出力 db.close() # SingleStoreとの接続を停止する logger.info(LOG_MESSAGE7) # 処理の過程をログで出力 # # メインの処理 # try: print("Prefect検証開始 : " + datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")) schedule = IntervalSchedule(interval=timedelta(seconds=Offset_Sec)) # スケジュールの設定 with Flow(FLOW_NAME, schedule) as flow: Prefect_Batch_Task() flow.run() except KeyboardInterrupt: print('!!!!! 割り込み発生 !!!!!') finally: print("Prefect検証終了 : " + datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")) ################################################################################################### 最初の検証を実行してみる 最初は、前回動作検証でも使用したRaw形式のクエリを、当該クエリ結果の先頭から指定行数と、当該結果全行の処理に分けて実行してみます。 各検証作業の前には、前回作成したなんちゃって物販データ生成ツールをコマンドラインで起動しておきます。 Raw形式で指定行数のクエリ結果を表示した例・・・ (base) apple@appurunoMacBook-Pro Prefect % /Users/apple/opt/anaconda3/bin/python /Users/apple/Desktop/Prefect/P01.py Prefect検証開始 : 2022/02/01 10:03:14 [2022-02-01 10:03:14+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T01:03:30+00:00 [2022-02-01 10:03:30+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Demo_FLOW' [2022-02-01 10:03:30+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Starting task run... [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続開始 [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続完了 [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStore上のテーブル名をクエリします [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Batch_Task | 現在登録されているカラム情報は以下の通りです（一部抜粋） 1 2022-02-01 10:03:07.620292 2022-02-01 10:03:10.606413 家電 エアコン 64800 1 関東中央物流センター Mastercard 5168403885421431 64800 6480 村上 陽一 266-4452 埼玉県 埼玉県中央区羽折町41丁目24番12号 関東 070-9568-3237 otasayuri@yahoo.com 6480 ///////////////////////////////////// 2 2022-02-01 10:03:09.629209 2022-02-01 10:03:10.610422 雑貨 女性用品 3580 4 伊丹物流センター 現金 N/A 14320 1432 橋本 くみ子 689-3204 大阪府 大阪府横浜市青葉区木立41丁目21番15号 近畿 090-2569-5835 hashimotohideki@fujita.net 716 ///////////////////////////////////// [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreとの接続を終了 [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStore連携処理の終了 [2022-02-01 10:03:30+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Finished task run for task with final state: 'Success' [2022-02-01 10:03:30+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded [2022-02-01 10:03:30+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T01:04:00+00:00 [2022-02-01 10:04:00+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Demo_FLOW' [2022-02-01 10:04:00+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Starting task run... [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続開始 [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続完了 [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStore上のテーブル名をクエリします [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Batch_Task | 現在登録されているカラム情報は以下の通りです（一部抜粋） 12 2022-02-01 10:03:31.701516 2022-02-01 10:03:51.661462 雑貨 ペットフード 980 3 甲州物流センター 現金 N/A 2940 294 伊藤 洋介 097-1867 山梨県 山梨県横浜市旭区四区町5丁目22番6号 独鈷沢コーポ409 中部 090-5952-2300 morinaoto@hashimoto.jp 147 ///////////////////////////////////// 13 2022-02-01 10:03:34.687769 2022-02-01 10:03:52.664464 家電 エアコン 64800 1 伊丹物流センター 現金 N/A 64800 6480 小林 亮介 199-5637 三重県 三重県調布市無栗屋15丁目7番12号 柿木沢新田コート001 近畿 080-4587-0462 wshimizu@nakamura.jp 6480 ///////////////////////////////////// [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreとの接続を終了 [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStore連携処理の終了 [2022-02-01 10:04:00+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Finished task run for task with final state: 'Success' [2022-02-01 10:04:00+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded [2022-02-01 10:04:00+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T01:04:30+00:00 無事に処理されている様なので、今度は対象データ全部を表示してみます。 Raw形式で対象全行数のクエリ結果を表示した例・・・ (base) apple@appurunoMacBook-Pro Prefect % /Users/apple/opt/anaconda3/bin/python /Users/apple/Desktop/Prefect/P01.py Prefect検証開始 : 2022/02/01 09:58:54 [2022-02-01 09:58:54+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T00:59:00+00:00 [2022-02-01 09:59:00+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Demo_FLOW' [2022-02-01 09:59:00+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Starting task run... [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続開始 [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続完了 [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStore上のテーブル名をクエリします [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Batch_Task | 現在登録されているカラム情報は以下の通りです（全部表示） 1 2022-02-01 09:58:54.182879 2022-02-01 09:59:00.157370 雑貨 女性用品 3580 4 平戸物流センター 現金 N/A 14320 1432 井上 春香 444-5180 佐賀県 佐賀県横浜市青葉区権現堂36丁目16番15号 コート細野801 九州・沖縄 09-7535-1393 kumikoyamada@suzuki.jp 716 ===================================== 2 2022-02-01 09:58:55.186754 2022-02-01 09:59:08.164371 雑貨 女性用品 3580 4 東北物流センター JCB 15 digit 349894719659340 14320 1432 田中 健一 033-4260 秋田県 秋田県白井市油井10丁目2番3号 竜泉シティ833 東北 94-7751-5917 naokisato@yahoo.com 716 ===================================== 3 2022-02-01 09:58:58.194997 2022-02-01 09:59:09.167376 書籍 フィクション 1400 3 関東中央物流センター Mastercard 4098910139916154 4200 420 佐藤 晃 199-3030 茨城県 茨城県八王子市幸手4丁目22番1号 コート箪笥町413 関東 070-4562-0870 wtanaka@aoki.jp 126 ===================================== [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreとの接続を終了 [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStore連携処理の終了 [2022-02-01 09:59:00+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Finished task run for task with final state: 'Success' [2022-02-01 09:59:00+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded [2022-02-01 09:59:00+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T00:59:30+00:00 [2022-02-01 09:59:30+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Demo_FLOW' [2022-02-01 09:59:30+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Starting task run... [2022-02-01 09:59:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続開始 [2022-02-01 09:59:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続完了 [2022-02-01 09:59:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStore上のテーブル名をクエリします [2022-02-01 09:59:30+0900] INFO - prefect.Prefect_Batch_Task | 現在登録されているカラム情報は以下の通りです（全部表示） 4 2022-02-01 09:59:00.200775 2022-02-01 09:59:18.169379 酒類 ビール 490 3 道央物流センター 現金 N/A 1470 147 小林 直子 157-9184 北海道 北海道足立区方京22丁目24番2号 コート脚折375 北海道 090-5992-4661 mikakonakamura@ito.jp 29 ===================================== 5 2022-02-01 09:59:01.206683 2022-02-01 09:59:24.178385 酒類 スコッチ 3500 10 讃岐物流センター 現金 N/A 35000 3500 村上 あすか 828-6801 徳島県 徳島県川崎市宮前区細竹9丁目1番13号 四国 50-3008-9131 hideki21@sasaki.jp 700 ===================================== 6 2022-02-01 09:59:03.212448 2022-02-01 09:59:25.178389 書籍 歴史 1500 3 広島臨港物流センター VISA 16 digit 180012400034853 4500 450 福田 あすか 866-9954 島根県 島根県大島町池之端41丁目14番12号 コート押上694 中国 070-2245-5515 watanabesotaro@okamoto.org 135 ===================================== こちらも無事に動いている様ですね。 では、少し重たいSQLを動かしてみる・・・ 今回のメイン検証になりますが、集計系のクエリを数パターン動かしてみたいと思います。 SQL的にはその時点の総数を対象にするクエリと、その時点の範囲対象に含まれるデータに対して実行されるクエリになります。 その時点までの総数を対象にするSQL SELECT SUM(Payment) as PP, Category FROM Prefect_Table GROUP BY Category ORDER BY PP DESC; SELECT SUM(Units) as UU, Category FROM Prefect_Table GROUP BY Category ORDER BY UU DESC; SELECT SUM(Units) as LL, Logistics FROM Prefect_Table GROUP BY Logistics ORDER BY LL DESC; その時点の対象データ範囲で集計を行うSQL SELECT SUM(Payment) as PP, Category FROM Prefect_Table WHERE ts_SS BETWEEN '此処から' AND '此処まで' GROUP BY Category ORDER BY PP DESC; SELECT SUM(Units) as UU, Category FROM Prefect_Table WHERE ts_SS BETWEEN '此処から' AND '此処まで' GROUP BY Category ORDER BY UU DESC; SELECT SUM(Units) as LL, Logistics FROM Prefect_Table WHERE ts_SS BETWEEN '此処から' AND '此処まで' GROUP BY Logistics ORDER BY LL DESC; では、実際にPrefectでFLOW実行させてみます。 集計系のクエリ結果を表示した例・・・ 総数系のSQL文が「実行時総数」になる関係上微妙に異なる結果にはなりますが、最初の段階では、総数系と範囲系は想定通りにほぼ同じ結果を返してきます。SQL文を改良すればより厳密な検証が可能だと思いますが、先を急ぐ・・・・と言うことでこのまま作業を続行します。 (base) apple@appurunoMacBook-Pro Prefect % /Users/apple/opt/anaconda3/bin/python /Users/apple/Desktop/Prefect/P01.py Prefect検証開始 : 2022/02/01 10:08:14 [2022-02-01 10:08:14+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T01:08:30+00:00 [2022-02-01 10:08:30+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Demo_FLOW' [2022-02-01 10:08:30+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Starting task run... [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続開始 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続完了 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStore上のテーブル名をクエリします [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の期間売上 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('137600'), '家電', Decimal('17432'), '雑貨', Decimal('17400'), 'DVD/CD', Decimal('7480'), '書籍'] [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の総売上 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('137600'), '家電', Decimal('17432'), '雑貨', Decimal('17400'), 'DVD/CD', Decimal('7480'), '書籍'] [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の期間出荷 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('12'), '雑貨', Decimal('6'), 'DVD/CD', Decimal('4'), '書籍', Decimal('3'), '家電'] [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の総出荷 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('12'), '雑貨', Decimal('8'), 'DVD/CD', Decimal('4'), '書籍', Decimal('3'), '家電'] [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | 配送センター別の期間出荷 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('9'), '讃岐物流センター', Decimal('7'), '伊丹物流センター', Decimal('3'), '平戸物流センター', Decimal('3'), '広島臨港物流センター', Decimal('1'), '道央物流センター', Decimal('1'), '関東中央物流センター', Decimal('1'), '東北物流センター'] [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | 配送センター別の総出荷 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('11'), '讃岐物流センター', Decimal('7'), '伊丹物流センター', Decimal('3'), '平戸物流センター', Decimal('3'), '広島臨港物流センター', Decimal('1'), '関東中央物流センター', Decimal('1'), '東北物流センター', Decimal('1'), '道央物流センター'] [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreとの接続を終了 [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Batch_Task | SingleStore連携処理の終了 [2022-02-01 10:08:30+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Finished task run for task with final state: 'Success' [2022-02-01 10:08:30+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded [2022-02-01 10:08:30+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T01:09:00+00:00 暫く生成処理が進むと、総数側の処理結果と範囲指定の結果の幅が大きくなり始めますが、30秒間隔でデータの今を見る事が普通に出来る事が確認できました。 [2022-02-01 10:09:00+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Demo_FLOW' [2022-02-01 10:09:00+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Starting task run... [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続開始 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreと接続完了 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStore上のテーブル名をクエリします [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の期間売上 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('130400'), '家電', Decimal('28460'), '酒類', Decimal('19920'), 'DVD/CD', Decimal('14280'), '書籍', Decimal('8912'), '雑貨'] [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の総売上 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('268000'), '家電', Decimal('37320'), 'DVD/CD', Decimal('28460'), '酒類', Decimal('26344'), '雑貨', Decimal('21760'), '書籍'] [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の期間出荷 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('13'), '酒類', Decimal('9'), '書籍', Decimal('8'), '雑貨', Decimal('8'), 'DVD/CD', Decimal('3'), '家電'] [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | カテゴリ別の総出荷 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('20'), '雑貨', Decimal('14'), 'DVD/CD', Decimal('13'), '書籍', Decimal('13'), '酒類', Decimal('6'), '家電'] [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | 配送センター別の期間出荷 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('10'), '関東中央物流センター', Decimal('9'), '伊丹物流センター', Decimal('8'), '讃岐物流センター', Decimal('8'), '平戸物流センター', Decimal('4'), '広島臨港物流センター', Decimal('2'), '甲州物流センター'] [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | 配送センター別の総出荷 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | [Decimal('17'), '讃岐物流センター', Decimal('16'), '伊丹物流センター', Decimal('11'), '関東中央物流センター', Decimal('11'), '平戸物流センター', Decimal('7'), '広島臨港物流センター', Decimal('2'), '甲州物流センター', Decimal('1'), '道央物流センター', Decimal('1'), '東北物流センター'] [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStoreとの接続を終了 [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Batch_Task | SingleStore連携処理の終了 [2022-02-01 10:09:00+0900] INFO - prefect.TaskRunner | Task 'Prefect_Batch_Task': Finished task run for task with final state: 'Success' [2022-02-01 10:09:00+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded [2022-02-01 10:09:00+0900] INFO - prefect.Prefect_Demo_FLOW | Waiting for next scheduled run at 2022-02-01T01:09:30+00:00 簡単なPython処理をタスクに設定し、Prefectのワークフローで普通に活用出来そうですね。 今回のまとめ 今回は、以前のPythonスクリプトを雪だるま方式で改造し、少し複雑なSQLをSingleStoreに処理させるワークフローを想定・検証してみました。結果的には非常にシンプル且つ簡単にワークフローをPythonワールドで書く事が出来ると同時に、Prefect自体も確実に狙った操作をしてくれる事が解りました。 クエリ結果を取り出す事が出来るようになったので、次回は追加で解析系ライブラリとの連携ワークフローに挑戦してみたいと思います。

忘れてました・・・検証で使用する道具はこれです。。 今回の検証で使用する「なんちゃって物販データ」生成用のツールはこれになります。毎度で恐縮ですが、NDA（ノン・ダメ出し・アグリーメント）ベースでご自由にお使いください。 # # Prefect検証用・仮想物販データ生成ツール # # Python 3版 # # 初期設定 import sys stdout = sys.stdout sys.stdout = stdout import pymysql.cursors import datetime import time import random import re # Generate_Data = 1000　 # 生成するデータの数 Wait_Time = 5　　　　　　　　　　　　　　　 # 生成開始までの待ち時間 Table_Name = "Prefect_Demo_TABLE"　# SingleStoreに作成するテーブルの名前 # # SigleStoreとの接続情報（適宜変更） SS_Host = "zzz.zzz.zzz.zzz" SS_Port = 3306 SS_User = "zzzzz" SS_Pass = "zzzzz" SS_DB = "zzzzz" SS_Char = "utf8" # # 物販情報で使うメタデータの定義 # カテゴリ名 Category_Name = ["酒類","家電","書籍","DVD/CD","雑貨"] # 酒類の商品情報 Product_Name0 = ["日本酒","バーボン","ビール","芋焼酎","赤ワイン","白ワイン","スコッチ","ブランデー","泡盛","テキーラ"] Product_Price0 = [1980, 2500, 490, 2000, 3000, 2500, 3500, 5000, 1980, 2000] # 家電の商品情報 Product_Name1 = ["テレビ","洗濯機","ラジオ","ステレオ","電子レンジ","パソコン","電池","エアコン","乾燥機","掃除機"] Product_Price1 = [49800, 39800, 2980, 88000, 29800, 64800, 198, 64800, 35800, 24800] # 書籍の商品情報 Product_Name2 = ["週刊誌","歴史","写真集","漫画","参考書","フィクション","経済","自己啓発","月刊誌","新刊"] Product_Price2 = [280, 1500, 2500, 570, 1480, 1400, 1800, 1540, 980, 1980] # DVD/CDの商品情報 Product_Name3 = ["洋楽","演歌","Jポップ","洋画","アイドル","クラッシック","邦画","連続ドラマ","企画","アニメ"] Product_Price3 = [1980, 2200, 2500, 3500, 2980, 1980, 3800, 2690, 1980, 2400] # 雑貨の商品情報 Product_Name4 = ["洗剤","電球","贈答品","医薬部外品","ペットフード","乾電池","文房具","男性用品","女性用品","季節用品"] Product_Price4 = [498, 198, 1980, 398, 980, 248, 398, 2980, 3580, 1980] # #　地域名ルックアップ情報（キーは都道府県名） Area_Data={'北海道':'北海道','青森県':'東北','岩手県':'東北','宮城県':'東北','秋田県':'東北','山形県':'東北','福島県':'東北', '茨城県':'関東','栃木県':'関東','群馬県':'関東','埼玉県':'関東','千葉県':'関東','東京都':'関東','神奈川県':'関東', '新潟県':'中部','富山県':'中部','石川県':'中部','福井県':'中部','山梨県':'中部','長野県':'中部','岐阜県':'中部','静岡県':'中部','愛知県':'中部', '三重県':'近畿','滋賀県':'近畿','京都府':'近畿','大阪府':'近畿','兵庫県':'近畿','奈良県':'近畿','和歌山県':'近畿', '鳥取県':'中国','島根県':'中国','岡山県':'中国','広島県':'中国','山口県':'中国', '徳島県':'四国','香川県':'四国','愛媛県':'四国','高知県':'四国', '福岡県':'九州・沖縄','佐賀県':'九州・沖縄','長崎県':'九州・沖縄','熊本県':'九州・沖縄','大分県':'九州・沖縄','宮崎県':'九州・沖縄','鹿児島県':'九州・沖縄','沖縄県':'九州・沖縄'} # 物流センタールックアップ情報（キーは地域名） Logi_Data = {'北海道':'道央物流センター','東北':'東北物流センター','関東':'関東中央物流センター', '中部':'甲州物流センター','近畿':'伊丹物流センター','中国':'広島臨港物流センター','四国':'讃岐物流センター','九州・沖縄':'平戸物流センター'} # 購入ポイント情報（カテゴリ名の順番に設定 Point_Data = [0.02, 0.1, 0.03, 0.02, 0.05] # 消費税率の設定 Tax_Data = 0.1 # # テーブル定義 DC0 = "id BIGINT AUTO_INCREMENT, ts_SS TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6), ts DATETIME(6), " DC1 = "Category VARCHAR(20), Product VARCHAR(20), Price INT, Units INT, Logistics VARCHAR(20), " DC2 = "Card VARCHAR(40), Number VARCHAR(30), Payment INT, Tax INT, " DC3 = "User VARCHAR(20), Zip VARCHAR(10), Prefecture VARCHAR(10), Address VARCHAR(60), Area VARCHAR(10), Tel VARCHAR(15), Email VARCHAR(40), Point INT, " DC4 = "SHARD KEY (Logistics), PRIMARY KEY(id,Logistics)" # # 書き込み用のカラム設定 DL0 = "ts, " # タイムスタンプ情報 DL1 = "Category, Product, Price, Units, Logistics, " # ビジネス情報 DL2 = "Card, Number, Payment, Tax, " # 支払い情報 DL3 = "User, Zip, Prefecture, Address, Area, Tel, Email, Point" # 顧客情報 # # テーブル初期化SQL Table_Init = "DROP TABLE IF EXISTS " + Table_Name # # テーブル作成用SQL Table_Create = "CREATE TABLE IF NOT EXISTS " + Table_Name + "(" + DC0 + DC1 + DC2 + DC3 + DC4 + ")" # # データベースとの接続 # def Open_DB(): db = pymysql.connect(host = SS_Host, port = SS_Port, user = SS_User, password = SS_Pass, db = SS_DB, charset = SS_Char, cursorclass=pymysql.cursors.DictCursor) return(db) try: # Fakerの初期化 from faker import Faker fakegen = Faker('ja_JP') Faker.seed(fakegen.random_digit()) print("検証用テーブル作成開始 : " + datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")) db = Open_DB() # SingleStoreに接続 with db.cursor() as cursor: # 検証作業用テーブルの初期化 cursor.execute(Table_Init) db.commit() #　新規にテーブルを作成 cursor.execute(Table_Create) db.commit() print("検証用テーブル作成終了 : " + datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")) print(str(Wait_Time) + "秒後にSingleStore上へのデータの生成を開始します。") time.sleep(Wait_Time) print("検証用データ生成開始 : " + datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")) dt_now = datetime.datetime.now() # 時間情報を生成する起点を確保 Loop_Counter = 0 # ループカウンターの初期化 while Loop_Counter < Generate_Data: # 検証データの生成 # ts用のデータを生成 Sec = fakegen.random_digit() MIL_Sec = fakegen.random_digit() MIC_Sec = fakegen.random_digit() # 現実的なタイムスタンプ情報として利用します（秒単位でデータをズラしてBI等で使い易くする） ts_now = dt_now + datetime.timedelta(seconds=Sec, milliseconds=MIL_Sec, microseconds=MIC_Sec) dt_now = ts_now #　生成データを次回の起点に変更 # ランダムに書き込む商材の種類と商品IDを選択 Category_ID = fakegen.random_digit() if Category_ID > 4: Category_ID = Category_ID - 5 Product_ID = fakegen.random_digit() # 割り当てられたカテゴリ内の商材IDを選択 Category = Category_Name[Category_ID] # カテゴリ名の設定 # カラム情報の設定 if Category_ID == 0: # 酒類 Product = Product_Name0[Product_ID] Price = Product_Price0[Product_ID] Units = fakegen.random_digit() + 1 # リアルっぽく調整しています Point = Price * Units * Point_Data[Category_ID] elif Category_ID == 1: # 家電 Product = Product_Name1[Product_ID] Price = Product_Price1[Product_ID] Units = 1 # リアルっぽく調整しています Point = Price * Units * Point_Data[Category_ID] elif Category_ID == 2: # 書籍 Product = Product_Name2[Product_ID] Price = Product_Price2[Product_ID] Units = fakegen.random_digit() + 1 if Units >3: Units = 3 # リアルっぽく調整しています Point = Price * Units * Point_Data[Category_ID] elif Category_ID == 3: # DVD/CD Product = Product_Name3[Product_ID] Price = Product_Price3[Product_ID] Units = fakegen.random_digit() + 1 if Units >2: Units = 2 # リアルっぽく調整しています Point = Price * Units * Point_Data[Category_ID] else: # 雑貨 Product = Product_Name4[Product_ID] Price = Product_Price4[Product_ID] Units = fakegen.random_digit() + 1 if Units >4: Units = 4 # リアルっぽく調整しています Point = Price * Units * Point_Data[Category_ID] # 支払い情報の設定 if str(fakegen.pybool()) == "True": Card = "現金" else: Card = fakegen.credit_card_provider() Number = fakegen.credit_card_number() if Card == "現金": Number = "N/A" # 支払い総額と消費税 Payment = Units * Price Tax = int(Payment * Tax_Data) # 購入者情報の生成 User = fakegen.name() Zip = fakegen.zipcode() Address = fakegen.address() Tel = fakegen.phone_number() Email = fakegen.ascii_email() # 都道府県情報の抽出 pattern = u"東京都|北海道|(?:京都|大阪)府|.{2,3}県" m = re.match(pattern , Address) if m: Prefecture = m.group() # 地域名と物流センター名を取得 Area = Area_Data.get(Prefecture) Logistics = Logi_Data.get(Area) #　SQLで使用するデータ列の作成 DV0 = str(ts_now)+"','" DV1 = Category + "','" + Product + "','" + str(Price) + "','" + str(Units) + "','" + Logistics + "','" DV2 = Card + "','" + Number + "','" + str(Payment) + "','" + str(Tax) + "','" DV3 = User + "','" + Zip + "','" + Prefecture + "','" + Address + "','" + Area + "','" + Tel + "','" + str(Email) + "','" + str(Point) SQL = "INSERT INTO " + Table_Name +"(" + DL0 + DL1 + DL2 + DL3 + ") VALUES('" + DV0 + DV1 + DV2 + DV3 + "')" # クエリ用のSQLを作成 # SingleStoreへの書き込み cursor.execute(SQL) db.commit() time.sleep(random.randint(1,3)) # １－3の乱数で次のデータ生成を待機する(適宜調整可) Loop_Counter = Loop_Counter + 1 # ループカウンタの更新 except KeyboardInterrupt: print('!!!!! 割り込み発生 !!!!!') finally: db.close() # データベースコネクションを閉じる print("生成したデータの総数 : " + str(Loop_Counter)) print("検証用データ作成終了 : " + datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")) ################################################################################################### 今回の検証では、このツールをコマンドラインで起動して使います。 今回のまとめ 今回は、検証に使用するツールを「超シンプル」に紹介させて頂きました。 次回は、前回のPrefect検証スクリプトを使って「雪だるま式開発手法」にて少し込み入ったSQLを複数投げるワークフローに挑戦してみたいと思います。

今回は・・・ 前回は、Pythonワールドでワークフローの仕組みが書けそうな・・・Prefectについての環境導入を行いましたので、今回からは少し深掘り検証モードに入って行きたいと思います。 シンプルにデータベースを転がせるか？・・・・・ 最初の検証は、指定したデータベースの状況を定期的に処理する、超簡単系ワークフローを作ってみたいと思います。 まずは、彼らのホームページから検証の叩き台をお借りして、少しカスタマイズを行います。 なお、コードに関しては・・・NDA（ノン・ダメ出し・アグリーメント）を前提にさせて頂きますので、自由にカスタマイズして使って頂いて結構です。 # # Prefectの検証たたき台（彼らのホームページより） # import prefect from prefect import task, Flow FLOW_NAME = "Prefect_Test01" LOG_MESSAGE = "Hello world!" @task def test01_task(): logger = prefect.context.get("logger") logger.info(LOG_MESSAGE) with Flow(FLOW_NAME) as flow: test01_task() flow.run() サクッと！実行してみます。 (base) apple@appurunoMacBook-Pro Prefect % /Users/apple/opt/anaconda3/bin/python /Users/apple/Desktop/Prefect/prefect01.py [2022-01-30 09:54:19+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Test01' [2022-01-30 09:54:19+0900] INFO - prefect.TaskRunner | Task 'test01_task': Starting task run... [2022-01-30 09:54:19+0900] INFO - prefect.test01_task | Hello world! [2022-01-30 09:54:19+0900] INFO - prefect.TaskRunner | Task 'test01_task': Finished task run for task with final state: 'Success' [2022-01-30 09:54:19+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded (base) apple@appurunoMacBook-Pro Prefect % 無事に動いている様なので、以前に作成したSingleStore向けMySQLのフリして繋げる・・を追加してみます。 SingleStoreに関しては、検証用に導入済みのMBPのローカルDocker版を使っています。 現在も引き続き条件付きフリー使用が継続している様なので、ご興味があれば是非メアドを登録して使ってみてください。 また、使用するテーブルも以前の検証で作成した「なんちゃって物販データ」の既存テーブルを転用してアクセスしてみます。 # Prefectの検証たたき台 # # お約束の初期設定 import prefect from prefect import task, Flow # # MySQL接続を利用する準備 import pymysql.cursors # # 広域変数設定 Yes = 1 No = 0 # Console_Out_ALL = No # コンソールに出力するパターンを設定(Yes:全ての情報 No:3行だけ) Column_Number = 21 # 今回の参照テーブルのカラム数 Data_Lines = 2 # 一部出力の際のデータ行数 # # 処理フローの名前 FLOW_NAME = "Prefect_Test02" # # ログ出力用メッセージ LOG_MESSAGE1 = "SingleStoreと接続開始" LOG_MESSAGE2 = "SingleStoreと接続完了" LOG_MESSAGE3 = "SingleStore上のテーブル名をクエリします" LOG_MESSAGE4 = "現在登録されているカラム情報は以下の通りです（全部表示）" LOG_MESSAGE5 = "現在登録されているカラム情報は以下の通りです（一部抜粋）" LOG_MESSAGE6 = "SingleStoreとの接続を終了" LOG_MESSAGE7 = "SingleStore連携処理の終了" # # SigleStoreとの接続情報（適宜変更） SS_Host = "zzz.zzz.zzz.zzz" SS_Port = 3306 SS_User = "zzzzz" SS_Pass = "zzzzz" SS_DB = "zzzzz" SS_Char = "utf8" # # SQLクエリで使うSQL文 SQL = "SELECT * FROM RTS_Demo_Table ORDER BY id_SS" # # SingleStoreに接続して処理に必要な情報を返す # def Open_DB(): db = pymysql.connect( host = SS_Host, port = SS_Port, user = SS_User, password = SS_Pass, db = SS_DB, charset = SS_Char, cursorclass = pymysql.cursors.DictCursor) return(db) # データベース処理に必要な情報を戻す # # コンソールに1行単位で表示する # def Print_Query_Data(Data, i, Type): if i % Column_Number == 0: print(Data) if Type == Yes: print("=====================================") else: print("/////////////////////////////////////") return("") # # Prefectで使用するタスクを作成 # @task def test02_task(): logger = prefect.context.get("logger") # ログ出力の準備 Tmp_Data=[] logger.info(LOG_MESSAGE1) # 処理の過程をログで出力 db = Open_DB() # SingleStoreと接続 logger.info(LOG_MESSAGE2) # 処理の過程をログで出力 with db.cursor() as cursor: # 処理の過程をログで出力 logger.info(LOG_MESSAGE3) # SingleStoreにSQLクエリを発行する cursor.execute(SQL) db.commit() # クエリ結果を取得して利用可能な状態にする for Query_Data in cursor.fetchall(): for item in Query_Data.values(): Tmp_Data.append(item) # クエリ結果の表示 i = 1 DB_Data = "" if Console_Out_ALL == Yes: # 全部を表示 logger.info(LOG_MESSAGE4) for loop in Tmp_Data: DB_Data = DB_Data + " " + str(Tmp_Data[i-1]) if i % Column_Number == 0: DB_Data = Print_Query_Data(DB_Data, i, Yes) i = i + 1 else: # 指定された行数を表示 logger.info(LOG_MESSAGE5) for loop in range(Column_Number * Data_Lines): DB_Data = DB_Data + " " + str(Tmp_Data[i-1]) if i % Column_Number == 0: DB_Data = Print_Query_Data(DB_Data, i, No) i = i + 1 logger.info(LOG_MESSAGE6) # 処理の過程をログで出力 db.close() # SingleStoreとの接続を停止する logger.info(LOG_MESSAGE7) # 処理の過程をログで出力 # # メインの処理 # with Flow(FLOW_NAME) as flow: test02_task() flow.run() ################################################ サクッと！動かしてみます。 (base) apple@appurunoMacBook-Pro Prefect % /Users/apple/opt/anaconda3/bin/python /Users/apple/Desktop/Prefect/prefect02.py [2022-01-30 09:59:44+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Test02' [2022-01-30 09:59:44+0900] INFO - prefect.TaskRunner | Task 'test02_task': Starting task run... [2022-01-30 09:59:44+0900] INFO - prefect.test02_task | SingleStoreと接続開始 [2022-01-30 09:59:44+0900] INFO - prefect.test02_task | SingleStoreと接続完了 [2022-01-30 09:59:44+0900] INFO - prefect.test02_task | SingleStore上のテーブル名をクエリします [2022-01-30 09:59:44+0900] INFO - prefect.test02_task | 現在登録されているカラム情報は以下の通りです（一部抜粋） 1 2021-11-19 10:24:09.441558 2021-11-19 10:24:16.423722 雑貨 贈答品 1980 3 297 5940 594 American Express 180091589062760 高橋 篤司 113-2169 JP-38 愛媛県 愛媛県袖ケ浦市芝公園7丁目10番7号 アーバン高田馬場436 090-5030-6067 itoyui@aoki.com 四国 讃岐物流センター ///////////////////////////////////// 2 2021-11-19 10:24:10.339402 2021-11-19 10:24:16.424731 DVD/CD 洋画 3500 2 140 7000 700 Mastercard 4819783616319228 佐藤 英樹 699-9934 JP-18 福井県 福井県神津島村土呂部8丁目23番13号 080-2980-5899 shota95@yamada.jp 中部 甲州物流センター ///////////////////////////////////// [2022-01-30 09:59:44+0900] INFO - prefect.test02_task | SingleStoreとの接続を終了 [2022-01-30 09:59:44+0900] INFO - prefect.test02_task | SingleStore連携処理の終了 [2022-01-30 09:59:44+0900] INFO - prefect.TaskRunner | Task 'test02_task': Finished task run for task with final state: 'Success' [2022-01-30 09:59:44+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded (base) apple@appurunoMacBook-Pro Prefect % 表示の関係で最初から2行分を出力していますが、無事に通常のPython系MySQLアクセスが出来る事を確認できました。 因みに100行全部出力すると・・・・ (base) apple@appurunoMacBook-Pro Prefect % /Users/apple/opt/anaconda3/bin/python /Users/apple/Desktop/Prefect/prefect02.py [2022-01-30 10:15:39+0900] INFO - prefect.FlowRunner | Beginning Flow run for 'Prefect_Test02' [2022-01-30 10:15:39+0900] INFO - prefect.TaskRunner | Task 'test02_task': Starting task run... [2022-01-30 10:15:39+0900] INFO - prefect.test02_task | SingleStoreと接続開始 [2022-01-30 10:15:39+0900] INFO - prefect.test02_task | SingleStoreと接続完了 [2022-01-30 10:15:39+0900] INFO - prefect.test02_task | SingleStore上のテーブル名をクエリします [2022-01-30 10:15:39+0900] INFO - prefect.test02_task | 現在登録されているカラム情報は以下の通りです（全部表示） 1 2021-11-19 10:24:09.441558 2021-11-19 10:24:16.423722 雑貨 贈答品 1980 3 297 5940 594 American Express 180091589062760 高橋 篤司 113-2169 JP-38 愛媛県 愛媛県袖ケ浦市芝公園7丁目10番7号 アーバン高田馬場436 090-5030-6067 itoyui@aoki.com 四国 讃岐物流センター ===================================== 2 2021-11-19 10:24:10.339402 2021-11-19 10:24:16.424731 DVD/CD 洋画 3500 2 140 7000 700 Mastercard 4819783616319228 佐藤 英樹 699-9934 JP-18 福井県 福井県神津島村土呂部8丁目23番13号 080-2980-5899 shota95@yamada.jp 中部 甲州物流センター ===================================== 3 2021-11-19 10:24:10.974893 2021-11-19 10:24:25.424737 雑貨 乾電池 248 1 12 248 24 現金 N/A 佐々木 千代 579-1085 JP-42 長崎県 長崎県長生郡一宮町松浦町26丁目17番12号 月島コーポ240 070-2421-9200 ytanaka@yamamoto.com 九州・沖縄 平戸物流センター ===================================== 4 2021-11-19 10:24:11.704003 2021-11-19 10:24:29.426744 DVD/CD アイドル 2980 2 119 5960 596 JCB 16 digit 180011672869244 中村 あすか 283-8664 JP-12 千葉県 千葉県横浜市西区細竹11丁目25番8号 070-3277-9886 sayurisasaki@ota.com 関東 関東中央物流センター ===================================== ----- 途中省略 ------- 95 2021-11-19 10:25:15.206949 2021-11-19 10:30:05.831141 雑貨 医薬部外品 398 4 79 1592 159 Diners Club / Carte Blanche 30280001215195 藤井 加奈 451-8413 JP-26 京都府 京都府杉並区南赤田40丁目15番9号 クレスト松が谷520 090-3696-8259 xfujita@hayashi.jp 近畿 伊丹物流センター ===================================== 96 2021-11-19 10:25:15.712581 2021-11-19 10:30:10.837144 家電 テレビ 49800 1 4980 49800 4980 VISA 13 digit 4583175589485926 坂本 涼平 173-7986 JP-15 新潟県 新潟県印旛郡本埜村平須賀21丁目10番17号 権現堂クレスト469 070-4714-1060 reitakahashi@gmail.com 中部 甲州物流センター ===================================== 97 2021-11-19 10:25:16.605000 2021-11-19 10:30:14.840148 書籍 漫画 570 3 51 1710 171 現金 N/A 斎藤 結衣 426-0651 JP-31 鳥取県 鳥取県印旛郡栄町虎ノ門虎ノ門ヒルズ森タワー28丁目19番6号 080-0512-0117 gfukuda@hotmail.com 中国 広島臨港物流センター ===================================== 98 2021-11-19 10:25:17.408374 2021-11-19 10:30:23.847156 書籍 新刊 1980 3 178 5940 594 Discover 3568815083887561 小川 あすか 052-3746 JP-40 福岡県 福岡県八丈島八丈町隼町8丁目24番13号 湯宮アーバン379 83-8825-5209 kobayashitomoya@hotmail.com 九州・沖縄 平戸物流センター ===================================== 99 2021-11-19 10:25:18.221668 2021-11-19 10:30:31.853157 DVD/CD Jポップ 2500 2 100 5000 500 現金 N/A 福田 陽子 328-2222 JP-04 宮城県 宮城県長生郡長柄町細竹26丁目16番17号 幸手コーポ025 070-1979-9296 khasegawa@maeda.jp 東北 東北物流センター ===================================== 100 2021-11-19 10:25:18.771961 2021-11-19 10:30:32.860161 雑貨 電球 198 4 39 792 79 JCB 15 digit 6569962741550363 佐藤 浩 703-9742 JP-38 愛媛県 愛媛県川崎市宮前区勝どき27丁目26番12号 070-0402-6939 nishimurajun@hotmail.com 四国 讃岐物流センター ===================================== [2022-01-30 10:15:39+0900] INFO - prefect.test02_task | SingleStoreとの接続を終了 [2022-01-30 10:15:39+0900] INFO - prefect.test02_task | SingleStore連携処理の終了 [2022-01-30 10:15:39+0900] INFO - prefect.TaskRunner | Task 'test02_task': Finished task run for task with final state: 'Success' [2022-01-30 10:15:39+0900] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded (base) apple@appurunoMacBook-Pro Prefect % となります。Docker利用とはいえ、SingleStoreのインメモリ性能が高いので、今後の「少し複雑系SQLアクセス」でも非常に期待が持てるかと思います。 余談ですが、国内のSingleStore性能評価で「十数億行のデータに対して抽出・集計系のSQLを仕掛けて、3秒を切る性能」が出ているとの事です。 今回のまとめ 今回は、初めてのPrefect第2弾として「Python経験のMySQL接続でSingleStoreが接続出来るか？」を検証してみました。結果的には、実にサクッと！使える事が確認でき、次回以降の展開が非常に期待出来る状況になってきました。 次回は、今回の「雪だるま方式開発手法」（勝手に命名。。）で作成・検証を行ったコードに、新たに「スケジュール機能」を組み込んでみたいと思います。また、その検証としてSingleStoreにランダムにデータを生成・挿入する仕組みを併用し、「初歩の短サイクルバッチ処理」的な作業を行いたいと思います。

バッチ処理問題・・・ Equalumの紹介・検証を行っていると、「バッチ処理・・」という質問・確認を頂くケースが増えており、そろそろ「それなりの提案」をしないといけないな・・と覚悟を決め始めている今日この頃なので、今回はその辺にフォーカスした検証等を実施してみたいと思います。 Equalum環境でのバッチ処理について Equalumの強みは、圧倒的に高い即時性と低遅延を優れたCDC技術や、最先端の実績あるクラウド系技術と有機的に組み合わせる事により、極めてハイスピードでノーコード設定されているFLOW処理を行いながら、指定されたターゲットのデータベース上にストリーミング処理を行える点に有ります。 また、ストリーミングの過程のタイミングを活かして、結構な種類のETL的FLOW処理をノーコードで設定出来ますので、かなりの領域までEqualumだけでデータ利活用をサポート出来るのでは？と考えております・・・ でも！ But!・・しかし！！！！ オリジナル側のデータベースの変化が、即時に同期されてし状況がターゲットデータベースに反映されても、存在しないデータを勝手に作る事はできませんので、集計系を中心に「ストリーミングだけでは可能にならない」作業も依然として存在することも事実です。 ですので、これらの「貯めてドン！系作業」をターゲットのデータシステム上で上手く実行する方法・・・を何とか出来ないかという発想と、Python環境で一般的に利用可能な「各種のデータ関連モジュール」もついでに組み込めれば助かる！という事で、今回のPrefect検証の実施に至りました。 そこで今回は、「ストリーミング後の利活用側データベース」上で効率良く「データ移動の為のバッチ処理」ではなく、「王道のデータ処理の為のバッチ（的）処理」環境構築・・という感じで検証を進めて行きたいと思います。 Prefectなる謎のソリューション・・・？？ 色々とGoogle先生にお伺いを立てていたところ、PrefectなるPython環境が有る！という事をご提案頂き、今回はその線でPrefectの検証を行ってみたいと思います。 彼らのソリューションは、ホームページから頂くと・・・・ イノベーションのためのエンジン と書かれており、概要的にはPrefectのPythonライブラリを何時もの様に環境に導入し、Pythonワールドでワークフローをアプリケーションレベルで設計、構築、テスト、実行するための全ての必要なモノが提供されている・・・・との事です。（素晴らしい！） また、専用のUI環境も有るとの事なので、稼働後の維持管理も楽そうな感じなので・・・・非常に期待度大！なソリューションだと言えるでしょう。 では、先ずは環境の導入から・・・・ 今回は、オンプレミス環境でのお試しを想定していますので、彼らのオープンソース環境をローカルに導入する事にします。 トップページのPRODUCTからGET THE CODEを選択します。 左側のGET THE CODEボタンを選択します。 無事にGitHubへ移動できました。 一応GitHubが確認できましたので、Prefectの環境を実際に導入したいと思います。 お約束のPiP更新から・・ % pip install --upgrade pip Prefectのインストール % pip install prefect 因みに事務所のWindows環境で、ユーザ云々のエラーが出ましたので、 $ pip install --user prefect のオプション付きを使って導入しました。Macの環境は最初のパターンで導入出来ると思います。 こんな感じで導入が始まります。 無事にインストールが終わったら、導入されたバージョンを確認してみます。 因みに、その他の導入方法としては・・・・ % conda install -c conda-forge prefect % pipenv install --pre prefect が有る様なので、環境に合わせて選択が可能です。 % prefect version この時点では0.15.13が入ってきました。 無事に導入出来た様なので、次は彼らの概要にも書かれているUI環境を確認してみたいと思います。 UI環境を確認してみる・・・・ 彼らのドキュメントによれば、通常導入の場合デフォルトが彼らのクラウド環境を、利用の際のバックエンドとして使用するようになっている様なので、コマンドを使ってPrefect Serverをローカルで動く様に構成します。 % prefect backend server クラウドにする場合は % prefect backend cloud とすれば良い様です。 バックエンドが動き始めた様です。 次に、Prefectのサーバを起動します。 % prefect server start 必要な環境構築が自動的に始まります。 暫くすると自動的にサーバがローカルで立ち上がりますので、http://localhost:8080をアクセスしてUIを見てみます。 まっさらなダッシュボードが出てきました。 エージェントなるものを走らせてみる・・・・ 先程のダッシュボードに表示されている中央下段のロボットアイコンのagentが、現状起動されていない！という赤色表示ですので、コマンドを使って1個ローカルにエージェントを起動させてみます。 先程のサーバを起動したターミナルとは別のターミナルを起動し以下のコマンドを使います。 % prefect agent local start 非常にシンプルで解り易いコマンドを発行すると・・・・・ サクッと！エージェントが立ち上がってきました。 では！お約束のHello World!を走らせてみる・・・・・ では、今回の最後の検証として「お約束のHello World!」を走らせてみたいと思います。 ソースコードは、彼らのドキュメントから使わせて頂きます。 import prefect from prefect import task, Flow @task def hello_task(): logger = prefect.context.get("logger") logger.info("Hello world!") with Flow("hello-flow") as flow: hello_task() flow.run() サクッ！と起動すると・・・・ 無事に動いた様です。 今回のまとめ 今回は、検証環境としてのPrefect導入と、彼らの公式ドキュメントに出ているHello Worldを動かしてみました。次回以降では、もう少し処理ステップを複雑・高度化し、既存のPython環境とのコラボレーションに関する可能性を深掘りしてみたいと思います。 狙いとしては、Equalumでオリジナル側から即時・低遅延同期で展開されるターゲット側のテーブルを、Python環境をフルに活用しての「バッチ処理的」仕組みの検証まで行ければと考えております。 なお、下記の先輩記事も参考にさせて頂きました。この場をお借りして御礼申し上げます。

はじめに poetry は Python のパッケージマネージャとしてデファクトになりつつある（個人的な感想）。 poetry では、以下の二つのファイルを使って、ライブラリを管理している * pyproject.toml ・・・ 使っている Python やライブラリのバージョンを管理 * poetry.lock ・・・ 各ライブラリの実際のバージョンや依存関係を管理 poetry.lock ではライブラリのハッシュ値も持つことで、 poetry install 時のライブラリの探索時間を短くしている。 問題点 poetry.lock はライブラリごとのハッシュ値を持っているので、インストールする先のOSが異なったりすると、以下のようなエラーになる。 • Installing libclang (12.0.0): Failed RuntimeError Retrieved digest for link libclang-12.0.0-2-py2.py3-none-win_amd64.whl(sha256:46414009fcee8375ba64ea6c2c43c5b80a63e3a8b679f4293e00aa605b7265aa) not in poetry.lock metadata ['sha256:3b0585dbdc3f3b372340f1efe7c28bf4a5b658d2fb8dc0544a7ef0d2bef40618', 'sha256:6df2f8a2cb75181e3c1e99e5dfca2fd44bb7f84ed12d5112541e03c10384f306', 'sha256:b828cb52cf3f02fb0e0a8fddb9dece7e2ed006f8b4d54ee811cef0471d414367', 'sha256:275126823c60ab5c9fae6a433cbb6a47e4d1b5f668a985fbd6065553dbc7efcc', 'sha256:fadad3bf5fbab50c996eb151adc58c3a7cbee45a9135060c416be7b640372112'] at /usr/local/lib/python3.8/site-packages/poetry/installation/chooser.py:113 in _get_links 109│ 110│ selected_links.append(link) 111│ 112│ if links and not selected_links: → 113│ raise RuntimeError( 114│ "Retrieved digest for link {}({}) not in poetry.lock metadata {}".format( 115│ link.filename, h, hashes 116│ ) 117│ ) poetry.lock を削除した状態で poetry install を行うと、上記エラーは発生しなくなるが、 その代わりに各ライブラリの依存関係などをそのタイミングで確認するので、すごく時間がかかる。 解決策 にあるように、 poetry export をして、 pip install -rを行えばよい。 ただ、上記のIssueにあるように最後に poetry install をすると問題は解決しないので、 pip install までで止めておく必要がある。 poetry export --without-hashes --dev --output poetry-requirements.txt pip install -r poetry-requirements.txt 上記コマンドの補足として、 --without-hashes は、poetry.lock で持っているハッシュを出力しないようにするオプション。これをやらないと poetry install と同様な問題に陥ってしまう --dev は、dev-dependencies 分も含めるというオプション。通常の poetry install では含んでいる部分なので、入れておいた方が良いかも（productionレベルだと不要かも） --outout は出力するファイルを指定するオプション。リダイレクトで保存するより、こっちの方がシンプルで良いかなと。 最後に あくまでも workaround なので、他に良い方法はいくらでもあるかなと思います。 「これいいよ」というのがあれば、コメント頂けると幸いです。

はじめに 移植やってます。 ( from python 3.7 to ruby 2.7 ) 関数の属性 (Python) def make_func(x): def func1(): return '1' def func2(): return '2' func1.__doc__ = x func1.other = func2 return func1 my_func = make_func("func1") print(my_func()) print(my_func.__doc__) print(my_func.other()) # output 1 func1 2 関数の属性という説明が正しいか自信はないのですが、特殊属性の代表である__doc__ドキュメンテーション文字列の他、自作関数を属性として追加し、呼び出すことができます。 どうする (Ruby) make_func = lambda do |x = nil, *args, **kwargs| def func1() return '1' end def func2() return '2' end if x.nil? func1 elsif x == 'other' || x == :other func2 end end my_func = make_func puts my_func.call() puts my_func.call('other') # output 1 2 __doc__はPythonの文化なので使用しないと思いますが、必要があればdef __doc__の追加で対応できそうです。 メモ Python の 関数の属性 を学習した 百里を行く者は九十里を半ばとす

参考 Python実践入門 方法 import uuid for i in range(10): random_file_names = str(uuid.uuid4()) print(random_file_names) bce14cf8-2858-46fa-abf4-d761f5c1c7f8 0b205a5a-7a3f-4769-9ddf-8e9c223f3203 49ec4902-b1cb-41a2-8e20-d05d15c3edd7 566bace9-65ad-43c4-ab58-8a534871aaf9 adc4730f-8199-4d2e-b257-1186e589a066 20365cda-3d29-448d-9c84-18321ae6ed35 6ae549c8-dcb3-4dfb-ab16-c63d6724a7cb 68ad81ba-57f3-4ceb-9033-338438b8d0e0 a1621dca-b6f0-4fd5-96c3-08635314aac8 02ed3124-fdf7-4a3a-90f7-7278f8c12702

参考 Python実践入門 方法1 uuid 標準モジュール import uuid for i in range(10): print(str(uuid.uuid4().hex)) ad37a981e8774484823d3446c1d7cf3a 69b36e62cebd49afa138241c1c9d6282 8329326e86f4417c817bfe372769a96f 2fb71832bb7242c6a7ba7d78c5405cf4 e4bcea0b3f4e45e1a093923145f2488c 8ee8734a6abd4e50a21c923c83f3c1eb 3980c4574fc044dcb18e0802abaf11e1 a3f655f41a6a4c7dafa9d722676a0a3c a97c6c33963f4795907b71b3d4f04b2b 040e31e4ac804de7b8fd81cf8308a3c4 方法2 names サードパーティー製モジュール pip install names import names for i in range(10): print(names.get_full_name()) Amanda Settle Kevin Bejar Luella Inman Sara Runyon Levi Rezentes Cesar Smith Constance Coates Patricia Schrag Jessica Cauthen William Conkle

Wordleとは？ フィードバックを頼りに、隠された 5 文字の英単語を当てるゲームです。 要は「マスターマインド」や「ヌメロン」等と呼ばれるゲームの英単語版です。 知らない方でも以下のような奇妙なツイートは目にしたことがあるかもしれません。 目的 「順番に入力していくだけで解が 1 つに定まるような英単語の組」が存在するかを Python を用いて検証していきます。 もしそのような組が存在すれば、ソルバーを使わず人力で Wordle を攻略できることになります。 先にネタバレしておくと、 - 4 回入力するだけで解が平均 1.1 個まで絞られ、 - 96% 以上の確率で 5 回目で正答でき、 - 6 回以内には確実に正答できる ような 4 つの英単語があることがわかりました。 方法 色々な英単語の組み合わせに対し「それらを入力した後に解の候補として残る単語数」を計算していきます。 しかし Wordle で入力可能な単語は約 13000 語（そのうち答えになるのは 2315 語）もあるため、全単語の組み合わせについて計算するのは困難です。 そこで今回は、様々なヒューリスティックを用い、優秀そうな単語や組み合わせに絞って探索していくことにします。 まずはどのような組み合わせであるべきかを考察していきます。 使用文字が重複していないこと できるだけ多くのフィードバックを得るためには、可能な限り同じ文字を使わないことが重要と考えられます。 そのため使用文字が被っていない単語の組についてだけ検討します。 要は、候補文字を重複なくローラーしていくような組を探していくということです。 4 語の組であること アルファベットは 26 文字あるので、文字が被らないように 5 文字の単語を作るとき、最大 5 つまで作ることが可能です。 しかし Wordle では 6 回しか回答チャンスがないため、5 回もローラーに費やすのは悪手と思われます。 また 4 回目の時点で 20 文字分のフィードバックが得られており、残り 6 文字についてもある程度は判明してきます。 （そして出現頻度の低いj, k, q, w, x, z辺りは実際情報が得られる可能性が低いと考えられます） そういった理由から最大でも4つあれば十分と判断しました。 またプレイした実感から 3 語では答えを絞りこめないと思われたため、今回はひとまず 4 語からなる組のみを探すことにします。 以上が単語セットに課す条件です。 続いては単語自体に課す条件を見ていきます。 Wordleで入力できる単語であること これは当然ですね。こちらはデータ解析の目的で Wordle 自身から取得しました。 # ローカルに落として整形した Wordle の単語リストを変数 word_list に格納する with open('path/to/wordle-dict.txt', mode='r') as f: word_list = f.read().split('\n') print(len(word_list)) # 12972 同じ文字を1度しか使わない単語であること Wordle の単語リストには例えば「class」のような、同じ文字を2回以上使う単語が入っています。 そういう単語では多くの場合 4 文字分以下しかフィードバックが得られず、期待できる情報量が少なそうなのであらかじめ除外します。 word_list = [word for word in word_list if len(set(word)) == 5] これにより 8322 個まで単語が減りました。 クイズコーナー 記事の主題とは関係ないですが、同じ文字を 3 回使う 5 文字の単語もありました。 一つはエンジニアがよく目にする単語なのですが、何だかわかりますか？ （答えは記事の最後に） よく知られた単語であること 探索のためには 8322 語だとまだ多過ぎるため、一般的な単語に絞ります。 また例えば「waqfs」とか「vozhd」といった耳慣れない単語をアリにするのも何となく面白味に欠けます。 そのためには、Googleの一般的な10000の単語リストをダウンロードし、その中に含まれる単語のみを使うようにしました。 # あらかじめ単語リストは落としておく with open('path/to/common_word_list.txt', mode='r') as f: common_word_list = f.read().split('\n') word_list = [word for word in word_list if word in common_word_list] これで最終的に 837 個まで単語数を減らすことができました。 あとは残った 837 語から先述の条件を満たす組を探していきます。 ここは愚直にループを回しているだけなのでコードは割愛します。 （一応リポジトリはこちらです。あまり整理されていないです） 結果 837語の候補単語に対し、上述の条件を満たす組を探索したところ、 album, depth, rocky, swing など計 2264 組が見つかりました。 結果をさらに見るalbum, depth, rocky, swing album, fight, rocky, spend awful, bring, empty, shock awful, might, rocky, spend badge, lunch, proxy, swift badly, grove, punch, swift badly, quick, sperm, thong batch, funky, pixel, sword black, depth, forum, swing black, fight, sperm, wound black, judge, proxy, shift black, judge, proxy, smith black, judge, proxy, swift black, sperm, width, young blame, crowd, funky, sight blame, crowd, funky, sixth blame, dutch, proxy, swing blame, funky, pitch, sword blank, judge, proxy, shift blank, judge, proxy, smith blank, judge, proxy, swift blank, judge, proxy, witch blend, proxy, quick, shaft blink, judge, match, proxy blink, judge, proxy, shaft blink, judge, proxy, watch block, empty, fraud, swing brick, empty, flash, wound brick, empty, flush, wagon brown, depth, lucky, sigma brown, empty, flash, quick bunch, medal, proxy, swift cable, funky, might, sword child, funky, grove, stamp climb, forth, judge, spank climb, forty, judge, spank climb, judge, party, shown climb, judge, proxy, shaft ... 最も優秀な組はどれか？ この 2264 組について「4語すべて入力した後に残る解候補数の期待値」を計算しました。 解候補数の期待値が最も少なかったのは、 cyber, might, spank, would で、平均 1.07 語まで絞れるという結果になりました （使用しない文字はf, j, q, v, x, z） 93.3%の確率で解候補数が 1 つに定まるため、ほとんどの場合 5 回目で正答することができます。 残る単語数は最大でも 3 語で、その最大値を取る解は 18 個だけありました。 もう一つ、「入力した後に解候補数が2語以下になる確率」も計算してみました。 解候補数が 2 語以下というのは、つまり残り 2 回はそれらを入力するだけで確実に解けるということです（それらの単語が思いつけばですが）。 解候補数が 2 つ以下になる確率が最も高かったのは、 moved, plans, rugby, witchでした （使用しない文字はf, j, k, q, x, z） 93.0%の確率で解候補が1つになり、99.7%の確率で解候補が2個以下になりました。 差し引き 6.7 % の確率で候補が 2 つ残る計算で、2 つ残った場合は 1/2 の確率で当たるので、 少なくとも 93.0% + (6.7 / 2)% = 96.4% の確率で 5 回で正答できることになります。 3 語以上残る場合（解となる単語 2315 語のうち）8 語のみでした。 残る単語のパターンとしては、 - ['aloft', 'float', 'allot'] - ['tease', 'stake', 'skate', 'feast', 'state'] の2通りしかなく、どちらもあと一回で解を判別できます。 そのためこれら 8 語が解の場合も 6 回までには確実に正答できます。 つまり理論的にはmoved, plans, rugby, witchと入力すれば、 - 96% 以上の確率で 5 回で正答できる - 100％ の確率で 6 回までに正答できる ということです。 期待値や確率の計算に使ったコードは一応こちらにあります（あまり整理はしていないのでご参考程度に）。 まとめ わりと単純なヒューリスティックスとコードで探索しましたが「確実に Wordle を攻略できる手順を見つける」という目的は果たせました。 ただ「6回までに正答できる」というのは、戦略なしの場合と比べて非常に優れているというわけでもないので、さらに強い単語の組を見つけるのが今後の課題です。 誤りや改善点等があればコメントなどでお知らせいただけると幸いです。 クイズ「同じ文字を3文字含む5文字の単語は？」の答えを見るerror, mammy, daddy, geese など

Wordleとは？ フィードバックを頼りに、隠された 5 文字の英単語を当てるゲームです。 要は「マスターマインド」や「ヌメロン」等と呼ばれるゲームの英単語版です。 知らない方でも以下のような奇妙なツイートは目にしたことがあるかもしれません。 目的 「順番に入力していくだけで正解が 1 つに定まるような英単語の組」が存在するかを Python を用いて検証していきます。 （要はソルバーを使わず人力で Wordle を完全攻略したいというモチベーションです） 先にネタバレしておくと、 - ある 4 つの単語を入力するだけで正解が平均 1.1 個まで絞られ、 - 96% 以上の確率で 5 回で正答でき、 - 100% の確率で 6 回以内に正答できる ことがわかりました。 方法 色々な英単語の組み合わせに対し「それらを入力した後に解の候補として残る単語数」を計算していきます。 しかし Wordle で入力可能な単語は約 13000 語（そのうち答えになるのは 2315 語）もあるため、全単語の組み合わせについて計算するのは困難です。 そこで今回は、様々なヒューリスティックを用い、優秀そうな単語や組み合わせに絞って探索していくことにします。 まずはどのような組み合わせであるべきかを考察していきます。 使用文字が重複していないこと できるだけ多くのフィードバックを得るためには、可能な限り同じ文字を使わないことが重要と考えられます。 そのため使用文字が被っていない単語の組についてだけ検討します。 要は、候補文字を重複なくローラーしていくような組を探していくということです。 4 語の組であること アルファベットは 26 文字あるので、文字が被らないように 5 文字の単語を作るとき、最大 5 つまで作ることが可能です。 しかし Wordle では 6 回しか回答チャンスがないため、5 回もローラーに費やすのは悪手と思われます。 また 4 回目の時点で 20 文字分のフィードバックが得られており、残り 6 文字についてもある程度は判明してきます。 （そして出現頻度の低いj, k, q, w, x, z辺りは実際情報が得られる可能性が低いと考えられます） そういった理由から最大でも4つあれば十分と判断しました。 またプレイした実感から 3 語では答えを絞りこめないと思われたため、今回はひとまず 4 語からなる組のみを探すことにします。 以上が単語セットに課す条件です。 続いては単語自体に課す条件を見ていきます。 Wordleで入力できる単語であること これは当然ですね。こちらはデータ解析の目的で Wordle 自身から取得しました。 # ローカルに落として整形した Wordle の単語リストを変数 word_list に格納する with open('path/to/wordle-dict.txt', mode='r') as f: word_list = f.read().split('\n') print(len(word_list)) # 12972 同じ文字を1度しか使わない単語であること Wordle の単語リストには例えば「class」のような、同じ文字を2回以上使う単語が入っています。 そういう単語では多くの場合 4 文字分以下しかフィードバックが得られず、期待できる情報量が少なそうなのであらかじめ除外します。 word_list = [word for word in word_list if len(set(word)) == 5] これにより 8322 個まで単語が減りました。 クイズコーナー 記事の主題とは関係ないですが、同じ文字を 3 回使う 5 文字の単語もありました。 一つはエンジニアがよく目にする単語なのですが、何だかわかりますか？ （答えは記事の最後に） よく知られた単語であること 探索のためには 8322 語だとまだ多過ぎるため、一般的な単語に絞ります。 また例えば「waqfs」とか「vozhd」といった耳慣れない単語をアリにするのも何となく面白味に欠けます。 そのためには、Googleの一般的な10000の単語リストをダウンロードし、その中に含まれる単語のみを使うようにしました。 # あらかじめ単語リストは落としておく with open('path/to/common_word_list.txt', mode='r') as f: common_word_list = f.read().split('\n') word_list = [word for word in word_list if word in common_word_list] これで最終的に 837 個まで単語数を減らすことができました。 あとは残った 837 語から先述の条件を満たす組を探していきます。 ここは泥臭くループを回しているだけなのでコードは割愛します。 （一応リポジトリはこちらです。あまり整理されていないです） 結果 837語の候補単語に対し、上述の条件を満たす組を探索したところ、 album, depth, rocky, swing など計 2264 組が見つかりました。 結果をさらに見るalbum, depth, rocky, swing album, fight, rocky, spend awful, bring, empty, shock awful, might, rocky, spend badge, lunch, proxy, swift badly, grove, punch, swift badly, quick, sperm, thong batch, funky, pixel, sword black, depth, forum, swing black, fight, sperm, wound black, judge, proxy, shift black, judge, proxy, smith black, judge, proxy, swift black, sperm, width, young blame, crowd, funky, sight blame, crowd, funky, sixth blame, dutch, proxy, swing blame, funky, pitch, sword blank, judge, proxy, shift blank, judge, proxy, smith blank, judge, proxy, swift blank, judge, proxy, witch blend, proxy, quick, shaft blink, judge, match, proxy blink, judge, proxy, shaft blink, judge, proxy, watch block, empty, fraud, swing brick, empty, flash, wound brick, empty, flush, wagon brown, depth, lucky, sigma brown, empty, flash, quick bunch, medal, proxy, swift cable, funky, might, sword child, funky, grove, stamp climb, forth, judge, spank climb, forty, judge, spank climb, judge, party, shown climb, judge, proxy, shaft ... 最も優秀な組はどれか？ この 2264 組について「4語すべて入力した後に残る解候補数の期待値」を計算しました。 解候補数の期待値が最も少なかったのは、 cyber, might, spank, would で、平均 1.07 語まで絞れるという結果になりました （使用しない文字はf, j, q, v, x, z） 93.3%の確率で解候補数が 1 つに定まるため、ほとんどの場合 5 回目で正答することができます。 残る単語数は最大でも 3 語で、その最大値を取る解は 18 個だけありました。 100% 正解にたどりつける組 もう一つ、「入力した後に解候補数が2語以下になる確率」も計算してみました。 （解候補数が 2 語以下というのは、つまり残り 2 回はそれらを入力するだけで確実に解けるということです） 解候補数が 2 つ以下になる確率が最も高かったのは、 moved, plans, rugby, witchでした （使用しない文字はf, j, k, q, x, z） 93.0%の確率で解候補が1つになり、99.7%の確率で解候補が2個以下になりました。 差し引き 6.7 % の確率で候補が 2 つ残る計算で、2 つ残った場合は 1/2 の確率で当たるので、 少なくとも 93.0% + (6.7 / 2)% = 96.4% の確率で 5 回で正答できることになります。 3 語以上残る場合（解となる単語 2315 語のうち）8 語のみでした。 残る単語のパターンとしては、 - ['aloft', 'float', 'allot'] - ['tease', 'stake', 'skate', 'feast', 'state'] の2通りしかなく、どちらもあと一回で解を判別できます。 そのためこれら 8 語の場合も 6 回までには確実に正答できます。 つまり理論的にはmoved, plans, rugby, witchと入力すれば、 - 96% 以上の確率で 5 回で正答できる - 100％ の確率で 6 回までに正答できる ということです。 期待値や確率の計算に使ったコードは一応こちらにあります（あまり整理はしていないのでご参考程度に）。 まとめ わりと単純なヒューリスティックスとコードで探索しましたが「確実に Wordle を攻略できる手順を見つける」という目的は果たせました。 ただ「5回目までにほぼ正答できる」というのは、戦略なしの場合と比べて非常に優れているわけでもないので、さらに強い単語の組を見つけるのが今後の課題です。 誤りや改善点等があればコメントなどでお知らせいただけると幸いです。 クイズ「同じ文字を3文字含む5文字の単語は？」の答えを見るerror, mammy, daddy, geese など

Wordleとは？ フィードバックを頼りに、隠された 5 文字の英単語を当てるゲームです。 要は「マスターマインド」や「ヌメロン」等と呼ばれるゲームの英単語版です。 知らない方でも以下のような奇妙なツイートは目にしたことがあるかもしれません。 目的 「順番に入力していくだけで正解が 1 つに定まるような英単語の組」が存在するかを Python を用いて検証していきます。 （要はソルバーを使わず人力で Wordle を完全攻略したいというモチベーションです） 先にネタバレしておくと、 - ある 4 つの単語を入力するだけで正解が平均 1.1 個まで絞られ、 - 96% 以上の確率で 5 回で正答でき、 - 100% の確率で 6 回以内に正答できる ことがわかりました。 方法 色々な英単語の組み合わせに対し「それらを入力した後に解の候補として残る単語数」を計算していきます。 しかし Wordle で入力可能な単語は約 13000 語（そのうち答えになるのは 2315 語）もあるため、全単語の組み合わせについて計算するのは困難です。 そこで今回は、様々なヒューリスティックを用い、優秀そうな単語や組み合わせに絞って探索していくことにします。 まずはどのような組み合わせであるべきかを考察していきます。 使用文字が重複していないこと できるだけ多くのフィードバックを得るためには、可能な限り同じ文字を使わないことが重要と考えられます。 そのため使用文字が被っていない単語の組についてだけ検討します。 要は、候補文字を重複なくローラーしていくような組を探していくということです。 4 語の組であること アルファベットは 26 文字あるので、文字が被らないように 5 文字の単語を作るとき、最大 5 つまで作ることが可能です。 しかし Wordle では 6 回しか回答チャンスがないため、5 回もローラーに費やすのは悪手と思われます。 また 4 回目の時点で 20 文字分のフィードバックが得られており、残り 6 文字についてもある程度は判明してきます。 （そして出現頻度の低いj, k, q, w, x, z辺りは実際情報が得られる可能性が低いと考えられます） そういった理由から最大でも4つあれば十分と判断しました。 またプレイした実感から 3 語では答えを絞りこめないと思われたため、今回はひとまず 4 語からなる組のみを探すことにします。 以上が単語セットに課す条件です。 続いては単語自体に課す条件を見ていきます。 Wordleで入力できる単語であること これは当然ですね。こちらはデータ解析の目的で Wordle 自身から取得しました。 # ローカルに落として整形した Wordle の単語リストを変数 word_list に格納する with open('path/to/wordle-dict.txt', mode='r') as f: word_list = f.read().split('\n') print(len(word_list)) # 12972 同じ文字を1度しか使わない単語であること Wordle の単語リストには例えば「class」のような、同じ文字を2回以上使う単語が入っています。 そういう単語では多くの場合 4 文字分以下しかフィードバックが得られず、期待できる情報量が少なそうなのであらかじめ除外します。 word_list = [word for word in word_list if len(set(word)) == 5] これにより 8322 個まで単語が減りました。 クイズコーナー 記事の主題とは関係ないですが、同じ文字を 3 回使う 5 文字の単語もありました。 一つはエンジニアがよく目にする単語なのですが、何だかわかりますか？ （答えは記事の最後に） よく知られた単語であること 探索のためには 8322 語だとまだ多過ぎるため、一般的な単語に絞ります。 また例えば「waqfs」とか「vozhd」といった耳慣れない単語をアリにするのも何となく面白味に欠けます。 そのためには、Googleの一般的な10000の単語リストをダウンロードし、その中に含まれる単語のみを使うようにしました。 # あらかじめ単語リストは落としておく with open('path/to/common_word_list.txt', mode='r') as f: common_word_list = f.read().split('\n') word_list = [word for word in word_list if word in common_word_list] これで最終的に 837 個まで単語数を減らすことができました。 あとは残った 837 語から先述の条件を満たす組を探していきます。 ここは泥臭くループを回しているだけなのでコードは割愛します。 （一応リポジトリはこちらです。あまり整理されていないです） 結果 837語の候補単語に対し、上述の条件を満たす組を探索したところ、 album, depth, rocky, swing など計 2264 組が見つかりました。 結果をさらに見るalbum, depth, rocky, swing album, fight, rocky, spend awful, bring, empty, shock awful, might, rocky, spend badge, lunch, proxy, swift badly, grove, punch, swift badly, quick, sperm, thong batch, funky, pixel, sword black, depth, forum, swing black, fight, sperm, wound black, judge, proxy, shift black, judge, proxy, smith black, judge, proxy, swift black, sperm, width, young blame, crowd, funky, sight blame, crowd, funky, sixth blame, dutch, proxy, swing blame, funky, pitch, sword blank, judge, proxy, shift blank, judge, proxy, smith blank, judge, proxy, swift blank, judge, proxy, witch blend, proxy, quick, shaft blink, judge, match, proxy blink, judge, proxy, shaft blink, judge, proxy, watch block, empty, fraud, swing brick, empty, flash, wound brick, empty, flush, wagon brown, depth, lucky, sigma brown, empty, flash, quick bunch, medal, proxy, swift cable, funky, might, sword child, funky, grove, stamp climb, forth, judge, spank climb, forty, judge, spank climb, judge, party, shown climb, judge, proxy, shaft ... 最も優秀な組はどれか？ この 2264 組について「4語すべて入力した後に残る解候補数の期待値」を計算しました。 解候補数の期待値が最も少なかったのは、 cyber, might, spank, would で、平均 1.07 語まで絞れるという結果になりました （使用しない文字はf, j, q, v, x, z） 93.3%の確率で解候補数が 1 つに定まるため、ほとんどの場合 5 回目で正答することができます。 残る単語数は最大でも 3 語で、その最大値を取る解は 18 個だけありました。 100% 正解にたどりつける組 もう一つ、「入力した後に解候補数が2語以下になる確率」も計算してみました。 （解候補数が 2 語以下というのは、つまり残り 2 回はそれらを入力するだけで確実に解けるということです） 解候補数が 2 つ以下になる確率が最も高かったのは、 moved, plans, rugby, witchでした （使用しない文字はf, j, k, q, x, z） 93.0%の確率で解候補が1つになり、99.7%の確率で解候補が2個以下になりました。 差し引き 6.7 % の確率で候補が 2 つ残る計算で、2 つ残った場合は 1/2 の確率で当たるので、 少なくとも 93.0% + (6.7 / 2)% = 96.4% の確率で 5 回で正答できることになります。 3 語以上残る場合（解となる単語 2315 語のうち）8 語のみでした。 残る単語のパターンとしては、 - ['aloft', 'float', 'allot'] - ['tease', 'stake', 'skate', 'feast', 'state'] の2通りしかなく、どちらもあと一回で解を判別できます。 そのためこれら 8 語の場合も 6 回までには確実に正答できます。 つまり理論的にはmoved, plans, rugby, witchと入力すれば、 - 96% 以上の確率で 5 回で正答できる - 100％ の確率で 6 回までに正答できる ということです。 期待値や確率の計算に使ったコードは一応こちらにあります（あまり整理はしていないのでご参考程度に）。 まとめ わりと単純なヒューリスティックスとコードで探索しましたが「確実に Wordle を攻略できる手順を見つける」という目的は果たせました。 ただ「5回目までにほぼ正答できる」というのは、戦略なしの場合と比べて非常に優れているわけでもないので、さらに強い単語の組を見つけるのが今後の課題です。 誤りや改善点等があればコメントなどでお知らせいただけると幸いです。 クイズ「同じ文字を3文字含む5文字の単語は？」の答えを見るerror, mammy, daddy, geese など

Wordleとは？ フィードバックを頼りに、隠された 5 文字の英単語を当てるゲームです。 要は「マスターマインド」や「ヌメロン」等と呼ばれるゲームの英単語版です。 知らない方でも以下のような奇妙なツイートは目にしたことがあるかもしれません。 この記事は何？ Wordleにおいて、最初にある4つの単語を入力すれば 100% 正答にたどりつけることが Python を使った解析でわかりました。 具体的には、 - それら 4 つの単語を入力すれば解の候補が平均 1.1 個まで絞られる - 96% 以上の確率で 5 回で正答できる - 100% の確率で 6 回以内に正答できる といった感じです。 本記事では、それらの単語をどのように探したかをザックリ解説しつつ（ネタバレ防止のため）最後にその4つの単語をご紹介します。 （結果だけ見たい方はこちら） 方法 色々な英単語の組み合わせに対し「それらを入力したとき、平均して何語まで解の候補が絞れるか」を計算し、その値が最も少なくなる組を探します。 しかし Wordle で入力可能な単語は約 13000 語（そのうち答えになるのは 2315 語）もあるため、全単語の組み合わせについて計算するのは困難です。 そこで今回は、様々なヒューリスティックを用い、優秀そうな単語や組み合わせに絞って探索していくことにします。 まずはどのような組み合わせであるべきかを考察していきます。 使用文字が重複していないこと できるだけ多くのフィードバックを得るためには、可能な限り同じ文字を使わないことが重要と考えられます。 そのため使用文字が被っていない単語の組についてだけ検討します。 要は、候補文字を重複なくローラーしていくような組を探していくということです。 4 語の組であること アルファベットは 26 文字あるので、文字が被らないように 5 文字の単語を作るとき、最大 5 つまで作ることが可能です。 しかし Wordle では 6 回しか回答チャンスがないため、5 回もローラーに費やすのは悪手と思われます。 また 4 回目の時点で 20 文字分のフィードバックが得られており、残り 6 文字についてもある程度は判明してきます。 （そして出現頻度の低いj, k, q, w, x, z辺りは実際情報が得られる可能性が低いと考えられます） そういった理由から最大でも4つあれば十分と判断しました。 またプレイした実感から 3 語では答えを絞りこめないと思われたため、今回はひとまず 4 語からなる組のみを探すことにします。 以上が単語の組に課す条件です。 続いては単語自体に課す条件を見ていきます。 Wordleで入力できる単語であること これは当然ですね。こちらはデータ解析の目的で Wordle 自身から取得しました。 # ローカルに落として整形した Wordle の単語リストを変数 word_list に格納する with open('path/to/wordle-dict.txt', mode='r') as f: word_list = f.read().split('\n') print(len(word_list)) # 12972 同じ文字を1度しか使わない単語であること Wordle の単語リストには例えば「class」のような、同じ文字を2回以上使う単語が入っています。 そういう単語では多くの場合 4 文字分以下しかフィードバックが得られず、期待できる情報量が少なそうなのであらかじめ除外します。 word_list = [word for word in word_list if len(set(word)) == 5] これにより 8322 個まで単語が減りました。 クイズコーナー 記事の主題とは関係ないですが、同じ文字を 3 回使う 5 文字の単語もありました。 一つはエンジニアがよく目にする単語なのですが、何だかわかりますか？ （答えは記事の最後に） よく知られた単語であること 探索のためには 8322 語だとまだ多過ぎるため、一般的な単語に絞ります。 また「waqfs」とか「vozhd」といった耳慣れない単語を使うのも何となく面白味に欠けます。 そのためには、Googleの一般的な10000の単語リストをダウンロードし、その中に含まれる単語のみを使うようにしました。 # あらかじめ単語リストは落としておく with open('path/to/common_word_list.txt', mode='r') as f: common_word_list = f.read().split('\n') word_list = [word for word in word_list if word in common_word_list] これで最終的に 837 個まで単語数を減らすことができました。 あとは残った 837 語から先述の条件を満たす組を探していきます。 ここは泥臭くループを回しているだけなのでコードは省略します。 （一応リポジトリはこちらです。あまり整理されていないです） 結果 837語の候補単語に対し、上述の条件を満たす組を探索したところ、 album, depth, rocky, swing など計 2264 組が見つかりました。 結果をさらに見るalbum, depth, rocky, swing album, fight, rocky, spend awful, bring, empty, shock awful, might, rocky, spend badge, lunch, proxy, swift badly, grove, punch, swift badly, quick, sperm, thong batch, funky, pixel, sword black, depth, forum, swing black, fight, sperm, wound black, judge, proxy, shift black, judge, proxy, smith black, judge, proxy, swift black, sperm, width, young blame, crowd, funky, sight blame, crowd, funky, sixth blame, dutch, proxy, swing blame, funky, pitch, sword blank, judge, proxy, shift blank, judge, proxy, smith blank, judge, proxy, swift blank, judge, proxy, witch blend, proxy, quick, shaft blink, judge, match, proxy blink, judge, proxy, shaft blink, judge, proxy, watch block, empty, fraud, swing brick, empty, flash, wound brick, empty, flush, wagon brown, depth, lucky, sigma brown, empty, flash, quick bunch, medal, proxy, swift cable, funky, might, sword child, funky, grove, stamp climb, forth, judge, spank climb, forty, judge, spank climb, judge, party, shown climb, judge, proxy, shaft ... 最も優秀な組はどれか？ この 2264 組について「4語すべて入力した後に残る解候補数の期待値」を計算しました（ここで「期待値」は正解となりうる各単語についての平均の意味です）。 解候補数の期待値が最も少なかったのは、 cyber, might, spank, would で、平均 1.07 語まで絞れるという結果になりました （使用しない文字はf, j, q, v, x, z） 93.3%の確率で解候補数が 1 つに定まるため、ほとんどの場合 5 回目で正答することができます。 （もちろん解の単語をそもそも知らない場合等は考慮していません。あくまで理論的に1つに絞り込めるという話です） 残る単語数は最大でも 3 語で、その最大値を取る解は 18 個だけありました。 この 18 個について「残り 2 回のチャンスで確実に正答できるか」を検証するのは面倒だったので「解候補数が2つ以下になる確率が最も高い組」を探し、それについて検証することにします。 100% 正解にたどりつける組 解候補数が 2 つ以下になる確率が最も高かったのは、 moved, plans, rugby, witchでした （使用しない文字はf, j, k, q, x, z） 93.0%の確率で解候補が1つになり、99.7%の確率で解候補が2個以下になりました。 差し引き 6.7 % の確率で候補が 2 個になる計算で、その場合でも一回の当てずっぽうで 1/2 の確率で当たるので、5 回で正答できる確率は少なくとも 93.0% + (6.7 / 2)% = 96.4% 以上あることになります。 三つ以上候補が残るような正解単語は 8 語だけしかなく、 残る単語のパターンとしては、 - ['aloft', 'float', 'allot'] - ['tease', 'stake', 'skate', 'feast', 'state'] の2通りしかありませんでした。 そして、どちらもあと一回で解を判別できることが簡単に検証できます。 そのため 3 つ以上候補が残った場合も 6 回までには確実に正答できます。 つまり理論的にはmoved, plans, rugby, witchと入力すれば、 - 96% 以上の確率で 5 回で正答できる - 100％ の確率で 6 回までに正答できる ということです。 期待値や確率の計算に使ったコードは一応こちらにあります（あまり整理はしていないのでご参考程度に）。 まとめ わりと単純なヒューリスティックスとコードで探索しましたが「確実に Wordle を攻略できる手順を見つける」という目的は果たせました。 ただ「5回目までにほぼ正答できる」というのは、戦略なしの場合と比較しそれ程優れているわけでもないので、さらに強い単語の組を見つけるのが今後の課題です。 誤りや改善点等があればコメントなどでお知らせいただけると幸いです。 クイズ「同じ文字を3文字含む5文字の単語は？」の答えを見るerror, mammy, daddy, geese など

Wordleとは？ フィードバックを頼りに、隠された 5 文字の英単語を当てるゲームです。 要は「マスターマインド」や「ヌメロン」等と呼ばれるゲームの英単語版です。 知らない方でも以下のような奇妙なツイートは目にしたことがあるかもしれません。 この記事は何？ Wordleにおいて、最初にある4つの単語を入力すれば 100% 正答にたどりつけることが Python を使った解析でわかりました。 具体的には、 - それら 4 つの単語を入力すれば解の候補が平均 1.1 個まで絞られる - 96% 以上の確率で 5 回で正答できる - 100% の確率で 6 回以内に正答できる といった感じです。 本記事では、それらの単語をどのように探したかをザックリ解説しつつ（ネタバレ防止のため）最後にその4つの単語をご紹介します。 （結果だけ見たい方はこちら） 方法 色々な英単語の組み合わせに対し「それらを入力したとき、平均して何語まで解の候補が絞れるか」を計算し、その値が最も少なくなる組を探しました。 しかし Wordle で入力可能な単語は約 13000 語（そのうち答えになるのは 2315 語）もあるため、全単語の組み合わせについて計算するのは困難です。 そこで今回は、様々なヒューリスティックを用い、優秀そうな単語や組み合わせに絞って探索していくことにします。 まずはどのような組み合わせであるべきかを考察していきます。 使用文字が重複していないこと できるだけ多くのフィードバックを得るためには、可能な限り同じ文字を使わないことが重要と考えられます。 そのため使用文字が被っていない単語の組についてだけ検討します。 要は、候補文字を重複なくローラーしていけるような組を探していくということです。 4 語の組であること アルファベットは 26 文字あるので、文字が被らないように 5 文字の単語を作るとき、最大 5 つまで作ることが可能です。 しかし Wordle では 6 回しか回答チャンスがないため、5 回もローラーに費やすのは悪手と思われます。 また 4 回目の時点で 20 文字分のフィードバックが得られており、残り 6 文字についてもある程度は判明してきます。 （そして出現頻度の低いj, k, q, w, x, z辺りは実際情報が得られる可能性が低いと考えられます） そういった理由から最大でも4つあれば十分と判断しました。 またプレイした実感から 3 語では答えを絞りこめないと思われたため、今回はひとまず 4 語からなる組のみを探すことにします。 以上が単語の組に課す条件です。 続いては単語自体に課す条件を見ていきます。 Wordleで入力できる単語であること これは当然ですね。こちらは Wordle 自身から取得しました。 同じ文字を1度しか使わない単語であること Wordle の単語リストには例えば「class」のような、同じ文字を2回以上使う単語が入っています。 そういう単語では多くの場合 4 文字分以下しかフィードバックが得られず、期待できる情報量が少なそうなのであらかじめ除外します。 word_list = [word for word in word_list if len(set(word)) == 5] これにより 8322 個まで単語が減らせました。 クイズコーナー 記事の主題とは関係ないですが、同じ文字を 3 回使う 5 文字の単語もありました。 一つはエンジニアがよく目にする単語なのですが、何だかわかりますか？ （答えは記事の最後に） よく知られた単語であること 探索のためには 8322 語だとまだ多過ぎるため、一般的な単語に絞ることにします。 （「waqfs」とか「vozhd」といった耳慣れない単語を使うのが何となく面白味に欠けるというのもあります） そのためには、Googleの一般的な10000の単語リストをダウンロードし、その中に含まれる単語のみを使うようにしました。 # あらかじめ単語リストは落としておく with open('path/to/common_word_list.txt', mode='r') as f: common_word_list = f.read().split('\n') word_list = [word for word in word_list if word in common_word_list] これで最終的に 837 個まで単語数を減らすことができました。 あとは残った 837 語から先述の条件を満たす組を探していきます。 ここは泥臭くループを回しているだけなのでコードは省略します。 一つ工夫を上げるとすれば「同じ文字は一回しか使わない」という条件により、一度使った文字を含む単語を探索候補から消しています。 （一応コードはこちらにあります。あまり整理はされていないです） 結果 837語の候補単語に対し条件を満たす組を探索したところ、 album, depth, rocky, swing など計 2264 組が見つかりました。 結果をさらに見るalbum, depth, rocky, swing album, fight, rocky, spend awful, bring, empty, shock awful, might, rocky, spend badge, lunch, proxy, swift badly, grove, punch, swift badly, quick, sperm, thong batch, funky, pixel, sword black, depth, forum, swing black, fight, sperm, wound black, judge, proxy, shift black, judge, proxy, smith black, judge, proxy, swift black, sperm, width, young blame, crowd, funky, sight blame, crowd, funky, sixth blame, dutch, proxy, swing blame, funky, pitch, sword blank, judge, proxy, shift blank, judge, proxy, smith blank, judge, proxy, swift blank, judge, proxy, witch blend, proxy, quick, shaft blink, judge, match, proxy blink, judge, proxy, shaft blink, judge, proxy, watch block, empty, fraud, swing brick, empty, flash, wound brick, empty, flush, wagon brown, depth, lucky, sigma brown, empty, flash, quick bunch, medal, proxy, swift cable, funky, might, sword child, funky, grove, stamp climb, forth, judge, spank climb, forty, judge, spank climb, judge, party, shown climb, judge, proxy, shaft ... 最も優秀な組はどれか？ この 2264 組について「4語すべて入力した後に残る解候補数の期待値」を計算しました。 解候補数の期待値が最も少なかったのは、 cyber, might, spank, would で、平均 1.07 語まで絞れるという結果になりました。 （使用しない文字はf, j, q, v, x, z） 93.3%の確率で解候補数が 1 つに定まるため、ほとんどの場合 5 回目で正答することができます。 （もちろん解の単語をそもそも知らない場合等は考慮していません。あくまで理論的に1つに絞り込めるという話です） 残る単語数は最大でも 3 語で、その最大値を取る正解単語は 18 個だけありました。 この 18 個について「残り 2 回のチャンスで確実に正答できるか」を検証するのは面倒だったので「解候補数が2つ以下になる確率が最も高い組」を探し、それについて検証することにします。 100% 正解にたどりつける組 解候補数が 2 つ以下になる確率が最も高かったのは、 moved, plans, rugby, witchでした。 （使用しない文字はf, j, k, q, x, z） 93.0%の確率で解候補が1つになり、99.7%の確率で解候補が2個以下になりました。 差し引き 6.7 % の確率で候補が 2 個になる計算で、その場合でも一回の当てずっぽうで 1/2 の確率で正解できるため、5 回で正答できる確率は少なくとも 93.0% + (6.7 / 2)% = 96.4% 以上あることになります。 三つ以上候補が残るような正解単語は 8 語だけしかなく、 残る単語のパターンとしては、 - ['aloft', 'float', 'allot'] - ['tease', 'stake', 'skate', 'feast', 'state'] の2通りしかありませんでした。 そして、どちらもあと一回で解を判別できることが簡単に検証できます。 そのため 3 つ以上候補が残った場合も 6 回までには確実に正答できます。 つまり理論的にはmoved, plans, rugby, witchと入力すれば、 - 96% 以上の確率で 5 回で正答できる - 100％ の確率で 6 回までに正答できる ということです。 期待値や確率の計算に使ったコードは一応こちらにあります（あまり整理はしていないのでご参考程度に）。 まとめ わりと単純なヒューリスティックスとコードで探索しましたが「確実に Wordle を攻略できる手順を見つける」という目的は果たせました。 ただ「5回目までにほぼ正答できる」というのは、戦略なしの場合と比較しそれ程優れているわけでもないので、さらに強い単語の組を見つけるのが今後の課題です。 クイズ「同じ文字を3文字含む5文字の単語は？」の答えを見るerror, mammy, daddy, geese など

ABC237のA,B,C,D,E,F問題を、Python3でなるべく丁寧に解説していきます。 ただ解けるだけの方法ではなく、次の3つのポイントを満たす解法を解説することを目指しています。 シンプル：余計なことを考えずに済む 実装が楽：ミスやバグが減ってうれしい 時間がかからない：パフォが上がって、後の問題に残せる時間が増える ご質問・ご指摘はコメントかツイッター、マシュマロ、Discordサーバーまでお気軽にどうぞ！ Twitter: u2dayo マシュマロ： https://marshmallow-qa.com/u2dayo ほしいものリスト: https://www.amazon.jp/hz/wishlist/ls/2T9IQ8IK9ID19?ref_=wl_share Discordサーバー（質問や記事の感想・リクエストなどどうぞ！） : https://discord.gg/jZ8pkPRRMT よかったらLGTMや拡散していただけると喜びます！ 目次 ABC237 まとめ A問題『Not Overflow』 B問題『Matrix Transposition』 C問題『kasaka』 D問題『LR insertion』 E問題『Skiing』 F問題『|LIS| = 3』 アプリ AtCoderFacts を開発しています コンテストの統計データを見られるアプリ『AtCoderFacts』を作りました。 現在のところ、次の3つのデータを見ることができます。 レート別問題正解率 パフォーマンス目安 早解きで上昇するパフォーマンス 今後も機能を追加していく予定です。使ってくれると喜びます。 ABC237 まとめ 全提出人数: 9245人 パフォーマンス パフォ AC 点数 時間 順位(Rated内) 200 AB------ 300 54分 6146(5902)位 400 ABC----- 600 112分 4924(4680)位 600 ABC----- 600 41分 4007(3765)位 800 ABCD---- 1000 86分 3100(2866)位 1000 ABCD---- 1000 50分 2294(2060)位 1200 ABCDE--- 1500 148分 1635(1411)位 1400 ABCDE--- 1500 77分 1136(917)位 1600 ABCDE--- 1500 52分 779(567)位 1800 ABCDEF-- 2000 124分 513(322)位 2000 ABCDEF-- 2000 66分 338(170)位 2200 ABCDEFG- 2600 124分 216(83)位 2400 ABCDEFG- 2600 87分 144(38)位 色別の正解率 色 人数 A B C D E F G Ex 灰 3243 91.0 % 82.4 % 42.2 % 18.9 % 2.2 % 0.3 % 0.1 % 0.0 % 茶 1388 98.3 % 97.0 % 85.3 % 65.0 % 9.6 % 0.3 % 0.3 % 0.0 % 緑 1053 97.4 % 96.4 % 93.3 % 89.9 % 37.9 % 2.0 % 1.1 % 0.1 % 水 665 98.0 % 97.6 % 96.2 % 96.1 % 72.3 % 13.5 % 5.3 % 0.0 % 青 372 98.9 % 97.6 % 97.6 % 97.3 % 83.1 % 51.1 % 22.3 % 0.3 % 黄 182 90.7 % 89.0 % 89.0 % 90.1 % 84.6 % 72.0 % 58.8 % 8.8 % 橙 41 97.6 % 97.6 % 97.6 % 97.6 % 97.6 % 92.7 % 82.9 % 31.7 % 赤 24 91.7 % 91.7 % 95.8 % 87.5 % 91.7 % 91.7 % 91.7 % 54.2 % ※表示レート、灰に初参加者は含めず A問題『Not Overflow』 問題ページ：A - Not Overflow 灰コーダー正解率：91.0 % 茶コーダー正解率：98.3 % 緑コーダー正解率：97.4 % 入力 $N$ : 整数（$-2^{63}\le{N}\lt{2^{63}}$） 考察 $N$ が $-2^{31}$ 以上 かつ、$2^{31}$ 未満であるか判定すればいいです。境界値（以下か未満か）のミスは非常によくあるので、よく問題文を読んで間違えないように注意してください。 一般的な言語では64bit整数型を使用しないとオーバーフローしますが、Pythonは多倍長整数を使用しているため、考える必要はありません。 コード C = 2 ** 31 N = int(input()) print("Yes" if -C <= N < C else "No") # 以上、未満です B問題『Matrix Transposition』 問題ページ：B - Matrix Transposition 灰コーダー正解率：82.4 % 茶コーダー正解率：97.0 % 緑コーダー正解率：96.4 % 入力 $H,W$ : $H$ 行 $W$ 列の行列が与えられる $A_{i,j}$ : $A$ の $i$ 行 $j$ 列目の要素 考察 $A$ を二次元リストで受け取ります。Pythonでは、At = list(zip(*A)) と書くと転置行列になるので、これを出力すればいいです。 コード H, W = map(int, input().split()) A = [list(map(int, input().split())) for _ in range(H)] At = list(zip(*A)) for r in At: print(*r) NumPy 全く使う必要はありませんが、NumPyを使う場合はこうなります。（AtCoderのPyPyにはNumPyが入っていないので、Pythonで提出する必要があります） import numpy as np H, W = map(int, input().split()) A = np.array([list(map(int, input().split())) for _ in range(H)]) At = np.transpose(A) for r in At: print(*r) C問題『kasaka』 問題ページ：C - kasaka 灰コーダー正解率：42.2 % 茶コーダー正解率：85.3 % 緑コーダー正解率：93.3 % 入力 $S$ : 英小文字からなる文字列 考察 回文とは逆から読んでも同じ文字列のことですから、自分自身を反転したものと一致すれば回文です。回文判定は S == S[::-1] で簡単にできます。 回文になるとすれば、少なくとも『$S$ を先頭から見て a 以外の文字が出るまで a が連続で出てくる文字数』『$S$ を末尾から見て a 以外の文字が出るまで a が連続で出てくる文字数』が一致している必要があります。この条件を満たしていても回文とは限りませんが、満たしていなければ回文にはなりません。 先頭・末尾の a の連続数が合うように、先頭にaを付け加えて回文判定を行えばいいです。ただし、a は先頭に付け加えることしかできないので、先頭の連続数のほうが多い場合は回文にはできません。 例1 aabaaa は、先頭から見て a が $2$ 文字、末尾から見て a が $3$ 文字連続で出てきます。先頭にaを $1$ 文字付け加えて、aaabaaa にすると回文になります。 例2 abcaa は先頭から見て a が $1$ 文字、末尾から見て a が $2$ 文字連続で出てきます。先頭にa を $1$ 文字付け加えて aabcaa にしても回文にはなりません。 例3 aaaaba は先頭から見て a が $4$ 文字、末尾から見て a が $1$ 文字連続で出てきます。a は先頭に付け加えることしかできないので、回文にはできません。 解法 この問題は以下のアルゴリズムで解けます。 aが先頭・末尾それぞれ何文字連続で出てくるか調べる 先頭・末尾のaの文字数が合うように、先頭にaを付け加えて判定する（先頭のほうが多い場合は不可能） コード def judge(): S = input() l = len(S) - len(S.lstrip('a')) # 先頭の連続数 r = len(S) - len(S.rstrip('a')) # 末尾の連続数 if l > r: return False T = "a" * (r - l) + S return T == T[::-1] print("Yes" if judge() else "No") D問題『LR insertion』 問題ページ：D - LR insertion 灰コーダー正解率：18.9 % 茶コーダー正解率：65.0 % 緑コーダー正解率：89.9 % 入力 $N$ : 操作を行う回数 $s_i$ : $i$ 回目の操作では $s_i$ がLかRのどちらかに応じて、 $i$ を $i-1$ のすぐ左かすぐ右に挿入する 考察 $A={N}$ から操作を逆向きに行い、$s_i$ が L なら $i-1$ を $A$ の右端に追加し、R なら $A$ の左端に追加することで簡単に解けます。 なぜこれでうまくいくかを考えます。$s_i$ が L のとき、$i$ 以上の要素はすべて $i-1$ より左に追加されます。$i$ 以上の要素をすべて数列に追加してから、$i-1$ を $A$ の右端に追加すれば正しい位置に $i-1$ を追加できます。$s_i$ が R の場合も同じです。 先頭から見る方法や、双方向連結リストを使う解法もありますが、省略します。 実装 数列の左端と右端両方向の挿入が必要なので、両端キューのdeque を使います。 コード from collections import deque def main(): N = int(input()) S = input() ans = deque((N,)) for i in reversed(range(N)): if S[i] == "L": ans.append(i) else: ans.appendleft(i) print(*ans) main() E問題『Skiing』 問題ページ：E - Skiing 灰コーダー正解率：2.2 % 茶コーダー正解率：9.6 % 緑コーダー正解率：37.9 % 入力 $N,M$ : 広場の数は $N$ 個で、坂の数は $M$ 本 $H_i$ : 広場 $i$ の標高 $U_i,V_i$ : $i$ 本目の坂は 広場 $U_i$ と $V_i$ を結ぶ どの頂点も互いに坂を使ってたどり着ける（グラフは連結） 考察 この問題をわかりやすくいうと、$X$ と $Y$ の標高差の絶対値 $|H_X-H_Y|$ を使って 広場 $X$ から今より標高が高い広場 $Y$ に坂を使って移動すると、楽しさが $2|H_X-H_Y|$ 減る 広場 $X$ から今より標高が低い広場 $Y$ に坂を使って移動すると楽しさが $|H_X-H_Y|$ 増える 頂点 $1$ から 楽しさの初期値 $0$ ではじめたときの、楽しさを最大化せよ という問題です。 最短経路問題とみなす 楽しさの正負を反転して、距離として『楽しさの $-1$ 倍』を考え、辺を以下のように貼ったグラフを考えます。 自分より高い広場には、コスト $2|H_X-H_Y|$ の辺を貼る 自分より低い広場には、コスト $-|H_X-H_Y|$ の辺を貼る このグラフ上で、頂点 $1$ からはじめた全頂点対最短経路問題を解き、距離（楽しさの $-1$ 倍）が最小の頂点の値を $-1$ 倍すると答えになります。 自分より低い広場に対してはコストが負の辺を貼ることになりますが、負閉路は絶対に出現しないことに注意してください。（同じ場所に戻ってくることはできますが、楽しさの減少量を増加量が上回るので、『楽しさの $-1$ 倍』である距離は減少しません） 負辺が邪魔 どのアルゴリズムを使えば解けるか考えてみましょう。 ダイクストラ法 : 負辺がある場合、$O(2^N)$ になるケースを作成可能なため不可（コンテスト中はACできてしまいましたが、after contestのテストケースでTLEになります） ベルマン・フォード法 : 計算量が $O(NM)$ のため、不可 どちらのアルゴリズムも使えないので、このままのグラフで最短経路問題を解くのは難しいです。 ポテンシャルを考える なぜ負のコストが出てきたかというと、距離として『楽しさの $-1$ 倍』を使ったからです。『楽しさの $-1$ 倍』は坂を下ると減少します。絶対に減少することがない、別の値を距離として使うことを考えます。 ここで、「楽しさが増える量は、標高が減った量と同じである」ことに着目します。言い換えると、坂を下っても『楽しさ+標高』は変化しません。そこで、距離として『楽しさ+標高』の $-1$ 倍を使った以下のグラフを考えます。 自分より高い広場には、コスト $|H_X-H_Y|$ の辺を貼る（楽しさが $2|H_X-H_Y|$ 減って、標高が $|H_X-H_Y|$ 増えるため、差し引き $|H_X-H_Y|$ です） 自分より低い広場には、コスト $0$ の辺を貼る このグラフ上で、頂点 $1$ の距離の初期値を $-H_1$ として、頂点 $1$ からダイクストラ法を行います。求めた頂点 $1$ から 頂点 $i$ への最短距離を $dist_i$ とすると、『頂点 $i$ で終わるときの最大の楽しさ』は $- dist_{i} - H_i$ で求められるので、全頂点に対して調べて、最大の値が答えになります。 （$楽しさ = 『楽しさ+標高』-標高=-『(楽しさ+標高)の-1 倍』 - 標高$ です） コード from heapq import heappop, heappush INF = 1 << 62 - 1 def main(): def dijkstra(G, start): N = len(G) dist = [INF] * N dist[start] = -H[0] pq = [(-H[0], start)] while pq: cost, u = heappop(pq) if dist[u] < cost: continue for v, d in G[u]: new_cost = cost + d if dist[v] > new_cost: dist[v] = new_cost heappush(pq, (new_cost, v)) return dist N, M = map(int, input().split()) H = list(map(int, input().split())) G = [[] for _ in range(N)] for _ in range(M): a, b = map(int, input().split()) a, b = a - 1, b - 1 G[a].append((b, max(0, H[b] - H[a]))) G[b].append((a, max(0, H[a] - H[b]))) dist = dijkstra(G, 0) ans = max(-dist[i] - H[i] for i in range(N)) print(ans) main() F問題『|LIS| = 3』 問題ページ：F - |LIS| = 3 灰コーダー正解率：0.3 % 茶コーダー正解率：0.3 % 緑コーダー正解率：2.0 % 入力 $N$ : 数列の長さ $M$ : 数列の各項の値は $1$ 以上 $M$ 以下 考察 長さ $3$ に制限したLISの配列を状態として、動的計画法を行います。$i$ 番目まで決めてLIS配列が $(a,b,c)$ である数列の $i+1$ 文字目を $x$ にしたとき、LIS配列がどうなるかは、$i$ によらず一意に決まり、また更新されるのは $(a,b,c)$ のいずれか一箇所のみです。 状態数は $O(NM^3)$ です。遷移は時間計算量 $O(1)$ でできるので、計算量 $O(NM^3)$ でこの問題を解くことができました。 コード PyPyで提出してください。 MOD = 998244353 def main(): N, M = map(int, input().split()) prev = [[[0] * (M + 1) for _ in range(M + 1)] for _ in range(M + 1)] prev[M][M][M] = 1 for _ in range(N): dp = [[[0] * (M + 1) for _ in range(M + 1)] for _ in range(M + 1)] for x in range(M): for a in range(M + 1): for b in range(M + 1): for c in range(M + 1): # LIS配列としてありえないものも含まれますが、問題ありません p = prev[a][b][c] if x <= a: dp[x][b][c] += p dp[x][b][c] %= MOD elif x <= b: dp[a][x][c] += p dp[a][x][c] %= MOD elif x <= c: dp[a][b][x] += p dp[a][b][x] %= MOD prev = dp ans = 0 for a in range(M): for b in range(a + 1, M): for c in range(b + 1, M): ans += prev[a][b][c] ans %= MOD print(ans) main()

ABC237のA,B,C,D,E,F問題を、Python3でなるべく丁寧に解説していきます。 ただ解けるだけの方法ではなく、次の3つのポイントを満たす解法を解説することを目指しています。 シンプル：余計なことを考えずに済む 実装が楽：ミスやバグが減ってうれしい 時間がかからない：パフォが上がって、後の問題に残せる時間が増える ご質問・ご指摘はコメントかツイッター、マシュマロ、Discordサーバーまでお気軽にどうぞ！ Twitter: u2dayo マシュマロ： https://marshmallow-qa.com/u2dayo ほしいものリスト: https://www.amazon.jp/hz/wishlist/ls/2T9IQ8IK9ID19?ref_=wl_share Discordサーバー（質問や記事の感想・リクエストなどどうぞ！） : https://discord.gg/jZ8pkPRRMT よかったらLGTMや拡散していただけると喜びます！ 目次 ABC237 まとめ A問題『Not Overflow』 B問題『Matrix Transposition』 C問題『kasaka』 D問題『LR insertion』 E問題『Skiing』 F問題『|LIS| = 3』 アプリ AtCoderFacts を開発しています コンテストの統計データを見られるアプリ『AtCoderFacts』を作りました。 現在のところ、次の3つのデータを見ることができます。 レート別問題正解率 パフォーマンス目安 早解きで上昇するパフォーマンス 今後も機能を追加していく予定です。使ってくれると喜びます。 ABC237 まとめ 全提出人数: 9245人 パフォーマンス パフォ AC 点数 時間 順位(Rated内) 200 AB------ 300 54分 6146(5902)位 400 ABC----- 600 112分 4924(4680)位 600 ABC----- 600 41分 4007(3765)位 800 ABCD---- 1000 86分 3100(2866)位 1000 ABCD---- 1000 50分 2294(2060)位 1200 ABCDE--- 1500 148分 1635(1411)位 1400 ABCDE--- 1500 77分 1136(917)位 1600 ABCDE--- 1500 52分 779(567)位 1800 ABCDEF-- 2000 124分 513(322)位 2000 ABCDEF-- 2000 66分 338(170)位 2200 ABCDEFG- 2600 124分 216(83)位 2400 ABCDEFG- 2600 87分 144(38)位 色別の正解率 色 人数 A B C D E F G Ex 灰 3243 91.0 % 82.4 % 42.2 % 18.9 % 2.2 % 0.3 % 0.1 % 0.0 % 茶 1388 98.3 % 97.0 % 85.3 % 65.0 % 9.6 % 0.3 % 0.3 % 0.0 % 緑 1053 97.4 % 96.4 % 93.3 % 89.9 % 37.9 % 2.0 % 1.1 % 0.1 % 水 665 98.0 % 97.6 % 96.2 % 96.1 % 72.3 % 13.5 % 5.3 % 0.0 % 青 372 98.9 % 97.6 % 97.6 % 97.3 % 83.1 % 51.1 % 22.3 % 0.3 % 黄 182 90.7 % 89.0 % 89.0 % 90.1 % 84.6 % 72.0 % 58.8 % 8.8 % 橙 41 97.6 % 97.6 % 97.6 % 97.6 % 97.6 % 92.7 % 82.9 % 31.7 % 赤 24 91.7 % 91.7 % 95.8 % 87.5 % 91.7 % 91.7 % 91.7 % 54.2 % ※表示レート、灰に初参加者は含めず A問題『Not Overflow』 問題ページ：A - Not Overflow 灰コーダー正解率：91.0 % 茶コーダー正解率：98.3 % 緑コーダー正解率：97.4 % 入力 $N$ : 整数（$-2^{63}\le{N}\lt{2^{63}}$） 考察 $N$ が $-2^{31}$ 以上 かつ、$2^{31}$ 未満であるか判定すればいいです。境界値（以下か未満か）のミスは非常によくあるので、よく問題文を読んで間違えないように注意してください。 一般的な言語では64bit整数型を使用しないとオーバーフローしますが、Pythonは多倍長整数を使用しているため、考える必要はありません。 コード C = 2 ** 31 N = int(input()) print("Yes" if -C <= N < C else "No") # 以上、未満です B問題『Matrix Transposition』 問題ページ：B - Matrix Transposition 灰コーダー正解率：82.4 % 茶コーダー正解率：97.0 % 緑コーダー正解率：96.4 % 入力 $H,W$ : $H$ 行 $W$ 列の行列が与えられる $A_{i,j}$ : $A$ の $i$ 行 $j$ 列目の要素 考察 $A$ を二次元リストで受け取ります。Pythonでは、At = list(zip(*A)) と書くと転置行列になるので、これを出力すればいいです。 コード H, W = map(int, input().split()) A = [list(map(int, input().split())) for _ in range(H)] At = list(zip(*A)) for r in At: print(*r) NumPy 全く使う必要はありませんが、NumPyを使う場合はこうなります。（AtCoderのPyPyにはNumPyが入っていないので、Pythonで提出する必要があります） import numpy as np H, W = map(int, input().split()) A = np.array([list(map(int, input().split())) for _ in range(H)]) At = np.transpose(A) for r in At: print(*r) C問題『kasaka』 問題ページ：C - kasaka 灰コーダー正解率：42.2 % 茶コーダー正解率：85.3 % 緑コーダー正解率：93.3 % 入力 $S$ : 英小文字からなる文字列 考察 回文とは逆から読んでも同じ文字列のことですから、自分自身を反転したものと一致すれば回文です。回文判定は S == S[::-1] で簡単にできます。 回文になるとすれば、少なくとも『$S$ を先頭から見て a 以外の文字が出るまで a が連続で出てくる文字数』『$S$ を末尾から見て a 以外の文字が出るまで a が連続で出てくる文字数』が一致している必要があります。この条件を満たしていても回文とは限りませんが、満たしていなければ回文にはなりません。 先頭・末尾の a の連続数が合うように、先頭にaを付け加えて回文判定を行えばいいです。ただし、a は先頭に付け加えることしかできないので、先頭の連続数のほうが多い場合は回文にはできません。 例1 aabaaa は、先頭から見て a が $2$ 文字、末尾から見て a が $3$ 文字連続で出てきます。先頭にaを $1$ 文字付け加えて、aaabaaa にすると回文になります。 例2 abcaa は先頭から見て a が $1$ 文字、末尾から見て a が $2$ 文字連続で出てきます。先頭にa を $1$ 文字付け加えて aabcaa にしても回文にはなりません。 例3 aaaaba は先頭から見て a が $4$ 文字、末尾から見て a が $1$ 文字連続で出てきます。a は先頭に付け加えることしかできないので、回文にはできません。 解法 この問題は以下のアルゴリズムで解けます。 aが先頭・末尾それぞれ何文字連続で出てくるか調べる 先頭・末尾のaの文字数が合うように、先頭にaを付け加えて判定する（先頭のほうが多い場合は不可能） コード def judge(): S = input() l = len(S) - len(S.lstrip('a')) # 先頭の連続数 r = len(S) - len(S.rstrip('a')) # 末尾の連続数 if l > r: return False T = "a" * (r - l) + S return T == T[::-1] print("Yes" if judge() else "No") D問題『LR insertion』 問題ページ：D - LR insertion 灰コーダー正解率：18.9 % 茶コーダー正解率：65.0 % 緑コーダー正解率：89.9 % 入力 $N$ : 操作を行う回数 $s_i$ : $i$ 回目の操作では $s_i$ がLかRのどちらかに応じて、 $i$ を $i-1$ のすぐ左かすぐ右に挿入する 考察 $A={N}$ から操作を逆向きに行い、$s_i$ が L なら $i-1$ を $A$ の右端に追加し、R なら $A$ の左端に追加することで簡単に解けます。 なぜこれでうまくいくかを考えます。$s_i$ が L のとき、$i$ 以上の要素はすべて $i-1$ より左に追加されます。$i$ 以上の要素をすべて数列に追加してから、$i-1$ を $A$ の右端に追加すれば正しい位置に $i-1$ を追加できます。$s_i$ が R の場合も同じです。 先頭から見る方法や、双方向連結リストを使う解法もありますが、省略します。 実装 数列の左端と右端両方向の挿入が必要なので、両端キューのdeque を使います。 コード from collections import deque def main(): N = int(input()) S = input() ans = deque((N,)) for i in reversed(range(N)): if S[i] == "L": ans.append(i) else: ans.appendleft(i) print(*ans) main() E問題『Skiing』 問題ページ：E - Skiing 灰コーダー正解率：2.2 % 茶コーダー正解率：9.6 % 緑コーダー正解率：37.9 % 入力 $N,M$ : 広場の数は $N$ 個で、坂の数は $M$ 本 $H_i$ : 広場 $i$ の標高 $U_i,V_i$ : $i$ 本目の坂は 広場 $U_i$ と $V_i$ を結ぶ どの頂点も互いに坂を使ってたどり着ける（グラフは連結） 考察 この問題をわかりやすくいうと、$X$ と $Y$ の標高差の絶対値 $|H_X-H_Y|$ を使って 広場 $X$ から今より標高が高い広場 $Y$ に坂を使って移動すると、楽しさが $2|H_X-H_Y|$ 減る 広場 $X$ から今より標高が低い広場 $Y$ に坂を使って移動すると楽しさが $|H_X-H_Y|$ 増える 頂点 $1$ から 楽しさの初期値 $0$ ではじめたときの、楽しさを最大化せよ という問題です。 最短経路問題とみなす 楽しさの正負を反転して、距離として『楽しさの $-1$ 倍』を考え、辺を以下のように貼ったグラフを考えます。 自分より高い広場には、コスト $2|H_X-H_Y|$ の辺を貼る 自分より低い広場には、コスト $-|H_X-H_Y|$ の辺を貼る このグラフ上で、頂点 $1$ からはじめた全頂点対最短経路問題を解き、距離（楽しさの $-1$ 倍）が最小の頂点の値を $-1$ 倍すると答えになります。 自分より低い広場に対してはコストが負の辺を貼ることになりますが、負閉路は絶対に出現しないことに注意してください。（同じ場所に戻ってくることはできますが、楽しさの減少量を増加量が上回るので、『楽しさの $-1$ 倍』である距離は減少しません） 負辺が邪魔 どのアルゴリズムを使えば解けるか考えてみましょう。 ダイクストラ法 : 負辺がある場合、$O(2^N)$ になるケースを作成可能なため不可（コンテスト中はACできてしまいましたが、after contestのテストケースでTLEになります） ベルマン・フォード法 : 計算量が $O(NM)$ のため、不可 どちらのアルゴリズムも使えないので、このままのグラフで最短経路問題を解くのは難しいです。 ポテンシャルを考える なぜ負のコストが出てきたかというと、距離として『楽しさの $-1$ 倍』を使ったからです。『楽しさの $-1$ 倍』は坂を下ると減少します。絶対に減少することがない、別の値を距離として使うことを考えます。 ここで、「楽しさが増える量は、標高が減った量と同じである」ことに着目します。言い換えると、坂を下っても『楽しさ+標高』は変化しません。そこで、距離として『楽しさ+標高』の $-1$ 倍を使った以下のグラフを考えます。 自分より高い広場には、コスト $|H_X-H_Y|$ の辺を貼る（楽しさが $2|H_X-H_Y|$ 減って、標高が $|H_X-H_Y|$ 増えるため、差し引き $|H_X-H_Y|$ です） 自分より低い広場には、コスト $0$ の辺を貼る このグラフ上で、頂点 $1$ の距離の初期値を $-H_1$ として、頂点 $1$ からダイクストラ法を行います。求めた頂点 $1$ から 頂点 $i$ への最短距離を $dist_i$ とすると、『頂点 $i$ で終わるときの最大の楽しさ』は $- dist_{i} - H_i$ で求められるので、全頂点に対して調べて、最大の値が答えになります。 （$楽しさ = 『楽しさ+標高』-標高=-『(楽しさ+標高)の-1 倍』 - 標高$ です） コード from heapq import heappop, heappush INF = 1 << 62 - 1 def main(): def dijkstra(G, start): N = len(G) dist = [INF] * N dist[start] = -H[0] pq = [(-H[0], start)] while pq: cost, u = heappop(pq) if dist[u] < cost: continue for v, d in G[u]: new_cost = cost + d if dist[v] > new_cost: dist[v] = new_cost heappush(pq, (new_cost, v)) return dist N, M = map(int, input().split()) H = list(map(int, input().split())) G = [[] for _ in range(N)] for _ in range(M): a, b = map(int, input().split()) a, b = a - 1, b - 1 G[a].append((b, max(0, H[b] - H[a]))) G[b].append((a, max(0, H[a] - H[b]))) dist = dijkstra(G, 0) ans = max(-dist[i] - H[i] for i in range(N)) print(ans) main() F問題『|LIS| = 3』 問題ページ：F - |LIS| = 3 灰コーダー正解率：0.3 % 茶コーダー正解率：0.3 % 緑コーダー正解率：2.0 % 入力 $N$ : 数列の長さ $M$ : 数列の各項の値は $1$ 以上 $M$ 以下 考察 長さ $3$ に制限したLISの配列を状態として、動的計画法を行います。$i$ 番目まで決めてLIS配列が $(a,b,c)$ である数列の $i+1$ 文字目を $x$ にしたとき、LIS配列がどうなるかは、$i$ によらず一意に決まり、また更新されるのは $(a,b,c)$ のいずれか一箇所のみです。 状態数は $O(NM^3)$ です。遷移は時間計算量 $O(1)$ でできるので、計算量 $O(NM^3)$ でこの問題を解くことができました。 コード PyPyで提出してください。 MOD = 998244353 def main(): N, M = map(int, input().split()) prev = [[[0] * (M + 1) for _ in range(M + 1)] for _ in range(M + 1)] prev[M][M][M] = 1 for _ in range(N): dp = [[[0] * (M + 1) for _ in range(M + 1)] for _ in range(M + 1)] for x in range(M): for a in range(M + 1): for b in range(M + 1): for c in range(M + 1): # LIS配列としてありえないものも含まれますが、問題ありません p = prev[a][b][c] if x <= a: dp[x][b][c] += p dp[x][b][c] %= MOD elif x <= b: dp[a][x][c] += p dp[a][x][c] %= MOD elif x <= c: dp[a][b][x] += p dp[a][b][x] %= MOD prev = dp ans = 0 for a in range(M): for b in range(a + 1, M): for c in range(b + 1, M): ans += prev[a][b][c] ans %= MOD print(ans) main()

from manimlib.imports import * import numpy as np class Sierpinski(MovingCameraScene): def construct(self): def b_afin(Cx,Cy): b_afin = np.array([ [1,0,-Cx], [0,1,-Cy], [0,0,1], ]) return b_afin def a_afin(Cx,Cy): a_afin = np.array([ [1,0,Cx], [0,1,Cy], [0,0,1] ]) return a_afin def rotate_matrix(rad): r_matrix = np.array([ [np.cos(rad),-1*np.sin(rad),0], [np.sin(rad),np.cos(rad),0], [0,0,1], ]) return r_matrix def create_tri(a,b,c): length = b[0]-a[0] c = [0,0,0] c[0] = a[0] + length / 2 c[1] = a[1] + np.sqrt(3)*length / 2 c = np.array([c[0],c[1],1]) n1 = (a + c)/2 n2 = (b + a)/2 n3 = (c + b )/2 line1 = Line(n1,n2,stroke_width =1).set_color(BLUE_A) line2 = Line(n2,n3,stroke_width =1).set_color(BLUE_A) line3 = Line(n3,n1,stroke_width =1).set_color(BLUE_A) length = np.linalg.norm(n2-a) length2 = np.linalg.norm(b-n2) length3 = np.linalg.norm(n3-n1) dots = [a,n2,n1,n2,b,n3,n1,n3,c] if length < 0.125/2: return else: lines = VGroup(line1,line2,line3) self.add(lines) self.wait(0.1) create_tri(a,n2,n1) create_tri(n2,b,n3) create_tri(n1,n3,c) return dots a = np.array([-3,-2,1]) Ax = a[0] Ay = a[1] b = [0,0,1] b[0] = a[0] + 6 b[1] = a[1] b[2] = 1 b = np.array([b[0],b[1],b[2]]) c = a_afin(Ax,Ay)@(rotate_matrix(np.pi/3)@(b_afin(Ax,Ay)@b)) tri = Polygon(a,b,c,stroke_width =1) self.play(ShowCreation(tri)) create_tri(a,b,c) self.wait(1)