"""This module contains classes, used by parsers of the output files"""
import os
import re
import importlib
from . import readers
from ..common.decorators import cached_property, cached_class_property, maybeclassmethod
from ..common.process_output_reader import run_coro_sync
from ..potentials.potentials import Potential
from ..input_parameters import input_parameters as input_parameters
from ..output_files.output_files import OutputFile
from pathlib import Path
import platform
import subprocess
import shutil
[docs]
class ResultValue:
[docs]
def __init__(self, name, value):
self.name = name
self._value = value
def __call__(self):
return self._value
[docs]
def value_label(self):
return self._value
[docs]
def actions(self):
return []
[docs]
class OutputFileResultValue:
[docs]
def __init__(self, name, type, filename):
self.name = name
self.type = type
self.filename = filename
def __call__(self):
return OutputFile.from_file(self.filename, try_only=self.type)
[docs]
def definition(self):
return OutputFile.definitions[self.type]
[docs]
def value_label(self):
return f"<{self.type} file>"
[docs]
def actions(self):
out = ["open", "save"]
if self.definition().result_class.can_be_plotted():
out.append("plot")
return out
[docs]
def save(self, parent=None):
try:
from PyQT6.QtWidgets import QFileDialog
except ImportError:
raise Exception("PyQt6 is required for saving the output file")
file_path, _ = QFileDialog.getSaveFileName(
parent, "Save File", "", f" (*.{self.definition().extension});;All Files (*)"
)
shutil.copy(self.filename(), file_path)
[docs]
def plot(self, parent=None):
self().plot()
[docs]
def open(self, parent=None):
if platform.system() == "Windows":
os.startfile(self.filename)
elif platform.system() == "Darwin": # macOS
subprocess.run(["open", self.filename])
else: # Linux
subprocess.run(["xdg-open", self.filename])
[docs]
class TaskResult:
"""A base class for a result of a runned task (kkrscf executable)"""
[docs]
def __init__(self, input_parameters, calculator, directory, output_file=None, input_file=None):
self._input_parameters = input_parameters
self._calculator = calculator
self.files = {}
self.output_file = output_file
if output_file:
self.files["output"] = output_file
self._directory = os.path.realpath(directory) if directory else None
self.input_file = input_file
[docs]
@cached_property
def directory(self):
def file_name(f):
if isinstance(f, str):
return f
if hasattr(f, "name"):
return f.name
return f
return self._directory or os.path.dirname(file_name(self.files.get("output")) or "") or os.getcwd()
[docs]
def path_to(self, file):
"""return full path to a given file
..doctest::
>>> t = TaskResult(None, None, "/example")
>>> t.files["input"] = "input.txt"
>>> t.path_to("input")
'/example/input.txt'
"""
file = self.files[file]
if Path(file).is_absolute():
return file
return os.path.join(self.directory, file)
@property
def task_name(self):
return self.input_parameters.task_name
[docs]
@cached_property
def potential_filename(self):
"""New (output) potential file name"""
potfil = self.input_parameters.CONTROL.POTFIL()
if not potfil:
potfil = self.files.get("potential", None)
if not potfil:
raise ValueError("Please set CONTROL.POTFIL of the input_parameters to read the potential")
if self.directory:
potfil = os.path.join(self.directory, potfil)
return potfil
[docs]
@cached_property
def potential(self):
"""The new (output) potential - that contains the converged charge density etc."""
return Potential.from_file(self.potential_filename)
[docs]
def new_task(self, task):
out = self._calculator.copy_with_potential(self.potential_filename)
out.input_parameters = task
if isinstance(task, str) and task.lower() == "jxc":
out.input_parameters.set("EMIN", self.last_iteration.energy.EMIN())
return out
[docs]
def complete(self, error, return_code):
self.error = error
self.return_code = return_code
@property
def atoms(self):
return self.potential.atoms
@cached_class_property
def _match_task_regex(self):
return re.compile(r" TASK\s+ = ([A-Z]+)\s+\n")
[docs]
@classmethod
def from_file(cls, file):
def read_output_for(task):
process = KkrOutputReader.class_for_task(task or "")
process = process(None, None, os.path.dirname(file))
f.seek(0)
return process.read_from_file(f)
with open(file, "rb") as f:
raw_out = f.read()
matches = cls._match_task_regex.search(raw_out.decode("utf8"))
typ = matches[1] if matches and matches[1] != "NONE" else None
if typ is None:
if b"Components of Dzyaloshinski-Moriya vector Dij" in raw_out:
typ = "jxc"
out = read_output_for(typ)
if typ is None:
if "DOS" in out.files:
with open(file, "rb") as f:
out = read_output_for("DOS")
return out
[docs]
class KkrProcess:
[docs]
def __init__(self, reader, output_parser, coroutine, result, callback=None):
self.reader = reader
self.output_parser = output_parser
self.coroutine = coroutine
self.result = result
self.callback = callback
[docs]
def run(self):
result = self.result
try:
result, error, return_code = run_coro_sync(self.coroutine)
self.result.complete(error, return_code)
finally:
if self.callback:
self.callback(self.result)
return self.result
[docs]
def stop_the_process(self):
self.output_parser.stop_the_process()
[docs]
async def run_async(self):
result, error, return_code = await self.coroutine
self.result.complete(error, return_code)
return self.result
[docs]
class KkrOutputReader:
"""Class, that runs a process and reads its output using the underlying
output parser (see :class:`ase2sprkkr.common.process_output_reader.ProcessOutputParser`)
and return the appropriate TaskResult.
Descendants should define parser_class and result_class property.
"""
[docs]
def __init__(self, input_parameters, calculator, directory, print_output=False, read_callback=None):
self.input_parameters = input_parameters
""" Input parameters, that command to read the output (thus probably the ones, that
run the process that produced the output. It is used e.g. for determining the potential file,
which belongs to the output.
"""
self.calculator = calculator
self.directory = directory
""" Calculator, that can be used for further processing of the results. """
self.parser = self.parser_class(print_output, read_callback)
[docs]
def _create_result(self, output_file, input_file=None):
"""Create an object that stores results of the KKR output parsing."""
return self.result_class(
self.input_parameters, self.calculator, self.directory, output_file=output_file, input_file=input_file
)
[docs]
def _create_process(self, coroutine, result, callback=None):
"""Create an object that can run the desired process (either reading from file or running a process)
Returns
-------
KkrProcess
An object that can run the process and return the result.
"""
return KkrProcess(self, self.parser, coroutine, result, callback=callback)
[docs]
def create_process(self, cmd, outfile, input_file=None, callback=None, **kwargs):
"""Create an object that takes care of running the command and parsing the results"""
result = self._create_result(getattr(outfile, "name", None), input_file)
coroutine = self.parser.run_async(cmd, outfile, self.directory, [result], **kwargs)
return self._create_process(coroutine, result, callback=callback)
[docs]
@maybeclassmethod
def read_from_file(self, cls, output, error=None, return_code=0, input_file=None, directory=None):
"""Creates an object that takes care of reading and parsing of the output of a sprkkr process"""
if self is None:
self = cls(None, None, directory or os.path.dirname(output))
if directory:
output = os.path.join(directory, output) if output else None
result = self._create_result(output, input_file)
coroutine = self.parser.read_output_file(output, error, [result], return_code)
return self._create_process(coroutine, result).run()
[docs]
@staticmethod
def class_for_task(task):
if not task:
return DefaultOutputReader
try:
mod = importlib.import_module(f".{task.lower()}", readers.__name__)
clsname = task.title() + "OutputReader"
cls = getattr(mod, clsname)
if not cls:
raise Exception(
f"Can not determine the class to read the results of task {task}"
"No {clsname} class in the module {oo.__name__}.{task}"
)
except ModuleNotFoundError:
cls = DefaultOutputReader
return cls
from ..outputs.readers.default import DefaultOutputReader # NOQA