1.初めに Pythonでデータを可視化する際、matplotlibとseabornが定番だが、表示範囲やスケールを変える際の手間がかかる。 インタラクティブに表示範囲・スケールを変更できる可視化ツールとして、今回Plotlyの使い方を簡単にまとめた。 2.プログラム visualize.py import pandas as pd import numpy as np df_passanger = pd.read_csv("AirportPassengers.csv", sep=";", index_col=0) print(df_passanger.head()) # Using plotly.express import plotly.express as px #fig = px.line(df_passanger, x="Month", y="Passengers", title='Life expectancy in Canada') fig = px.line(df_passanger, y="Passengers", title='Air Passengers') fig.show() read_csvでファイルを読み込む。その後、plotly.expressを使い、折れ線グラフでデータを表示。 グラフ部分の表示例を以下に示す。 output.py fig.write_html("test.html") グラフ画像は、グラフエリア右上のボタンから画像ファイルとして手動で保存できるが、上記のようにhtml方式で保存もできる。html方式の場合、スケールなども変更できる。 3.参考資料 公式ドキュメント Line Plots with plotly.express Plotly.express.line Qiita記事 [Python] Plotlyでぐりぐり動かせるグラフを作る

せっかくQiitaのアカウントを取得しましたので、AtCoderの問題の解説を書いてみたいと思います。 問題 HHKB プログラミングコンテスト 2022（AtCoder Beginner Contest 235） 考察 やりたいこと まずは落ち着いて問題文を読む。 長さNの数列 A がある。たとえば、A = [ 1 1 2 3 1 2 ] (入力例1の場合) 。 この数列に対して、以下のようなクエリが飛んでくるので、順に答える。 ・「1が1番目に出てくる場所はどこか?」　→　「1番目です。」 ・「2が2番目に出てくる場所はどこか?」　→　「6番目です。」 ・「3が2番目に出てくる場所はどこか?」　→　「ありません。(3は1個しかない)」 ・「5が1番目に出てくる場所はどこか?」　→　「ありません。(5は一つもない)」 愚直な探索の計算量 一つのクエリが来たら、Aの先頭から順に要素を見ていって、たとえば「3が2番目に出てくる場所」がどこかを調べれば、Aの要素へのアクセスを $2\times10^5$ 回程度することで、クエリに答えることが出来る。 しかし、クエリは $2\times10^5$ 個あるので、最悪で $(2\times10^5)\times(2\times10^5)=4\times10^{10}$ 回のAの要素へのアクセスが必要となる。 計算の効率化 上記の方法だと、毎回先頭から順に要素を見ている部分が効率が悪く感じられる。 極端な話、仮に最初に「1が3番目に出てくる場所はどこか」というクエリがあったなら、先頭から順に要素を調べて、3つめの「1」が見つかった時点で、「1が1番目に出てくる場所」「2が2番目に出てくる場所」は把握できている。 そもそも、先頭から順に調べる際に、「1」だけに注目するのではなく、他の数字もどこにあったかを覚えておくことも出来る。 それならいっそのこと、最初に先頭から最後までAを走査(スキャン)して、どの数字が、何番目と何番目にあったかのメモを作ればよい。 そうすると、クエリに対して(一つ一つAの要素を順に見ずに)即座に答えられる。 ・「2が2番目に出てくる場所はどこか?」 　→　2は「3,6」番目にあるので、答えは「6番目」。 ・「3が2番目に出てくる場所はどこか?」 　→　3は「4」番目にある。このリストは要素数が1なので、2番目の「3」は「ありません」。 ・「5が1番目に出てくる場所はどこか?」 　→　メモの中には5は登場していないので、「ありません」。 このようなデータ構造を作ることが出来れば、効率的に解ける。 必要な計算量は、 ・メモの作成の際に、Aの要素に合計 $2\times10^5$ 回アクセスして内容を読み取る。 ・$2\times10^5$ 個のクエリに対して、即座に答える。 なので、時間制限に十分間に合いそうだ。 実装 上記のようなメモは、「リストの辞書」を使うと実装できる。利用時に ・memo[1] → [1, 2, 5] ・memo[2] → [3,6] ・memo[3] → [4] ・memo[5] → (要素なし) のように使えるようなイメージだ。 memoを作る手順は以下のようにする。先頭から順番にAの要素を見ていって、 ・「1」番目は 1 だった　→　memo[1] のリストに「1」を追加。 ・「2」番目は 1 だった　→　memo[1] のリストに「2」を追加。 ・「3」番目は 2 だった　→　memo[2] のリストに「3」を追加。 ・ (以下同様...) のようにすればよい。 このような場合は、collections の defaultdict が役に立つ。 詳細は、下記のコードを見ていただければわかるかと思います。 解答 清書した解答 【提出 #28589877】 ABC235C.py from collections import defaultdict # 入力を受け取る n,q = tuple(map(int, input().split())) a = list(map(int, input().split())) xk = [ tuple(map(int, input().split())) for _ in range(q)] # メモの辞書 memo = defaultdict(list) # メモを作る for i,ai in enumerate(a): memo[ai].append((i+1)) # python は 0-indexed なので、i に 1 を足しておく # 答えを格納する配列 ans = [] for x,k in xk: # A の中に x が存在する個数 = memo[x] の要素数 となる。 # A の中に x が存在しない場合、 memo[x] = [] (空リスト) になるので、問題ない。 if len(memo[x])<k: ans.append(-1) else: ans.append(memo[x][k-1]) # python は 0-indexed なので、k から 1 を引く # 答えを出力 print(*ans, sep='\n')

新たにpandasを学習し始めたので、学習した内容を記録しておきます。間違った内容を書いてしまう場合もございますのでその際はご指摘していただけるとありがたいです。 記事の内容はpandasを使用する上での最低限の内容になります。 Pandasとは Pythonで最もポピュラーなライブラリの一種で,主に表計算などのデータ処理が得意でSeriesやDataFrameなどの独自のデータ形式を持っています。 Excleやcsvなどの表形式のデータを読み込んで、計算を行うことができます。 DataFrameとは Pandasでは表形式のデータのことをDtaFrameと言い、Headerの中にcolumnがあり横１行のことをrecordと言います。 そして縦一列または横一行のことをSeriesといい、このSeriesが集まってできたものをDataFrameと呼びます。 DataFrame基本操作 まず初めに、Pandasをimportします。 import pandas as pd 次に、kaggleからデータをread_csvインポートします、今回使用するのはTMDB5000　moviedatasetです。（映画の興行収入やレビューなどが記録されています。） read_csvと入力して()にPATHを指定しることでDataFrameを持ってくることができます。 read_は他にもデータ形式を指定できます。(Excle, SQL, JSON etc...) df変数に格納します。 基本的にデータサイエンスでは、データ量が膨大なので同じ変数をできるだけ使いまわします。 df = pd.read_csv('tmdb_5000_movies.csv') .head()で括弧内に数字を打ち込むとその数だけデータを表示します。 df.head(3) .describe()で統計量を表示します。 df.describe() .columnsでカラムの一覧を取得します。 df.columns Seriesをリストで表示するとDataFrameで返ってきます。 df[['revenue', 'original_title']] .iloc[index]で特定のSeriesを取得します。 df.iloc[10] .drop(index)で行やカラムを取り除くことができる。 df.drop(0) df.drop('id', axis=1, inplace=True) inplace=Trueで元のdfを更新します。 DataFrameのフィルタ 設定した条件に該当するデータを抽出する。

はじめに Cloud Functions を python で使ってみるための調査をしました。いまどき、サーバレスでなくてはね、ということで、老体に鞭を打ち、休日に作業しています。戦いの証をこちらに掲げています。 Quick Start から CDまで まずはQuick Start Quick start は、コンソールだけでdeploy するところまでできます。テストも楽にできるんだなーというのが、印象的です。 その後、gcloud による操作をここを読んでざっと理解しました。 これらをもとに、Cloudbuild でCI/CDできるように頑張りました。 作業内容 準備 準備します。Python3.9 にしているのは、使ってみたいからです。 $ virtualenv -p python3.9 venv $ source ./venv/bin/activate (venv) $ $ python -V Python 3.9.5 ローカルで動かすために必要なものを入れます。 $ python -m pip install functions-framework ... # python -m pip list $ python -m pip list Package Version ------------------------------ --------- ... Flask 2.0.2 functions-framework 3.0.0 google-api-core 2.4.0 google-auth 2.3.3 grpc-google-iam-v1 0.12.3 grpcio 1.43.0 ... という感じでした。 ローカルで動かす 関数 my_hello_http が実装してあるフォルダで、function-framework を実行します。 実装はサンプルにあるもののコピペです。 main.py from flask import escape import functions_framework @functions_framework.http def my_hello_http(request): request_json = request.get_json(silent=True) request_args = request.args if request_json and 'name' in request_json: name = request_json['name'] elif request_args and 'name' in request_args: name = request_args['name'] else: name = 'World' return 'Hello {}!'.format(escape(name)) これを functions-framework で実行します。 $ functions-framework --target my_hello_http --signature-type=http --port=8080 --debug * Serving Flask app 'hello_http' (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: on * Running on all addresses. WARNING: This is a development server. Do not use it in a production deployment. * Running on http://172.25.125.68:8080/ (Press CTRL+C to quit) * Restarting with watchdog (inotify) * Debugger is active! * Debugger PIN: 111-222-666 別のターミナルで接続してみる。 $ curl -H "accept: application/json" -H "Content-Type: application/json" -d '{"name": "baby."}' http://172.25.125.68:8080/my_hello_sample Hello baby.! 動いた。よしよし。 手動で Deploy deploy は gcloud functions deploy で行ける。覚えられるかな。 $ gcloud functions deploy my_hello_http --runtime python39 --trigger-http --allow-unauthenticated Deploying function (may take a while - up to 2 minutes)...⠛ For Cloud Build Logs, visit: https://console.cloud.google.com/cloud-build/builds;region=us-central1/xxxxxxxxxxxxxxxxxxxxx?project=1234567890123 Deploying function (may take a while - up to 2 minutes)...done. availableMemoryMb: 256 buildId: xxxxxxxxxxxxxxxxxxxxxxxxxxx buildName: projects/1234567890123/locations/us-central1/builds/xxxxxxxxxxxxxxxxxxxxxxxxxxx entryPoint: my_hello_http httpsTrigger: securityLevel: SECURE_ALWAYS url: https://us-central1-myproject.cloudfunctions.net/my_hello_http ingressSettings: ALLOW_ALL labels: deployment-tool: cli-gcloud name: projects/myproject/locations/us-central1/functions/my_hello_http runtime: python39 serviceAccountEmail: ambmonitordev@appspot.gserviceaccount.com sourceUploadUrl: https://storage.googleapis.com/gcf-upload-us-central1-yyyyyyyyyyyyyyyyyyy/zzzzzzzzzzzzzzzzzzzzzzzzzzz.zip status: ACTIVE timeout: 60s updateTime: '2022-01-16T11:16:18.258Z' versionId: '1' 内部ではどのように動いているのかな。build したり、storage の設定が出てきたり、entryPoint の文字が見えたり。 この情報は、以下でも表示できる。 $ gcloud functions describe my_hello_http deploy しているserverless application を忘れたら、list する。 $ gcloud functions list NAME STATUS TRIGGER REGION my_hello_http ACTIVE HTTP Trigger us-central1 CloudBuild でDeploy 今のコマンドをcloudbuild.yaml に書いてみる。 steps: - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk' args: - gcloud - functions - deploy - my_hello_http - --region=asia-northeast1 - --source=. - --trigger-http - --runtime=python39 とりあえずローカルで実行してみる。今、main.py と同じディレクトリで下記を実行することを想定していますが、そうでなければ"source=..."の値を書き換えてください。 $ gcloud builds submit --config cloudbuild.yml . がエラーになってしまった。 - '@type': type.googleapis.com/google.rpc.ErrorInfo domain: googleapis.com metadata: consumer: projects/12345678901234 service: cloudresourcemanager.googleapis.com reason: SERVICE_DISABLED ERROR と言われたので、Cloud Resouce Manager API を有効にしました。そうしたら成功はしたのですが、 Deploying function (may take a while - up to 2 minutes)... WARNING: Setting IAM policy failed, try `gcloud alpha functions add-iam-policy-binding my_hello_http --region=asia-northeast1 --member=allUsers --role=roles/cloudfunctions.invoker` と言われたので、素直にコマンドを叩いておきました。 $ gcloud alpha functions add-iam-policy-binding my_hello_http --region=asia-northeast1 --member=allUsers --role=roles/cloudfunctions.invoker bindings: - members: - allUsers role: roles/cloudfunctions.invoker etag: BwXVssdV2mE= version: 1 いよいよCloud Build にtrigger を作ります。いろいろエラーが出たのですが、 CloudBuild のサービスアカウントに CloudFunctions 開発者のロールが無効だった。これはコンソールのCloudBuild の設定から変更できます。 build trigger を実行するサービスアカウントの権限がLogging になかった。これはせってしない方が良かったのかもしれない。 で最後は無事にdeploy できました!! (^^)/ まとめ GCPのサーバレスをpython で利用するために、Cloud Functions を python 使い方を調査し、動作確認をとった。 ローカルで動かすことも可能。(functions-) CloudBuild から deploy 可能 いろいろ作っていくのか。location(region?) が違うと別のものとして扱われているようだった。 来週も生き延びれるかな。。。 メモ 参考にさせていただきました。PubSubもさくっと動作確認されている。 ローカルに作ったパッケージを呼び出す。 google 様のサンプル: python-doc-samples/functions/ 以下に全部ある。 gcloud の使い方入門

概要 前回、気温を予測するAIを作りましたが、天気を予測してみようと思い作りました 環境 OS:Windows 言語:Python エディター:Geany データを集める AIに学習させるための気象データを集めましょう。 今回は気象庁のHPから過去の気象データをダウンロードします。 年月日,平均気温(℃),降水量の合計(mm),日照時間(時間),最深積雪(cm),平均風速(m/s),最多風向(16方位),平均蒸気圧(hPa),平均現地気圧,等のデータをダウンロードしましょう。 説明変数がテキストだった場合どうすればよいか？ 今回は説明変数の中に風向き(16方向)がありますが、そのままでは使えません。なのでテキストを種類ごとにナンバリングしてくれる機能を使います。 from sklearn import preprocessing label = preprocessing.LabelEncoder() label.fit(data) labels_id = le.transform(data) 上2行はLabelEncoderの宣言。 label.fit(data)はラベルとラベルIDの対応づけを行います。この文字列は0にしよう，みたいなことを決めています。これはle.fit(ラベルの一次元リスト)で行います。 ラベルとラベルIDの対応づけの変換することができます。これはtransform(ラベルの一次元リスト)で行います。返り値に配列が来ます。 AIモデルを決める Scikit-learnのアルゴリズムチートシートを見てモデルを決めます。 今回は「サンプル数は50以上」→「カテゴリー予測」→「ラベル付きのデータを持っている」→「サンプル数は10万以下」としてLinerSVC(線形)やkneighbors classifier(非線形),ensemble classifier(非線形)等が選べます。 ３つ試したところensemble classifierのスコアが一番良かったのでensemble classifierのコードのみ載せておきます。 ensemble classifier(非線形) 作成したAIモデルにスコアをつけるプログラム。 値が大きいほど良い (0<score<1) 今回はあらかじめ説明変数のファイルと目的変数のファイルを分けています。(data.csv、targetGOZEN.csv) tenkiAI # -*- coding: Shift-JIS -*- import re import numpy as np from sklearn.model_selection import train_test_split from sklearn import svm,linear_model,preprocessing,ensemble from sklearn.neighbors import KNeighborsClassifier np.set_printoptions(threshold=np.inf) #年月日,平均気温(℃),降水量の合計(mm),日照時間(時間),最深積雪(cm),平均風速(m/s),最多風向(16方位),平均蒸気圧(hPa),平均現地気圧(hPa) #天気概況(昼：06時～18時) 天気概況(夜：18時～翌日06時) #気象データを説明変数として読み込む f=open('data.csv','r') data=re.split('[,\n/]',f.read()) f.close() del data[-1] print(data) del data[::11]#西暦を消す #方角をナンバリングする dire=[] for i in data[7::10]: dire.append(i) le = preprocessing.LabelEncoder() le.fit(dire) labels_id = le.transform(dire) #方角データをナンバリング方角に置き換える j=0 for i in range(len(data)): if(i%10==0): data[i+7]=labels_id[j] j+=1 #二次元numpy配列にする data = np.array(data, dtype='float64') data=data.reshape(int(len(data)/10),10) #天気データを目的変数とする f=open('targetGOZEN.csv','r') target=re.split('[,\n/]',f.read()) f.close() del target[-1] target = np.array(target, dtype='str') #明日の天気を予測するために説明変数と目的変数を１日分ずらす data=np.delete(data,len(target)-1,0) target=np.delete(target,0) #学習用データとテストデータに分ける traind, testd, traint, testt =train_test_split(data,target, test_size=0.2) #学習モデルの生成と学習 clf = ensemble.RandomForestClassifier(criterion='entropy', n_estimators=1000, random_state=1, n_jobs=2,class_weight="balanced") clf.fit (traind,traint) #スコアの表示 print(clf.score(testd,testt)) 結果はこうなります。 実行結果 スコア:0.1761904761904762 まとめ スコアが低いですが今回はクラス分類での正答率をスコアにしているので、例えば本来の天気が「晴れ」に対して「晴れのち曇り」という予測が出てきた場合、当たらずとも遠からずといった感じですがスコアでは不正解にカウントされています。 自分で予測を確認したところそこまで大きな間違いは感じられませんでしたし、逆にぴったり当たっている予測が17%ほどあるということはそこそこなスコアだと思います。

Pythonのmplfinanceはローソク足チャートを作成するにあたって非常に便利なライブラリですが、描画するスタイルを自分の好みに編集する方法をまとめているサイトが全然なかったので、備忘録として記しておきます。 pandas_datareaderを使ってデータを取得する ローソク足を使うデータは以下のコードで取得したものを使用します。 import pandas as pd import numpy as np import matplotlib.pyplot as plt from pandas_datareader import data # yahooから2021/01/01から2021/04/01の期間の日経平均のデータを取得 df_master = data.DataReader('^N225', 'yahoo', '2021-01-01', '2021-04-01') 1. mplfinanceに用意されているスタイルを利用する mplfinanceライブラリに用意されているスタイル一覧は以下のコードで確認できます。 import mplfinance as mpf mpf.available_styles() output: [ 'binance','blueskies','brasil','charles','checkers', 'classic','default','ibd','kenan','mike', 'nightclouds','sas','starsandstripes','yahoo' ] これらのいずれかを指定することにより、あらかじめ用意されたスタイルに簡単に変更できます。 チートシート import mplfinance as mpf mpf.plot( df_master, # 使用するデータフレームを第一引数に指定 type='candle', # グラフ表示の種類を指定 ローソク足チャートであれば"candle" style='yahoo', # ここにスタイル一覧の中から好みのものを指定 例では"yahoo"を指定 # チャートのサイズの設定 figratio=(20, 10), # 図のサイズをタプルで指定 figscale=1.0, # 図の大きさの倍率を指定、デフォルトは1 tight_layout=False, # 図の端の余白を狭くして最適化するかどうかを指定 # 軸の設定 datetime_format='%Y/%m/%d', # X軸の日付の表示書式を指定 # xlim=('2021-01-01', '2021-02-01'), # X軸の日付の範囲をタプルで指定 指定無しならデータフレームを元に自動で設定される # ylim=(25000, 30000), # Y軸の範囲をタプルで指定 指定無しなら自動で設定される xrotation=45, # X軸の日付ラベルの回転角度を指定 デフォルトは45度 axisoff=False, # 軸を表示するかどうかを指定 デフォルトは"False" # データの表示設定 volume=False, # ボリュームを表示するかどうかを指定 デフォルトは"False" show_nontrading=False, # データがない日付を表示するかどうかを指定 デフォルトは"False" # ラベルの設定 title='Title', # チャートのタイトル ylabel='ylabel', # チャートのY軸ラベル ylabel_lower='ylabel_lower', # ボリュームを表示する場合は、ボリュームのグラフのY軸ラベル ) このソースコードを実行すると、以下のようなグラフが得られます。 2. mplfinanceのチャートのスタイルを自作する mplfinanceに用意されているスタイルではなく、自分好みにスタイルを作成することもできます。辞書形式に指定するだけなので、容易に変更できるようになっています。 チートシート # ローソクの色の設定 marketcolors = mpf.make_marketcolors( up='#00BFFF', # 上昇時のろうそくの塗りつぶし色 down='#DC143C', # 下降時のろうそくの塗りつぶし色 edge='lightgray', # ろうそくの端の線の色 wick={ # 辞書形式で、ろうそく足の真の色を指定 'up':'#00BFFF', # 上昇時のろうそくの芯の色 'down':'#DC143C' # 下降時のろうそくの芯の色 } ) # グラフ全体の設定 mpfstyle = mpf.make_mpf_style( marketcolors=marketcolors, gridcolor='lightgray', # チャートのグリッドの色 facecolor='white', # チャートの背景の色 edgecolor='black', # チャートの外枠の色 figcolor='white', # チャートの外側の色 gridstyle='-', # チャートのグリッドの種類 "--":実線, "--":破線, ":":点線, "-.":破線と点線の組み合わせ gridaxis='both', # チャートのグリッドの有無を指定 both:縦横双方, horizontal:横線のみ, vertical:縦線のみ y_on_right=False, # y軸を右に表示するかどうかを指定 rc = { 'xtick.color': 'black', # X軸の色 'xtick.labelsize': 8, # X軸の文字サイズ 'ytick.color': 'black', # Y軸の色 'ytick.labelsize': 8, # Y軸の文字サイズ 'axes.labelsize': 10, # 軸ラベルの文字サイズ 'axes.labelcolor': 'black', # 軸ラベルの色 # 'font.family': 'IPAexGothic', # タイトル,ラベルのフォントを指定 } ) mpf.plot( df_master, # 使用するデータフレームを第一引数に指定 type='candle', # グラフ表示の種類を指定 ローソク足チャートであれば"candle" style=mpfstyle, # 自作したスタイルを指定 # チャートのサイズの設定 figratio=(20, 10), # 図のサイズをタプルで指定 figscale=1.0, # 図の大きさの倍率を指定、デフォルトは1 tight_layout=False, # 図の端の余白を狭くして最適化するかどうかを指定 # 軸の設定 datetime_format='%Y/%m/%d', # X軸の日付の表示書式を指定 # xlim=('2021-01-01', '2021-02-01'), # X軸の日付の範囲をタプルで指定 指定無しならデータフレームを元に自動で設定される # ylim=(25000, 30000), # Y軸の範囲をタプルで指定 指定無しなら自動で設定される xrotation=45, # X軸の日付ラベルの回転角度を指定 デフォルトは45度 axisoff=False, # 軸を表示するかどうかを指定 デフォルトは"False" # データの表示設定 volume=False, # ボリュームを表示するかどうかを指定 デフォルトは"False" show_nontrading=False, # データがない日付を表示するかどうかを指定 デフォルトは"False" # ラベルの設定 title='Title', # チャートのタイトル ylabel='ylabel', # チャートのY軸ラベル ylabel_lower='ylabel_lower', # ボリュームを表示する場合は、ボリュームのグラフのY軸ラベル ) 色は、RGBで指定するか、またはメインの色であれば色の名前で指定します。 このソースコードを実行すると、以下のようなグラフが得られます。 以上となります。 次回以降は、mplfinanceにて作ったグラフにテクニカル指標についてのグラフを表示する方法についてまとめる予定です。

