はじめに
著者はこれまでシミュレーションコードを自作してきましたが，最近商用CAEソフトウェアを使う機会に恵まれました．ポストプロセスの過程で，すべての結果からある高さの断面の物理量を取り出す必要が生じたので，ユーザスクリプトを作成することにしました．ポストプロセッサが標準でサポートしているスクリプト用言語はVBScriptだったのですが，一応Pythonインタフェースも用意されていました．
いつも通り，一通りの処理をするスクリプトを作成した後，改良している最中に，「このスクリプトの改良過程を残しておくと後々役立つかも」と思ったので，残しておくことにしました．
環境
- scFLOWpost 2023.1
- Python 3.12.3 (Anaconda 24.9.2)
作業内容
やりたいことは，上でも書きましたが，すべての結果からある高さの断面の物理量を取り出すことです．
結果は，.fphファイルに出力され，ファイル名には0から100刻みで100000まで番号がふられています．GUIで作業をするときは，結果ファイルの一つをポストプロセスで開き，ある高さに切断面とよばれる面を設け，そこで表示する物理量を選択し，その状態をファイルに出力します．このとき出力されるファイルには，変数出力ファイルという名前が付けられています．
この作業をスクリプトで再現するには，下記の手順を踏むことになります．ここで出力したい変数は，VOF関数値（液相の体積率）です．
- ポストプロセッサを起動する
- ファイルを一つ読み込む
- 切断面を設定する
- 切断面に表示する変数を選択する
- 表示設定を行う（切断面の表か裏か，両面に表示するか）
- 変数出力ファイルとして保存する
- ファイルを閉じる
- ポストプロセッサを終了する
この流れを，ファイル名の連番を変更しながら実行していきます．
ユーザスクリプトの作成
一つのファイルを読むスクリプト
いきなり全部のファイルを対象とせず，一つのファイルに対して上記手順を実行するスクリプトを作成します．これはVBScriptからの移植でもあるので，Python的でないところがあちこちに存在しますが，それは気にしないでください．
単一ファイルを読み込んで切断面のVOF関数値を保存するスクリプト
import win32com.client
from win32com.client import VARIANT
import pythoncom
# scFLOWpostを起動してラップされたCOMオブジェクトを作成
PostApp = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")
print("scFLOWpost起動")
# fphファイルを読み込んでfieldオブジェクトを作成
filename = "結果ファイル_100000.fph"
Fld = PostApp.CreateObjectFLD(
"D:\\path\\to\\" + filename
)
if Fld is not None:
print(f"読み込み: {filename}")
# 切断面を取得もしくは作成
CutPlane = Fld.GetObjectByType("PLANE", 1)
print("デフォルトの切断面取得")
if CutPlane is None:
Fld._FlagAsMethod("CreateObjectCutplane")
CutPlane = Fld.CreateObjectCutplane()
print("デフォルトの切断面がなかったので新規作成")
# 切断面の位置，変数，面塗の設定
success = CutPlane.SetPosition(0.0, 0.0, 1.0, -2.0) # aX+bY+cZ+D=0の平面
print(f"切断面位置設定 success: {success}")
success = CutPlane.SetScalarVariable("VOF") # スカラ量としてVOF関数値を設定
print(f"表示変数設定 success: {success}")
success = CutPlane.SetScalarFillDisplay(3) # 両面を塗りつぶす
print(f"面の塗りつぶし設定 success: {success}")
# 出力ファイルの設定
varout_filename = "vof_z"
output_filetype = "xml" # xml形式で出力
output_items = ["coords", "scalar"] # 座標値とスカラ量を出力
output_filename = (
"D:\\path\\to\\"
+ varout_filename
+ "."
+ output_filetype
)
# VBScriptのVARIANT型配列（参照渡し）に変換
output_items = VARIANT(
pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT, output_items
)
# 切断面を出力
success = PostApp.SaveVariableOutput(output_filename, output_items)
print(f"変数出力ファイル{output_filename}を出力: {success}")
# scFLOWpsot終了
del CutPlane
del Fld
PostApp._FlagAsMethod("Quit")
PostApp.Quit()
del PostApp
print("scFLOWpost終了")
VBScriptに対して用意されたPythonインタフェースを用いているので，独特の書き方や制約があります．重要なのは，以下の項目くらいでしょうか．
-
PostApp = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")でポストプロセッサを起動し，PostApp.Quit()でポストプロセッサを終了する -
Fld = PostApp.CreateObjectFLD("D:\\path\\to\\" + filename)で結果ファイル（fphファイル）を読み込み，del Fldでファイルを閉じる- ファイルを閉じるメソッドがあるかもしれませんが，リファレンスを見る限り見当たりません
-
PostApp.SaveVariableOutput(output_filename, output_items)のoutput_itemsで出力する項目を指定し，output_filenameの拡張子で出力されるファイル拡張子が決まる -
VARIANT(pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT, output_items)ではPythonのListをVBScriptの配列に変換しており，VARIANT型で参照渡しされる配列に変換する- あくまで，
SaveVariableOutput関数にVARIANT型配列を参照渡ししないといけないためであり，関数ごとにリファレンスを確認する必要がある
- あくまで，
- 引数のないメソッドは，
Fld._FlagAsMethod()でメソッドであることを設定する必要がある- これをしない場合，メソッドではなく論理型変数と認識され，論理型変数を呼ぼうとしているというエラーがでる
この実装で一つのファイルを読んで切断面を出力できるのですが，スクリプト実行中に何らかの問題が起こってスクリプトが中断されたり，ユーザが強制終了すると，ポストプロセッサのプロセスがゾンビとなって残り続けます．商用CAEソフトウェアでは，ソフトウェアを起動できる数（や使えるスレッド・プロセス数）はライセンスで制限されており，使いもしないライセンスを握り続けるのは大ひんしゅくです．再発防止に問題解決，横展，社長報告などで3日は仕事ができなくなるでしょう．
そこで，スクリプトが中断されたときには，ポストプロセッサが閉じられるようにします．
try-except-finallyの導入
try-except-finallyを導入し，中断が発生したときにはポストプロセッサが閉じられるように改良したスクリプトを示します．中断時点ではCutPlaneやFldオブジェクトが存在しない可能性があるので，delもtry-exceptで囲んでいます．PostAppについては何もしていませんが，問題なのは起動しているポストプロセッサがゾンビプロセスになることなので，PostAppが存在しないということは，そもそもポストプロセッサが起動されていないことを意味するので，特に例外処理はなくてもよいでしょう．（よくないですが）
中断に対して安全にしたスクリプト
import win32com.client
from win32com.client import VARIANT
import pythoncom
try:
# scFLOWpostを起動してラップされたCOMオブジェクトを作成
PostApp = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")
print("scFLOWpost起動")
# fphファイルを読み込んでfieldオブジェクトを作成
filename = "結果ファイル_100000.fph"
Fld = PostApp.CreateObjectFLD(
"D:\\path\\to\\" + filename
)
if Fld is not None:
print(f"読み込み: {filename}")
# 切断面を取得もしくは作成
CutPlane = Fld.GetObjectByType("PLANE", 1)
print("デフォルトの切断面取得")
if CutPlane is None:
Fld._FlagAsMethod("CreateObjectCutplane")
CutPlane = Fld.CreateObjectCutplane()
print("デフォルトの切断面がなかったので新規作成")
# 切断面の位置，変数，面塗の設定
success = CutPlane.SetPosition(0.0, 0.0, 1.0, -2.0)
print(f"切断面位置設定 success: {success}")
success = CutPlane.SetScalarVariable("VOF")
print(f"表示変数設定 success: {success}")
success = CutPlane.SetScalarFillDisplay(3) # 裏表を塗りつぶす
print(f"面の塗りつぶし設定 success: {success}")
# 出力ファイルの設定
varout_filename = "vof_z"
output_filetype = "xml" # xml形式で出力
output_items = ["coords", "scalar"] # 座標値とスカラ量を出力
output_filename = (
"D:\\path\\to\\"
+ varout_filename
+ "."
+ output_filetype
)
# VBScriptのVARIANT型配列（参照渡し）に変換
output_items = VARIANT(
pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT, output_items
)
# 切断面を出力
success = PostApp.SaveVariableOutput(output_filename, output_items)
print(f"変数出力ファイル{output_filename}を出力: {success}")
except Exception as e:
print("Exception:")
print(e)
finally:
# scFLOWpsot終了
try:
del CutPlane
except NameError:
pass
try:
del Fld
except NameError:
pass
PostApp._FlagAsMethod("Quit")
PostApp.Quit()
del PostApp
print("scFLOWpost終了")
複数のファイルを順番に処理するスクリプト
単一のファイルに対する処理はこんな感じで良さそうなので，次は複数のファイルに対して処理をするように，全体をforで囲みます．
0から100000まで結果ファイルを読んで切断面を出力するスクリプト
import win32com.client
from win32com.client import VARIANT
import pythoncom
import numpy as np
step_begin = 0
step_end = 100000
step_stride = 100
num_files = (step_end - step_begin) // step_stride + 1
serial_number = np.arange(step_begin, step_end + 1, step_stride)
for num in serial_number:
try:
# scFLOWpostを起動してラップされたCOMオブジェクトを作成
PostApp = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")
print("scFLOWpost起動")
# fphファイルを読み込んでfieldオブジェクトを作成
filename = f"結果ファイル_{num}.fph"
Fld = PostApp.CreateObjectFLD(
"D:\\path\\to\\" + filename
)
if Fld is not None:
print(f"読み込み: {filename}")
# 切断面を取得もしくは作成
CutPlane = Fld.GetObjectByType("PLANE", 1)
print("デフォルトの切断面取得")
if CutPlane is None:
Fld._FlagAsMethod("CreateObjectCutplane")
CutPlane = Fld.CreateObjectCutplane()
print("デフォルトの切断面がなかったので新規作成")
# 切断面の位置，変数，面塗の設定
success = CutPlane.SetPosition(0.0, 0.0, 1.0, -2.0)
print(f"切断面位置設定 success: {success}")
success = CutPlane.SetScalarVariable("VOF")
print(f"表示変数設定 success: {success}")
success = CutPlane.SetScalarFillDisplay(3) # 裏表を塗りつぶす
print(f"面の塗りつぶし設定 success: {success}")
# 出力ファイルの設定
varout_filename = f"vof_z_{num}"
output_filetype = "xml" # xml形式で出力
output_items = ["coords", "scalar"] # 座標値とスカラ量を出力
output_filename = (
"D:\\path\\to\\"
+ varout_filename
+ "."
+ output_filetype
)
# VBScriptのVARIANT型配列（参照渡し）に変換
output_items = VARIANT(
pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT, output_items
)
# 切断面を出力
success = PostApp.SaveVariableOutput(output_filename, output_items)
print(f"変数出力ファイル{output_filename}を出力: {success}")
except Exception as e:
print("Exception:")
print(e)
finally:
# scFLOWpsot終了
try:
del CutPlane
except NameError:
pass
try:
del Fld
except NameError:
pass
PostApp._FlagAsMethod("Quit")
PostApp.Quit()
del PostApp
print("scFLOWpost終了")
これで番号を0から10000まで100ずつ変化させてファイルを読み込み，切断面の値を出力するスクリプトが完成しました．
このスクリプトを実行すると，画面に実行状況が表示されていきます．どうせならプログレスバーを出したり，見た目にもこだわりたいので，進行状況を表示するtqdmを使ってみます．
tqdmの簡単な使用例は下記の通りで，これを実行すると，進行状況に応じてプログレスバーが伸びていくので，どのくらい処理が進んでいるかを直感的に把握できます．
from tqdm import tqdm
import time
def process_with_tqdm(items):
for item in tqdm(items, desc="Processing"):
# 処理を実行
time.sleep(0.1)
if __name__ == "__main__":
items = list(range(20))
process_with_tqdm(items)
Processing: 100%|██████████| 20/20 [00:02<00:00, 9.86it/s]
どうやらfor num in serial_number:のところを変更するだけでよさそうです．また，出力されるメッセージは，tqdmのオブジェクトを作って，set_postfix_strで表示できるようです．tqdmを導入したスクリプトは次のようになりました．
tqdmで進捗状況を表示するようにしたスクリプト
import win32com.client
from win32com.client import VARIANT
import pythoncom
import numpy as np
from tqdm import tqdm
step_begin = 0
step_end = 100000
step_stride = 100
num_files = (step_end - step_begin) // step_stride + 1
serial_number = np.arange(step_begin, step_end + 1, step_stride) # exclusive ending
pbar = tqdm(serial_number)
for num in pbar:
try:
# scFLOWpostを起動してラップされたCOMオブジェクトを作成
PostApp = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")
pbar.set_postfix_str("scFLOWpost起動")
# fphファイルを読み込んでfieldオブジェクトを作成
filename = f"結果ファイル_{num}.fph"
Fld = PostApp.CreateObjectFLD(
"D:\\path\\to\\" + filename
)
if Fld is not None:
pbar.set_postfix_str(f"読み込み: {filename}")
# 切断面を取得もしくは作成
CutPlane = Fld.GetObjectByType("PLANE", 1)
pbar.set_postfix_str("デフォルトの切断面取得")
if CutPlane is None:
Fld._FlagAsMethod("CreateObjectCutplane")
CutPlane = Fld.CreateObjectCutplane()
pbar.set_postfix_str("デフォルトの切断面がなかったので新規作成")
# 切断面の位置，変数，面塗の設定
success = CutPlane.SetPosition(0.0, 0.0, 1.0, -2.0)
pbar.set_postfix_str(f"切断面位置設定 success: {success}")
success = CutPlane.SetScalarVariable("VOF")
pbar.set_postfix_str(f"表示変数設定 success: {success}")
success = CutPlane.SetScalarFillDisplay(3) # 裏表を塗りつぶす
pbar.set_postfix_str(f"面の塗りつぶし設定 success: {success}")
# 出力ファイルの設定
varout_filename = f"vof_z_{num}"
output_filetype = "xml" # xml形式で出力
output_items = ["coords", "scalar"] # 座標値とスカラ量を出力
output_filename = (
"D:\\path\\to\\"
+ varout_filename
+ "."
+ output_filetype
)
# VBScriptのVARIANT型配列（参照渡し）に変換
output_items = VARIANT(
pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT, output_items
)
# 切断面を出力
success = PostApp.SaveVariableOutput(output_filename, output_items)
pbar.set_postfix_str(f"変数出力ファイル{output_filename}を出力: {success}")
except Exception as e:
print("Exception:")
print(e)
finally:
# scFLOWpsot終了
try:
del CutPlane
except NameError:
pass
try:
del Fld
except NameError:
pass
PostApp._FlagAsMethod("Quit")
PostApp.Quit()
del PostApp
pbar.set_postfix_str("scFLOWpost終了")
step_endを減らして実行してみた結果が下記の通りです．いい感じです．
73%|█████████████████████████████████████████████████████████▍ | 8/11 [01:04<00:24, 8.13s/it, scFLOWpost終了]
ユーザスクリプトの大幅な改良
さて，これで一通り所望の作業を行うスクリプトは完成しました．しかし，色々と問題があります．例えば，異なる高さの切断面を出力したときに，切断面の式をファイル名に付けたくなったとしても，今の実装ではかなり大変です．他にも，変数出力ファイルの指定内容に間違いがないかを事前に確認することができず，実行してからエラーが出て初めて気づきます．
他にも，切断面の設定をもっとすっきりとさせたいとか，細かい作業はログに出力して，tqdmの進行状況には大まかな作業を表示したいとか，色々と改良したいところが出てきます．
変数出力ファイルの指定内容の正誤確認
まずは，変数出力ファイルの指定内容に間違いがないかをチェックする機能を作ります．まずは利用可能な項目をリストAVAILABLE_VARIABLEOUTPUT_ITEMSとして定義しておき，validate_items関数内で，AVAILABLE_VARIABLEOUTPUT_ITEMSに列挙されていない項目が引数内にないかを確認します．また，SaveVariableOutputの仕様として，出力項目にallが渡されるとすべての項目を出力するので，allが渡された場合は有効であることを示す結果を返します．
# 全ての利用可能な項目を定義
AVAILABLE_VARIABLEOUTPUT_ITEMS = {
"title": "オブジェクト名",
"coords": "座標",
"normal": "法線ベクトル",
"scalar": "スカラー変数値",
"vector": "ベクトル変数値",
"elem": "要素番号",
"node": "節点番号",
"rank": "スレッド識別番号",
}
# 項目の検証機能
def validate_items(items):
"""
指定された項目が有効かどうかをチェック
Args:
items (list): チェックする項目のリスト
Returns:
tuple: (bool, str) - (有効かどうか, エラーメッセージ)
"""
if items == "all":
return True, ""
if not isinstance(items, list):
return False, "変数出力ファイルの項目はリストで指定する必要があります"
invalid_items = [
item for item in items if item not in AVAILABLE_VARIABLEOUTPUT_ITEMS
]
if invalid_items:
return False, f"無効な変数出力項目が含まれています: {invalid_items}"
return True, ""
変数出力ファイルの指定内容の確認は，下記のように実行します．
output_items = ["coords", "scalar"]
is_valid, error_message = validate_items(output_items)
if not is_valid:
# ここで例外をraise
固有の例外の追加
変数出力ファイルの内容が有効でないという例外は，ポストプロセッサ固有の問題ですから，新たに例外を定義することにします．
class ScFLOWpostProcessingError(Exception):
"""scFLOWpostに関する独自の例外クラス"""
pass
この例外を使えば，変数出力ファイルの指定内容に誤りがあったときに，ポストプロセッサ固有の例外と共に，その内容を確認できます．
output_items = ["coords", "elements"]
is_valid, error_message = validate_items(output_items)
if not is_valid:
raise ScFLOWpostProcessingError(error_message)
Traceback (most recent call last):
File "D:\path\to\script.py", line xxx, in <module>
raise ScFLOWpostProcessingError(error_message)
ScFLOWpostProcessingError: 無効な変数出力項目が含まれています: ['elements']
切断面を表す型の追加
次に，切断面の情報を扱いやすくするために，切断面（3次元空間における平面）の情報を取り扱うクラスを定義します．このクラスの主な役割は，切断面の式を文字列に変換することなので，__str__以外はほとんど実装していません．
class Plane:
"""平面の方程式 ax + by + cz + d = 0 を表現するクラス"""
def __init__(self, a: float, b: float, c: float, d: float):
"""
平面の方程式の係数を初期化
Parameters
----------
a : float
x係数
b : float
y係数
c : float
z係数
d : float
定数項
"""
self.a = float(a) # 数値型に変換
self.b = float(b)
self.c = float(c)
self.d = float(d)
def _format_term(self, coef: float, var: str = "") -> str:
"""係数と変数名を用いて項を整形する補助関数"""
if coef == 0:
return ""
# 係数の絶対値が1の場合は係数を表示しない
coef_str = "" if abs(coef) == 1 and var else str(abs(coef))
return f"{coef_str}{var}"
def __str__(self) -> str:
"""
平面の方程式を文字列として返す
Returns
-------
str
ax + by + cz + d = 0 の形式の文字列
"""
terms = [(self.a, "x"), (self.b, "y"), (self.c, "z"), (self.d, "")]
# 0でない項を文字列に変換
formatted = [self._format_term(coef, var) for coef, var in terms if coef != 0]
if not formatted:
return "0=0"
# 初項が負の場合はマイナス記号を付ける
result = f"-{formatted[0]}" if terms[0][0] < 0 else formatted[0]
# 2項目以降を符号付で追加
for i, term in enumerate(formatted[1:], 1):
coef = terms[i][0]
result += f"+{term}" if coef > 0 else f"-{term}"
# 右辺（=0）をつけて返す
return f"{result}=0"
切断面を表す型は，以下のように使います．
plane = Plane(0, 0, 1, -2) # 0x+0y+z-2=0の平面
SetPosition(plane.a, plane.b, plane.c, plane.d)
print(plane) # "z-2.0=0"を返す
ログの設定
切断面出力のために行うそれぞれの手順は，さほど時間のかかる作業ではありません．そのため，tqdmに作業内容を出力していると，出力が頻繁に切り替わっｔ目がチカチカするので，ログへ出力することにします．tqdmの出力には，現在どのファイルを処理しているかがわかればよいでしょう．ログ出力を設定し，set_postfix_strに出力していた作業内容をすべてログファイルに出力するように設定します．
import logging
# ロギングの設定
logging.basicConfig(
level=logging.INFO,
encoding="utf-8",
format="%(asctime)s - %(levelname)s - %(message)s",
filename="scflowpost_processing.log",
)
全体の処理の再考
細かい機能の追加をしたので，全体の処理の流れを考え直してみます．
とりあえず，スクリプトに処理をべた書きするのは止めて，mainメソッドとしてまとめます．mainメソッドも，例外を捕まえるようにしておきます．
import numpy as np
from pathlib import Path
from tqdm import tqdm
def main():
"""メイン処理"""
# パラメータ設定
TASK_DESCRIPTION = "切断面の抽出"
# 必ずフルパス
BASE_PATH = Path(
"D:/path/to"
) # ここのセパレータは`/`
BASE_NAME = "結果ファイル_"
STEP_BEGIN = 0
STEP_END = 100000
STEP_STRIDE = 100
# 保存する切断面（ax+by+cz+d=0の形式）
PLANE = Plane(0, 0, 1, -2)
# 保存する物理量と項目の設定
OUTPUT_VARIABLE = "VOF"
OUTPUT_ITEMS = ["coords", "scalar"]
# 作業ディレクトリの確認
if not BASE_PATH.exists():
raise ScFLOWpostProcessingError(f"作業ディレクトリが存在しません: {BASE_PATH}")
# 処理対象のファイル番号を生成
serial_numbers = np.arange(STEP_BEGIN, STEP_END + STEP_STRIDE, STEP_STRIDE)
# メイン処理ループ
pbar = tqdm(serial_numbers, desc=TASK_DESCRIPTION)
for number in pbar:
try:
pbar.set_postfix_str(f"Processing file number {number}")
# ここで1ファイルを読み込んで切断面を出力する処理を書く
except Exception as e:
logging.error(f"ファイル {BASE_NAME}{number}.fph の処理に失敗: {e}")
continue
if __name__ == "__main__":
try:
main()
except Exception as e:
logging.critical(f"プログラムが異常終了しました: {e}")
raise
# ここで1ファイルを読み込んで切断面を出力する処理を書くと書いてある箇所でメソッドを呼び出せばよいわけですが，Pythonの機能を導入することで，リソース管理を効率的に行えるようにします．
コンテキストマネージャの導入
上で作成したユーザスクリプトでは，ポストプロセッサを起動した結果作られるオブジェクトPostAppがとても重要でした．この管理を誤ってしまうと，3日は仕事ができなくなります．そこで，リソースの確保と解放を自動的に行えるコンテキストマネージャを導入することで，安全で効率的なスクリプトにできます．
import win32com.client
from contextlib import contextmanager
@contextmanager
def create_post_app():
"""PostAppのコンテキストマネージャ"""
post_app = None
try:
post_app = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")
logging.info("scFLOWpostを起動しました")
yield post_app
except Exception as e:
logging.error(f"PostAppの作成中にエラーが発生: {e}")
raise ScFLOWpostProcessingError(f"PostApp作成エラー: {e}")
finally:
if post_app is not None:
try:
post_app._FlagAsMethod("Quit")
post_app.Quit()
del post_app
logging.info("scFLOWpostを終了しました")
except Exception as e:
logging.error(f"PostAppの終了中にエラーが発生: {e}")
PostAppの作成，解放を管理するコンテキストマネージャを作成し，wtihステートメントと併用するとこで，スクリプトを非常に簡潔に書けるようになります．
for number in pbar:
try:
with create_post_app() as post_app:
pbar.set_postfix_str(f"Processing file number {number}")
# ここで1ファイルを読み込んで切断面を出力する処理を書く
with create_post_app() as post_app:のうち，as post_appはpost_appがcreate_post_app()の戻り値であることを示しています．post_appはwithステートメントの中で使用でき，withステートメントを抜ける際に，finallyの処理が自動で呼ばれて解放されます．このようにwithステートメントとコンテキストマネージャを使って生成したpost_appは，# ここで1ファイルを読み込んで切断面を出力する処理を書くで行った処理が正常に終了しても，異常終了してもきちんと解放されるので，ゾンビプロセスを作って激詰めされる心配が無くなります．
1ファイルの処理
最後に，1ファイルを読み込んで切断面を出力する処理を整理します．メソッドの名前を，安直ですがprocess_fileとしておくと，post_appはprocess_fileへ引数として渡すことになります．他にも，処理するファイルの番号や平面の式，出力項目や名前なども引数として渡します．
from win32com.client import VARIANT
import pythoncom
def process_file(
post_app, file_number, plane, output_variable, output_items, base_path, base_name
):
"""1ファイルを読み込んで切断面を出力する"""
try:
# ファイルパスの構築
fph_filename = base_name + f"{file_number}.fph"
file_path = str(Path(base_path) / fph_filename) # str()が必要
# FLDオブジェクトの作成
fld = post_app.CreateObjectFLD(file_path)
if fld is None:
raise ScFLOWpostProcessingError(f"ファイルを開けません: {fph_filename}")
try:
# 切断面の作成と設定
cut_plane = create_cut_plane(fld)
# 切断面の設定
settings = [
(cut_plane.SetPosition, (plane.a, plane.b, plane.c, plane.d)),
(
cut_plane.SetScalarVariable,
(output_variable,),
), # output_variable, のカンマは単一要素のタプルを作るために必要
(cut_plane.SetScalarFillDisplay, (3,)), # 両面を塗りつぶす場合は3
]
for method, args in settings:
if not method(*args):
raise ScFLOWpostProcessingError(
f"切断面設定エラー: {method.__name__}"
)
# 出力ファイルの設定
OUTPUT_FILETYPE = (
"xml" # xmlとtxtを選べるが，txtだと出力項目の情報が反映されない
)
varout_filename = (
f"{output_variable}_{file_number}_{plane}.{OUTPUT_FILETYPE}"
)
file_path = str(Path(base_path) / varout_filename) # str()が必要
is_valid, error_message = validate_items(output_items)
if not is_valid:
raise ScFLOWpostProcessingError(error_message)
# *VARIANT型**配列*の*参照渡し*
output_items = VARIANT(
pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT,
output_items,
)
if not post_app.SaveVariableOutput(file_path, output_items):
raise ScFLOWpostProcessingError("変数出力の保存に失敗しました")
logging.info(f"変数出力ファイルを保存しました: {varout_filename}")
finally:
# リソースの解放
try:
del cut_plane
except NameError:
pass
try:
del fld
except NameError:
pass
except Exception as e:
logging.error(
f"ファイル {base_name}{file_number}.fph の処理中にエラーが発生: {e}"
)
raise
処理の流れに大きな変更はありませんが，切断面の作成は単一のメソッドに切り出しています．
def create_cut_plane(fld):
"""切断面の作成と検証"""
cut_plane = fld.GetObjectByType("PLANE", 1)
if cut_plane is not None:
logging.info("既存の切断面を取得しました")
return cut_plane
else:
fld._FlagAsMethod("CreateObjectCutplane")
cut_plane = fld.CreateObjectCutplane()
if cut_plane is None:
raise ScFLOWpostProcessingError("切断面の作成に失敗しました")
logging.info("新しい切断面を作成しました")
return cut_plane
また，切断面の設定は，少し技巧的な方法を使っています．メソッドと引数のリストを作り，for method, args in settings:中のmethod(*args)でリスト中のメソッドを一つずつ順番に実行し，もし結果がFalseなら例外を発生させます．この処理は必ずリストの順番に実行されます．戻り値の真偽値を判定して例外処理を何個も書くのが面倒なときに使える技巧です．
settings = [
(cut_plane.SetPosition, (plane.a, plane.b, plane.c, plane.d)),
(cut_plane.SetScalarVariable, (output_variable,), ), # output_variable, のカンマは単一要素のタプルを作るために必要
(cut_plane.SetScalarFillDisplay, (3,)), # 両面を塗りつぶす場合は3
]
for method, args in settings:
if not method(*args):
raise ScFLOWpostProcessingError(f"切断面設定エラー: {method.__name__}")
その他
コンテキストマネージャとwtihステートメントについて説明しましたが，pbar = tqdm(serial_numbers, desc=TASK_DESCRIPTION)もwithステートメントを用いてwith tqdm(serial_numbers, desc=TASK_DESCRIPTION) as pbar:と書き換えることができます．
post_appと同じように，fldやcut_planeもコンテキストマネージャを用いればリソースの解放が簡潔にはなるのですが，ちょっとやり過ぎかと思ったので，今回はfinally句に解放をべた書きしました．
まとめ
適当に書きがちなユーザスクリプトを改良しながら，色々なPythonの機能や概念を導入してみました．導入した機能・概念をまとめておきます．
- 概念を表す型を作ることで，関係する情報と処理を閉じ込める
- 取り得る値を列挙しておき，設定が有効かを確認する
- 画面出力は最低限にしてログを取る
- 進行状況を直感的に把握できるようにする
- アプリケーション固有の例外を導入する
- コンテキストマネージャと
withステートメントでリソース管理を効率化する - 戻り値が真偽値の場合，メソッドと引数をまとめたリストを作ることで記述を簡略化できる
ユーザスクリプト全体
import win32com.client
from win32com.client import VARIANT
import pythoncom
import numpy as np
from pathlib import Path
from tqdm import tqdm
import logging
from contextlib import contextmanager
# ロギングの設定
logging.basicConfig(
level=logging.INFO,
encoding="utf-8",
format="%(asctime)s - %(levelname)s - %(message)s",
filename="scflowpost_processing.log",
)
class ScFLOWpostProcessingError(Exception):
"""scFLOWpostに関する独自の例外クラス"""
pass
class Plane:
"""平面の方程式 ax + by + cz + d = 0 を表現するクラス"""
def __init__(self, a: float, b: float, c: float, d: float):
"""
平面の方程式の係数を初期化
Parameters
----------
a : float
x係数
b : float
y係数
c : float
z係数
d : float
定数項
"""
self.a = float(a) # 数値型に変換
self.b = float(b)
self.c = float(c)
self.d = float(d)
def _format_term(self, coef: float, var: str = "") -> str:
"""係数と変数名を用いて項を整形する補助関数"""
if coef == 0:
return ""
# 係数の絶対値が1の場合は係数を表示しない
coef_str = "" if abs(coef) == 1 and var else str(abs(coef))
return f"{coef_str}{var}"
def __str__(self) -> str:
"""
平面の方程式を文字列として返す
Returns
-------
str
ax + by + cz + d = 0 の形式の文字列
"""
terms = [(self.a, "x"), (self.b, "y"), (self.c, "z"), (self.d, "")]
# 0でない項を文字列に変換
formatted = [self._format_term(coef, var) for coef, var in terms if coef != 0]
if not formatted:
return "0=0"
# 初項が負の場合はマイナス記号を付ける
result = f"-{formatted[0]}" if terms[0][0] < 0 else formatted[0]
# 2項目以降を符号付で追加
for i, term in enumerate(formatted[1:], 1):
coef = terms[i][0]
result += f"+{term}" if coef > 0 else f"-{term}"
# 右辺（=0）をつけて返す
return f"{result}=0"
# 全ての利用可能な項目を定義
AVAILABLE_VARIABLEOUTPUT_ITEMS = {
"title": "オブジェクト名",
"coords": "座標",
"normal": "法線ベクトル",
"scalar": "スカラー変数値",
"vector": "ベクトル変数値",
"elem": "要素番号",
"node": "節点番号",
"rank": "スレッド識別番号",
}
# 項目の検証機能
def validate_items(items):
"""
指定された項目が有効かどうかをチェック
Args:
items (list): チェックする項目のリスト
Returns:
tuple: (bool, str) - (有効かどうか, エラーメッセージ)
"""
if items == "all":
return True, ""
if not isinstance(items, list):
return False, "変数出力ファイルの項目はリストで指定する必要があります"
invalid_items = [
item for item in items if item not in AVAILABLE_VARIABLEOUTPUT_ITEMS
]
if invalid_items:
return False, f"無効な変数出力項目が含まれています: {invalid_items}"
return True, ""
def create_cut_plane(fld):
"""切断面の作成と検証"""
cut_plane = fld.GetObjectByType("PLANE", 1)
if cut_plane is not None:
logging.info("既存の切断面を取得しました")
return cut_plane
else:
fld._FlagAsMethod("CreateObjectCutplane")
cut_plane = fld.CreateObjectCutplane()
if cut_plane is None:
raise ScFLOWpostProcessingError("切断面の作成に失敗しました")
logging.info("新しい切断面を作成しました")
return cut_plane
def process_file(
post_app, file_number, plane, output_variable, output_items, base_path, base_name
):
"""1ファイルを読み込んで切断面を出力する"""
try:
# ファイルパスの構築
fph_filename = base_name + f"{file_number}.fph"
file_path = str(Path(base_path) / fph_filename) # str()が必要
# FLDオブジェクトの作成
fld = post_app.CreateObjectFLD(file_path)
if fld is None:
raise ScFLOWpostProcessingError(f"ファイルを開けません: {fph_filename}")
try:
# 切断面の作成と設定
cut_plane = create_cut_plane(fld)
# 切断面の設定
settings = [
(cut_plane.SetPosition, (plane.a, plane.b, plane.c, plane.d)),
(
cut_plane.SetScalarVariable,
(output_variable,),
), # "VOF", のカンマは単一要素のタプルを作るために必要
(cut_plane.SetScalarFillDisplay, (3,)), # 常に3
]
for method, args in settings:
if not method(*args):
raise ScFLOWpostProcessingError(
f"切断面設定エラー: {method.__name__}"
)
# 出力ファイルの設定
OUTPUT_FILETYPE = (
"xml" # xmlとtxtを選べるが，txtだと出力項目の情報が反映されない
)
varout_filename = (
f"{output_variable}_{file_number}_{plane}.{OUTPUT_FILETYPE}"
)
file_path = str(Path(base_path) / varout_filename) # str()が必要
is_valid, error_message = validate_items(output_items)
if not is_valid:
raise ScFLOWpostProcessingError(error_message)
# *VARIANT型**配列*の*参照渡し*
output_items = VARIANT(
pythoncom.VT_ARRAY | pythoncom.VT_BYREF | pythoncom.VT_VARIANT,
output_items,
)
if not post_app.SaveVariableOutput(file_path, output_items):
raise ScFLOWpostProcessingError("変数出力の保存に失敗しました")
logging.info(f"変数出力ファイルを保存しました: {varout_filename}")
finally:
# リソースの解放
try:
del cut_plane
except NameError:
pass
try:
del fld
except NameError:
pass
except Exception as e:
logging.error(
f"ファイル {base_name}{file_number}.fph の処理中にエラーが発生: {e}"
)
raise
@contextmanager
def create_post_app():
"""PostAppのコンテキストマネージャ"""
post_app = None
try:
post_app = win32com.client.Dispatch("scFLOWpost_Sx64net.Application.2023")
logging.info("scFLOWpostを起動しました")
yield post_app
except Exception as e:
logging.error(f"PostAppの作成中にエラーが発生: {e}")
raise ScFLOWpostProcessingError(f"PostApp作成エラー: {e}")
finally:
if post_app is not None:
try:
post_app._FlagAsMethod("Quit")
post_app.Quit()
del post_app
logging.info("scFLOWpostを終了しました")
except Exception as e:
logging.error(f"PostAppの終了中にエラーが発生: {e}")
def main():
"""メイン処理"""
# パラメータ設定
TASK_DESCRIPTION = "切断面の抽出"
# 必ずフルパス
BASE_PATH = Path(
"D:/path/to"
) # ここのセパレータは`/`
BASE_NAME = "結果ファイル_"
STEP_BEGIN = 0
STEP_END = 100000
STEP_STRIDE = 100
# 保存する切断面（ax+by+cz+d=0の形式）
# z=2の面を保存する場合は0x+0y+1z=2->(0, 0, 1, -2)
PLANE = Plane(0, 0, 1, -2)
# 保存する物理量と項目の設定
OUTPUT_VARIABLE = "VOF"
OUTPUT_ITEMS = ["coords", "scalar"]
# 作業ディレクトリの確認
if not BASE_PATH.exists():
raise ScFLOWpostProcessingError(f"作業ディレクトリが存在しません: {BASE_PATH}")
# 処理対象のファイル番号を生成
serial_numbers = np.arange(STEP_BEGIN, STEP_END + STEP_STRIDE, STEP_STRIDE)
# メイン処理ループ
with tqdm(serial_numbers, desc=TASK_DESCRIPTION) as pbar:
for number in pbar:
try:
with create_post_app() as post_app:
pbar.set_postfix_str(f"Processing file number {number}")
process_file(
post_app,
number,
PLANE,
OUTPUT_VARIABLE,
OUTPUT_ITEMS,
BASE_PATH,
BASE_NAME,
)
except Exception as e:
logging.error(f"ファイル {BASE_NAME}{number}.fph の処理に失敗: {e}")
continue
if __name__ == "__main__":
try:
main()
except Exception as e:
logging.critical(f"プログラムが異常終了しました: {e}")
raise