[Day22]~[Day23]我們準備利用Streamlit建立一個job submitter project(我們取名叫stem)，可以在Windows WSL2下搭配LS-DYNA small system運行，且最多同時跑2個smp job。

會有這個想法，是有一些朋友詢問，是否有可能在WSL2下執行LS-DYNA？因為有些問題用Windows無法求解，但在Linux下可以得出結果(難處是他們平常只有Windows可以用呀)。

我們的回答是WSL2下的確可以執行LS-DYNA，但是目前好像沒有看到有job submitter可以在Windows下直接操控WSL2的job。

在回覆的當下，我們的coding魂被莫名點燃，既然找不到，何不自己build一個勒？由於我們只想做一個非常lightweight的App，不想使用到資料庫，正巧那陣子我們與Streamlit打得火熱，利用Streamlit cache來作為一個小型資料庫的想法，不知怎地就出現在我們腦海中，然後code就這樣一點一點生出來了。

今天我們講解一些概念，[Day23]再分享詳細的實作。

Disclaimer

開打前，照慣例下個警語。[Day22]~[Day23]的code非常...怎麼說呢...原汁原味吧？沒有經過實戰的考驗，純粹是我們的side project(雖然我們的確有拿來跑一些小job)。老話一句，請自行評估後果再決定是否取用。有任何死當、閃退、license佔用或無法運行等疑難雜症，拜託請不要來找我們，我們以分享概念為主。

虛擬環境venv

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

python3 -m venv venv
venv\Scripts\activate
pip install -r requirements.txt

#requirements.txt
streamlit
streamlit_autorefresh
pandas

Folder Picker

雖然Streamlit提供了file_uploader，但卻沒有folder picker這種widget(註1)。好在我們從Streamlit GitHub Issues找到了以下方法。

#app.py
import tkinter as tk

root = tk.Tk()
root.withdraw()
root.wm_attributes('-topmost', 1)

這樣我們就可以透過filedialog的askopenfilename及askdirectory得到檔案或資料夾的路徑。

Task

我們將每個job當作一個Task object，為一pydantic model。

• id為一獨特的識別碼，設定為八位數的uuid4 str。
• cmd為執行該task所需的指令。
• cp為sentinel或是subprocess運行的狀態。
• status是指該task目前運行的狀態，為一個Enum，共有四種狀態，預設為staging：
  • staging
  • running
  • notOK
  • finished

#schemas.py
class Task(BaseModel):
    id: str
    cmd: str
    cp: Any
    status: TaskStatus = TaskStatus.staging

    
class TaskStatus(str, Enum):
    staging = 'staging'
    running = 'running'
    notOK = 'notOK'
    finished = 'finished'

Session State

Session State為Streamlit的快取機制之一，用法很直觀，可想像為一dict來使用。

於stem中我們共使用了三個快取變數：

• tasks為一list，收集各task的資訊。
• sentinel為一獨特值，我們使用object()，用來檢查task當前求解的狀態。
• insertable_idx為當前可插入task的index。

#app.py
if 'tasks' not in st.session_state:
    st.session_state['tasks'] = []

if 'sentinel' not in st.session_state:
    st.session_state['sentinel'] = object()

if 'insertable_idx' not in st.session_state:
    st.session_state['insertable_idx'] = 0

並寫了三個getter function，方便我們取得這三個快取(註2)。

#app.py
def get_tasks():
    return st.session_state.tasks


def get_sentinel():
    return st.session_state.sentinel


def get_insertable_idx():
    return st.session_state.insertable_idx

env.py

env.py內有兩個變數：

• ST_AUTO_REFRESH_INTERVAL代表多久refresh一次，單位為毫秒。例如你想每五秒refresh一次，需輸入ST_AUTO_REFRESH_INTERVAL=5000。
• MAX_CONCURRENT_LIMIT代表最多可以concurrent的執行幾個task。這需要根據small system所擁有的資源設定，一般為1~2。

#env.py
ST_AUTO_REFRESH_INTERVAL = 5000
MAX_CONCURRENT_LIMIT = 2

Helper Functions

create_task_id

_create_task_id預設回傳前八個uuid4.hex的字串。

#helpers.py
def _create_task_id(n=8):
    return str(uuid4().hex)[:n]

create_task_id會回傳一個獨特的task_id，其中的while迴圈是防止會有重複的task_id所作的措施。

#app.py
def create_task_id():
    task_ids = get_task_ids()
    while True:
        task_id = _create_task_id()
        if task_id not in task_ids:
            break
    return task_id

get_disk_id

