"""
"System" Module
===============
A portal to load/run tasks of PanPython.
"""
import os.path
import time
from panpython._libs.syslib import SystemArgs
from panpython.sdk.htc.htc_core import run_all_batchfiles_lib
[docs]def run_all_batchfiles(pandat_exe:str, path_to_folder:str,
save_console_output:bool=True,
show_progress:bool=True,
time_limit_in_min:float=5.0,
include_subfolders:bool=False,
n_process:int=1,
save_table_as_dat:bool=True,
append_batchname_to_tablename:bool=True):
"""Run all batchfiles in a given folder
Args:
pandat_exe (str): Full path of Pandat executable
path_to_folder (str): Full path to the target folder
save_console_output (bool, optional): Optionally save console output. Defaults to True.
show_progress (bool, optional): Optionally print the progress of the calculations on screen. Defaults to True.
time_limit_in_min (float, optional): Time limit for each simulation. Defaults to 5.0.
include_subfolders (bool, optional): Optionally include batchfiles in subfolders. Defaults to False.
n_process (int, optional): Number of parallel threads to use. Defaults to 1.
save_table_as_dat (bool, optional): Save the table in the batchfiles as *.dat file. Defaults to True
append_batchname_to_tablename (bool, optional): Rename the table by appending the batchfile name. Defaults to True
"""
run_all_batchfiles_lib(pandat_exe,path_to_folder,
save_console_output=save_console_output,
show_progress=show_progress,
time_limit_in_min=time_limit_in_min,
include_subfolders=include_subfolders,
n_process=n_process,
save_table_as_dat=save_table_as_dat,
append_batchname_to_tablename=append_batchname_to_tablename)
[docs]class System:
"""
Parameters
----------
dump_path: str
the path where task result is saved
pandat: str, default=''
a full path of pandat.exe
time_limit_in_min: str, default=15.0
time limit (minutes) for each pandat calculation
Note
----
Initialize "System" instance with dump path and full path of Pandat.exe, and time limit (minutes) of each Pandat calculation.
Full path of Pandat.exe is necessary only when Pandat calculations are performed, while this path can be left empty when doing post-processing.
Time limit of Pandat calculation is necessary for discarding abnormal calculations.
Examples
--------
>>> import os
>>> from panpython.system import System
>>> from panpython.task.htc import HtcMesh
>>> pandat = "YOUR_PATH_TO_PANDAT_EXE/Pandat.exe"
>>> dump_path = "./output/"
>>> batch_file = "./template_batchfile.pbfx"
>>> config_file = "./Configuration.JSON"
>>>
>>> m_system = System(pandat=pandat, dump_path=dump_path) # initialize system
>>> m_system.add_task(task_definition=HtcMesh(batch_file=batch_file, config_file=config_file, thread_num=1)) # add task
>>> m_system.run() # run task
"""
def __init__(self, dump_path, pandat='', time_limit_in_min=15.0):
"""
Initialize "System" instance with dump path and full path of Pandat.exe, and time limit (minutes) of each Pandat calculation.
Full path of Pandat.exe is necessary only when Pandat calculations are performed, while this path can be left empty when doing post-processing.
Time limit of Pandat calculation is necessary for discarding abnormal calculations.
Parameters
----------
dump_path: str
the path where task result is saved
pandat: str, optional
a full path of pandat.exe
time_limit_in_min: str, optional
time limit (minutes) for each pandat calculation
Returns
-------
"""
print('\n==================================================================')
print('\t\t welcome to PanPython')
print('\t\t Package updated at 10:23 AM, Dec 15, 2020')
print('==================================================================')
self._system_args = SystemArgs()
self._system_args.pandat = pandat
self._system_args.dump_path = dump_path
if not os.path.exists(dump_path):
os.makedirs(dump_path)
self._system_args.ready_to_run = False
self._system_args.time_limit_in_min = time_limit_in_min
self._task_id: int = 0
self._tasks: dict = {}
self._tasks_status: dict = {}
print('> System defined')
return
@property
def system_args(self):
"""arguments of the calculation"""
return self._system_args
@system_args.setter
def system_args(self, obj):
raise ValueError("system_args is not allowed to be set outside system instance")
[docs] def add_task(self, task_instance):
"""
Add task instance to "System"
Parameters
----------
task_instance: object
A task instance
Returns
-------
"""
self._tasks[int(self._task_id)] = task_instance # a method instance
self._tasks_status[int(self._task_id)] = False
self._task_id += 1
print('> Task loaded: ', task_instance.task_type)
self._compile()
return
def _compile(self) -> bool:
"""
check status of a task
@return: bool
"""
if self._task_id < 0:
print('> Task is empty')
return False
self._tasks_status[int(self._task_id - 1)] = True
print('> Task compiled')
return True
[docs] def run(self) -> bool:
"""
Run all added tasks in "System"
Parameters
----------
Returns
-------
bool
True if all task is success, False is any task failed
"""
t0 = time.time()
if self._task_id < 0:
print('>>> Task is empty. but ready to run')
return False
for i, s in self._tasks_status.items():
if not s:
print('>>> Task[' + str(i) + '] is not ready to run')
return False
for i, task in self._tasks.items():
task.run(self)
print('> Task[' + str(i) + '] completes')
print('> All tasks complete')
t1 = time.time()
time_cost = "\n>>> Time cost of running all tasks: {sec:.4f} seconds."
print(time_cost.format(sec=t1 - t0))
return True