AWSマネジメントコンソールのEC2インスタンスのページにはCPUの使用率やネットワーク入力量等の情報を最大で2週間分チャート表示する機能があります。職場の業務として2週間に1回、2週間分のチャート画面をキャプチャしてExcelファイルのレポートを作成するというタスクを行っているのですが、これをPythonを使って自動化(RPA?)してみました。
現在手動で行ってる作業について
行ってる作業を順番に書くと以下のような流れとなってます。
実際の業務内容をここには書けないので少しぼかしてます。
Excelのテンプレートも実務のものは使えないので適当なものを自作してます。
- AWSにサインイン
- EC2インスタンス一覧を開き、モニタリング対象のインスタンスのみを選択した状態で「モニタリング」タブをクリック(↓の画像の青枠部分)
- CPU使用率のチャート情報を取得
- ネットワーク入力のチャート情報を取得
- 表示期間を「過去 2週間」に変更
- PrintScreen→チャート部分を切り抜いて保存
ファイル名:netin.png - 目視でY軸の最も高い数値を示してる箇所の値と時刻を取得
チャートの頂点をマウスオーバーすると表示されるポップアップをスクリーンショットで取得
ファイル名:netin1.png - チャートを閉じる
- ネットワーク出力のチャート情報を取得
- 表示期間を「過去 2週間」に変更
- PrintScreen→チャート部分を切り抜いて保存
ファイル名:netout.png - 目視でY軸の最も高い数値を示してる箇所の値と時刻を取得
チャートの頂点をマウスオーバーすると表示されるポップアップをスクリーンショットで取得
ファイル名:netout1.png - チャートを閉じる
- Excelのレポート作成
だいたいこんな感じです。1インスタンス辺り10分くらい掛かってるかと思います。
プログラムについて
今回のプログラムは非公開のリポジトリで作ってて公開の予定もないので、以下に開発時に苦労した点をいくつか。
Chromeを起動、フルスクリーン化
ブラウザの操作の自動化は当然Seleniumを使います。
キャプチャするチャート画像はそれなりに大きいのでプログラムで自動制御するChromeはフルスクリーン化しています。
また、AWSのログイン画面はChromeのセッションを引き継ぐとログインIDを入力する画面が省略されたりして動作が不安定だったので常にゲストモードで開くようにしています。
options = webdriver.ChromeOptions() options.add_experimental_option('excludeSwitches', ['enable-logging']) # ゲストモードで起動 # ユーザーデータを設定すると、メールアドレスが入力済みだったり、ログイン中のセッションを引き継いだりしてしまうため、ゲストモードに切り替えセッションを初期化する # シークレットモード(--incognito)より--guestの方が良い?? options.add_argument('--guest') driver = webdriver.Chrome(executable_path = config['chrome_executable_path'], options = options) # フルスクリーン化 driver.maximize_window()
Seleniumの待機処理について
Seleniumではページ遷移の待機処理についてWebDriverWaitを使うのがベストかと思いますが、AWSの管理画面ではいくつかうまく行かないところがあり、(私の知識では)time.wait()によるスリープ処理を使わざるを得ない箇所がありました。
例えばEC2インスタンス一覧画面を開いてモニタリング対象のインスタンスをチェックするときのプログラムは以下のような感じ。
# 「インスタンスの作成」ボタンがクリックできる状態まで待機中 WebDriverWait(driver, 60).until(EC.element_to_be_clickable((By.ID, 'gwt-debug-button-launch-instance'))) # 選択可能なインスタンスが存在するか判定。.GB5AGWGDAFHがdisplay:noneでないとき、インスタンスがないと判定する if (driver.find_element_by_css_selector('.GB5AGWGDAFH').is_displayed()): logger.error("インスタンスが一覧画面に1件も存在しないため処理を中断します") chrome_end(driver) return # 非同期のインスタンス一覧の読み込みが原因なのかsleep入れないと以下のエラーが出る # selenium.common.exceptions.StaleElementReferenceException: Message: stale element reference: element is not attached to the page document time.sleep(2) WebDriverWait(driver, 60).until(EC.element_to_be_clickable((By.CSS_SELECTOR, '#gwt-debug-gridTable .GB5AGWGDEEG table'))) ec2_instance_list_table = driver.find_element_by_css_selector('#gwt-debug-gridTable .GB5AGWGDEEG table')
チャートの一番高い部分の取得について
今回一番苦労したところです。
画面表示されたチャートから最も高い数値を取得する方法についてはテンプレートマッチングを使用しました。
- チャートの領域をキャプチャ
- Seleniumで切り抜きたい要素を取得
- screenshot_as_pngという関数でキャプチャ
- チャートの頂点に小さい〇が表示されてるのでこれをテンプレートマッチング
- 小さい画像なので閾値をかなり厳しく設定する(0.99~↑くらい)
- n個見つかったらY座標の数値順でソート
- n=0番目の要素が一番高い座標、n=1番目の要素が二番目に高い座標となる
- ActionChainsで画像上の指定座標をマウスオーバー
- 表示されたポップアップ要素をスクレイピング
という流れになります。
cpu_png = chart_div.screenshot_as_png with open(WORK_DIR + 'cpu.png', 'wb') as f: f.write(cpu_png) # テンプレートマッチング cpu_frame = cv2.imread(WORK_DIR + 'cpu.png', cv2.IMREAD_GRAYSCALE) cpu_frame = cpu_frame[CHART_ROI[1]:CHART_ROI[3], CHART_ROI[0]:CHART_ROI[2]] res = cv2.matchTemplate(cpu_frame, BINARY_POINT, cv2.TM_CCORR_NORMED) locs = np.where(res >= 0.99) locs = zip(*locs[::-1]) locs = sorted(locs, key = lambda x: x[1]) cpu1_location = cpu2_location = None if (len(locs) <= 0): chrome_end(driver) return elif (len(locs) >= 2): cpu1_location = locs[0] cpu2_location = locs[1] else: cpu1_location = locs[0] # CPU1の座標にカーソルを持っていきマウスオーバー時に表示されるポップアップ要素を取得 if cpu1_location is not None: point_x = cpu1_location[0] point_y = CHART_ROI[1] + cpu1_location[1] actions = ActionChains(driver) actions.move_to_element(chart_div) actions.move_by_offset(-(chart_div.size['width'] / 2), -(chart_div.size['height'] / 2)) actions.move_by_offset(point_x, point_y).perform() time.sleep(0.5) popup_element = driver.find_element_by_css_selector(".gwt-PopupPanel.dataPointPopup table") if (popup_element): cpu1_value = popup_element.find_element_by_css_selector("tr:nth-of-type(1) .gwt-Label").text cpu1_date = popup_element.find_element_by_css_selector("tr:nth-of-type(2) .gwt-Label").text result_dict['cpu1_value'] = cpu1_value result_dict['cpu1_date'] = cpu1_date
リソースファイルについて
テンプレートマッチングで使用する画像や、取得したデータをExcelのレポートにまとめる際に使用するテンプレートファイルについて実行ファイル内に含めるべくnpzファイル化しました。
画像ファイルは先日作成したプログラムの方でも実施してましたが、Excelファイルのようなバイナリファイルも以下のような形で含めることができました。
import numpy as np import os import cv2 SAMPLE_DIR = 'sample_data' + os.sep EXCEL_TEMPLATE_DIR = 'templates' + os.sep EXCEL_TEMPLATE_FILE = EXCEL_TEMPLATE_DIR + 'result_template.xlsx' NPZ_FILE = 'resources' + os.sep + 'bundle.npz' # チャートの頂点のサンプルデータ BINARY_POINT = cv2.imread(SAMPLE_DIR + 'point.png', cv2.IMREAD_GRAYSCALE) # Excelテンプレートデータ EXCEL_TEMPLATE_DATA = np.fromfile(EXCEL_TEMPLATE_FILE, dtype = 'int8') # npz保存 np.savez_compressed(NPZ_FILE, point = BINARY_POINT, excel_template = EXCEL_TEMPLATE_DATA )
Excelファイルの編集について
PythonによるExcelファイルの編集はopenpyxlで行っています。
テンプレートはCPU、NETIN、NETOUTという名前のシートがある単純なExcelファイルです。
キャプチャした画像を添付したり、テンプレートマッチングで求めた数値を所定位置のセルに書き込んだりしてます。
添付した画像の外枠に線を引きたかったですが、少し調べた感じではわかりませんでした。。
work_file = WORK_DIR + 'result.xlsx' shutil.copy(EXCEL_TEMPLATE_FILE, work_file) logger.debug("Excelファイルを開く") wb = openpyxl.load_workbook(work_file) logger.debug("CPUシートを編集") ws = wb["CPU"] # 画像の枠線を引く方法がわからない(できないのかも?) ws.add_image(openpyxl.drawing.image.Image(WORK_DIR + 'cpu.png'), 'B3') ws['Q3'] = result_dict['cpu1_date'] ws['Q4'] = result_dict['cpu1_value'] ws['Q6'] = result_dict['cpu2_date'] ws['Q7'] = result_dict['cpu2_value'] ws = wb["NETIN"] ws.add_image(openpyxl.drawing.image.Image(WORK_DIR + 'netin.png'), 'B3') ws['Q3'] = result_dict['netin1_date'] ws['Q4'] = result_dict['netin1_value'] ws = wb["NETOUT"] ws.add_image(openpyxl.drawing.image.Image(WORK_DIR + 'netout.png'), 'B3') ws['Q3'] = result_dict['netout1_date'] ws['Q4'] = result_dict['netout1_value'] wb.save(work_file)
ロギングについて
今回は過剰なくらいロギングしました。
特にActionChainsの座標なんかは実際に目に見えるマウスカーソルが移動するわけではないので念入りに出力しています。
全体的にはinfoログで大まかなロギングをしてdebugログで詳細な出力を行うようにしています。
以下のような感じでプログラム実行時にclickによるコマンドラインパーサーでdebugオプションを指定したかどうかでログレベルの制御を行っています。
@click.command(context_settings = dict(help_option_names = ['-h', '--help'])) @click.option('--debug', is_flag = True, help = "debugログを出力します") @click.option('--config-name', default = 'default', help = "コンフィグ名を指定します(任意)") def main(debug, config_name): logger.info("aws-ec2-monitor-rpa start.") if debug: logzero.loglevel(logging.DEBUG) logger.info("workディレクトリに本日分の作業フォルダ作成") if not os.path.exists(WORK_DIR): os.makedirs(WORK_DIR) logger.info("コンフィグ取得") config = get_config(config_name)
まとめ
Pythonで単純作業を自動化できました。
処理時間は全体で1分くらいなので手作業でやってた頃の1/10になりました。
ちなみにこれRPA(Robotic Process Automation)って呼んで良いんですかね?RPA自体使ったことがないのでよくわかってない(^^;