forked from NatLabRockies/GEOPHIRES-X
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy path__init__.py
More file actions
82 lines (63 loc) · 2.76 KB
/
Copy path__init__.py
File metadata and controls
82 lines (63 loc) · 2.76 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import os
import re
import sys
import tempfile
import uuid
from pathlib import Path
from typing import Any
# noinspection PyProtectedMember
from geophires_x_client.common import _get_logger
from hip_ra import HIP_RA
class HipRaInputParameters:
def __init__(self, file_path_or_params_dict: Any):
if isinstance(file_path_or_params_dict, dict):
tmp_file_path = Path(tempfile.gettempdir(), f'hip-ra-params_{uuid.uuid1()}.txt')
with open(tmp_file_path, 'w') as f:
f.writelines(
[', '.join([str(p) for p in param_item]) + '\n' for param_item in file_path_or_params_dict.items()]
)
file_path_or_params_dict = str(tmp_file_path)
elif not isinstance(file_path_or_params_dict, (str, Path)):
raise ValueError('Provide either file path (as Path or string) or parameters as dict')
self._input_file_path = Path(file_path_or_params_dict)
self._output_file_path = Path(
tempfile.gettempdir(), f'hip-ra-result_{self._input_file_path.stem}_{uuid.uuid1()!s}.out'
)
def as_file_path(self) -> Path:
return self._input_file_path
@property
def output_file_path(self) -> Path:
return self._output_file_path
class HipRaResult:
def __init__(self, output_file_path):
self.output_file_path = output_file_path
self.result = self._parse_fields()
def _parse_fields(self):
with open(self.output_file_path) as f:
text = f.read()
pattern = re.compile(r'(.+?):\s+([0-9eE.+-]+)\s*(\S+)*?\n')
matches = re.findall(pattern, text)
result = {
key.strip(): {'value': float(value), 'unit': unit.strip() if unit else None}
for key, value, unit in matches
}
return result
class HipRaClient:
def __init__(self, enable_caching=True, logger_name='root'):
self._logger = _get_logger(logger_name=logger_name)
def get_hip_ra_result(self, input_params: HipRaInputParameters) -> HipRaResult:
stash_cwd = Path.cwd()
stash_sys_argv = sys.argv
sys.argv = ['', input_params.as_file_path(), input_params.output_file_path]
try:
HIP_RA.main(enable_hip_ra_logging_config=False)
except Exception as e:
raise RuntimeError(f'HIP-RA encountered an exception: {e!s}') from e
except SystemExit:
raise RuntimeError('HIP-RA exited without giving a reason') from None
finally:
# Undo HIP-RA internal global settings changes
sys.argv = stash_sys_argv
os.chdir(stash_cwd)
self._logger.info(f'HIP-RA output file: {input_params.output_file_path}')
return HipRaResult(input_params.output_file_path)