import os
import json
from panpython.sdk.parser.htc_condition_parser import JsonParser
from panpython.sdk.htc.htc_core import Htc
[docs]class PropertyOpt:
def __init__(self, batch_file, config_file):
self.batch_file: str = batch_file
self.config_file: str = config_file
self.task_type: str = 'Property Optimization'
print(self.task_type + ' is defined!')
pass
[docs] def run(self, system):
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
# --- These files and variables... where to get and set?
# 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) # Hard-code now
# 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
# opt_condition_file: str = os.path.join(dump_path, 'opt_condition.json')
# parser.write_opt_condition_file()
task_map: dict = parser.task_map
units: dict = parser.units
task_type: str = parser.task_type
opt_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)
opt_tasks.set_optimization(parser=parser)
if not opt_tasks.has_batch_file():
raise FileNotFoundError('>>> Batch-file template is not defined or found!')
if not system.system_args.pandat:
raise SystemError('>>> Pandat is required to perform optimization!')
opt_tasks.run_htc()
with open(os.path.join(system.system_args.dump_path, 'failed_calculation.json'), 'w', encoding='utf-8') as fp:
json.dump(opt_tasks.failed_task, fp, indent=4)
print(self.task_type + ' completes!')
pass