FastAPI可以說是這幾年發展最快速，最受歡迎的Python API FrameWork。

今天我們來試試ANSA API是不是能跟FastAPI擦出點火花呢？

我們的核心概念，是希望能夠從Python使用subprocess呼叫ANSA的command line，進而執行一個FastAPI app，裡面有我們想使用的與ANSA操作相關的endpoint。

虛擬環境venv

建立一個Python虛擬環境venv，並於啟動後安裝requirements.txt內package。

echo -e "fastapi\nuvicorn" >> requirements.txt 
python3 -m venv venv 
source venv/bin/activate 
pip install -r requirements.txt 

env.py

大家需要根據ANSA的安裝環境，調整ANSA_ENV及ANSAPATH。

ANSA_ENV = '/home/user/BETA_CAE_Systems/shared_v23.0.0/python/linux64/python3.8/'
ANSAPATH = '/home/user/BETA_CAE_Systems/ansa_v23.0.0/ansa64.sh'
API_PORT = 8104

server.py

code應該滿直觀的，組合需要的command後，呼叫subprocess.Popen執行。

import subprocess
from pathlib import Path

from env import ANSAPATH


def run():
    api_file = (Path(__file__).parent / 'api.py').as_posix()
    cmd = []
    cmd.append(Path(ANSAPATH).as_posix())
    cmd.append('-nogui')
    cmd.append('-exec')
    cmd.append(f'"load_script: {api_file}"')
    cmd = ' '.join(str(entry) for entry in cmd)
    cp = subprocess.Popen(cmd, stdin=subprocess.PIPE, shell=True)
    print(f'{cp=}')


if __name__ == '__main__':
    run()

schemas.py

• NodeEntityModel為一個pydantic model，方便我們建立node Entity。
• LSDYNAType是老朋友了啊。

#schemas.py
from enum import Enum

from pydantic import BaseModel


class NodeEntityModel(BaseModel):
    X: float
    Y: float
    Z: float


class LSDYNAType(str, Enum):
    ALL: str = '__ALL_ENTITIES__'
    PART: str = 'ANSAPART'
    NODE: str = 'NODE'
    ELEMENT: str = '__ELEMENTS__'
    SOLID: str = 'ELEMENT_SOLID'
    SHELL: str = 'ELEMENT_SHELL'
    PROPERTY: str = '__PROPERTIES__'
    MATERIAL: str = '__MATERIALS__'
    SET: str = 'SET'
    CONTACT: str = 'CONTACT'
    SEGMENT: str = 'SEGMENT'

api.py

準備工作

• 記得一開頭就要記得導入ANSA_ENV，並使用sys.path.append將其添加到搜尋路徑。
• 建立app = FastAPI()。

import asyncio
import functools
import sys
from concurrent.futures import ThreadPoolExecutor
from random import randint

import ansa
import uvicorn
from ansa import base, constants
from fastapi import FastAPI

from env import ANSA_ENV, API_PORT
sys.path.append(ANSA_ENV)
from schemas import LSDYNAType, NodeEntityModel


app = FastAPI()
deck = constants.LSDYNA

Helper Functions

get_entity_card_values

get_entity_card_values回傳該Entity的card value。

#api.py
def get_entity_card_values(entity, fields=None, deck=None):
    deck = deck or constants.LSDYNA
    fields = fields or entity.card_fields(deck)
    return entity.get_entity_values(deck, fields=fields)

_create_node

_create_node回傳該node Entity的card value。

#api.py
def _create_node(node: NodeEntityModel):
    ent = ansa.base.CreateEntity(deck, LSDYNAType.NODE, node.dict())
    return get_entity_card_values(ent)

_get_nodes

_get_nodes回傳當下ANSA環境內的所有node Entity。

#api.py
def _get_nodes():
    return ansa.base.CollectEntities(deck, None, LSDYNAType.NODE)

_get_all_ents

