"""
"HTC Core" Module
======================
The core module of running concurrency HTC
"""
from typing import Dict
from panpython._libs.htc import HtcCoreInstance,run_all_batchfiles_lib
from panpython.sdk.parser.htc_condition_parser import JsonParser
[docs]class Htc:
"""
Parameters
----------
batch_file: str
full path of batch file
task_map: dict
a dict of tasks
units: Union[list, dict]
units of Pandat
target_path: str
full path of output folder
exe_path: str
full path of Pandat.exe
json_config_type: str
calculation type from JSON configuration file
time_limit_in_min: float, default=15.0
time limitation of a single Pandat.exe
thread_num: int, default=1
number of concurrency threads
Examples
--------
>>> from panpython.sdk.htc.htc_core import Htc
>>> from panpython.sdk.parser.htc_condition_parser import JsonParser
>>> parser: JsonParser = JsonParser('htc_condition.JSON')
>>> task_map: dict = parser.task_map
>>> htc_tasks = Htc(batch_file='example.pbfx',
... task_map=parser.task_map,
... units=parser.units,
... target_path='output/intermediate/',
... exe_path='pandat.exe',
... json_config_type='point',
... time_limit_in_min=10.0,
... thread_num=1)
>>> htc_tasks.run_htc()
"""
def __init__(self, batch_file, task_map, units, target_path, exe_path,
json_config_type,
time_limit_in_min=15.0, thread_num=1):
"""Construct a concurrency HTC instance
Parameters
----------
batch_file: str
full path of batch file
task_map: dict
a dict of tasks
units: Union[list, dict]
units of Pandat
target_path: str
full path of output folder
exe_path: str
full path of Pandat.exe
json_config_type: str
calculation type from JSON configuration file
time_limit_in_min: float
time limitation of a single Pandat.exe
thread_num: int
number of concurrency threads
Returns
-------
"""
self.htc_core_instance = HtcCoreInstance(batch_file=batch_file, task_map=task_map, units=units,
target_path=target_path, exe_path=exe_path,
json_config_type=json_config_type,
time_limit_in_min=time_limit_in_min,
thread_num=thread_num)
self._failed_task: Dict[str, Dict[str, float]] = {}
return
@property
def failed_task(self):
"""a dictionary of failed task info"""
return self._failed_task
@failed_task.setter
def failed_task(self, val):
raise ValueError("Assigning 'failed_task' is not allowed")
[docs] def has_batch_file(self):
"""
Return False if batch file does not exist or is empty.
Parameters
----------
Returns
-------
bool
True if batch file exists, False otherwise
"""
return self.htc_core_instance.has_batch_file()
[docs] def set_optimization(self, parser: JsonParser):
"""
Set up optimization conditions for HTC
Parameters
----------
parser: object
HTC parser
Returns
-------
"""
self.htc_core_instance.set_optimization(parser.opt_condition, parser.opt_protocol, parser.opt_bounds)
pass
[docs] def run_htc(self, htc_type: str = '', start_id: int = 0):
"""
Run a batch of Pandat calculations
Parameters
----------
htc_type: str
type of htc calculation
start_id: int
starting index of HTC calculation
Returns
-------
"""
try:
self.htc_core_instance.create_and_run_batch_modified(htc_type, start_id)
except RuntimeError:
self.htc_core_instance.opt_property_with_constrain(htc_type)
self._failed_task = self.htc_core_instance.failed_task
return