モチベーション Windows で Wi-Fi の SSID と BSSID を取得したいと思ったのがきっかけ。 はじめはコマンドラインで netsh wlan show networks mode=bssid で Wi-Fi の一覧を取得しようかと考えていたのですが、参考資料にある通り、何度実行しても同一のSSID/BSSIDになってしまったり、複数あるはずのSSIDが接続中のSSID1つだけになってしまったりしました。 解決方法を以下に引用します。 C#を使ってWindowsAPIを呼び出すことでWifiの電波強度一覧を取得することができました。 使用したAPIはWlanScan()です。NativeWifiというオープンソースのライブラリを使って使用することができました。 上記を Python でもできないか調べたところ、win32wifi を見つけました。 環境 Windows 10 Python 3.6 追加ライブラリ win32wifi ※ 3.7 以上で実行しようとすると import comtypes でエラーが生じます。参考資料の方法で回避できるかもしれません。 プログラム 実行をする場合は中身を確認しながら自己責任でお願いします。 WlanRegisterNotification にて通知がきても Wi-Fi 情報が空の場合があるため3回実行しています。 (何か使い方が間違っているのでしょうか...) from win32wifi import Win32Wifi as ww import threading def get_ssid(trial=3): ## Wlanが変更を登録するEvent th_ev = threading.Event() def callback(wnd, p): th_ev.set() interfaces = ww.getWirelessInterfaces() handle = ww.WlanOpenHandle() ## 試行回数 for i in range(trial): ssid_dict = {} for interface in interfaces: ## 検索 ws = ww.WlanScan(handle, interface.guid) ## 登録完了を最大10秒待機する。 cb = ww.WlanRegisterNotification(handle, callback) th_ev.wait(10) th_ev.clear() networks = ww.getWirelessNetworkBssList(interface) for network in networks: ssid = network.ssid.decode() bssid = network.bssid quality = network.link_quality ## ssid がなければリストを追加 if ssid not in ssid_dict.keys(): ssid_dict[ssid] = [] ## bssid と電波強度を追加 ssid_dict[ssid].append((bssid, quality)) ## 空でなければ break if ssid_dict != {}: break ww.WlanCloseHandle(handle) return ssid_dict if __name__ == "__main__": print(get_ssid()) 実行結果(例) 実行毎に更新し、周囲の SSID と BSSID、電波強度を取得できました。 win32wifi のソースコードを確認するとより多くの情報を取得できると思います。 {'SampleSSID1': [('XX:XX:XX:XX:XX:XX', 100)], 'SampleSSID2': [('YY:YY:YY:YY:YY:YY',80)], ..... } 参考資料 「netsh wlan show networks mode=bssid」コマンドで表示される同一SSIDのBSSIDの情報が省略される。 Syntax error when trying to import comtypes on Python 3.6.7 win32wifi (Python Windows Wifi)

フランスの大学院で先生の仕事を行いながら、生徒のためにいつも新たな教え方を探しています。僕はプログラミングのレッソンを差し上げるので、そのお返しに生徒は応用問題をしなきゃいけません。 会社に仕事でも、学校の教育でも、先輩からもらったの感謝は第一な事って僕が絶対に思います。今回は先生として、その先輩になるだからこそ、感謝の正しい伝え方を探しに行きました。生徒の努力がちゃんと認めたら、自然に頑張ります。 生徒は得点を持ったなければならなくても、その先、応用問題がよくできたかどうかは伝えたい。でも結局、それは十分じゃないだと思いました。最初は、応用問題に関して生徒の努力をはっきり感謝したい。そうしたら頑張らなかった生徒は悔しいくて盛り上がって、よく頑張った生徒はもっと頑張ります。 得点のかわりは、何？ 勇気付けられる物を考えたら「音ゲー」を思いだした、すなわち、リズムゲームです。一曲を遊んだ後で、リザルト画面は文字スコアで性能を見せます。 あまりできなくたら「B」スコアを当たって、普通にクリアしたら「A」スコアがあって、もっと上手なできたら「S」スコアがあって、それに完璧だったら「SSS」スコアもあります。これで性能はわかりやすいし、なら次回はもっと頑張りたいの感じがします。 これを思い出したとたん、文字スコアの利用をすぐに決めました。 「チューニズム」アケードゲームのリザルト画面（©SEGA)️ 上のイメージ通り、文字スコアは一番大きいテキストですね。その他は、細かいことについてのデータが見せます。そのリザルト画面は基本的に明るくて、かっこくて、そして前回からの伸びがわかります。 文字スコアだけじゃなく、応用問題についての細かい説明がいいでしょう。例えばソースコードに関して評価が差し上げますね。応用問題の目的がちゃんとできたかどうか、構文規則が悪くかどうか、バグが多いかどうか。 「maimai」アケードゲームのリザルト画面（©SEGA)️ その上maimaiの画面は、プレイの時間と場所が見えます。じゃあ、応用問題の名前と日付も書きましょうか。 リザルト画面のやり方 最初はどこかで生徒の評価を書きなければなりません。簡単な記入のために、エクセル表計算のほうがいいだっと思った。その表計算は後でPythonで読みやすいし、そして余白イメージにデータを印刷するつもりです。 テンプレートのイメージ さって、その印刷されられるイメージを作りましょう。いつも通り僕はAffinity Designerを使いますが、別のソフトでも大丈夫です。 後でPythonはイメージにテキストを印刷するので、空欄は必要でしょう。なんかリズムゲームのように感じがしたいから、カラフルと読みやすいのデザインでやりました。このファイルはtemplate.pngというにしましょう。 例のために日本語に変えちゃった テキストを印刷ために、空欄の座標を書いておきましょう。イメージ本体のサイズは2048px*1024pxで、応用問題の名前の空欄は1296*196から始まるし、日付の空欄は1298*902から始まるし、などなど。 そして、全部の可能の文字スコアは別のイメージでおいて置きましょう。 文字スコアのイメージ 表計算の用意 評価はある表計算にまとめるつもりので、準備しましょう。上側には応用問題の名前、日付、関してレッソンの名前を書いておきます。次は、一行一人で生徒の評価が書いておきます。 例の表計算 Pythonを使うときのため、データはどこに置いたを覚えましょう。応用問題の名前は「B1」セル、日付は「D1」セル、レッソンは「F1」セルで置いておきました。そして二番の列のあとは、生徒の評価。一番目の生徒のランクは「C3」セル、構文規則ランクは「D3」セル、などなど。 このファイルはnotes.xslxというにしましょう。 Pythonで読むと印刷 テンプレートイメージと表計算が終わったと、Pythonで生徒の結果のイメージが印刷できます。 スクリプト import os import argparse from PIL import Image, ImageDraw, ImageFont from openpyxl import load_workbook #色んなサイズでフォントの準備 font_bold_60 = ImageFont.truetype("MPLUS1p-Bold.ttf", 60) font_bold_52 = ImageFont.truetype("MPLUS1p-Bold.ttf", 52) font_bold_28 = ImageFont.truetype("MPLUS1p-Bold.ttf", 28) font_medium_32 = ImageFont.truetype("MPLUS1p-Medium.ttf", 28) #argparseのおかげで使いやすいになります parser = argparse.ArgumentParser(description="ある応用問題の結果を印刷差し上げます。") parser.add_argument("spreadsheet", help="表計算ファイルの名前") args = parser.parse_args() #ファイルを開く excel = load_workbook(args.spreadsheet) #一番目の表計算を使う sheet = excel.active #データを読んでおきます exercice_name = sheet["B1"].value exercice_date = sheet["D1"].value exercice_lesson = sheet["F1"].value #3行目から表計算を読んで、10列目まで for row in sheet.iter_rows(min_row=3, max_col=10, values_only=True): #一つ行からデータを引き出す lastname, firstname, rank, syntax_rank, syntax_note, objective_rank, objective_note, quality_rank, quality_note, notes = row #ランクがない場合は多分、結果の打ち込みがまだ if rank is None: print(f"空きランク、スキップ...") continue print(f"--> {lastname} {firstname} ● ランク {rank}") #テンプレートのイメージを開く with Image.open("template.png") as im: draw = ImageDraw.Draw(im) # テキストの印刷 draw.text((92, 80), f"{lastname}\n{firstname}", font=font_bold_52, fill=(0,0,0)) #anchor="mm"の意味は、中央揃えで書く draw.text((1296, 196), exercice_name, font=font_bold_60, anchor="mm", fill=(255,255,255)) draw.text((1298, 902), exercice_date.strftime("%d/%m/%Y"), font=font_bold_28, fill=(0,0,0)) draw.text((1610, 902), exercice_lesson, font=font_bold_28, fill=(0,0,0)) #ここからは複数行テキスト draw.multiline_text((500, 384), syntax_note or "", font=font_medium_32, anchor="lm", fill=(0,0,0)) draw.multiline_text((500, 544), objective_note or "", font=font_medium_32, anchor="lm", fill=(0,0,0)) draw.multiline_text((500, 704), quality_note or "", font=font_medium_32, anchor="lm", fill=(0,0,0)) draw.multiline_text((280, 870), notes or "", font=font_medium_32, anchor="lm", fill=(255,255,255)) #文字ランクのイメージのサイズが違っているので、 #中央揃えのために、正しいx位置を書いておこう ranks_x_position = { "B": 1496, "A": 1466, "S": 1490, "SS": 1384, "SSS": 1278 } #文字ランクのイメージの印刷 with Image.open(f"rank_{rank}.png") as im_rank: im.paste(im_rank, (ranks_x_position[rank], 486), im_rank) #構文規則ランクのイメージの印刷 with Image.open(f"rank_{syntax_rank}.png") as im_syntax_rank: im_syntax_rank.thumbnail((94, 94)) im.paste(im_syntax_rank, (380 if syntax_rank != "A" else 375, 340), im_syntax_rank) #目的ランクのイメージの印刷 with Image.open(f"rank_{objective_rank}.png") as im_objective_rank: im_objective_rank.thumbnail((94, 94)) im.paste(im_objective_rank, (380 if objective_rank != "A" else 375, 500), im_objective_rank) #コード質のイメージの印刷 with Image.open(f"rank_{quality_rank}.png") as im_quality_rank: im_quality_rank.thumbnail((94, 94)) im.paste(im_quality_rank, (380 if quality_rank != "A" else 375, 660), im_quality_rank) #最後はイメージを保存する im.save(f"{firstname} {lastname} - {exercice_name}.png") スクリプトを実行する前に、フォントのファイルをダウンロードしなきゃいけません。たとえばM+1フォントはGoogle Fontsからみつけます。そして、スクリプトはprinter.pyで呼ばれたら、notes.xlsxの表計算を使いたら、ターミナルで実行のはこういうになります。 $ python printer.py notes.xlsx 数秒の後で、一人生徒で一枚のイメージが保存されました。 できたイメージはこういうです。 生徒たちに影響 多分、一番知りたいことでしょうか。基本的に、生徒が自分の能力をちゃんとわかるように、そして後でどうしたらいい、役に立っただと思います。 それなのに、次の応用問題をもらったときに、たまに同じミスや悪いやり方を見つかりました。それはねえ、原因はいろいろだと思う。おそらく僕の説明が明らかなかったとか、締め切りの先に生徒が忙しい過ぎたとか、だからこそ責任は生徒だけじゃなく、学校や先生も責任があると思います。 とりあえず、結果の新しい伝え方法がみつけてうれしいです。せめて生徒が応用問題の結果にもっと気になって、次の応用問題に関して役に立って、よかったです。 読んでありがたいです。そのトピックで話したいなら、ツイッターは@komanakun、Wantedlyはこちらです。