_get_nodes回傳當下ANSA環境內的所有Entity。

#api.py
def _get_all_ents():
    return ansa.base.CollectEntities(deck, None, LSDYNAType.ALL)

Endpoints

home

home作為一個測試的endpoint，方便我們快速判斷FastAPI有沒有啟動成功。
此處的deck為ansa.constants.LSDYNA，其值為2。如果啟動FastAPI能在首頁看到{"Hello":"current deck : deck=2"}，代表我們已經成功與ANSA對接了。

#api.py
@app.get('/', status_code=200)
async def home():
    return {'Hello': f'current deck : {deck=}'}

create_node

create_node作為_create_node的endpoint，藉由post method建立node Entity。

#api.py
@app.post('/create-node', status_code=201)
async def create_node(node: NodeEntityModel):
    return _create_node(node)

create_nodes

create_nodes會利用建立asyncio+concurrent.futures.ThreadPoolExecutor建立 n個random座標的node Entity。

#api.py
@app.post('/create-nodes', status_code=201)
async def create_nodes(n: int = 100):
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        tasks = [loop.run_in_executor(pool,
                                      functools.partial(_create_node,
                                                        NodeEntityModel(**{'X': randint(0, 10),
                                                                           'Y': randint(0, 10),
                                                                           'Z': randint(0, 10)})))
                 for _ in range(n)]
        results = await asyncio.gather(*tasks)
    return len(results)

cal_nodes_len

cal_nodes_len回傳當下ANSA環境內的node Entity數量。

#api.py
@app.get('/cal-nodes-len', status_code=200)
async def cal_nodes_len():
    return len(_get_nodes())

show_ents

show_ents可以回傳當前所有Entity的card value。

#api.py
@app.get('/show-ents', status_code=200)
async def show_ents():
    content = []
    ents = _get_all_ents()
    for ent in ents:
        try:
            content.append(get_entity_card_values(ent))
        except:
            print(f'Can not get the card values of {ent}')
    return content

delete_ents

delete-ents可以刪除全部的Entity。

#api.py
@app.delete('/delete-ents', status_code=204)
async def delete_ents():
    ents = _get_all_ents()
    base.DeleteEntity(ents)

if name == 'main'

最後使用uvicorn啟動FastAPI。

if __name__ == '__main__':
    uvicorn.run('api:app', port=API_PORT, workers=1)

docs

FastAPI附有/docs endpoint，讓我們可以做簡單的API測試。

舉例來說，下圖是create-node的測試結果，可以看到我們得到了201的status code以及成功回傳了node1 Entity的card value。

Limitation

• 目前還找不出方法將Gunicorn搭配uvicorn來使用。
• 當結束程式後，背後的subprocess不會關閉，會佔用到ANSA license，需手動kill process。
• 不確定是不是每個Python3.8.x版本都可以跑得起來，我們用的是3.8.10(ANSA的Python版本為3.8.1)。

感恩

如果本文對您有任何一丁點兒啟發，請與我一起感謝Lasso GmbH Germany及C. Diez。

他們多年前發表的LS-DYNA conference paper qd – Build your own LS-DYNA® Tools Quickly in Python讓我對CAE有了全新的觀點(也帶我走向學習Python及做二次開發的不歸路...XD)。

後來他們開源了lasso-python，其中的lasso.ansa.rest又讓我對如何利用ANSA API建立一個RESTful API有了新的體悟，本文的概念大致上都取自觀察其原始碼。

最近他們又準備再戰江湖，開源了open-lasso-python，除了支援LS-DYNA外，也預計將支援OpenRadioss，這真是令CAE從業人員振奮的消息呀！

此外他們也很用心地在YouTube上分享2022 FREE Professional Python Course，免費教大家學習Python。

感謝他們無私的分享，希望我們能夠承接他們的精神，一起讓CAE領域變得更好。

Code

本日程式碼傳送門。