get_disk_id擷取Windows路徑下第一個字元並轉為小寫，如C => c。

#helpers.py
def get_disk_id(one_path):
    # ex: C: => c
    return str(one_path)[0].lower()

tk_2_wsl2

tk_2_wsl2幫助我們將於tkinter中取得的路徑轉換為WSL2下的路徑。

#helpers.py
def tk_2_wsl2(one_path, my_sep='/'):
    '''
    C:/Users/username/Desktop/LS_DYNA/airbag_deploy.k
    => /mnt/c/Users/username/Desktop/LS_DYNA/airbag_deploy.k
    '''
    disk_id = get_disk_id(one_path)
    return f'/mnt/{disk_id}/' + '/'.join(one_path.split(my_sep)[1:])

parse_dyna_folder

parse_dyna_folder幫忙取出task_cmd中第二項的LS-DYNA solver所在資料夾，並忽略前兩個字元，即i=。

#helpers.py
def parse_dyna_folder(cmd):
    _, i, *_ = cmd.split()
    return Path(i[2:]).parent.as_posix()  # ignore 'i=

parse_task_cmd

parse_task_cmd幫助我們取出task_cmd內各項目的值，接著組合為於dashboard顯示的型態。例如ncpu=4，會擷取出4。

#helpers.py
def parse_task_cmd(task_cmd):
    solver_, deck_, ncpu, memory, *consoles = task_cmd.split()
    *_, solver_ver, solver_name = solver_.split('/')
    solver = '_'.join([solver_name[:3], solver_name[-1],  solver_ver])
    deck = '/'.join(deck_.split('/')[-2:])
    ncpu = ncpu.split('=')[-1]
    memory = memory.split('=')[-1]
    return solver, deck, ncpu, memory, consoles

get_win_user

get_win_user幫助我們取得Windows的用戶名。

#helpers.py
@st.cache
def get_win_user():
    return getpass.getuser()

get_solver_dir

get_solver_dir為將放置LS-DYNA solver檔案夾的Windows路徑換為WSL2下的路徑。

#helpers.py
@st.cache
def get_solver_dir():
    '''
    C:\\Users\\username\\Desktop\\LS_DYNA\\program
    '''
    cwd = os.getcwd()
    disk_id = get_disk_id(cwd)
    win_user = get_win_user()
    return f'/mnt/{disk_id}/Users/{win_user}/Desktop/LS_DYNA/program'

convert_df

convert_df我們直接取用自st.download_button的說明文件。

#helpers.py
@st.cache
def convert_df(df):
    # IMPORTANT: Cache the conversion to prevent computation on every rerun
    return df.to_csv().encode('utf-8')

get_csv_filename

get_csv_filename回傳一個獨特的csv檔名。

#helpers.py
def get_csv_filename():
    return datetime.now().strftime('%Y%m%d_%H%M%S') + '.csv'

mappings.py

• solver_type_mapping為於前端選擇solver type 對應的字串。
• solver_precision_mapping為於前端選擇solver precision 對應的字串。
• solver_version_pool列出於前端可選擇的solver version。
• emogi_mapping可以讓我們依task的status於dashboard中顯示對應的emogi。

from schemas import TaskStatus

solver_type_mapping = {'smp': 'smp', 'mpp': 'mpp', 'hybrid': 'hyb'}
solver_precision_mapping = {'single': 's', 'double': 'd'}
solver_version_pool = ('12.0', '8.0', '8.1', '9.0', '9.1', '9.2', '10.0',
                       '10.1', '11.0', '11.1', '11.2', '12.0', '12.1', '13.0', 'Daily')
emogi_keys = [t.name for t in TaskStatus]
emogi_values = [e.encode('utf-8') for e in ('?', '?', '❌', '✅')]
emogi_mapping = dict(zip(emogi_keys, emogi_values))

Limitation

• 雖然是透過Streamlit起server，即使不在本機，理論上應該也可以連到。但由於我們需要使用folder picker，所以當要選取檔案的時候，跳出的選取視窗只會出現在本機端，所以目前這只是一個透過瀏覽器操控本機作業的App。
• 當按下重新整理後，就真的重新整理了，沒有執行過job的資訊或是log。
• 使用Auto-refresh的確可以達成我們想要的，但似乎不是很有效率的作法。

備註

註1：想一想好像也合理，如果有這種功能的話，那麼這個Streamlit app就可以透過瀏覽器，取得你本機端的資料夾路徑...好像有點資安疑慮？

註2：st.session_state.foobar或st.session_state['foobar']兩種語法都可以使用唷。

Code

本日程式碼傳送門。