はじめに Pythonでデータ分析やスクレイピングや画像認識などをやっているのですが、Webアプリのバックエンドとして仕事で本格的に利用したいと思い、いろいろ調べてみると、AWSの LambdaでFastAPIを動かすとの話題が多く目に留まり、ふむふむ、と進めていくと、SQLModelも一緒に利用したいと思い、半日ほど試行錯誤してみたのでまとめてみたいと思います。 ソースはgithubに登録しております。 SQLModelの概要 SQLModelは、直感的で使いやすく、互換性が高く、堅牢になるように設計されており、Pythonの型アノテーションに基づき、PydanticとSQLAlchemyを利用しています。 SQLModelの公式ページ 利用例 公式サイトのサンプルコードに出てくるHeroクラスとTeamクラスを対象とする利用例となります。 ファイル構成は以下のとおりです。 プロジェクトルート ├─sample │ │─common_const.py │ │─common_function.py │ │─hero.py │ │─operation_hero.py │ │─operation_hero_team.py │ │─team.py ├─tests │ │─__init__.py │ │─common_function_test.py │ │─operation_hero_team_test.py │ │─operation_hero_test.py ファイル名 説明 common_const.py 共通定義的な情報を格納 common_function.py 共通処理的を含むモジュール hero.py Heroテーブルに対応するHeroクラスのモジュール operation_hero.py Heroテーブルの処理を含むモジュール operation_hero_team.py HeroテーブルとTeamテーブル処理を含むモジュール team.py Teamテーブルに対応するTeamクラスを含むモジュール common_function.py common_function.pyのテストモジュール operation_hero_team_test.py operation_hero.pyのテストモジュール operation_hero_test.py operation_hero_team.pyのテストモジュール エンティティクラス hero.py from typing import Optional from sqlmodel import Field, SQLModel class Hero(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) name: str secret_name: str age: Optional[int] = None team_id: Optional[int] = Field(default=None, foreign_key="team.id") idはORマッパーでよくある主キーで自動採番されるあれです。 team_idはteamテーブルのidをに関連する外部キーです。default=Noneなので必須(Not null制約付きではない)ではありません。 team.py from typing import Optional from sqlmodel import Field, SQLModel class Team(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) name: str = Field(index=True) headquarters: str common_const.py ヒーロー名など、各種処理で共通に利用する文字列を定義しています。 common_const.py import sys class _heroConst: class ConstError(TypeError): pass def __setattr__(self, name, value): if name in self.__dict__: raise self.ConstError("Can't rebind const (%s)" % name) self.__dict__[name] = value sys.modules[__name__] = _heroConst() _heroConst.HERO_NAME_DEADPOND = "Deadpond" _heroConst.HERO_NAME_SPIDER_BOY = "Spider-Boy" _heroConst.HERO_NAME_RUSTY_MAN = "Rusty-Man" _heroConst.HERO_SECRET_NAME_DEADPOND = "Dive Wilson" _heroConst.HERO_SECRET_NAME_SPIDER_BOY = "Pedro Parqueador" _heroConst.HERO_SECRET_NAME_RUSTY_MAN = "Tommy Sharp" _heroConst.TEAM_NAME_PREVENTERS = "Preventers" _heroConst.TEAM_NAME_Z_FORCE = "Z-Force" common_function.py common_function.pyでは、delete_all:heroとteamの全削除、初期データの登録処理、セッションの作成処理を実装しています。 全体の内容は以下のとおりです。 common_const.py from typing import List from sample.hero import Hero from sample.team import Team import sample.common_const as HeroConst from sqlmodel import SQLModel, Session, create_engine, delete from logging import getLogger, StreamHandler, DEBUG import traceback import sqlalchemy from sqlalchemy.future import Engine from sqlalchemy.orm import sessionmaker logger = getLogger(__name__) handler = StreamHandler() handler.setLevel(DEBUG) logger.setLevel(DEBUG) logger.addHandler(handler) logger.propagate = False def delete_all() -> bool: engine = get_engine() with create_session(engine) as session: try: statement_hero = delete(Hero) delete_hero_result = session.exec(statement_hero) statement_team = delete(Team) delete_team_result = session.exec(statement_team) session.commit() logger.debug(f"delete_hero_result.rowcount={delete_hero_result.rowcount}") logger.debug(f"delete_team_result.rowcount={delete_team_result.rowcount}") except (sqlalchemy.exc.OperationalError, BaseException) as error: # テーブルが存在しないときにsqlalchemy.exc.OperationalErrorが発生するが問題問題なし if type(error) == sqlalchemy.exc.OperationalError: logger.error("Exception occurred") return True trace = traceback.format_exception_only(type(error), error) logger.debug(trace) return False return True def init_data() -> List[Hero]: logger.debug("init_data start") engine = get_engine() SQLModel.metadata.create_all(engine) teams = generate_initdata_teams() heroes = generate_initdata_heroes() create_teams(teams, engine) # PreventersのidをDeadpondとSpider-Boyのteam_idセット heroes[0].team_id = teams[0].id heroes[1].team_id = teams[0].id # Z-ForceのidをRusty-Manのteam_idセット heroes[2].team_id = teams[1].id create_heroes(heroes, engine) logger.debug("init_data end") return heroes def generate_initdata_teams() -> List[Team]: team_preventers = Team( name=HeroConst.TEAM_NAME_PREVENTERS, headquarters="Sharp Tower" ) team_z_force = Team( name=HeroConst.TEAM_NAME_Z_FORCE, headquarters="Sister Margaret’s Bar" ) return [team_preventers, team_z_force] def generate_initdata_heroes() -> List[Hero]: hero_1 = Hero( name=HeroConst.HERO_NAME_DEADPOND, secret_name=HeroConst.HERO_SECRET_NAME_DEADPOND, age=30, ) hero_2 = Hero( name=HeroConst.HERO_NAME_SPIDER_BOY, secret_name=HeroConst.HERO_SECRET_NAME_SPIDER_BOY, ) hero_3 = Hero( name=HeroConst.HERO_NAME_RUSTY_MAN, secret_name=HeroConst.HERO_SECRET_NAME_RUSTY_MAN, age=48, ) return [hero_1, hero_2, hero_3] def create_teams(teams: List[Team], engine: Engine): logger.debug(f"create_teams start teams size={len(teams)}") with create_session(engine) as session: session.add_all(teams) session.commit() for team in teams: # refreshでなくても参照するだけで反映される。アクセスすると遅延ロードしてくれる session.refresh(team) logger.debug(f"team.id={team.id}") logger.debug("create_teams end") def create_heroes(heros: List[Hero], engine: Engine): logger.debug(f"create_heroes start heros size={len(heros)}") with create_session(engine) as session: session.add_all(heros) session.commit() logger.debug("create_heroes end") def get_engine() -> Engine: return create_engine("sqlite:///database.db", echo=True) def create_session(engine: Engine) -> Engine: return Session(engine) """ session_factory = sessionmaker( bind=engine, expire_on_commit=False, autocommit=False ) return session_factory() """ セッションの作成処理 sqlite利用する部分はお決まりの処理ですので、説明は省略させていただきます。 セッションはsqlmodelのSessionのコンストラクタにEngineを渡せばOKです。 common_function.py def get_engine() -> Engine: return create_engine("sqlite:///database.db", echo=True) def create_session(engine: Engine) -> Engine: return Session(engine) """ session_factory = sessionmaker( bind=engine, expire_on_commit=False, autocommit=False ) return session_factory() """ コメント化しているのですが、sqlmodelはデフォルトだとcommitするとセッション内のインスタンスが全て期限切れになるので、関連付けられていたセッションから遅延ロードしようとして、sqlalchemy.orm.exc.DetachedInstanceErrorが発生して読めなくなります。 これを回避するために、sqlalchemyのsessionmakerでexpire_on_commit=Falseを指定してセッションを作成し、commitしてもインスタンスが利用できるようにする必要があるのですが、後程登場するSessionクラスの検索処理時のexecメソッドが呼び出せなくなりますので、create_sessionのテストでインスタンスの中身の検証を行うときのみコメント部分の方を有効にする必要があります。まあ、実際のプロダクションコードではセッションを使いまわすので、問題になる事はないです。 更新系のexecメソッドだと実行可能なのは謎ですが・・・ heroとteamの全削除 HeroテーブルとTeamテーブルの全データを削除しています。 common_function.py def delete_all() -> bool: engine = get_engine() with create_session(engine) as session: try: statement_hero = delete(Hero) delete_hero_result = session.exec(statement_hero) statement_team = delete(Team) delete_team_result = session.exec(statement_team) session.commit() logger.debug(f"delete_hero_result.rowcount={delete_hero_result.rowcount}") logger.debug(f"delete_team_result.rowcount={delete_team_result.rowcount}") except (sqlalchemy.exc.OperationalError, BaseException) as error: # テーブルが存在しないときにsqlalchemy.exc.OperationalErrorが発生するが問題問題なし if type(error) == sqlalchemy.exc.OperationalError: logger.error("Exception occurred") return True trace = traceback.format_exception_only(type(error), error) logger.debug(trace) return False return True delete(Hero)とエンティティを指定して得られた処理をsession.execに渡せばHeroテーブルの全データが削除されます。 statement_hero = delete(Hero).where(Hero.name == "HOGE") のように条件を指定しての削除も可能です。whereで条件を指定する方法は、検索処理のところで説明させていただきます。 初期データの登録処理 common_function.py def init_data() -> List[Hero]: logger.debug("init_data start") engine = get_engine() SQLModel.metadata.create_all(engine) teams = generate_initdata_teams() heroes = generate_initdata_heroes() create_teams(teams, engine) # PreventersのidをDeadpondとSpider-Boyのteam_idセット heroes[0].team_id = teams[0].id heroes[1].team_id = teams[0].id # Z-ForceのidをRusty-Manのteam_idセット heroes[2].team_id = teams[1].id create_heroes(heroes, engine) logger.debug("init_data end") return heroes SQLModel.metadata.create_all(engine)を実行すれば、ロードしているエンティティに対応するテーブルが存在しない場合は、DBにテーブルを作成してくれます。 generate_initdata_teamsとgenerate_initdata_heroesはテーブルに登録するエンティティのインスタンスを含むリストを返却する処理となります。 create_teamsとcreate_heroesは上述のエンティティのリストをDBに登録する処理となります。 create_teams実行後にteamsの各要素のidに値がセットされますので、 heroes[0].team_id = teams[0].idのように、Heroのteam_idにTeamのidをセットしています。 def generate_initdata_teams() -> List[Team]: team_preventers = Team( name=HeroConst.TEAM_NAME_PREVENTERS, headquarters="Sharp Tower" ) team_z_force = Team( name=HeroConst.TEAM_NAME_Z_FORCE, headquarters="Sister Margaret’s Bar" ) return [team_preventers, team_z_force] def generate_initdata_heroes() -> List[Hero]: hero_1 = Hero( name=HeroConst.HERO_NAME_DEADPOND, secret_name=HeroConst.HERO_SECRET_NAME_DEADPOND, age=30, ) hero_2 = Hero( name=HeroConst.HERO_NAME_SPIDER_BOY, secret_name=HeroConst.HERO_SECRET_NAME_SPIDER_BOY, ) hero_3 = Hero( name=HeroConst.HERO_NAME_RUSTY_MAN, secret_name=HeroConst.HERO_SECRET_NAME_RUSTY_MAN, age=48, ) return [hero_1, hero_2, hero_3] ですので、hero_1とhero_2はteam_preventers(name=Preventers)、hero_3はteam_z_force(name=Z-Force)に属するようになります。 実際のデータ登録処理は以下のとおりです。 def create_teams(teams: List[Team], engine: Engine): logger.debug(f"create_teams start teams size={len(teams)}") with create_session(engine) as session: session.add_all(teams) session.commit() for team in teams: # refreshでなくても参照するだけで反映される。アクセスすると遅延ロードしてくれるので session.refresh(team) logger.debug(f"team.id={team.id}") logger.debug("create_teams end") def create_heroes(heroes: List[Hero], engine: Engine): logger.debug(f"create_heroes start heros size={len(heroes)}") with create_session(engine) as session: session.add_all(heroes) session.commit() logger.debug("create_heroes end") session.add_all(teams)のようにエンティティのインスタンスのリストを指定すると、一括でInsert可能です。 for team in teams: session.add(teams) のように単一インスタンスを指定してInsertすることも可能です。 common_function.pyのテスト init_data_fixtureでデータを削除後に初期データ登録し、delete_allとinit_dataを呼び出すテストとなります。 検証ほぼないですが・・・ common_function_test.py from multiprocessing.dummy.connection import Listener import sys import pytest from logging import getLogger, StreamHandler, DEBUG from sample.common_function import delete_all, init_data import sample.common_const as HeroConst logger = getLogger(__name__) handler = StreamHandler() handler.setLevel(DEBUG) logger.setLevel(DEBUG) logger.addHandler(handler) logger.propagate = False @pytest.fixture def init_data_fixture(): # sqlite:///database.dbとローカルファイル指定なので全データ削除 delete_all() # 前提データの作成 init_data() def test_delete_all(init_data_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") assert delete_all() logger.debug(f"{sys._getframe().f_code.co_name} end") def test_init_data(): logger.debug(f"{sys._getframe().f_code.co_name} start") heroes = init_data() assert len(heroes) == 3 # 関連付けられていたセッションから遅延ロードしようとして、sqlalchemy.orm.exc.DetachedInstanceErrorが発生して読めない # デフォルトだとcommitするとセッション内のインスタンスが全て期限切れになるので # sqlalchemyのsessionmakerでexpire_on_commit=Falseを指定してセッションを作成しないと以下の検証は実行できません。 """ assert heroes[0].name == HeroConst.HERO_NAME_DEADPOND assert heroes[1].name == HeroConst.HERO_NAME_SPIDER_BOY assert heroes[2].name == HeroConst.HERO_NAME_RUSTY_MAN """ logger.debug(f"{sys._getframe().f_code.co_name} end") コメントにも記載しておりますが、init_dataが返却するList[Hero]の値の検証は、sessionmakerで生成したセッションでないと成功しません。lenだけは大丈夫なんですよね・・・ テストを実行したときのログ(一部)は以下のようになります。 --------------------------------------------------- Captured stdout setup ---------------------------------------------------- 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine DELETE FROM team 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine [generated in 0.00010s] () 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine COMMIT 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("hero") 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine [raw sql] () 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("team") 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine [raw sql] () 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine COMMIT 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine INSERT INTO team (name, headquarters) VALUES (?, ?) 2022-01-16 17:40:13,595 INFO sqlalchemy.engine.Engine [generated in 0.00012s] ('Preventers', 'Sharp Tower') 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine INSERT INTO team (name, headquarters) VALUES (?, ?) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine [cached since 0.001709s ago] ('Z-Force', 'Sister Margaret’s Bar') 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine COMMIT 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine SELECT team.id, team.name, team.headquarters FROM team WHERE team.id = ? 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine [generated in 0.00013s] (1,) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine SELECT team.id, team.name, team.headquarters FROM team WHERE team.id = ? 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine [cached since 0.0009577s ago] (2,) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine ROLLBACK 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine INSERT INTO hero (name, secret_name, age, team_id) VALUES (?, ?, ?, ?) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine [generated in 0.00012s] ('Deadpond', 'Dive Wilson', 30, 1) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine INSERT INTO hero (name, secret_name, age, team_id) VALUES (?, ?, ?, ?) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine [cached since 0.001575s ago] ('Spider-Boy', 'Pedro Parqueador', None, 1) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine INSERT INTO hero (name, secret_name, age, team_id) VALUES (?, ?, ?, ?) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine [cached since 0.001779s ago] ('Rusty-Man', 'Tommy Sharp', 48, 2) 2022-01-16 17:40:13,611 INFO sqlalchemy.engine.Engine COMMIT --------------------------------------------------- Captured stderr setup ---------------------------------------------------- delete_hero_result.rowcount=0 delete_team_result.rowcount=2 init_data start create_teams start teams size=2 team.id=1 team.id=2 create_teams end create_heroes start heros size=3 create_heroes end init_data end Heroテーブルの検索と更新処理 Heroテーブルの全件検索、name指定での検索、nameとsecret_name指定(AND)での検索、nameとsecret_name指定(OR)での検索、条件のageより大きい年齢のHeroの検索、nameを条件にageを更新 との処理を含みます。 operation_hero_test.py from typing import List from sample.hero import Hero from sample.team import Team from sample.common_function import get_engine, create_session from sqlmodel import ( select, or_, ) from logging import exception, getLogger, StreamHandler, DEBUG logger = getLogger(__name__) handler = StreamHandler() handler.setLevel(DEBUG) logger.setLevel(DEBUG) logger.addHandler(handler) logger.propagate = False def select_all() -> List[Hero]: engine = get_engine() with create_session(engine) as session: heroes = session.exec(select(Hero)).all() return heroes def select_by_name(name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where(Hero.name == name) return session.exec(statement).all() def select_by_name_and_secret_name(name: str, secret_name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = ( select(Hero).where(Hero.name == name).where(Hero.secret_name == secret_name) ) # 単一のwhereでも指定可能 # statement = select(Hero).where(Hero.name == name, Hero.secret_name == secret_name) return session.exec(statement).all() def select_by_name_or_secret_name(name: str, secret_name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where( or_(Hero.name == name, Hero.secret_name == secret_name) ) return session.exec(statement).all() def select_by_age_above(age: int) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where(Hero.age > age) return session.exec(statement).all() def update_age_by_name(name: str, age: int) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where(Hero.name == name) # 条件にあう先頭のレコードを取得 hero = session.exec(statement).one() # heroインスタンスのageを更新してsession.addとsession.commitで更新 hero.age = age session.add(hero) session.commit() Heroテーブルの全件検索 select(Hero)で検索処理に必要なオブジェクトが返却されますので、これをsession.execに渡し、結果に対してallを呼び出します。 def select_all() -> List[Hero]: engine = get_engine() with create_session(engine) as session: heroes = session.exec(select(Hero)).all() return heroes name指定での検索 select(Hero).where(Hero.name == name)でHero.nameの検索条件が指定されたオブジェクトが返却されますので、session.execに渡し、結果に対してallを呼び出します。 def select_by_name(name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where(Hero.name == name) return session.exec(statement).all() nameとsecret_name指定(AND)での検索 whereを複数指定することでANDで条件が指定可能となります。 コメントでも記載済みですが、単一のwhere内で複数条件を指定することで同様の動作が実現可能です。 def select_by_name_and_secret_name(name: str, secret_name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = ( select(Hero).where(Hero.name == name).where(Hero.secret_name == secret_name) ) # 単一のwhereでも指定可能 # statement = select(Hero).where(Hero.name == name, Hero.secret_name == secret_name) return session.exec(statement).all() nameとsecret_name指定(OR)での検索 whereにor_(Hero.name == name, Hero.secret_name == secret_name)を指定することで、Hero.nameとHero.secret_nameのORの条件で検索しています。 def select_by_name_or_secret_name(name: str, secret_name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where( or_(Hero.name == name, Hero.secret_name == secret_name) ) return session.exec(statement).all() 条件のageより大きい年齢のHeroの検索 素直な仕様ですね、説明は不要と思います。 def select_by_age_above(age: int) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where(Hero.age > age) return session.exec(statement).all() nameを条件にageを更新 条件にあうレコードに対応するインスタンスを取得し、値を更新、session.add後にcommitすると更新できます。 def update_age_by_name(name: str, age: int) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero).where(Hero.name == name) # 条件にあう先頭のレコードを取得 hero = session.exec(statement).one() # heroインスタンスのageを更新してsession.addとsession.commitで更新 hero.age = age session.add(hero) session.commit() Heroテーブルの検索と更新処理のテスト sqlalchemy.orm.exc.DetachedInstanceErrorの関係で検証はlenしかしておりません。 operation_hero_test.py from multiprocessing.dummy.connection import Listener import sys import pytest from logging import getLogger, StreamHandler, DEBUG from typing import List, Tuple import sample.common_const as HeroConst from sample.operation_hero import ( select_all, select_by_name, select_by_name_and_secret_name, select_by_name_or_secret_name, select_by_age_above, update_age_by_name, ) from sample.hero import Hero from sample.common_function import delete_all, init_data logger = getLogger(__name__) handler = StreamHandler() handler.setLevel(DEBUG) logger.setLevel(DEBUG) logger.addHandler(handler) logger.propagate = False @pytest.fixture def init_data_fixture(): # sqlite:///database.dbとローカルファイル指定なので全データ削除 delete_all() # 前提データの作成 init_data() @pytest.fixture(params=[(HeroConst.HERO_NAME_DEADPOND, 1), ("Hoge", 0)]) def test_select_by_name_fixture(request) -> Tuple[str, int]: request.getfixturevalue("init_data_fixture") return (request.param[0], request.param[1]) def test_select_by_name(test_select_by_name_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") name, expected_result = test_select_by_name_fixture result = select_by_name(name) output_name(result) assert len(result) == expected_result logger.debug(f"{sys._getframe().f_code.co_name} start") @pytest.fixture( params=[ (HeroConst.HERO_NAME_DEADPOND, HeroConst.HERO_SECRET_NAME_DEADPOND, 1), (HeroConst.HERO_NAME_DEADPOND, HeroConst.HERO_SECRET_NAME_SPIDER_BOY, 0), (HeroConst.HERO_NAME_SPIDER_BOY, HeroConst.HERO_SECRET_NAME_DEADPOND, 0), ] ) def test_select_by_name_and_secret_name_fixture(request) -> Tuple[str, str, int]: request.getfixturevalue("init_data_fixture") return (request.param[0], request.param[1], request.param[2]) def test_select_by_name_and_secret_name(test_select_by_name_and_secret_name_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") ( name, secret_name, expected_result, ) = test_select_by_name_and_secret_name_fixture result = select_by_name_and_secret_name(name, secret_name) output_name(result) assert len(result) == expected_result logger.debug(f"{sys._getframe().f_code.co_name} start") @pytest.fixture( params=[ (HeroConst.HERO_NAME_DEADPOND, HeroConst.HERO_SECRET_NAME_DEADPOND, 1), (HeroConst.HERO_NAME_DEADPOND, HeroConst.HERO_SECRET_NAME_SPIDER_BOY, 2), (HeroConst.HERO_NAME_SPIDER_BOY, HeroConst.HERO_SECRET_NAME_DEADPOND, 2), (HeroConst.HERO_NAME_SPIDER_BOY, "Hoge Hoge", 1), ("Hoge", HeroConst.HERO_SECRET_NAME_SPIDER_BOY, 1), ] ) def test_select_by_name_or_secret_name_fixture(request) -> Tuple[str, str, int]: request.getfixturevalue("init_data_fixture") return (request.param[0], request.param[1], request.param[2]) def test_select_by_name_or_secret_name(test_select_by_name_or_secret_name_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") ( name, secret_name, expected_result, ) = test_select_by_name_or_secret_name_fixture result = select_by_name_or_secret_name(name, secret_name) output_name(result) error_message = ( f"name={name} secret_name={secret_name}で検索した結果の要素数の期待値は{expected_result}です。" ) assert len(result) == expected_result, error_message logger.debug(f"{sys._getframe().f_code.co_name} start") @pytest.fixture(params=[(29, 2), (30, 1), (47, 1), (48, 0)]) def test_select_by_age_above_fixture(request) -> Tuple[int, int]: request.getfixturevalue("init_data_fixture") return (request.param[0], request.param[1]) def test_select_by_age_above(test_select_by_age_above_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") ( age, expected_result, ) = test_select_by_age_above_fixture result = select_by_age_above(age) output_name(result) assert len(result) == expected_result logger.debug(f"{sys._getframe().f_code.co_name} start") def test_select_all(init_data_fixture): logger.debug("test_select_all start") result = select_all() output_name(result) assert len(result) == 3 logger.debug("test_select_all end") def test_update_age_by_name(init_data_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") target_name = "Rusty-Man" update_age_by_name(target_name, 50) heroes = select_by_name(target_name) assert len(heroes) == 1 assert heroes[0].age == 50 logger.debug(f"{sys._getframe().f_code.co_name} end") def output_name(target_list: List[Hero]): for index, hero in enumerate(target_list): logger.debug(f"index={index} hero.name={hero.name}") HeroテーブルとTeamテーブルをjoinする処理 Team.nameを条件にHeroを検索する処理を実装しています。 operation_hero_team.py from typing import List from sample.hero import Hero from sample.team import Team from sample.common_function import get_engine, create_session from sqlmodel import Field, Session, SQLModel, create_engine, select, delete, or_ from logging import getLogger, StreamHandler, DEBUG from sqlalchemy.orm.session import sessionmaker logger = getLogger(__name__) handler = StreamHandler() handler.setLevel(DEBUG) logger.setLevel(DEBUG) logger.addHandler(handler) logger.propagate = False def select_heroes_by_team_name(team_name: str) -> List[Hero]: engine = get_engine() with create_session(engine) as session: statement = select(Hero, Team).where( Hero.team_id == Team.id, Team.name == team_name ) return session.exec(statement).all() select(Hero, Team)で対象テーブルがHeroとTeamであることを指定、whereでHero.team_id == Team.idと指定された`team_nameでTeam.nameを絞り込むとの処理となります。 テスト実行時のログ(抜粋)は以下のようになります。 INFO sqlalchemy.engine.Engine:log.py:117 BEGIN (implicit) INFO sqlalchemy.engine.Engine:log.py:117 SELECT hero.id, hero.name, hero.secret_name, hero.age, hero.team_id, team.id AS id_1, team.name AS name_1, team.headquarters FROM hero, team WHERE hero.team_id = team.id AND team.name = ? INFO sqlalchemy.engine.Engine:log.py:117 [no key 0.00015s] ('Preventers',) INFO sqlalchemy.engine.Engine:log.py:117 ROLLBACK join句は利用せず、fromに複数テーブルを指定し、where句でHeroテーブルとTeamテーブルの結合条件とTeamのnameの条件指定を行っているSQLになっていることが分かります。 HeroテーブルとTeamテーブルをjoinする処理のテスト operation_hero_team_test.py from multiprocessing.dummy.connection import Listener import sys import pytest from logging import getLogger, StreamHandler, DEBUG import sample.common_const as HeroConst from sample.operation_hero_team import ( select_heroes_by_team_name, ) from sample.hero import Hero from sample.common_function import delete_all, init_data logger = getLogger(__name__) handler = StreamHandler() handler.setLevel(DEBUG) logger.setLevel(DEBUG) logger.addHandler(handler) logger.propagate = False @pytest.fixture def init_data_fixture(): delete_all() init_data() def test_select_heroes_by_team_name(init_data_fixture): logger.debug(f"{sys._getframe().f_code.co_name} start") result = select_heroes_by_team_name(HeroConst.TEAM_NAME_PREVENTERS) assert len(result) == 2 result = select_heroes_by_team_name(HeroConst.TEAM_NAME_Z_FORCE) assert len(result) == 1 logger.debug(f"{sys._getframe().f_code.co_name} end") まとめ ざっくりと動作確認をした感想ですが、良い意味で、既視感の強い作りですので、多くの方がストレスなく利用できるプロダクトだと感じます。 今回は触りませんでしたが、async sessionにも対応しており、FastAPIと一緒に利用すると作業が捗りそうですね。 懸念材料ですが、バージョンが0.0.6であり、まだまだ足りない部分も多いですし、「Breaking Changes」が行われる可能性が高いと感じます。これからのプロダクトですので、より良い物になっていくことを期待してSQLModelを選択する方も多いと思いますので、今後も注目していきたいです。 ソースはgithubに登録しております。

