"""
"Task: HTC" Module
===============================================================
A pre-installed example script which organizes HTC calculations
"""
import os
import json
from panpython.sdk.parser.htc_condition_parser import JsonParser
from panpython.sdk.htc.htc_core import Htc
[docs]class HtcMesh:
"""
Parameters
----------
batch_file: str
full path of batch file template
config_file: str
full path HTC configuration file
start_id: int, default=0
staring index of HTC
thread_num: int, default=1
number of concurrency thread number
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, batch_file, config_file, start_id=0, thread_num=1):
"""
Initialize a instance to handle mesh of HTC
Parameters
----------
batch_file: str
full path of batch file template
config_file: str
full path HTC configuration file
start_id: int, optional
staring index of HTC
thread_num: int, optional
number of concurrency thread number
Returns
-------
"""
self.thread_num: int = thread_num
self.batch_file: str = batch_file
self.config_file: str = config_file
self.task_type: str = 'Htc mesh'
self.start_id: int = start_id
print('htc mesh defined')
return
[docs] def run(self, system):
"""
Run a defined HTC calculation
Parameters
----------
system: panpy_dev.system.System
a reference to a system instance handling all existing task
Returns
-------
"""
try:
os.makedirs(system.system_args.dump_path)
except OSError:
print(system.system_args.dump_path, ' already exists!')
intermediate_path: str = os.path.join(system.system_args.dump_path, 'intermediate')
try:
os.makedirs(intermediate_path)
except OSError:
print(intermediate_path, ' already exists!')
pandat: str = system.system_args.pandat
dump_path: str = system.system_args.dump_path
time_limit_in_min: float = system.system_args.time_limit_in_min
htc_file: str = os.path.join(dump_path, 'PanHTC_interface.txt')
map_file: str = os.path.join(dump_path, 'task_id.txt')
stat_file: str = os.path.join(dump_path, 'statistics_log.JSON')
thermal_history_file: str = os.path.join(dump_path, 'thermal_history.JSON')
parser: JsonParser = JsonParser(self.config_file)
parser.write_thermal_history(thermal_history_file)
parser.write_PanHTC_interface(htc_file) # save htc interface file
parser.write_task_map(map_file) # save the task id for future analysis
parser.write_statistics_log(stat_file) # save a statistics log file of this HTC
task_map: dict = parser.task_map
units: dict = parser.units
task_type: str = parser.task_type
htc_tasks = Htc(batch_file=self.batch_file,
task_map=task_map,
units=units,
target_path=os.path.join(intermediate_path),
exe_path=pandat,
json_config_type=task_type,
time_limit_in_min=time_limit_in_min,
thread_num=self.thread_num)
if not htc_tasks.has_batch_file():
return
if not system.system_args.pandat:
print('>>> Pandat path is not specified. Only create necessary files. No calculation is performed.')
return
htc_tasks.run_htc(start_id=self.start_id)
with open(os.path.join(system.system_args.dump_path, 'failed_calculation.json'), 'w', encoding='utf-8') as fp:
json.dump(htc_tasks.failed_task, fp, indent=4)
print('htc mesh complete')
return