はじめ Lambdaで日付を20220116の形にし、5日後の日付20220121を引くと、5という数字を出すことが目的として、Lambdaを作成したため、まとめます。 Lambdaコードの流れ 当日の日付 2022-01-16 11:41:21.282791（タイムゾーンはUTC） ↓UTCをJSTに変換する 2022-01-16 11:41:21.282791（タイムゾーンはUTC） ↓タイムスタンプに変更 1642765665851 ↓年、月、日のみに変換。（時間、分、秒は、切り捨て） 20210116 5日後の日付 2022-01-16 11:41:21.282791（タイムゾーンはUTC） ↓日付を5日進める 2022-01-21 11:41:21.282791（タイムゾーンはUTC） ↓UTCをJSTに変換する 2022-01-21 11:41:21.282791（タイムゾーンはUTC） ↓タイムスタンプに変更 1642765665851 ↓年、月、日のみに変換。（時間、分、秒は、切り捨て） 20210121 Lambdaの設定 Node.jsとPythonそれぞれ作成します。 Node.js Node.js exports.handler = async (event) => { // UTCの日付 "2022-01-16T11:31:09.498Z" const dateUTC = new Date(); const fivedateUTC = new Date().setDate(dateUTC.getDate() + 5); // JSTに変更後、タイムスタンプに変える const dateJST = new Date(dateUTC.getTime() + 32400000); // 5日進める const FivedateJST = new Date(fivedateUTC + 32400000); // タイムスタンプを日付の年月日に変える。時間は切り捨て let year = dateJST.getFullYear(); let month = dateJST.getMonth()+1; let day = dateJST.getDate(); let FiveLateryear = FivedateJST.getFullYear(); let FiveLatermonth = FivedateJST.getMonth()+1; let FiveLaterday = FivedateJST.getDate(); // 値が1桁であれば '0'を追加 if (month < 10) month = '0' + month; if (day < 10) day = '0' + day; if (FiveLatermonth < 10) FiveLatermonth = '0' + FiveLatermonth; if (FiveLaterday < 10) FiveLaterday = '0' + FiveLaterday; //"20220116" const dateTimeString = year + month + day; //"20220121" const FiveDaysLaterTimeString = FiveLateryear + FiveLatermonth + FiveLaterday; //20220116 const today = Number(dateTimeString); //20220121 const FiveDaysLater = Number(FiveDaysLaterTimeString); return FiveDaysLater - today; }; Python Python from datetime import datetime, timedelta def lambda_handler(event, context): # 2022-01-16 20:54:05.134159 dateJST = datetime.today() + timedelta(hours=9) dateJSTFiveDaysLater = dateJST + timedelta(days=5) #現在の日本の日付 20220116 today = int(dateJST.strftime("%Y%m%d")) # 5日後 20220121 fiveDaysLater = int(dateJSTFiveDaysLater.strftime("%Y%m%d")) return fiveDaysLater - today Pythonの方がすでにLambdaでインストール済みのライブラリが豊富なため、コードが少なくていいですね。 どちらも20220121 - 20220116 = 5という結果になりました。 参考 Node.js Python

はじめ Lambdaで、当日の日付を20220116の形にし、5日後の日付20220121を引くと、5という数字が出せるようにすることを目的として、Lambdaを作成したため、まとめます。 Lambdaコードの流れ 当日の日付 2022-01-16 11:41:21.282791（タイムゾーンはUTC） ↓UTCをJSTに変換する 2022-01-16 11:41:21.282791（タイムゾーンはUTC） ↓タイムスタンプに変更 1642765665851 ↓年、月、日のみに変換。（時間、分、秒は、切り捨て） 20210116 5日後の日付 2022-01-16 11:41:21.282791（タイムゾーンはUTC） ↓日付を5日進める 2022-01-21 11:41:21.282791（タイムゾーンはUTC） ↓UTCをJSTに変換する 2022-01-21 11:41:21.282791（タイムゾーンはUTC） ↓タイムスタンプに変更 1642765665851 ↓年、月、日のみに変換。（時間、分、秒は、切り捨て） 20210121 Lambdaの設定 Node.jsとPythonそれぞれ作成します。 Node.js Node.js exports.handler = async (event) => { // UTCの日付 "2022-01-16T11:31:09.498Z" const dateUTC = new Date(); const fivedateUTC = new Date().setDate(dateUTC.getDate() + 5); // JSTに変更後、タイムスタンプに変える const dateJST = new Date(dateUTC.getTime() + 32400000); // 5日進める const FivedateJST = new Date(fivedateUTC + 32400000); // タイムスタンプを日付の年月日に変える。時間は切り捨て let year = dateJST.getFullYear(); let month = dateJST.getMonth()+1; let day = dateJST.getDate(); let FiveLateryear = FivedateJST.getFullYear(); let FiveLatermonth = FivedateJST.getMonth()+1; let FiveLaterday = FivedateJST.getDate(); // 値が1桁であれば '0'を追加 if (month < 10) month = '0' + month; if (day < 10) day = '0' + day; if (FiveLatermonth < 10) FiveLatermonth = '0' + FiveLatermonth; if (FiveLaterday < 10) FiveLaterday = '0' + FiveLaterday; //"20220116" const dateTimeString = year + month + day; //"20220121" const FiveDaysLaterTimeString = FiveLateryear + FiveLatermonth + FiveLaterday; //20220116 const today = Number(dateTimeString); //20220121 const FiveDaysLater = Number(FiveDaysLaterTimeString); return FiveDaysLater - today; }; Python Python from datetime import datetime, timedelta def lambda_handler(event, context): # 2022-01-16 20:54:05.134159 dateJST = datetime.today() + timedelta(hours=9) dateJSTFiveDaysLater = dateJST + timedelta(days=5) #現在の日本の日付 20220116 today = int(dateJST.strftime("%Y%m%d")) # 5日後 20220121 fiveDaysLater = int(dateJSTFiveDaysLater.strftime("%Y%m%d")) return fiveDaysLater - today Pythonの方がすでにLambdaでインストール済みのライブラリが豊富なため、コードが少なくていいですね。 どちらも20220121 - 20220116 = 5という結果になりました。 参考 Node.js Python

2022年からRiot Games APIから高ランクプレイヤーの試合情報を取得する方法が変わった為、備忘録として残す。 全体の手順 高ランクのプレイヤーのsummoerIdを取得する 各プレイヤーのpuuidを取得する puuidに紐づくmatchIdを取得する matchIdに紐づく試合情報を取得する matchIdに紐づくタイムラインを取得する ※ タイムライン: いつ、どんなアイテムを購入したか？ どのようなオブジェクトを破壊したか？など 2021年からの大きな変更点 日本サーバーの場合、エンドポイントはhttps://jp1.api.riotgames.com/lolが決め打ちだったが、2022年から、3のmatchId取得からhttps://asia.api.riotgames.com/lolに変更。 1. 高ランクのプレイヤーの情報を取得 手順： 取得したJSONの['entries']の中にある人数分のsummoerIdを抽出する 1. チャレンジャー帯 https://jp1.api.riotgames.com/lol/league/v4/challengerleagues/by-queue/RANKED_SOLO_5x5?api_key=[APIKEY] グランドマスター帯 https://jp1.api.riotgames.com/lol/league/v4/grandmasterleagues/by-queue/RANKED_SOLO_5x5?api_key=[APIKEY] マスター帯 https://jp1.api.riotgames.com/lol/league/v4/masterleagues/by-queue/RANKED_SOLO_5x5?api_key=[APIKEY] 2. 各プレイヤーのpuuidを取得 手順： 取得したJSONの中にあるpuuidを抽出する https://jp1.api.riotgames.com/lol/summoner/v4/summoners/[SUMMONER_ID]?api_key=[APIKEY] 3. puuidに紐づくmatchIdを取得する 手順： 取得したJSONからmatchIdのリストを取得する https://asia.api.riotgames.com/lol/match/v5/matches/by-puuid/[PUUID]/ids?queue=420&type=ranked&start=0&count=100&api_key=[APIKEY] 4. matchIdに紐づく試合情報を取得する 手順： 取得したJSONから必要な試合情報を取得する https://asia.api.riotgames.com/lol/match/v5/matches/[MATCH_ID]?api_key=[APIKEY] 5. matchIdに紐づくタイムラインを取得する 手順： 取得したJSONから試合のタイムラインを取得する https://asia.api.riotgames.com/lol/match/v5/matches/[MATCH_ID]?api_key=[APIKEY] 参考サイト Riot Games API https://developer.riotgames.com/apis APIを利用する開発者用ドキュメント https://developer.riotgames.com/docs/lol メモ 各サーバーのエンドポイント APIを利用する開発者用ドキュメントのPlatform Routing Valuesを参照 各サーバーのエンドポイント APIを利用する開発者用ドキュメントのRegional Routing Valuesを参照 queueのパラメータについて 下記を参照 https://static.developer.riotgames.com/docs/lol/queues.json

2022年からRiot Games APIから高ランクプレイヤーの試合情報を取得する方法が変わった為、備忘録として残す。 全体の手順 高ランクのプレイヤーのsummoerIdを取得する 各プレイヤーのpuuidを取得する puuidに紐づくmatchIdを取得する matchIdに紐づく試合情報を取得する matchIdに紐づくタイムラインを取得する ※ タイムライン: いつ、どんなアイテムを購入したか？ どのようなオブジェクトを破壊したか？など 2021年からの大きな変更点 日本サーバーの場合、エンドポイントはhttps://jp1.api.riotgames.com/lolが決め打ちだったが、2022年から、3のmatchId取得からhttps://asia.api.riotgames.com/lolに変更。 1. 高ランクのプレイヤーの情報を取得 手順： 取得したJSONの['entries']の中にある人数分のsummoerIdを抽出する チャレンジャー帯 https://jp1.api.riotgames.com/lol/league/v4/challengerleagues/by-queue/RANKED_SOLO_5x5?api_key=[APIKEY] グランドマスター帯 https://jp1.api.riotgames.com/lol/league/v4/grandmasterleagues/by-queue/RANKED_SOLO_5x5?api_key=[APIKEY] マスター帯 https://jp1.api.riotgames.com/lol/league/v4/masterleagues/by-queue/RANKED_SOLO_5x5?api_key=[APIKEY] 2. 各プレイヤーのpuuidを取得 手順： 取得したJSONの中にあるpuuidを抽出する https://jp1.api.riotgames.com/lol/summoner/v4/summoners/[SUMMONER_ID]?api_key=[APIKEY] 3. puuidに紐づくmatchIdを取得する 手順： 取得したJSONからmatchIdのリストを取得する https://asia.api.riotgames.com/lol/match/v5/matches/by-puuid/[PUUID]/ids?queue=420&type=ranked&start=0&count=100&api_key=[APIKEY] 4. matchIdに紐づく試合情報を取得する 手順： 取得したJSONから必要な試合情報を取得する https://asia.api.riotgames.com/lol/match/v5/matches/[MATCH_ID]?api_key=[APIKEY] 5. matchIdに紐づくタイムラインを取得する 手順： 取得したJSONから試合のタイムラインを取得する https://asia.api.riotgames.com/lol/match/v5/matches/[MATCH_ID]?api_key=[APIKEY] 参考サイト Riot Games API https://developer.riotgames.com/apis APIを利用する開発者用ドキュメント https://developer.riotgames.com/docs/lol メモ 各サーバーのエンドポイント APIを利用する開発者用ドキュメントのPlatform Routing Valuesを参照 各サーバーのエンドポイント APIを利用する開発者用ドキュメントのRegional Routing Valuesを参照 queueのパラメータについて 下記を参照 https://static.developer.riotgames.com/docs/lol/queues.json

常駐アプリを作ってみたい お手軽にタイトルみたいな事ができないかと調べていたときに pystray というライブラリを知ったので参考資料を見ながら作成してみました。 ご指摘やアドバイス等あればコメント頂きたいです。 環境 Windows 10 Python 3.9 追加ライブラリ pystray schedule プログラム 中身を確認しながら自己責任で実行してください。 アイコン (sample.jpeg) は ICOON MONO さんのフリー素材を使用しました。 tray.py from pystray import Icon, MenuItem, Menu from PIL import Image import time import threading import schedule class taskTray: def __init__(self, image): self.status = False ## アイコンの画像 image = Image.open(image) ## 右クリックで表示されるメニュー menu = Menu( MenuItem('Task', self.doTask), MenuItem('Exit', self.stopProgram), ) self.icon = Icon(name='nameTray', title='titleTray', icon=image, menu=menu) def doTask(self): print('実行しました。') def runSchedule(self): ## 5秒毎にタスクを実行する。 schedule.every(5).seconds.do(self.doTask) ## status が True である間実行する。 while self.status: schedule.run_pending() time.sleep(1) def stopProgram(self, icon): self.status = False ## 停止 self.icon.stop() def runProgram(self): self.status = True ## スケジュールの実行 task_thread = threading.Thread(target=self.runSchedule) task_thread.start() ## 実行 self.icon.run() if __name__ == '__main__': system_tray = taskTray(image="sample.jpeg") system_tray.runProgram() 実行結果 プログラムを実行すると5秒毎に print します。右クリックで Task をクリックしても実行します。 Exit をクリックすればプログラムを終了します。 参考資料 github: OtagoPolytechnic/CommandLineSpeechControl/Developers/systray.py

venv 使い方 環境の作成 $ python -m venv [環境名] 環境の有効化 $ .\[環境名]\Scripts\activate 環境の無効化 $ deactivate

本記事はProjectEulerの「100番以下の問題の説明は記載可能」という規定に基づいて回答のヒントが書かれていますので、自分である程度考えてみてから読まれることをお勧めします。 問題 32：パンデジタル数の積 原文 Problem 32: Digit fifth powers 問題の要約：39×186=7254のように1-9のすべてを使った掛け算になる式の積の合計を求めよ（重複は除く） いろいろ試してみると右辺の積は4桁、よって左辺は1桁x4桁か2桁x3桁であることが分かります。よって"123456789"の順列すべてを(1,4,4)と(2,3,4)に分割して探します。重複はセットを用いて除きます。 import itertools ans = set({}) for parr in itertools.permutations("123456789"): pstr = ''.join(parr) for (la,lb) in [(1,4),(2,3)]: a, b, c = int(pstr[:la]),int(pstr[la:la+lb]),int(pstr[la+lb:]) if a * b == c: print(a, b, c) ans.add(c) print(f"Answer = {sum(ans)} ({ans})") （開発環境：Google Colab）

本記事はProjectEulerの「100番以下の問題の説明は記載可能」という規定に基づいて回答のヒントが書かれていますので、自分である程度考えてみてから読まれることをお勧めします。 問題 31：コインの合計 原文 Problem 31: Coin sums 問題の要約：1p,2p,5p,10p,20p,50p,£1(100p),£2(200p)のコインを使って£2(200p)にするのは何通りあるか求めよ 再帰呼び出しでの実装です。使うコインの枚数はリストnumcに格納します。リストを引数にすると関数の中で値が変わってしまうのでcopyを使ってコピーを渡すようにしています。特に高速化はしなくても十分早いので行っていません。 このTotal＝10のコードではデバックしやすいように結果のコインリストを表示していますが、Total=200にしたときは大量に表示されるのでprint文をコメントアウトしたほうがいいですね。 from copy import copy Total = 10 # Total amount coins = [200,100, 50, 20,10, 5, 2, 1] # Coin values numc = [0 for i in coins] # store the number of coin used for each coin def coinsums(cp, numc, remvalue): if cp==len(coins)-1: # for the smallest coin if remvalue%coins[cp]== 0: # if there is no remeining numc[cp] = remvalue//coins[cp] # --use all of the smallest coin print("*",numc) return 1 else: return 0 # --otherwise return 0 ret = 0 for i in range((remvalue//coins[cp])+1): numc[cp] = i if remvalue-coins[cp]*i == 0: # no remaining value ret += 1 print("-",numc) else: # go to smaller coins ret += coinsums(cp+1, copy(numc), remvalue-coins[cp]*i) return ret # input: depth, num of coin, remaining value ans = coinsums(0, numc, Total) print(f"***Answer = {ans}") #* [0, 0, 0, 0, 0, 0, 0, 10] #* [0, 0, 0, 0, 0, 0, 1, 8] #* [0, 0, 0, 0, 0, 0, 2, 6] #* [0, 0, 0, 0, 0, 0, 3, 4] #* [0, 0, 0, 0, 0, 0, 4, 2] #- [0, 0, 0, 0, 0, 0, 5, 0] #* [0, 0, 0, 0, 0, 1, 0, 5] #* [0, 0, 0, 0, 0, 1, 1, 3] #* [0, 0, 0, 0, 0, 1, 2, 1] #- [0, 0, 0, 0, 0, 2, 0, 0] #- [0, 0, 0, 0, 1, 0, 0, 0] ***Answer = 11 （開発環境：Google Colab）

ABC235のA,B,C,D,E問題を、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や拡散していただけると喜びます！ 目次 ABC235 まとめ A問題『Rotate』 B問題『Climbing Takahashi』 C問題『The Kth Time Query』 D問題『Multiply and Rotate』 E問題『MST + 1』 アプリ AtCoderFacts を開発しています コンテストの統計データを見られるアプリ『AtCoderFacts』を作りました。 現在のところ、次の3つのデータを見ることができます。 レート別問題正解率 パフォーマンス目安 早解きで上昇するパフォーマンス 今後も機能を追加していく予定です。使ってくれると喜びます。 ABC235 まとめ 全提出人数: 9061人 パフォーマンス パフォ AC 点数 時間 順位(Rated内) 200 AB------ 300 23分 6170(5913)位 400 AB------ 300 6分 5116(4859)位 600 ABC----- 600 39分 4107(3855)位 800 ABC----- 600 17分 3182(2937)位 1000 ABCD---- 1000 85分 2344(2104)位 1200 ABCD---- 1000 42分 1679(1440)位 1400 ABCDE--- 1500 96分 1156(934)位 1600 ABCDE--- 1500 63分 791(574)位 1800 ABCDE--- 1500 45分 539(337)位 2000 ABCDE--- 1500 29分 361(183)位 2200 ABCDEF-- 2000 89分 225(91)位 2400 ABCDEF-- 2000 57分 134(44)位 色別の正解率 色 人数 A B C D E F G Ex 灰 3072 97.0 % 92.2 % 43.1 % 7.9 % 1.9 % 0.1 % 0.1 % 0.0 % 茶 1322 98.9 % 98.3 % 88.1 % 31.5 % 5.2 % 0.3 % 0.1 % 0.0 % 緑 1016 98.9 % 98.6 % 96.4 % 69.2 % 24.9 % 0.8 % 0.3 % 0.0 % 水 643 97.2 % 96.7 % 96.4 % 89.6 % 68.7 % 5.6 % 1.7 % 0.2 % 青 387 97.9 % 97.4 % 97.7 % 94.1 % 85.5 % 22.5 % 5.9 % 0.3 % 黄 195 89.2 % 89.7 % 89.7 % 86.7 % 86.2 % 49.2 % 21.5 % 4.6 % 橙 43 93.0 % 90.7 % 90.7 % 90.7 % 88.4 % 67.4 % 48.8 % 13.9 % 赤 27 88.9 % 88.9 % 88.9 % 88.9 % 85.2 % 81.5 % 70.4 % 51.9 % ※表示レート、灰に初参加者は含めず A問題『Rotate』 問題ページ：A - Rotate 灰コーダー正解率：97.0 % 茶コーダー正解率：98.9 % 緑コーダー正解率：98.9 % 入力 $abc$ : $0$ を 含まない $3$ 桁の整数 考察 $abc$ を文字列で受け取って、$bca$ と $cab$ を作り、int関数で整数に変換して足せばいいです。 コード a, b, c = input() # このように文字列を1文字ずつバラして受け取れます abc = a + b + c bca = b + c + a cab = c + a + b print(int(abc) + int(bca) + int(cab)) a, b, c を $1$ 文字ずつバラして受け取るのがよくわからなければ、次のコードのように普通に $3$ 文字の文字列として受け取って、インデックスで $bca$ と $cab$ を作ってもいいです。 abc = input() bca = abc[1] + abc[2] + abc[0] cab = abc[2] + abc[0] + abc[1] print(int(abc) + int(bca) + int(cab)) 別解 $a,b,c$ いずれも、$1, 10, 100$ の位に $1$ 度ずつ出現します。つまり、$111a+111b+111c=111(a+b+c)$ でも答えを求めることができます。 a, b, c = map(int, input()) print((a + b + c) * 111) B問題『Climbing Takahashi』 問題ページ：B - Climbing Takahashi 灰コーダー正解率：92.2 % 茶コーダー正解率：98.3 % 緑コーダー正解率：98.6 % 入力 $N$ : 台の数 $H_i$ : $i$ 番目の台の高さ 考察 問題文に書いてあるとおりに、移動できなくなるまでforループでシミュレーションを行えばいいです。 最初は左端の台 $1$ に乗っている（高さ $H_1$ ） 右隣の台が自分が今立っている台より高いとき、右隣の台に移動する。そうでなければ移動をやめる。この操作を繰り返す 今立っている台が右端の台（台 $N$ 高さ $H_N$）の場合も、右にもう台がないので移動をやめる $1$ 番目に、初期値として $ans=H_1$ とします。 $2$ 番目に、$H_2$ 以降の台をひとつずつ見ていきます。右隣の台の高さを $h$ とすると、『右隣の台が今立っている台より高い』は $ans\lt{h}$ という条件で表されます。これを満たすならば、$ans=h$ に変更し、その次の右隣の台と高さを比べるのを繰り返します。 満たさない場合は、その時点での $ans$ が答えですから、それを出力すればいいです。 コード def solve(): N = int(input()) H = list(map(int, input().split())) ans = H[0] # 問題文ではH_1ですが、PythonではH[0]が左端です for h in H[1:]: # H[1]以降と比較します if ans < h: ans = h # 右隣の台のほうが高いので、移動します else: return ans # 右隣の台に移動しないので、現時点のansが答えです return ans # 右端まで移動したとき（forループを抜けたとき）のために、ここでもansを返します print(solve()) C問題『The Kth Time Query』 問題ページ：C - The Kth Time Query 灰コーダー正解率：43.1 % 茶コーダー正解率：88.1 % 緑コーダー正解率：96.4 % 入力 $N$ : 数列の長さ $Q$ : クエリの数 $a_i$ : 数列 $A$ の $i$ 番目の値（$0\le{a_i}\le{10^9}$） $x_i, k_i$ : $i$ 番目のクエリでは、数列 $A$ で、$x_i$ が $k_i$ 回目に出現するのが何番目か答える。条件を満たす要素が存在しなければ、$-1$ を出力する 考察 あらかじめ数 $x$ が、$A$ の何番目の要素に出てくるか記憶しておけば、クエリに $O(1)$ で答えることができます。 $a_i$ の上限が $10^9$ と非常に大きいため、二次元リストで管理することはできません。そこで、連想配列（dict）にリストを入れ子にすることにします。 例えば、数列 $A$ が $\{1,1,2,3,1,2\}$ だとします。 $1$ は前から $1,2,5$ 番目、$2$ は $3,6$ 番目、 $3$ は $4$ 番目に出てきています。これを、以下の形で記録します。 $1 : \{1,2,5\}$ $2 : \{3,6\}$ $3 : \{4\}$ $1$ が $2$ 番目に出るのは $a_2$ 、$2$ が $2$ 番目に出るのは $a_6$、 $2$ 番目に出る $3$ は存在しないというように、クエリに対して高速に答えることができます。 実装 collectionsモジュールのdefaultdictを使います。基本的な動作はdict と同じですが、defaultdict(list) と書くことで、存在しないキーにアクセスした際に、自動的に空のリストで初期化してくれます。 コード from collections import defaultdict N, Q = map(int, input().split()) A = list(map(int, input().split())) D = defaultdict(list) # D[x] : xが出現するインデックスのリスト for i, x in enumerate(A, 1): # enumerate(A, 1)とすることで、A[0]を1番目として数え始めて、あとで問題文にあわせて1足す手間を省いています D[x].append(i) # D[x] にはじめてアクセスする場合も、勝手に空のリストで初期化してくれるので、appendできます for _ in range(Q): x, k = map(int, input().split()) if k <= len(D[x]): # x が出現する回数は len(D[x])回です print(D[x][k - 1]) # リストはインデックス0から数え始めるので、k番目に出現するインデックスは、D[x][k-1]です else: print(-1) D問題『Multiply and Rotate』 問題ページ：D - Multiply and Rotate 灰コーダー正解率：7.9 % 茶コーダー正解率：31.5 % 緑コーダー正解率：69.2 % 入力 $a$ : 操作 $1$ では今黒板に書いている数 $x$ を $a$ 倍して $a\times{x}$ に書き換える $N$ : 黒板に書いている $1$ から操作を繰り返して、$N$ に変化させるために必要な操作の最小回数を求める 考察 黒板に書かれた $1$ に、$2$ 種類の操作を繰り返して、$N$ に変化させるのにかかる最小の操作回数を求めるのが目標です。 はじめに結論を言うと、この問題は『$1$ から $10^6-1$ までの正の整数を頂点』『$2$ つの操作を 別の頂点へのコスト $1$ の辺』とみなすことで、最短経路問題として扱うことができるので、BFS（幅優先探索）を使うことにより解くことができます。 これ以降では、そこに至るまでの考察過程を書きます。 操作について 操作は以下の $2$ つです。 操作 $1$ : 黒板に書いてある数を $a$ 倍する 操作 $2$ : $x$ が $10$ で割り切れない（下一桁が $0$ ではない）ときに行える。列の末尾の数字を、列の先頭に移動させる 操作1 これは単純です。たとえば、$a=3$ 、今黒板に書かれた数 $x$ が $5$ だとします。この操作を行うと、黒板に書かれた数は $x\times{a}=5\times{3}=15$ に変化します。 操作2 こちらの操作に $a$ は関係ありません。 例えば、$x=12345$ とします。この操作を行うと末尾の $5$ が先頭に移動して、$x=51234$ になります。 なお、操作を行える条件は、以下の $2$ つです。 末尾が $0$ でないこと *$x$ が $10$ 以上であること * 末尾が $0$ の場合、例えば $x=120$ に対して操作を行ったとすると、$x = 012$ になってしまい、自然な十進法の数ではなくなってしまうからです。 $x$ が $10$ 以上でない、つまり $x$ が一桁の場合、操作を行っても元の数から変化せず、操作を行う意味がないからです。 重要なポイント：操作によって桁数が減ることは絶対にない この問題を解く方法を考えます。まず、重要な考察点として 操作 $1$ を行うと、$x$ の桁数は変わらないか増える 操作 $2$ を行っても、$x$ の桁数は変わらない まとめると、どちらの操作でも、$x$ の桁数が減ることはない というポイントがあります。 操作 $1$ については、正整数 $x$ を $a$ 倍（$a\ge2$）すると、元の $x$ よりも大きくなりますから、当然です。 操作 $2$ は、$x$ の末尾の $0$ でない数字を取り除いて先頭に移動すると、$1$ 桁減って $1$ 桁増えて、結局桁数は変わりません。（元の $x$ よりも小さくなる可能性はあります） 10^6以上の整数については考えなくていい ということは、一度 $x$ の桁数 が $N$ の桁数より大きくなると、二度と $N$ に到達することはありません。 $N$ の制約の上限は $N\lt10^6$ ですから、$N$ は最大でも $6$ 桁です。$7$ 桁以上の数字について一切考える必要はありません。$1$ から $10^6 -1$ までの、およそ $10^6$ 個の数字について考えればいいです。 最短経路問題とみなす $1$ からスタートして、$N$ に至るまでの最小操作回数を求めれば良いです。これは、BFS（幅優先探索）で解くことができます。 $1$ から $10^6 -1$ までの整数を頂点とする 各頂点 $x$ には、操作 $1$ 、操作 $2$ に対応する $2$ つのコスト $1$ の一方通行の辺がある $10^6$ 以上の頂点に移動するような辺は使わない 頂点 $1$ から 頂点$N$ の最短距離が答え、 到達不能な場合は $-1$ $10^6$ 以上の数を使わない枝刈りを忘れると、頂点数が文字通り無限に増えてしまうので、TLEになります。 頂点数はおよそ $10^6$ 個、辺の数は $2\times10^6$ 個未満ですから、十分高速です。 コード from collections import deque INF = 1 << 62 MAX_N = 10 ** 6 def solve(): a, N = map(int, input().split()) dist = [INF] * MAX_N dist[1] = 0 que = deque((1,)) while que: x = que.popleft() new_cost = dist[x] + 1 # ここから操作1 : xをa倍する nx1 = x * a if nx1 < MAX_N: # 10 ** 6以上の場合は移動しません if new_cost < dist[nx1]: dist[nx1] = new_cost que.append(nx1) # ここから操作2 : xの末尾が0でないとき、末尾を先頭に移動する if x % 10 != 0: nx2 = int(str(x % 10) + str(x // 10)) # 文字列で結合したあと、整数に戻すのが楽でしょう if new_cost < dist[nx2]: dist[nx2] = new_cost que.append(nx2) if dist[N] != INF: return dist[N] else: return -1 print(solve()) E問題『MST + 1』 問題ページ：E - MST + 1 灰コーダー正解率：1.9 % 茶コーダー正解率：5.2 % 緑コーダー正解率：24.9 % 入力 $N,M$ : $N$ 頂点 $M$ 辺の『重み付き』『無向』『連結』グラフが与えられる $a_i, b_i, c_i$ : $i$ 番目の辺は $a_i$ と $b_i$ を結び、重み（コスト）は $c_i$ である $Q$ : クエリの数 $u_i, v_i, w_i$ : $i$ 番目のクエリでは、元のグラフに $u_i$ と $v_i$ を結ぶ、重み（コスト） $w_i$ の辺 $e_i$ を追加したとき、この辺 $e_i$ が最小全域木に含まれるか判定する 元のグラフが連結であるため、最小全域木は必ず存在する 辺の重み $c_i$ はすべて異なる値である。クエリで追加される辺の重み $w_i$ は、元のグラフに存在する辺の重み $c_i$ とは異なる値である。そのため、最小全域木は一意に定まる 考察 問題文と入力が非常に長いですが、やることは単純です。 クラスカル法 最小全域木を求めるアルゴリズムに、クラスカル法 というものがあります。このアルゴリズムは、全頂点がバラバラで辺のない状態からスタートします。次に、辺のコストで辺をソートして、コストが小さい順に辺を $1$ つずつ見ていきます。この辺が結ぶ $2$ つの頂点が別の連結成分（まだつながっていない）であれば、この辺を使用して、$2$ つの頂点を連結させます。既に同じ連結成分であれば、この辺を使う意味はないので、使いません。 クラスカル法は、辺のコストが小さい順に使用できる辺を使う貪欲法のアルゴリズムです。貪欲に使っていくだけですから、理解さえすれば意外と単純なアルゴリズムです。 すべてのクエリに対して効率的に答える さて、クエリは『ある $1$ つの辺 $e_i$ を追加したときに、その辺が最小全域木に含まれるか』を答えるものです。各クエリは独立であり、他のクエリで問われる辺のことは一切考慮しません。 クエリごとに $Q$ 回、元のグラフに辺を $1$ つ追加したグラフでクラスカル法を行えば正解を出すことはできますが、計算量は $O(QMlogM)$ となり、当然TLEになります。 そこで、一度クエリをまとめて処理することを考えます。 クラスカル法で辺 $e_i$ が使われる条件は、『辺 $e_i$ よりコストが小さい辺についてすべて処理した時点で、辺 $e_i$ が結ぶ $2$ つの頂点が別の連結成分である』 **ことです。連結かどうかの判定さえできればクエリには答えられるので、実際に頂点の連結を行う必要はありません。** そこで、元の辺とクエリの辺を一緒に混ぜてソートをして、クラスカル法を行えば良いです。ただし、クエリの辺はその辺をが使われるかどうかだけを判定して、実際に連結は行いません。 また、元の辺かクエリの辺か区別できるようにする必要がありますが、クエリの番号がコストの順に並んでいるとは限らないので、何番目のクエリなのかもも区別できるようにして、最後にまとめて クエリ $1$ から順番に答えを出力する必要があります。 まとめると 元のグラフの辺と、クエリの辺を一緒にして辺のコストでソートしてしまう このとき、元のグラフの辺かクエリの辺かを区別できるようにする また、クエリである場合は、何番目のクエリであるかも区別できるようにする 普通にプリム法を行うが、クエリの辺である場合は判定だけして、実際に連結はしない 最後に、クエリ $1$ から順番に答えを出力する コード from operator import itemgetter from typing import List class UnionFind: """0-indexed""" def __init__(self, n): self.n = n self.parent = [-1] * n self.__group_count = n def unite(self, x, y): """xとyをマージ""" x = self.root(x) y = self.root(y) if x == y: return False self.__group_count -= 1 if self.parent[x] > self.parent[y]: x, y = y, x self.parent[x] += self.parent[y] self.parent[y] = x return True def is_same(self, x, y): """xとyが同じ連結成分か判定""" return self.root(x) == self.root(y) def root(self, x): """xの根を取得""" if self.parent[x] < 0: return x else: self.parent[x] = self.root(self.parent[x]) return self.parent[x] def size(self, x): """xが属する連結成分のサイズを取得""" return -self.parent[self.root(x)] def all_sizes(self) -> List[int]: """全連結成分のサイズのリストを取得 O(N)""" sizes = [] for i in range(self.n): size = self.parent[i] if size < 0: sizes.append(-size) return sizes def groups(self) -> List[List[int]]: """全連結成分の内容のリストを取得 O(N・α(N))""" groups = dict() for i in range(self.n): p = self.root(i) if not groups.get(p): groups[p] = [] groups[p].append(i) return list(groups.values()) @property def group_count(self) -> int: """連結成分の数を取得 O(1)""" return self.__group_count def main(): N, M, Q = map(int, input().split()) E = [] for _ in range(M): a, b, c = map(int, input().split()) E.append((-1, a, b, c)) # 1要素目を-1として、元のグラフの辺であると区別できるようにする for i in range(Q): u, v, w = map(int, input().split()) E.append((i, u, v, w)) # 1要素目をクエリの番号とする E.sort(key=itemgetter(3)) # そのままタプルをソートすると重いので、辺のコストc, wだけ使ってソートする ans = [False] * Q # n 番目のクエリの答えを記憶する配列 uf = UnionFind(N + 1) for n, a, b, c in E: if n == -1: # 元のグラフの辺の場合 uf.unite(a, b) # 本来最小全域木では既に同じ連結成分の辺は使用しませんが、答えに関係ないのでサボります else: # n >= 0, n 番目のクエリの答えを判定する if not uf.is_same(a, b): # a, bがまだ違う連結成分なら、この辺は使われる ans[n] = True # n 番目のクエリの答えを "Yes" にするが、実際に連結はしない for flag in ans: print("Yes" if flag else "No") if __name__ == '__main__': main()

目的 ・スマホの育成ゲームの結果をデータベースにまとめたい。 ・1つずつ手入力するのは手間である。 ・AWSに画像上の文字を認識して出力するサービスがあるので利用してみる。 （日本語は無理そうだが、アルファベットや数字は問題なさそう。） 実行内容 ①画像の準備（スマホでスクリーンショットで収集） ②画像の加工（OpenCVで大きさや不要な箇所を隠すなど実施） →文字認識結果をまとめやすくするため ③AWSのRekognitionを使ってテキスト検出を行う。 ④結果をデータベースにまとめる 記載内容 今回は③のRekognitionのテキスト検出について記述する。 参考 ①Qiita記事（コードの内容） https://qiita.com/banquet_kuma/items/560787299b83fb924ff7 ②AWS_Rekognition公式入門ガイド https://aws.amazon.com/jp/rekognition/resources/ テキスト検出結果 加工した画像 出力結果 ※一番最初には読み込んだ画像名を出力してます。（テキスト検出の範囲外） 　「NN」という文字はどこから検出したか不明ですが、収集後削除すれば良し。 コード code import boto3 import configparser region = "画像を保存しているS3のRegion" bucket = "画像を保存しているS3のバケット名" #アクセスキーが記述されているiniデータから必要な情報を抽出 ini = configparser.ConfigParser() ini.read("読み込む.iniデータ","utf-8") access_key = ini["AWS_KEY"]["aws_access_key_id"] secret_key = ini["AWS_KEY"]["aws_secret_key"] session = boto3.Session(aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) #S3に保存している画像一覧の取得 s3 = session.client("s3") objects = s3.list_objects_v2(Bucket=bucket) for filename in objects["Contents"]: filename = filename["Key"] idname = filename.split(".")[0].split("/")[1] #idnameは読み込んだデータ名 data = [idname] rekognition = session.client("rekognition") #テキストの検出 response = rekognition.detect_text(Image={"S3Object":{"Bucket":bucket,"Name":filename}}) textDetections = response["TextDetections"] for text in textDetections: data.append(text["DetectedText"]) #収集した結果を出力 print(data) 所感 ・精度がよくて、満足。 ・育成データの手入力が不要って、すごく便利！！ ・知らないうちに世の中がさらに便利になっていく、、、驚きです。 ※本記事には記載していないが、収集結果をデータベースにまとめて、欲しいデータをSELECT文で抽出してグラフで表示することまで行った。 以上。

ハロー、Qiita！いかがお過ごしでしょうか。 はい、というわけで今週もKivyMDのお時間となりました。お正月気分は残って いないでしょうか。投稿者はまんまと残っている限りであります。いやー、もう 1正月を迎えたい(切実)。 さて、新年早々のニュースとしては、とある企業が週休3日を考えているという なんとも羨ましいニュースがありましたが、なんと言ってもこのニュースが話題を 呼んでいるのではないでしょうか。 多くの方が恩恵を受けている中、ハッと目が覚めるようなニュースでしたね。当然 私も受けていますので、改めて寄付をしようかなとも思った次第ではあります。 だって、KivyMDがなければこの投稿もなかったもの。 まぁ、一旦個人の信条などは置いておいて、投稿することによって貢献をしていき たいと思います(解決には至ってない)。今週はというと、先週に引き続き初期公開 したアプリについて解説をしていきたいと思っております。それではレッツラゴ。 動き(繰り返し) コードだけ見ても、動きを見ておかないとなんだか分からんとなると思うので、 前回を一部だけ振り返っておきます。 あ、この文は先週と同じものです。はい。。 どんなアプリなのよ、と思われる方は上のキャプチャを見てもらえればなんとなく 雰囲気は伝わるかと思われます。(これも同じ) 説明 ここからが先週と少し変わっていて、まずはアプリのリポジトリを見てもらえれば と思います。 大きく分けて、アプリとしてはmain.(py|kv)の2種類で構成されています。先週 においては、主にKV側を説明していましたが今週はpythonコードについて説明を していきます。 main.py さっそくではありますが、pythonコードの全文を載せておきます。ちなみにですが、 コードを公開した日から変更はしておりません。 main.py # Kivy from kivy.lang import Builder from kivy.properties import StringProperty # KivyMD from kivymd.app import MDApp from kivymd.uix.boxlayout import MDBoxLayout from kivymd.uix.tab import MDTabsBase from kivymd.uix.list import ThreeLineAvatarIconListItem from kivymd.uix.button import MDFlatButton from kivymd.uix.dialog import MDDialog from kivymd.uix.snackbar import Snackbar # Standard import datetime class Tab(MDBoxLayout, MDTabsBase): text = StringProperty() class Task(ThreeLineAvatarIconListItem): text = StringProperty() secondary_text = StringProperty() tertiary_text = StringProperty() class DoneTask(ThreeLineAvatarIconListItem): text = StringProperty() secondary_text = StringProperty() tertiary_text = StringProperty() # not use class Content(MDBoxLayout): taskname = StringProperty() description = StringProperty() def get_taskname(self): return taskname def get_description(self): return description class Main(MDApp): dialog = None task_dialog = None def __init__(self, **kwargs): super(Main, self).__init__(**kwargs) def build(self): self.root.ids.taskname.bind( on_text_validate=self.set_error_message, on_focus=self.set_error_message, ) def set_error_message(self, instance_textfield): self.root.ids.taskname.error = True def on_tab_switch( self, instance_tabs, instance_tab, instance_tab_label, tab_text ): pass def send(self): taskname = self.root.ids.taskname.text description = self.root.ids.description.text if(len(taskname) == 0): Snackbar(text="Please input taskname!").open() return now = datetime.datetime.now() nowtime = now.strftime("%Y/%m/%d %H:%M:%S") self.root.ids.todo.add_widget( Task( text=taskname, secondary_text=description, tertiary_text=nowtime ) ) Snackbar(text="Task Created!").open() self.root.ids.taskname.text = "" self.root.ids.description.text = "" def remove_widget(self, instance): if("todo" == instance.parent.text): self.root.ids.todo.remove_widget(instance) elif("doing" == instance.parent.text): self.root.ids.doing.remove_widget(instance) else: self.root.ids.done.remove_widget(instance) Snackbar(text="Task Removed!").open() def change_task_status(self, instance): if not self.task_dialog: self.task_dialog = MDDialog( title="Move to Status?", buttons=[ MDFlatButton( text="NO", theme_text_color="Custom", text_color=self.theme_cls.primary_color, ), MDFlatButton( text="YES", theme_text_color="Custom", text_color=self.theme_cls.primary_color, on_release=lambda x: self.move_to_next( x, instance, instance.text, instance.secondary_text, instance.tertiary_text ) ) ], ) self.task_dialog.open() def move_to_next(self, button_instance, task_instance, text, secondary_text, tertiary_text): if("todo" == task_instance.parent.text): self.root.ids.doing.add_widget( Task( text=text, secondary_text=secondary_text, tertiary_text=tertiary_text ) ) else: self.root.ids.done.add_widget( DoneTask( text=text, secondary_text=secondary_text, tertiary_text=tertiary_text ) ) self.remove_widget(task_instance) self.task_dialog.dismiss() self.task_dialog = None Snackbar(text="Task Moved!").open() Main().run() import パッケージなりをインポートしている部分を説明していきます。該当部分を再掲 します。 # Kivy from kivy.lang import Builder from kivy.properties import StringProperty # KivyMD from kivymd.app import MDApp from kivymd.uix.boxlayout import MDBoxLayout from kivymd.uix.tab import MDTabsBase from kivymd.uix.list import ThreeLineAvatarIconListItem from kivymd.uix.button import MDFlatButton from kivymd.uix.dialog import MDDialog from kivymd.uix.snackbar import Snackbar # Standard import datetime 大きく分けて、KivyとKivymd、Standardという3つで構成されています。 Kivyなどは字の通りですが、Standardはこれも名の通り標準パッケージと しています。ちなみにですが、datetimeはタスク生成時の時刻を埋め込む ために使っています。 カスタムレイアウトたち これも該当部分を再掲しておきます。 class Tab(MDBoxLayout, MDTabsBase): text = StringProperty() class Task(ThreeLineAvatarIconListItem): text = StringProperty() secondary_text = StringProperty() tertiary_text = StringProperty() class DoneTask(ThreeLineAvatarIconListItem): text = StringProperty() secondary_text = StringProperty() tertiary_text = StringProperty() (略) これらについてはカスタムで作るウィジェットクラスとなります。Tabに ついてはあやしい部分がありますが、分けて書くのは少々面倒なので一緒 くた(出身がわかりそう)にしています。 おのおの、クラスを継承をしていることがわかるかと思います。あとは、 レイアウトで使うプロパティをここで書いています。タスクが2つ持って いるのは、役割が異なっているので分けるようにしました。すみません、 嘘です。実装してるときに詰んだ詰んだ!となって2つとなりました。 # タスクの種類については今後増えていく見込みです また、Contentクラスについては、気になさらずでお願いしたいところ です。これは先週伝えていたことですが、本当はこうやりたかったという 点とまさに繋がっているところになります。完全ダミークラスとなっていて 取り除くのを忘れていました。 Mainクラス さて、今日のメインディッシュとなるところです。 こちらもおのおのメソッド単位のコードの方を再掲しておきます。 初期状態 はい、いきなり矛盾が出てきて申し訳ないですが、initメソッドやbuild メソッドなどオブジェクトを作成するときにお馴染みの定義がを以下のように 抜粋しています。 class Main(MDApp): dialog = None task_dialog = None def __init__(self, **kwargs): super(Main, self).__init__(**kwargs) def build(self): self.root.ids.taskname.bind( on_text_validate=self.set_error_message, on_focus=self.set_error_message, ) def set_error_message(self, instance_textfield): self.root.ids.taskname.error = True def on_tab_switch( self, instance_tabs, instance_tab, instance_tab_label, tab_text ): pass お馴染みと言っているにも関わらず、set_error_messageやon_tab_ switchメソッドなんかはお馴染みではないだろ!と思われる方はその通り でらっしゃいます。 まず、buildメソッドとset_error_messageメソッドなんかはcreate タブのtasknameプロパティと結びついています。これは単にバリデーション をしているだけで、空値のままタスクの生成をしないようにしています。この あたりの詳しい触れ込みについては以下もしくは該当マニュアルが詳しいです。 続いて、on_tab_switchメソッドですが、これもTab篇ではお馴染みなのですが、 タブをスイッチするときに欠かせないものです。最初はタブスイッチするときに特に やることないので、省略していたのですが無くすると動かなくなるので定義をしてい ます。こちらも以下もしくは該当マニュアルが詳しいです。 後説明するところというと、このあと使うtask_dialog変数にNoneを入れ込んで いるだけになります。dialogというのもあるけど??ということも当てはまっており まして、これも単に消し忘れの部分になりますw sendメソッド def send(self): taskname = self.root.ids.taskname.text description = self.root.ids.description.text if(len(taskname) == 0): Snackbar(text="Please input taskname!").open() return now = datetime.datetime.now() nowtime = now.strftime("%Y/%m/%d %H:%M:%S") self.root.ids.todo.add_widget( Task( text=taskname, secondary_text=description, tertiary_text=nowtime ) ) Snackbar(text="Task Created!").open() self.root.ids.taskname.text = "" self.root.ids.description.text = "" ようやく、当初言っていることに返ってきて、メソッドの説明になります。 これは大まかに何やっているかをいうと、createタブでタスクを入力したあとで SENDボタンを押したときに発動する中身となります。やっていることも大したこと なく、単に受け取った値をTaskクラスに使用してタスク生成して、作ったあとはSn- ackbarインスタンスを生成して作ったよということを通知しています。 最後に空文字を入れているのは、createタブの中身を一旦クリアしているだけになり ます。あと触れていないところでいうと、tasknameはバリデーションしているのですが、 なぜかSENDボタンを押すとこのメソッドは発動されてしまうのでチェック処理を入れて いるのと、importのところで言っていたdatetimeオブジェクトをここで使用しています。 参照としては以下もしくは該当マニュアルが詳しいですね。 remove_widgetメソッド 続いてはremove_widgetメソッドになります。 def remove_widget(self, instance): if("todo" == instance.parent.text): self.root.ids.todo.remove_widget(instance) elif("doing" == instance.parent.text): self.root.ids.doing.remove_widget(instance) else: self.root.ids.done.remove_widget(instance) Snackbar(text="Task Removed!").open() ここについては言うまでもないくらいに大したことはやっていませんが、それぞれ(todo とかdoingとか)のステータスの際に分岐をしてタスクを消去していることをやっています。 タスクウィジェットが1つであれば、こんな処理は必要ないですがここに関しては今後の 課題ですかね。もしかするとやらないかもだけど。。あと、この部分に関してはsendメソ ッドのところでも貼り付けてあったList篇などが参照となります。 change_task_statusメソッド さらに続きますが、こちらはchange_task_statusメソッドになります。 def change_task_status(self, instance): if not self.task_dialog: self.task_dialog = MDDialog( title="Move to Status?", buttons=[ MDFlatButton( text="NO", theme_text_color="Custom", text_color=self.theme_cls.primary_color, ), MDFlatButton( text="YES", theme_text_color="Custom", text_color=self.theme_cls.primary_color, on_release=lambda x: self.move_to_next( x, instance, instance.text, instance.secondary_text, instance.tertiary_text ) ) ], ) self.task_dialog.open() これも長々と書いていますが、特に複雑なことはやっていません。 あとで参照を貼り付けはしますが、本当に参照の部分とほとんど変わんねーじゃんか と言われそうなくらいそのままのコードとなっています。というかKivyMDの特徴でも あるようなところでもありますね。実装に迷いがでなくなるというか。 改めてちゃんと説明すると、単にこちらはタスクの左にあるドキュメントアイコンを 押したときに発動する中身となります。アイコンを押すと、上記のようにダイアログを 生成させてタスクスタータス移動する？という案内を出します。今のところNOボタンを 押しても何もならないですが、これは今思うとダイアログをdismissしなければいけま せんでした。。これはあとあとの課題にしよっと。 もちろんYESボタンを押すと、TODOはDOINGに、DOINGはDONEにという感じでステー タスが変わっていきます。この中のon_releaseメソッドがミソになりますが、lambda を使ってタスクの中身が保持されるよう、次に説明するmove_to_nextメソッドに受け 継がれます。instanceとかは移動元のタスクインスタンスとなりますね。 ここも課題は課題なのですが、本当はアプリを閉じた場合に情報を保持しなければいけない ので、このやり方はあまり相応しくはありません。でもpythonはlocalStorageとかもない しでどうやるんだろ。まぁ色々考えなければいけません。 参照としては以下もしくは該当マニュアルが詳しいです。 move_to_nextメソッド 最後はというと、先程でもあったmove_to_nextメソッドとなります。 def move_to_next(self, button_instance, task_instance, text, secondary_text, tertiary_text): if("todo" == task_instance.parent.text): self.root.ids.doing.add_widget( Task( text=text, secondary_text=secondary_text, tertiary_text=tertiary_text ) ) else: self.root.ids.done.add_widget( DoneTask( text=text, secondary_text=secondary_text, tertiary_text=tertiary_text ) ) self.remove_widget(task_instance) self.task_dialog.dismiss() self.task_dialog = None Snackbar(text="Task Moved!").open() こちらは先程のダイアログでもあった、YESボタンを押下したときに発動される 中身となります。 ここもそれほど複雑ではなく、現状ではtodoタスクかdoingタスクの場合にそれぞれ doingリストもしくはdoneリストにタスクを追加することが前半部分です。今思うと、 何かしらラッパーを作れば良かったかなと思いました。これはリファクタリングとして の課題になりますね。 そして、作ったあとはもともとあったタスクを消去して、ダイアログも閉じてNoneを入れ、 最後にSnackbarで消したよ!と通知を出すという感じですね。やっていることは単純です。 まぁ、これも本来ならタスクが作られたことを確認して通知をした方がいいのもあるのです が、、課題ばっかりですね。あとみなさんもこれどうなの?とかここが分からんということが あればコメント頂ければと思います。 まとめ はい、今日も長々見て頂きありがとうございました。 これまでのことが分かっていれば、超ミニマムアプリなんかは作れるということが分かられた のではないでしょうか。本当にオリジナルの実装となったのは多くても1割程度であったような。。 まぁでも、逆にたくさんある課題も浮き彫りになったのも事実ではあります。これについては おいおいこなしていく所存ではあります。 課題を潰していくというのはもちろんあるのですが、実はというとやりたいことはこれだけでは ありません。というのも、やりたいことは他にもあってそのためにこのアプリを作ったということ がありました。やりたいことは以下の通りですね。 アプリビルド Linterや自動テスト実施 CIなどで固定作業効率化 firebase連携 etc.. というわけで、やりたいことは他にもあったようなとど忘れしてしまっているものもありますが、 ざっとやりたいことは上記の通りです。なんと言っても、アプリをビルドしないと実際にモバイル でどのようになるのかとか分かりませんし、みなさんも開発などで試すことが出来ません。なので、 これだけは早急にやりたいなぁと思っています。なんかこういうことやってくれ!というのがあり ましたら、コメントどしどしお寄せください。 はい、ということで今日は以上でアプリについての説明も以上となります。来週からはBehivors 篇に戻っていくこととします。ということで、最後まで見て頂きありがとうございましたー！ それではごきげんよう。 参照 KivyMD https://kivymd.readthedocs.io/en/latest/

はじめに 中の人はエンジニアのeの字も知らない初学者です。文章やコードに不備があるかもしれませんのであたたかい目でみていただければと思います。。 また，理解が不足している部分があればぜひコメントで教えて頂きたいです。 作成の経緯と参考サイト AIについて勉強していた折に，卒業アルバム作りにAIを活用したという記事を見まして。 せっかくなので最終目標を「擬似卒アル作成アプリ」として勉強を始めました。 FaceAPIというものを使うと顔認識が簡単にできるとのことなので，まずは初めの段階としてFaceAPIを使って顔を認識するアプリを作成してみました。 （FaceAPIにたどり着くまでにも紆余曲折を経たのですが，それについては別の記事を書ければと考えています。） 以下のサイトに大変お世話になりました。 【Microsoft Azure Face】Pythonで画像の中の顔を認識してみる(SDK) 1.Azure FaceAPIのダウンロード Microsoft Azureの公式サイト クイック スタート:Face クライアント ライブラリを使用する に従い，①Azureのアカウントを作成②Faceリソースの作成③キーとエンドポイントの取得 2.実行するコードを記述 main.py import streamlit as st from azure.cognitiveservices.vision.face import FaceClient from msrest.authentication import CognitiveServicesCredentials from PIL import Image, ImageDraw, ImageFont import os # タイトルを設定 st.title("顔検出アプリ") # ローカル環境変数よりキーとエンドポイントを取得 KEY = os.environ.get('KEY') ENDPOINT = "https://エンドポイントのURL/" # サブスクリプションキー情報を使用してインスタンス化 face_client = FaceClient(ENDPOINT, CognitiveServicesCredentials(str(KEY))) # Streamlitでアップローダーを作成 uploded_file = st.file_uploader("jpg画像をアップロードしてください。顔を検出します。", type="jpg") if uploded_file is not None: # 画像をtmb.jpgとして一時保存 img = Image.open(uploded_file) img.save("tmb.jpg") image_data = open("tmb.jpg", 'rb') # 画像をバイナリーデータに変換 # 顔の検出 detected_faces = face_client.face.detect_with_stream( image_data, return_face_landmarks=True, return_face_attributes=['accessories','age','emotion','gender','glasses','hair','makeup','smile']) if not detected_faces: raise Exception('画像から顔を検出できませんでした。') # 認識された顔周辺に四角を描く関数 def getRectangle(faceDictionary): rect = faceDictionary.face_rectangle left = rect.left top = rect.top right = left + rect.width bottom = top + rect.height return ((left, top), (right, bottom)) # 認識された顔の上に年齢を描く関数 def getAge(faceDictionary): rect = faceDictionary.face_rectangle left = rect.left top = rect.top - 30 return (left, top) # イメージオブジェクト生成 image_data = Image.open("tmb.jpg") drawing = ImageDraw.Draw(image_data) # 関数を呼び出して、顔に四角を描く for face in detected_faces: drawing.rectangle(getRectangle(face), outline='Red', width = 3) drawing.text(getAge(face), str(face.face_attributes.age), font = ImageFont.truetype("arial.ttf", size=30), align = 'Left', fill = 'Red') st.image(image_data, caption='Uploaded Image', use_column_width=True) 3.Herokuでのデプロイ ①必要なファイルを作成 必要なファイルは以下の6種類です。 ・main.py ・requirements.txt requirements.txt streamlit Pillow msrest azure-cognitiveservices-vision-face ・setup.sh setup.sh mkdir -p ~/.streamlit/ echo "\ [general]\n\ email = \"メールアドレス\"\n\ " > ~/.streamlit/credentials.toml echo "\ [server]\n\ headless = true\n\ enableCORS=false\n\ port = $PORT\n\ " > ~/.streamlit/config.toml ・Procfile web: sh setup.sh && streamlit run main.py ・フォントファイルのコピーたち（下記で説明しています。） ②Gitでデプロイ 作業ディレクトリに移動します。 $ cd "作業ディレクトリへのパス" Herokuにログイン $ heroku login create new appから新しくアプリを立ち上げます。 Settingsからアプリ名をコピーします。 $ heroku git:remote -a "アプリ名" サブスクリプションキーを環境変数に設定します。 $ heroku config:set KEY=xxx Gitの初期化とcommitをします。 $ git init $ git add. $ git commit -m "first commit" インスタンスを作成します。 $ heroku create pushして開きます。 $ git push heroku master $ heroku open アプリが正常に作動すれば完了です。 躓いた点と解決法 ①サブスクリプションキーを公開しない工夫→環境変数の設定 セキュリティ対策のため，初めはseacret.jsonを通してキーを得ようとしたのですが，うまくいきませんでした。 Herokuにデプロイする際に，ターミナルで以下を実行することで，ローカルに環境変数を設定することができました。 heroku config:set KEY="xxx" 恐らくWeb開発をしている方にとっては常識なんだろうなー…でもこの段階でも躓きました。 ②画像をバイナリーデータに変換できない→一度名前を付けて保存 FaceAPIに渡す画像データはバイナリデータでないといけないということで，ここもかなり苦労しました。当初は以下のコードでBytesIOを用いてバイナリデータを返そうとしました。 img = Image.open(uploded_file) with io.BytesIO() as output: img.save(output, 'jpeg') image_data = output.getvalue() が，以下のエラー AttributeError: 'bytes' object has no attribute 'read' どうやらbytesオブジェクトはreadに対応していませんよーといったことのようです。アップロードされた画像を，一度名前を付けてopenを使用して保存することによって解決できました。 # 画像をtmb.jpgとして一時保存 img = Image.open(uploded_file) img.save("tmb.jpg") image_data = open("tmb.jpg", 'rb') # 画像をバイナリーデータに変換 ③drawing.textのフォント指定がうまくいかない→Fontファイルもアップロード Streamlitはターミナルでstreamlit run "ファイル名".pyと実行することによってアプリのデモを見ることができます。デモ上でアプリが問題なく作動したのを確認して，いざデプロイ！完了して，動作確認のため画像をアップロードしたところ，エラーが。。。 エラーメッセージは特になかったと記憶しています。しかし，以下のコードで対応するfontが見つからないとのだろうなということは何となくわかりました。 # 関数を呼び出して、顔に四角を描く for face in detected_faces: drawing.rectangle(getRectangle(face), outline='Red', width = 3) drawing.text(getAge(face), str(face.face_attributes.age), font = ImageFont.truetype("arial.ttf", size=30), align = 'Left', fill = 'Red') 解決策として，Fontsファイルにあった該当の文字フォントファイルを，作業ディレクトリに丸ごとコピーしました。フォントファイルも一緒にデプロイすることで，なんとか解決できました。どこかで調べたわけではないので，正攻法ではないと思いますが… ④Streamlitへのデプロイで公開されない→Herokuでデプロイ Streamlitにてアプリをデプロイ後，念のため…と思って他のアカウントから開いてみたところ You do not have access to this app or it does not existとエラー。 恐らくGithubのレポジトリを非公開にしていることが原因かと思います。Githubのレポジトリを非公開にしているのは，.gitignoreファイルをうまく扱えずseacret.jsonが丸出しになってしまったためです。 以下のサイトを参考にしてHerokuでアプリをデプロイすることで解決しました。 【簡単爆速第2弾】Streamlitをherokuにデプロイ まとめ FaceAPIを用いてWebアプリを作ることができました。 正直ここに辿り着くまでに多くのエラーと戦い，かなり苦労しました。無事にアプリが作動した時にはそれはそれは嬉しかったです。叫びました笑 そして今回はStreamlitの力を借りてHTMLやCSSを使わずの実装だったので，まだまだ勉強の余地はありますね。 月並みな感想ですが，実際のエンジニアの方はすごいなあ。と思いました。 おわり

はじめに ・登山好きプログラミング初心者が衛星データを利用して山探しする企画です。 ・こう書いた方がより良いなどありましたら、勉強になりますのでぜひコメントください。 ・本企画は以下の４部構成を予定しており、今回は第３弾です。 ①日本の山リスト作成編 ②衛星データで周囲で一番高い山を探してみる編 ③衛星データで山周辺の地域の晴天率を調べてみる編 ④衛星データで眺望の良い、山頂がひらけている山を探してみる編 前回記事(②衛星データで周囲で一番高い山を探してみる編)で作成した日本の山リストに対して、気象の衛星データを使って自分たちの登りたい山を絞っていきます。 気象データの紹介と晴天率の求め方 今回使用する衛星データは「GSMaP」です。 https://developers.google.com/earth-engine/datasets/catalog/JAXA_GPM_L3_GSMaP_v6_operational こちらのデータのhourlyPrecipRateGCというバンドで降水量のデータを取得します。 GSMaPは1時間ごとの降水量のデータです。 本記事では晴天率の算出は以下の①〜③の流れで行います。 ① X日0時〜23時までの1時間ごとの降水量の合計を計算する ② 合計値が0のときには晴れとしてカウント ③ ①と②を1ヶ月分繰り返して、晴れ/日数で晴天率を算出 つまり、1日の中で少しでも雨が降っていればその日は晴れとしてカウントしません。少し厳しい条件ですが、簡単のためその条件で求めてみます。 ImageCollectionへのmap()の適用 晴天率の算出では、一つの山に対して24時間×1ヶ月のそれぞれの画像に対して画像内の平均値を計算し、1ヶ月分の合計値を計算する必要があります。 前回の記事では一枚の画像に対して最大値を取得していました。 しかし今回のように、同じ条件の複数の画像に対して同じ処理を行う場合には、ImageCollectionと呼ばれる画像の集まりに対してmap()で同じ処理を適用する方法があります。 例として、GSMaPのデータを使って2020年10月の富士山の周辺の降水量のデータを取得してみます。 まずはライブラリのインポートとGEEへのログインを行います。 import ee import pandas as pd import csv try: ee.Initialize() except Exception as e: ee.Authenticate() ee.Initialize() 関心領域と対象期間を指定して、ImageCollectionを取得します。 詳細は私もわかっていないのですが、ImageCollectionとはImageの集まりと考えれば問題ないと思います。 # 富士山周辺を指定 lon = 138.727363 lat = 35.360626 aoi = ee.Geometry.Rectangle([lon - 0.1, lat - 0.1, lon + 0.1, lat + 0.1]) # 対象期間を指定 from_date='2020-10-01' to_date='2020-11-01' GSMaP = ee.ImageCollection("JAXA/GPM_L3/GSMaP/v6/operational").filterDate(from_date, to_date) reducerによって平均値を取得する関数を定義します。 あるImageに対して、reducer、関心領域、バンドを指定してreduceRegionを適用することでImageの代表値を取得することができます。 今回は取得した平均値と日付を画像のデータに書き込んで返します。 def aoi_mean(img): mean = img.reduceRegion(reducer=ee.Reducer.mean(), geometry=aoi, scale=30).get('hourlyPrecipRateGC') return img.set('date', img.date().format()).set('mean',mean) 上で定義したreducerの関数を、ImageCollectionに対してmap()で適用することで、ImageCollection内のそれぞれのimageに平均値を取得するreducerが適用され、日付と平均値が各imageに付与されます。 更に、reduceColumnsによって各imageが持つデータを日付と平均値のみに減らし、 取得した日付と平均値はImageCollection内のそれぞれのImageが持っているため、そちらを取り出してきてリストにし、それをpandasのデータフレームに変換します。 aoi_reduced_imgs = GSMaP.map(aoi_mean) nested_list = aoi_reduced_imgs.reduceColumns(ee.Reducer.toList(2), ['date','mean']).values().get(0) df = pd.DataFrame(nested_list.getInfo(), columns=['date','mean']) df 出力された結果は以下のようになります。 24時間×31日の744個の日付と平均値のリストを取得することができました。 リストにある山の晴天率を求めてみる 上で述べたImageCollectionへのmap()の適用を用いて晴天率の算出を行います。 まずはライブラリのインポートとGEEへのログインを行います。 import ee import pandas as pd import csv try: ee.Initialize() except Exception as e: ee.Authenticate() ee.Initialize() 前回(②衛星データで周囲で一番高い山を探してみる編)取得した標高2000m以上かつ周囲で最も高い山のリストを読み込みます。 df_top = pd.read_csv("/content/drive/My Drive/mountain_list_top.csv") 対象期間を指定して、ImageCollectionを取得します。 # 対象期間を指定 from_date='2020-10-01' to_date='2020-11-01' GSMaP = ee.ImageCollection("JAXA/GPM_L3/GSMaP/v6/operational").filterDate(from_date, to_date) 上と同じく関心領域の平均値を取得する関数を定義します。 def aoi_mean(img): mean = img.reduceRegion(reducer=ee.Reducer.mean(), geometry=aoi, scale=30).get('hourlyPrecipRateGC') return img.set('date', img.date().format()).set('mean',mean) 晴天率を算出する関数を定義します。 引数として、上で作成した日付と平均値のリストを受け取るようにしています。データは、24時間×その月の日数分のデータ数を持っています。 簡単にいうと以下の①〜④の処理を行なっています。 ① データ数を24で割ってその月の日数を算出 ② 24時間の降水量(Imageごとの平均値)の合計値を計算 ③ 合計値が0であればカウント ④ ②と③を日数分繰り返す def get_sunny_rate(df): cnt = 0 ndays = int(len(df) / 24) for i in range(ndays): total_pa = 0 k = 0 for j in range(24): k = i * 24 + j total_pa += df.at[k, 'mean'] if total_pa == 0: cnt += 1 return cnt / ndays リストにある山に対して晴天率を取得します。 リストの緯度経度から関心領域を指定し、上で定義したreducerによる平均値のリストを取得する関数と、平均値のリストから晴天率を算出する関数を適用します。 算出した晴天率は元のデータフレームの’sunny_rate’列に書き込みます。 各山に上の処理を適用し、最後にcsvとして保存します。 for i in range(len(df_top)): lat = df_top.at[i, "緯度"] lon = df_top.at[i, "経度"] aoi = ee.Geometry.Rectangle([lon - 0.025, lat - 0.025, lon + 0.025, lat + 0.025]) aoi_reduced_imgs = GSMaP.map(aoi_mean) nested_list = aoi_reduced_imgs.reduceColumns(ee.Reducer.toList(2), ['date','mean']).values().get(0) df_pa = pd.DataFrame(nested_list.getInfo(), columns=['date','mean']) sr = get_sunny_rate(df_pa) df_top.at[i, 'sunny_rate'] = sr print('Done:', df_top.at[i, "山名"]) print(sr) df_top.to_csv("/content/drive/My Drive/add_sunny_rate.csv") 取得したcsvをエクセルを使ってグラフにしてみました。 晴天率は高くても50%程度のようです。1日の中で少しでも降水量があると雨という判定をしているので、このくらいの数字になってしまうと思われます。 判定に使用する時間帯を絞ったり、降水量も判定に使ったりより良い方法もあると思われますが、今回はとりあえずこれを晴天率とします。 まとめ 今回は対象の山周辺の晴天率を取得してみました。 ImageCollectionへmap()を使ってreducerを適用することで代表値のリストを作成する方法も紹介しました。これを使えば時系列的な比較が容易になります。 次回は最終章です。植生指数という衛星データを用いて山頂がはげている山を探していきます。 ※2022/01/23に次の記事を公開予定です。 参考文献 GEEデータカタログ：GSMaP Operational: Global Satellite Mapping of Precipitation https://developers.google.com/earth-engine/datasets/catalog/JAXA_GPM_L3_GSMaP_v6_operational?hl=en

いろんなモジュール（特に機械学習系）が入ってる環境で、このAuto-import候補がズラーーーーっと出てこられるとPCがカクつくし見ずらいし困ります。 ユーザセッティングでauto importで探したら見つかりました。以下をsettings.jsonに追記すればOKです。 settings.json { ... ... ... "python.analysis.autoImportCompletions": false, }

超初心者が２週間でpythonエンジニア認定基礎試験に合格してみた ★合格者のスペック★ ・国公立大学文系卒業 ・業務でpythonは全く使用しない ・2021年12月25日から勉強を開始して、2022年1月15日に合格 ★勉強期間★ 3週間 平日に２時間、休日は５時間程度実施 ★勉強手順★ 書籍スッキリわかるPython入門を2時間程度で流し読み →この段階ではわからなくてもOK 書籍スッキリわかるPython入門から大事そうなところをノートにまとめる →わからなくてもとりあえずノートに書くことを意識 Jupyter Notebookで遊んで動かしてみる。 →まとめてというよりも、読んでいる途中途中で書いて、動かして自分を感動させる UdemyのPython初心者講座を購入し、2倍速で流し見＋コードを書く →とにかく流れをつかむ 過去問サイトで繰り返し演習（合計１０回以上は回答した） →ほかのみなさんがあげている某サイトを使用。演習はこれだけでＯＫ。 過去問の解説は、ググる、もしくはyoutubeで解説をあげてくださっている方がいるのでそれを見る →この、自分に適した動画、まとめを見つけることが一番重要だと思います。それを ノートにまとめていく。UDEMYよりもわかりやすいものがたくさんあります。 寝る前にノートを読む。 通勤時間にＵＤＥＭＹを流す。 ★感想★ Pythonの基本的な構文はこれで大丈夫、というくらいまでは理解できた まずは慣れるまでがハードルなので、わからなくてもとりあえず聞く、流し読み する、が大事だと思います。 合格だけを目的にするなら、コードを記載するのはそんなに重要じゃない。

最新の機械学習は、Pytorchのラッパーライブラリを使うケースが多い ObjectDetectionタスクのモデルであれば、mmdetectionやdetectron2などのpytorchラッパーを通して使われていることが多い。 例えば、FaceBookResearchのモデルは、FaceBookReseachが使いやすいdetectron2を通して発表している。 以下のモデルはdetectron2で使えるモデルである。 DensePose: Dense Human Pose Estimation In The Wild Scale-Aware Trident Networks for Object Detection TensorMask: A Foundation for Dense Object Segmentation Mesh R-CNN PointRend: Image Segmentation as Rendering Momentum Contrast for Unsupervised Visual Representation Learning DETR: End-to-End Object Detection with Transformers Panoptic-DeepLab: A Simple, Strong, and Fast Baseline for Bottom-Up Panoptic Segmentation D2Go (Detectron2Go), an end-to-end production system for training and deployment for mobile platforms. Pointly-Supervised Instance Segmentation Unbiased Teacher for Semi-Supervised Object Detection Rethinking "Batch" in BatchNorm Per-Pixel Classification is Not All You Need for Semantic Segmentation ラッパーライブラリの使い方を知れば、最新モデルが使いやすい 高精度の最新モデルを使うには、これらのラッパーライブラリの基本的な使い方に慣れておくと便利である。 覚えるのめんどくさそう？ わりとかんたん そもそもライブラリ自体かんたんに推論などを行えるようにラップされているので、割と使い方はかんたんである。 使い方（Detectron2） 基本的には、 １、使いたいモデルのConfigファイルとWeightをAPIに渡してモデルをビルド ２、inference という手順である。 # build cfg = get_cfg() cfg.merge_from_file(model_zoo.get_config_file("COCO-InstanceSegmentation/mask_rcnn_R_50_FPN_3x.yaml")) cfg.MODEL.ROI_HEADS.SCORE_THRESH_TEST = 0.5 # set threshold for this model # Find a model from detectron2's model zoo. You can use the https://dl.fbaipublicfiles... url as well cfg.MODEL.WEIGHTS = model_zoo.get_checkpoint_url("COCO-InstanceSegmentation/mask_rcnn_R_50_FPN_3x.yaml") predictor = DefaultPredictor(cfg) # inference im = cv2.imread("./input.jpg") outputs = predictor(im) 覚えておくとお得 基本的に使いやすいように作ってくれたライブラリなので、覚えておくと便利である。 ? フリーランスエンジニアです。 お仕事のご相談こちらまで rockyshikoku@gmail.com Core MLやARKitを使ったアプリを作っています。 機械学習／AR関連の情報を発信しています。 Twitter Medium

VEML6075 という UV インデックスセンサーを入手して Raspberry Pi で動かしてみました。UV インデックスセンサーの比較は英語ですがここ が詳しそうです。UV インデックスの数字の意味に関してはこのあたりが参考になります。 しかし Adafruit をはじめ既に VEML6705 の製品はほぼ売っていません。SI1145 や GUVA-S12SD なら引き続き販売されているようですが Adafruit の SI1145 ページにも VEML6075 を推奨すると書いてあり UV Index のセンサーとしては VELM6075 が優秀なようです。しかしなぜかは不明ですが製造元の Vishay がが製造を辞めてしまったようです。* UV インデックスセンサー使ってみると言っても Adafruit のライブラリを使うので簡単です。まず CircuitPython ライブラリと adafruit-blinka (Blinka) をインストールします。CircuitPython とは MicroPython の一種で、adafruit-blinka は CircuitPython の API を Raspberry Pi では RPi.GPIO に変換するためのもののようです。Raspberry Pi だけではなく Arduino などでも使えるようです。* OS のパッケージ最新にして python3-pip をインストールします。 $ sudo apt update && sudo apt upgrade -y $ sudo apt-get install python3-pip $ sudo pip3 install --upgrade setuptools 次に adafruit-python-shell を pip3 でインストールし raspi-blinka.py という名前のスクリプトをダウンロードします。 $ cd ~ $ sudo pip3 install --upgrade adafruit-python-shell $ wget https://raw.githubusercontent.com/adafruit/Raspberry-Pi-Installer-Scripts/master/raspi-blinka.py $ sudo python3 raspi-blinka.py raspi-blinka.py を実行すると adafruit-blinka がインストールされます。最後は Y と答えて Enter を押すと Raspberry Pi が再起動します。 ~ $ sudo python3 raspi-blinka.py This script configures your Raspberry Pi and installs Blinka RASPBERRY_PI_ZERO_W detected. Updating System Packages <snip> Blinka Successfully installed Adafruit-PureIO-1.1.9 adafruit-blinka-6.18.0 pyftdi-0.53.3 pyusb-1.2.1 rpi-ws281x-4.3.1 sysv-ipc-1.1.0 DONE. Settings take effect on next boot. REBOOT NOW? [Y/n] Y adafruit-blinka が正常にインストールされたかはここのコードを使って確認することができます。 再起動が完了したら veml6075 用のライブラリ (adafruit-circuitpython-veml6075) をインストールします。同時に adafruit-circuitpython-busdevice もインストールされます。 ~ $ sudo pip3 install adafruit-circuitpython-veml6075 adafruit-circuitpython-veml6075 が正常にインストールされていれば以下のコマンドを実行した際に特にエラーなどは出力されません。 ~ $ python3 -m adafruit_veml6075 配線はここにある通りこんな感じです。SparkFun や Garvity のものは I2C のケーブルを使って接続できるので便利です。 コードは以下のような感じです。 import time, board, busio import adafruit_veml6075 interval = 3 i2c = busio.I2C(board.SCL, board.SDA) while True: try: veml = adafruit_veml6075.VEML6075(i2c, integration_time=100) print("UV index:", veml.uv_index) time.sleep(interval) except KeyboardInterrupt: break 日中帯に太陽の方向にセンサーを向けて (窓は開けたほうが良いかもしれません) 上記を実行すると冬なら以下のような値を返します。室内で実行すると 0 に近い値が返ってくると思います。 ~ $ python3 ./veml6075.py UV Index: 0.6220514149999998 UV Index: 0.7128853399999999 UV Index: 0.74213456

はじめに 新年早々、Netflexに加入しまして、以前からネットとかで海外ドラマのフレンズが英語の勉強にいいという噂は耳にしていたので、今年は素直にやってみようと思った次第です。 フレンズは1話あたり22分でシーズン1〜10まであり、全部で236話あるそうです。1日1話ずつ観ても8ヶ月くらいかかります。 フレンズの脚本のスクリプトも見ることができます。ただし、日本語訳はついていません。 DeepL API DeepLは、口語に近い文章に訳してくれるということで、フレンズを訳すのにちょうど良いかなと思って、ちょっとしたアプリケーションを作る上でDeepL APIを試してみたくなりました。 準備 1.開発者向けのアカウントを登録 DeepLAPIを使うためにDeepLProアカウントを作ります。 https://www.deepl.com/pro#developer 姓名メールアドレスやパスワード、住所などを入力し、登録します。 このサイトから「DeepL API Free」を選択して手順に沿って登録します。 ※クレジットカードを入力する項目がありますが、Freeプランであれば課金はされません。 ※DeepL API Freeだと50万文字/月が上限のようです。（執筆時点） 2.登録後のAPI認証キーをコピーしておく 登録完了後に認証キーが表示されるのでコピーしておきます。 「XXX~XXXX:fx」というキーです。 マイページ->プランでも見ることができます。 Python Google Colaboratoryで簡単に使用してみます。 API KEYを入力のところは、API認証キーに書き換えてください。 import requests # NOTE: put API KEY API_KEY:str = 'API KEYを入力' txt = 'This is a pen.' params = { "auth_key": API_KEY, "text": txt, "source_lang": 'EN', "target_lang": 'JA' } request = requests.post("https://api-free.deepl.com/v2/translate", data=params) result = request.json() print(result["translations"][0]["text"]) 結果 これはペンです。 JavaScript API KEYを入力のところは、API認証キーに書き換えてください。 <!DOCTYPE html> <html lang = "ja"> <head> <meta charset = "utf-8"> <title>JavaScript</title> <style> textarea { width: 500px; height:250px; } </style> </head> <body> <script> const API_KEY = 'API KEYを入力'; const API_URL = 'https://api-free.deepl.com/v2/translate'; function output() { const entext = document.getElementById("entext").value; let content = encodeURI('auth_key=' + API_KEY + '&text=' + entext + '&source_lang=EN&target_lang=JA'); let url = API_URL + '?' + content; fetch(url) .then(function(response) { if (response.ok) { return response.json(); } else { throw new Error("Could not reach the API: " + response.statusText); } }).then(function(data) { document.getElementById("jatext").value = data["translations"][0]["text"]; }).catch(function(error) { document.getElementById("jatext").value = error.message; }); }; </script> <textarea id="entext" placeholder="英語を入力してください"></textarea> <br> <input type="button" value="翻 訳" onclick="output()" /> <br> <textarea id="jatext"></textarea> </body> </html> 結果 試しにフレンズの一部を翻訳してみました。 English C’mon Daddy, listen to me! It’s like, it’s like, all of my life, everyone has always told me, ‘You’re a shoe! You’re a shoe, you’re a shoe, you’re a shoe!’. And today I just stopped and I said, ‘What if I don’t wanna be a shoe? What if I wanna be a- a purse, y’know? Or a- or a hat! No, I’m not saying I want you to buy me a hat, I’m saying I am a ha- It’s a metaphor, Daddy! Japanese パパ、聞いてよ！これまでの人生で、誰もが私に言ってきたのは、『お前は靴だ！』ということでした。あなたは靴だ！あなたは靴だ！あなたは靴だ！』ってね。そして今日、私はただ立ち止まり、こう言ったのです「もし私が靴になりたくないとしたら？もし私が靴になりたくなかったら、どうしよう？あるいは......あるいは帽子！』と。帽子を買って欲しいと言っているのではなくて、私は帽子だと言っているの......それは比喩よ、パパ！」 注意点 まとめて翻訳させようとすると、翻訳できなかったりします。 また、話者のところが「モニカです。」みたいになるので、本文のみを訳させるようにした方がいい。 嵌った点 最初、Python同様にPOSTでやろうとしたのですが、403エラー(type:cors)となり解決できず、諦めました。 サーバー側(node.js)ならPOSTできると思われますが、クライアント側はダメなのかな感じました。 最後に フレンズのスクリプトを訳す専用のツールを作ろうと思ってます。 あと、Chrome拡張の「Language Reactor」を使うとフレンズを英語字幕・日本語字幕を同時表示して視聴できるので便利です。

　はじめに 年があけ年賀状がちらほら届き「『さて郵便番号を書いて返信してやろうか』って、コイツ郵便番号書いてねぇぞ」となりましてググれば一発のところ、『住所を入力して郵便番号を逆引きできればな』と思い立ち、今回のハンズオン構築を行ってみました。 　参考資料 参考リンクには今回利用したAPIのホームページURLです。『郵便番号から住所の検索するAPI』は日本郵便様始め結構見つかったのですが『住所から郵便番号』はなかなか見当たらなかったのですが大変助かりました。 HeartRails Geo API 出典:「位置参照情報」(国土交通省)の加工情報・「HeartRails Geo API」(HeartRails Inc.) 　構成図 　ハンズオン 1：コードの記述 サンプルコード（コメントアウトで各動作の内容記述あり） Postcode_Search.py # ライブラリのインポート import requests #APIをリクエストのため import json #json形式で受け取ったデータの処理するため import os #環境変数のため import sys #exit()関数でプログラムを終了させるため from dotenv import load_dotenv #環境変数を利用するため load_dotenv() # 環境変数 yahoo_api_key = "&appid=" + os.getenv('YAHOO')#トークン直打ちでも可能 #yahooジオコーダ変数 geo_api_url = "https://map.yahooapis.jp/geocode/V1/geoCoder?" y_parm1 = "&output=json" #APIのレスポンスをjson形式に指示す y_parm2 = "&query=" #調べたい場所の引数のプレフィックス #緯度経度による町域情報一覧の変数 city_api_url = 'http://geoapi.heartrails.com/api/json?method=searchByGeoLocation' #緯度経度を抽出する def find_latitude_and_longitude(area): # 変数 geo = geo_api_url + yahoo_api_key + y_parm1 + y_parm2 + area # Yahoo!ジオコーダによる取得したい場所の情報をURLにする geo_info = requests.get(geo) #リクエストする geo_obj = json.loads(geo_info.text)#リクエストしたjsonの内容を解析する try: #入力した地点の緯度経度が有無で処理が分岐する geo_parm = geo_obj["Feature"][0]["Geometry"]["Coordinates"] # オブジェクトから座標情報を取得する geo_parm_list = geo_parm.split(",")# 取得した座標をカンマごとにリストで分ける longitude = geo_parm_list[0] #経度 latitude = geo_parm_list[1] #緯度 return(latitude,longitude) #後続関数に緯度・経度を返す except: #緯度経度が取得出来ない場合の処理 print("申し訳ございませんが、入力した地点の緯度経度が取得出来ませんでした") sys.exit() #対話型シェルを修了させるための関数 #緯度経度から町域情報を取得して郵便番号を抽出する def town_area_information(latitude,longitude): # 変数 t_parm1 = "&x=" + longitude t_parm2 = "&y=" + latitude cityarea = city_api_url + t_parm1 + t_parm2 cityarea_info = requests.get(cityarea)#リクエストする cityarea_obj = json.loads(cityarea_info.text)#リクエストしたjsonの内容を解析する cityarea_parm = cityarea_obj["response"]["location"][0]["postal"]# jsonから郵便番号だけを取得する return(cityarea_parm) if __name__ == "__main__": print("郵便番号を知りたい県名から市長村名までを入力ください\n例:埼玉県川口市\n") area = input() #area = "埼玉県川口市" print(area + "の郵便番号は以下の番号です") latitude,longitude = find_latitude_and_longitude(area) postcode = town_area_information(latitude,longitude) print("〒" + postcode[:3] + "-" + postcode[3:]) 2:部分的な説明 2.1 今回利用するライブラリをインストールする Postcode_Search.py # ライブラリのインポート import requests #APIをリクエストのため import json #json形式で受け取ったデータの処理するため import os #環境変数のため import sys #exit()関数でプログラムを終了させるため from dotenv import load_dotenv #環境変数を利用するため load_dotenv() 緯度経度が該当しない地名に関しては処理を修了させるためにimport sysを利用 2.2 .envの環境変数を代入する Postcode_Search.py # 環境変数 yahoo_api_key = "&appid=" + os.getenv('YAHOO')#トークン直打ちでも可能 .envファイルをから、load_dotenvでファイルの中身を読み取り環境変数として読み込む ディレクトリの構成 . ├── .env └── Postcode_Search.py 2.3 関数で利用するの変数を代入する Postcode_Search.py #yahooジオコーダ変数 geo_endpoint = "https://map.yahooapis.jp/geocode/V1/geoCoder?" y_parm1 = "&output=json" #APIのレスポンスをjson形式に指示す y_parm2 = "&query=" #調べたい場所の引数のプレフィックス #緯度経度による町域情報一覧の変数 city_endpoint = 'http://geoapi.heartrails.com/api/json?method=searchByGeoLocation' 2.4 find_latitude_and_longitude()関数 Postcode_Search.py #緯度経度を抽出する def find_latitude_and_longitude(area): # 変数 geo = geo_endpoint + yahoo_api_key + y_parm1 + y_parm2 + area # Yahoo!ジオコーダによる取得したい場所の情報をURLにする 引数areaを受け取りgeoに代入する Postcode_Search.py geo_info = requests.get(geo) #リクエストする geo_obj = json.loads(geo_info.text)#リクエストしたjsonの内容を解析する Yahoo!ジオコーダAPI レスポンスフィールドについての詳細(下記画面は一部抜粋) ブラウザでURLを押下した際の画面抜粋 ※緯度・経度の取得した順番に注意する Postcode_Search.py try: #入力した地点の緯度経度が有無で処理が分岐する geo_parm = geo_obj["Feature"][0]["Geometry"]["Coordinates"] # オブジェクトから座標情報を取得する geo_parm_list = geo_parm.split(",")# 取得した座標をカンマごとにリストで分ける longitude = geo_parm_list[0] #経度 latitude = geo_parm_list[1] #緯度 return(latitude,longitude) #後続関数に緯度・経度を返す except: #緯度経度がない場合の処理 print("申し訳ございませんが、入力した地点の緯度経度が取得出来ませんでした") sys.exit() #対話型シェルを修了させるための関数 例外処理としてtry,exceptを利用しており、下記のような動作とする try: Yahoo!ジオコーダAPIで緯度,経度取得 　　　　後続に引数（緯度,経度）を渡す except: ただし該当なしの場合は「取得出来ませんでしたコメント」及び 　　　　システムの修了 2.5 town_area_information()関数 Postcode_Search.py def town_area_information(latitude,longitude): # 変数 t_parm1 = "&x=" + longitude t_parm2 = "&y=" + latitude cityarea = city_endpoint + t_parm1 + t_parm2 cityarea_info = requests.get(cityarea)#リクエストする find_latitude_and_longitude()関数で取得した緯度,経度をAPIにリクエストする Postcode_Search.py cityarea_obj = json.loads(cityarea_info.text)#リクエストしたjsonの内容を解析する cityarea_parm = cityarea_obj["response"]["location"][0]["postal"]# jsonから郵便番号だけを取得する return(cityarea_parm) 今回ご利用させてもらっているAPIのリクエストパラメータ、レスポンスフィールド一覧 下記形式はXML形式で取得した際の見え方の一例 上記の内容を、入力した地域のjson形式でpostal部分だけを取得している 2.6 モジュールを直接実行したときだけ実行する動作の指定 Postcode_Search.py if __name__ == "__main__": print("郵便番号を知りたい県名から市長村名までを入力ください\n例:埼玉県川口市\n") area = input() #area = "埼玉県川口市" print(area + "の郵便番号は以下の番号です") latitude,longitude = find_latitude_and_longitude(area) postcode = town_area_information(latitude,longitude) print("〒" + postcode[:3] + "-" + postcode[3:]) 出力の見え方として、取得した郵便番号を整形"〒" + postcode[:3] + "-" + postcode[3:] 3：挙動の確認 該当する緯度・経度がある場合 ターミナルで押下 tetutetu214@mbp 0_Qiita_hanson % python Postcode_Search.py 郵便番号を知りたい県名から市長村名までを入力ください 例:埼玉県川口市 埼玉県川口市 埼玉県川口市の郵便番号は以下の番号です 〒332-0016 該当する緯度・経度がない場合 ターミナルで押下 tetutetu214@mbp 0_Qiita_hanson % python Postcode_Search.py 郵便番号を知りたい県名から市長村名までを入力ください 例:埼玉県川口市 あああああ あああああの郵便番号は以下の番号です 申し訳ございませんが、入力した地点の緯度経度が取得出来ませんでした ※確認中みつけてしまったこと tetutetu214@mbp 0_Qiita_hanson % python Postcode_Search.py 郵便番号を知りたい県名から市長村名までを入力ください 例:埼玉県川口市 なぜなの なぜなのの郵便番号は以下の番号です 〒894-0036 yahooジオコーダAPIにおいて入力された住所がない場合、上位のレベルで再検索などにより、よっぽど該当しない文言でない限り検索してしまうからか。 完全マッチしか取得出来ないようにパラメータに記述すれば、問題を防げそうだと思っています。が、今回は後回しにします。 　さいごに yahooジオコーダAPIの出力について問題ありますが、当初の想定であった県名市町村までを入力すれば郵便番号までは取得することはできました。もやもやは残っていますが。 try,exceptなどは今回始めて利用してみましたが、挙動としては動くものの記述する場所は正しいのか？など課題を残すような構築となりました。来週もまた再度コードを書きながら理解を深めていきたいと思います。

目次 1. はじめに 2. ライブラリのインストール 3. 地震応答解析 4. グラフの出力 5. おわりに 1-はじめに 構造系ではGrassHopperが人気ですが、Dynamoでも簡単な応答解析ができないか試してみます。あえてDynamoを使うメリットはとくにないです。 2-ライブラリのインストール Pythonで使う外部ライブラリをインストールします。 1. はじめに Dynamo を起動し、下記のコードで CPython のバージョンを確認します。 PythonScript import sys OUT = sys.version 2. Anaconda をダウンロードし、Anaconda Navigator から「CMD.exe Prompt」を実行します。 3. 下記のコードを入力して Dynamo 用の環境をつくります。 3.8.3は最初に確認した CPython のバージョンの番号です。 Prompt conda create --name Dynamo383 python=3.8.3 4. Proceed ([y]/n)?と聞かれるので Y を入力します。 5. いま作成した環境に切り替えます。 Prompt conda activate Dynamo383 6. 必要なライブラリをまとめてインストールします。 Prompt pip install numpy pandas scipy Pillow matplotlib 3-地震応答解析 せん断変形のみを考慮した多質点系モデルの振動解析を実装します。 波形データは東京都建設局で公開されているCSVデータを利用します。 ライブラリをインポートします。 PythonScript #ライブラリをロード sys.path.append(r'C:\Users\shota\.conda\envs\Dynamo383\Lib\site-packages') import sys, math, io import numpy as np import pandas as pd from scipy import interpolate import matplotlib.pyplot as plt from PIL import Image as Img import System.Drawing from System.Drawing import * from System.Drawing.Imaging import * from System.IO import MemoryStream コード中にIN[0]とかあると分かりにくいので入力値はInputクラスにまとめておきます。 PythonScript # インプット class Input: def __init__(self): self.g = IN[0] # 重力加速度 self.dt = IN[1] # 積分時間刻み self.beta = IN[2] # Newmarkβ法のβ self.damp_factor = IN[3] # 減衰定数 self.natural_frequency = IN[4] # 振動数 self.wave_path = IN[7] # CSVファイルのパス self.arr_m = IN[5] # 重量 self.arr_k = IN[6] # 剛性 波形データを読み込みます。 PythonScript # 波形データ def WaveData(input): with open(input.wave_path, 'r') as f: df = pd.read_csv(f) return list(df['acc'].values.tolist()) マトリクス生成 質量マトリクス、減衰マトリクス、剛性マトリクスをつくります。 PythonScript # 質量マトリクス def MassMatrix(arr_m, input): return np.diag(list(map(lambda x: x / input.g, arr_m))) # 剛性マトリクス def StiffMatrix(arr_k): size = len(arr_k) full = np.zeros((size, size)) for i in range(size): k = arr_k[i] part = [[k , -k], [-k, k]] for i_row in range(len(part)): for i_col in range(len(part)): target_row = i-1 + i_row target_col = i-1 + i_col if 0 <= target_row and 0 <= target_col: full[target_row][target_col] += part[i_row][i_col] return full # 減衰マトリクス def DampMatrix(k, input): omega = 2 * math.pi * input.natural_frequency return 2 * input.damp_factor * k / omega 質点系モデル作成 前項で生成したマトリクスを用いて質点系モデルをつくります。 PythonScript # 質点系モデル class Model: def __init__(self,input): arr_m = input.arr_m arr_k = input.arr_k self.size = len(arr_m) arr_k.reverse() arr_m.reverse() self.m = MassMatrix(arr_m, input) self.k = StiffMatrix(arr_k) self.c = DampMatrix(self.k, input) 解析 Newmarkβ法を用いて数値解析を行い、入力波形のステップごとに加速度・速度・変位を求めます。 PythonScript def dynamic_analysis(model, input): # 初期化 m = model.m k = model.k c = model.c acc0 = WaveData(input) unit_vector = np.ones(model.size) pre_acc0 = 0.0 time = 0.0 dis = np.zeros(model.size) vel = np.zeros(model.size) acc = np.zeros(model.size) ddis = np.zeros(model.size) dvel = np.zeros(model.size) dacc = np.zeros(model.size) dis_his = {} vel_his = {} acc_his = {} for i in range(0, model.size): dis_his[i] = [] vel_his[i] = [] acc_his[i] = [] time_his = [] # Newmarkβ法による数値解析（増分変位による表現） for i in range(0, len(acc0)): kbar = k + (1.0/(2.0*input.beta*input.dt)) * c + (1.0/(input.beta*input.dt**2.0)) * m dp1 = -1.0 * m.dot(unit_vector) * (acc0[i] - pre_acc0) dp2 = m.dot((1.0/(input.beta*input.dt))*vel + (1.0/(2.0*input.beta))*acc) dp3 = c.dot((1.0/(2.0*input.beta))*vel + (1.0/(4.0*input.beta)-1.0)*acc*input.dt) dp = dp1 + dp2 + dp3 ddis = np.linalg.inv(kbar).dot(dp) dvel = (1.0/(2.0*input.beta*input.dt))*ddis - (1.0/(2.0*input.beta))*vel - ((1.0/(4.0*input.beta)-1.0))*acc*input.dt dacc = (1.0/(input.beta*input.dt**2.0))*ddis - (1.0/(input.beta*input.dt))*vel - (1.0/(2.0*input.beta))*acc dis += ddis vel += dvel acc += dacc acc_abs = acc + [acc0[i] for n in range(1,model.size)] [dis_his[i].append(x) for i, x in enumerate(dis)] [vel_his[i].append(x) for i, x in enumerate(vel)] [acc_his[i].append(x) for i, x in enumerate(acc_abs)] time_his.append(time) time += input.dt pre_acc0 = acc0[i] return time_his, dis_his, vel_his, acc_his 結果出力 前項で作成したdynamic_analysisを実行して結果を求めます。 PythonScript # メイン input = Input() model = Model(input) time_his, dis_his, vel_his, acc_his = dynamic_analysis(model, input) 4-グラフの出力 matplotlibでグラフを作成します。 グラフをWatch Imageノードで表示するために、System.Drawing.Bitmapに変換します。 PythonScript # 画像変換 def convertToBitmap(fig): rgba_buf = fig.canvas.buffer_rgba() (w,h) = fig.canvas.get_width_height() npImgArray = np.frombuffer(rgba_buf, dtype=np.uint8).reshape((h,w,4)) bitmap_ = None # alphaチャンネルを削除 if npImgArray.ndim == 3 and npImgArray.shape[-1] == 4: npImgArray = npImgArray[:, :, :-1] img = Img.fromarray(npImgArray, "RGB") # 画像をバイナリデータに変換し、メモリ上に保存 byteIO = io.BytesIO() img.save(byteIO, format='BMP') byteArr = byteIO.getvalue() # convert to Net ByteArray netBytes = System.Array[System.Byte](byteArr) with MemoryStream(netBytes) as ms: bitmap_ = Bitmap(ms) return bitmap_ # グラフ fig = plt.figure(figsize=(10,10)) ax1 = fig.add_subplot(3,1,1) ax1.set_ylabel("acc") # 縦軸のタイトル ax1.plot(time_his, acc_his[1], label="1F", color="darkturquoise") # 1階のグラフ ax1.plot(time_his, acc_his[0], label="2F", color='limegreen') # 2階のグラフ ax1.legend(loc='lower right') # 凡例の位置 ax1.grid(True) # グリッド ax2 = fig.add_subplot(3,1,2) ax2.set_ylabel("vel") # 縦軸のタイトル ax2.plot(time_his, vel_his[1], label="1F", color="darkturquoise") # 1階のグラフ ax2.plot(time_his, vel_his[0], label="2F", color='limegreen') # 2階のグラフ ax2.legend(loc='lower right') # 凡例の位置 ax2.grid(True) # グリッド ax3 = fig.add_subplot(3,1,3) ax3.set_xlabel("time") # 横軸のタイトル ax3.set_ylabel("dis") # 縦軸のタイトル ax3.plot(time_his, dis_his[1], label="1F", color="darkturquoise") # 1階のグラフ ax3.plot(time_his, dis_his[0], label="2F", color='limegreen') # 2階のグラフ ax3.legend(loc='lower right') # 凡例の位置 ax3.grid(True) # グリッド fig.canvas.draw() OUT = convertToBitmap(fig) DateTime.Nowノードを使ってアニメーションにするのも面白いと思います。 5-おわりに スクリプト単体では使い道がないですが、データ操作やアニメーション出力の箇所はいろいろと応用できそうな気がします。

目次 1. はじめに 2. ライブラリのインストール 3. 地震応答解析 4. グラフの出力 5. おわりに 1-はじめに 構造系ではGrassHopperが人気ですが、Dynamoでも簡単な応答解析ができないか試してみます。あえてDynamoを使うメリットはとくにないです。 2-ライブラリのインストール Pythonで使う外部ライブラリをインストールします。 1. はじめに Dynamo を起動し、下記のコードで CPython のバージョンを確認します。 PythonScript import sys OUT = sys.version 2. Anaconda をダウンロードし、Anaconda Navigator から「CMD.exe Prompt」を実行します。 3. 下記のコードを入力して Dynamo 用の環境をつくります。 3.8.3は最初に確認した CPython のバージョンの番号です。 Prompt conda create --name Dynamo383 python=3.8.3 4. Proceed ([y]/n)?と聞かれるので Y を入力します。 5. いま作成した環境に切り替えます。 Prompt conda activate Dynamo383 6. 必要なライブラリをまとめてインストールします。 Prompt pip install numpy pandas scipy Pillow matplotlib 3-地震応答解析 せん断変形のみを考慮した多質点系モデルの振動解析を実装します。 波形データは東京都建設局で公開されているCSVデータを利用します。 条件値と波形データの入力 ライブラリをインポートします。 PythonScript #ライブラリをロード sys.path.append(r'C:\Users\shota\.conda\envs\Dynamo383\Lib\site-packages') import sys, math, io import numpy as np import pandas as pd from scipy import interpolate import matplotlib.pyplot as plt from PIL import Image as Img import System.Drawing from System.Drawing import * from System.Drawing.Imaging import * from System.IO import MemoryStream コード中にIN[0]とかあると分かりにくいので入力値はInputクラスにまとめておきます。 PythonScript # インプット class Input: def __init__(self): self.g = IN[0] # 重力加速度 self.dt = IN[1] # 積分時間刻み self.beta = IN[2] # Newmarkβ法のβ self.damp_factor = IN[3] # 減衰定数 self.natural_frequency = IN[4] # 振動数 self.wave_path = IN[7] # CSVファイルのパス self.arr_m = IN[5] # 重量 self.arr_k = IN[6] # 剛性 CSV形式の波形データを読み込みます。 PythonScript # 波形データ def WaveData(input): with open(input.wave_path, 'r') as f: df = pd.read_csv(f) return list(df['acc'].values.tolist()) マトリクス生成 質量マトリクス、減衰マトリクス、剛性マトリクスをつくります。 PythonScript # 質量マトリクス def MassMatrix(arr_m, input): return np.diag(list(map(lambda x: x / input.g, arr_m))) # 剛性マトリクス def StiffMatrix(arr_k): size = len(arr_k) full = np.zeros((size, size)) for i in range(size): k = arr_k[i] part = [[k , -k], [-k, k]] for i_row in range(len(part)): for i_col in range(len(part)): target_row = i-1 + i_row target_col = i-1 + i_col if 0 <= target_row and 0 <= target_col: full[target_row][target_col] += part[i_row][i_col] return full # 減衰マトリクス def DampMatrix(k, input): omega = 2 * math.pi * input.natural_frequency return 2 * input.damp_factor * k / omega 質点系モデル作成 前項で生成したマトリクスを用いて質点系モデルをつくります。 PythonScript # 質点系モデル class Model: def __init__(self,input): arr_m = input.arr_m arr_k = input.arr_k self.size = len(arr_m) arr_k.reverse() arr_m.reverse() self.m = MassMatrix(arr_m, input) self.k = StiffMatrix(arr_k) self.c = DampMatrix(self.k, input) 解析 Newmarkβ法を用いて数値解析を行い、入力波形のステップごとに加速度・速度・変位を求めます。 PythonScript def dynamic_analysis(model, input): # 初期化 m = model.m k = model.k c = model.c acc0 = WaveData(input) unit_vector = np.ones(model.size) pre_acc0 = 0.0 time = 0.0 dis = np.zeros(model.size) vel = np.zeros(model.size) acc = np.zeros(model.size) ddis = np.zeros(model.size) dvel = np.zeros(model.size) dacc = np.zeros(model.size) dis_his = {} vel_his = {} acc_his = {} for i in range(0, model.size): dis_his[i] = [] vel_his[i] = [] acc_his[i] = [] time_his = [] # Newmarkβ法で解析 for i in range(0, len(acc0)): kbar = k + (1.0/(2.0*input.beta*input.dt)) * c + (1.0/(input.beta*input.dt**2.0)) * m dp1 = -1.0 * m.dot(unit_vector) * (acc0[i] - pre_acc0) dp2 = m.dot((1.0/(input.beta*input.dt))*vel + (1.0/(2.0*input.beta))*acc) dp3 = c.dot((1.0/(2.0*input.beta))*vel + (1.0/(4.0*input.beta)-1.0)*acc*input.dt) dp = dp1 + dp2 + dp3 ddis = np.linalg.inv(kbar).dot(dp) dvel = (1.0/(2.0*input.beta*input.dt))*ddis - (1.0/(2.0*input.beta))*vel - ((1.0/(4.0*input.beta)-1.0))*acc*input.dt dacc = (1.0/(input.beta*input.dt**2.0))*ddis - (1.0/(input.beta*input.dt))*vel - (1.0/(2.0*input.beta))*acc dis += ddis vel += dvel acc += dacc acc_abs = acc + [acc0[i] for n in range(1,model.size)] [dis_his[i].append(x) for i, x in enumerate(dis)] [vel_his[i].append(x) for i, x in enumerate(vel)] [acc_his[i].append(x) for i, x in enumerate(acc_abs)] time_his.append(time) time += input.dt pre_acc0 = acc0[i] return time_his, dis_his, vel_his, acc_his 結果出力 前項で作成したdynamic_analysisを実行して結果を求めます。 PythonScript # メイン input = Input() model = Model(input) time_his, dis_his, vel_his, acc_his = dynamic_analysis(model, input) 4-グラフの出力 matplotlibでグラフを作成します。 グラフをWatch Imageノードで表示するために、System.Drawing.Bitmapに変換します。 PythonScript # 画像変換 def convertToBitmap(fig): rgba_buf = fig.canvas.buffer_rgba() (w,h) = fig.canvas.get_width_height() npImgArray = np.frombuffer(rgba_buf, dtype=np.uint8).reshape((h,w,4)) bitmap_ = None # alphaチャンネルを削除 if npImgArray.ndim == 3 and npImgArray.shape[-1] == 4: npImgArray = npImgArray[:, :, :-1] img = Img.fromarray(npImgArray, "RGB") # 画像をバイナリデータに変換し、メモリ上に保存 byteIO = io.BytesIO() img.save(byteIO, format='BMP') byteArr = byteIO.getvalue() # convert to Net ByteArray netBytes = System.Array[System.Byte](byteArr) with MemoryStream(netBytes) as ms: bitmap_ = Bitmap(ms) return bitmap_ # グラフ fig = plt.figure(figsize=(10,10)) ax1 = fig.add_subplot(3,1,1) ax1.set_ylabel("acc") # 縦軸のタイトル ax1.plot(time_his, acc_his[1], label="1F", color="darkturquoise") # 1階のグラフ ax1.plot(time_his, acc_his[0], label="2F", color='limegreen') # 2階のグラフ ax1.legend(loc='lower right') # 凡例の位置 ax1.grid(True) # グリッド ax2 = fig.add_subplot(3,1,2) ax2.set_ylabel("vel") # 縦軸のタイトル ax2.plot(time_his, vel_his[1], label="1F", color="darkturquoise") # 1階のグラフ ax2.plot(time_his, vel_his[0], label="2F", color='limegreen') # 2階のグラフ ax2.legend(loc='lower right') # 凡例の位置 ax2.grid(True) # グリッド ax3 = fig.add_subplot(3,1,3) ax3.set_xlabel("time") # 横軸のタイトル ax3.set_ylabel("dis") # 縦軸のタイトル ax3.plot(time_his, dis_his[1], label="1F", color="darkturquoise") # 1階のグラフ ax3.plot(time_his, dis_his[0], label="2F", color='limegreen') # 2階のグラフ ax3.legend(loc='lower right') # 凡例の位置 ax3.grid(True) # グリッド fig.canvas.draw() OUT = convertToBitmap(fig) DateTime.Nowノードを使ってアニメーションにするのも面白いと思います。 5-おわりに スクリプト単体では使い道がないですが、データ操作やアニメーション出力の箇所はいろいろと応用できそうな気がします。