swim/bo-ptycho
1
0
Fork 0
mirror of https://github.com/c-sooyoung/bo-ptycho.git synced 2026-09-17 21:29:07 +09:00
Code Issues Packages Projects Releases Wiki Activity

removed PtychoEngine unnecessary methods

Browse Source
This commit is contained in:
swim
2026-07-30 20:56:58 +09:00
parent 02293389a1
commit 19f3cff188
3 changed files with 42 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ptycho/fold_slice.py
+38 -68
View File
@@ -2,6 +2,7 @@ import os
import sys import sys
import shutil import shutil
import subprocess import subprocess
import re
from typing import Any from typing import Any
from scipy.io import loadmat from scipy.io import loadmat
import numpy as np import numpy as np
@@ -12,102 +13,71 @@ class FoldSlicePtychoEngine(PtychoEngine):
def __init__(self, config): def __init__(self, config):
super().__init__(config) super().__init__(config)
self.fold_slice_output_dir = os.path.join(config['io']['result_dir'], 'fold_slice') self._output_dir = os.path.join(config['io']['result_dir'], 'fold_slice')
self._fold_slice_path = self.config['ptycho']['path']
self._setup_txt_path = os.path.join(self._output_dir, 'setup.txt')
self._verbosity = self.config['io'].get('verbosity', 0)
def fold_slice_prepare(self) -> None: def run(self, run_id="") -> None:
config = self.config
fold_slice_output_dir = self.fold_slice_output_dir
fold_slice_dict = {} fold_slice_dict = {}
fold_slice_dict['raw_data'] = config['io']['input_data_path'] fold_slice_dict['raw_data'] = self.config['io']['input_data_path']
fold_slice_dict['result_dir'] = os.path.join(fold_slice_output_dir, '') fold_slice_dict['result_dir'] = os.path.join(self._output_dir, '')
fold_slice_dict.update(config['ptycho']['params']) fold_slice_dict.update(self.config['ptycho']['params'])
for key, value in fold_slice_dict.items():
if type(value) == bool:
fold_slice_dict[key] = str(value).lower()
if os.path.exists(os.path.join(fold_slice_output_dir)): if os.path.exists(os.path.join(self._output_dir)):
shutil.rmtree(os.path.join(fold_slice_output_dir)) shutil.rmtree(os.path.join(self._output_dir))
os.makedirs(os.path.join(fold_slice_output_dir)) os.makedirs(os.path.join(self._output_dir))
with open(os.path.join(fold_slice_output_dir, 'setup.txt'), 'w') as f: with open(os.path.join(self._output_dir, 'setup.txt'), 'w') as f:
f.write('\n\n') f.write('\n\n')
for key, value in fold_slice_dict.items(): for key, value in fold_slice_dict.items():
f.write(f"{key} {value}\n") f.write(f"{key} {value}\n")
def run(self, header="") -> None:
self.fold_slice_prepare()
config = self.config
fold_slice_path = config['ptycho']['path']
setup_txt = os.path.join(self.fold_slice_output_dir, 'setup.txt')
verbosity = config['io'].get('verbosity', 0)
matlab_commands = [ matlab_commands = [
f"cd('{fold_slice_path}');", f"cd('{self._fold_slice_path}');",
"cd('ptycho');", "cd('ptycho');",
f"prepare_data('{setup_txt}');", f"prepare_data('{self._setup_txt_path}');",
f"run_multislice_new('{setup_txt}');" f"run_multislice_new('{self._setup_txt_path}');"
] ]
p = subprocess.Popen( p = subprocess.Popen(
['matlab', '-batch', ' '.join(matlab_commands)], ['matlab', '-batch', ' '.join(matlab_commands)],
stdout=subprocess.PIPE if verbosity > 0 else subprocess.DEVNULL, stdout=subprocess.PIPE if self._verbosity > 0 else subprocess.DEVNULL,
stderr=subprocess.STDOUT if verbosity > 0 else subprocess.DEVNULL, stderr=subprocess.STDOUT if self._verbosity > 0 else subprocess.DEVNULL,
text=True text=True
) )
if not verbosity == 0: if self._verbosity > 0:
header = header + '[fold slice]'
for line in p.stdout: # type: ignore for line in p.stdout: # type: ignore
sys.stdout.write(f'{header} {line}') sys.stdout.write(f'[{run_id}] [fold_slice] {line}')
p.stdout.close() # type: ignore p.stdout.close() # type: ignore
else: else:
print(f"fold_slice running. Set verbosity > 0 for full fold_slice output.") print(f"fold_slice running. Set verbosity > 0 for full fold_slice output.")
p.wait() p.wait()
def get_output_mat_path(self) -> str:
config = self.config
fold_slice_output_dir = self.fold_slice_output_dir
roi_dir = os.path.join( roi_dir = os.path.join(
fold_slice_output_dir, self._output_dir,
f"{config['ptycho']['params']['scan_number']}", f"{self.config['ptycho']['params']['scan_number']}",
f"roi{config['ptycho']['params']['roi_label']}" f"roi{self.config['ptycho']['params']['roi_label']}"
) )
output_dir = os.path.join(roi_dir, next(os.walk(roi_dir))[1][0]) output_dir = os.path.join(roi_dir, next(os.walk(roi_dir))[1][0])
# image_path = os.path.join(output_dir, 'obj_phase_roi_sum', next(os.walk(os.path.join(output_dir, 'obj_phase_roi_sum')))[2][0]) mat_path = os.path.join(output_dir, f"Niter{self.config['ptycho']['params']['Niter']}.mat")
output_mat_path = os.path.join(output_dir, f"Niter{config['ptycho']['params']['Niter']}.mat") image_path = os.path.join(output_dir, 'obj_phase_roi_sum', f"obj_phase_roi_sum_Niter{self.config['ptycho']['params']['Niter']}.tiff")
return output_mat_path
def output(self) -> Any: if not os.path.exists(mat_path):
if self._output is None: raise FileNotFoundError(f"{mat_path} does not exist. Please check the fold_slice output.")
output_mat_path = self.get_output_mat_path() if not os.path.exists(image_path):
if not os.path.exists(output_mat_path): raise FileNotFoundError(f"{image_path} does not exist. Please check the fold_slice output.")
raise FileNotFoundError(f"Result directory {output_mat_path} does not exist. Please check the fold_slice output.")
self._output = loadmat(output_mat_path)
return self._output
def recon_object(self) -> np.ndarray: self._output = loadmat(mat_path)
if self._recon_object is None: self._metric = -np.log(float(self._output['outputs']['fourier_error_out'][0][0].squeeze()[-1]))
output = self.output()
self._recon_object = output['object'] # type: ignore
return self._recon_object
os.makedirs(os.path.join(self.config['io']['result_dir'], "mat"), exist_ok=True)
def recon_probe(self) -> np.ndarray: os.makedirs(os.path.join(self.config['io']['result_dir'], "tiff"), exist_ok=True)
if self._recon_probe is None: shutil.copy(mat_path, os.path.join(self.config['io']['result_dir'], "mat", f"{self._metric:.4f}_{run_id}.mat"))
output = self.output() shutil.copy(image_path, os.path.join(self.config['io']['result_dir'], "tiff", f"{self._metric:.4f}_{run_id}.tiff"))
self._recon_probe = output['probe'] # type: ignore
return self._recon_probe
def metric(self) -> float:
if self._metric is None:
output = self.output()
fourier_error = output['outputs']['fourier_error_out'][0][0].squeeze()
metric = float(fourier_error[-1])
self._metric = metric
return self._metric
ptycho/ptycho_base.py
+1 -21
View File
@@ -7,28 +7,8 @@ class PtychoEngine(ABC):
def __init__(self, config): def __init__(self, config):
self.config = config self.config = config
self._output = None
self._recon_object = None
self._recon_probe = None
self._metric = None
@abstractmethod @abstractmethod
def run(self) -> None: def run(self, run_id="") -> None:
pass
@abstractmethod
def output(self) -> Any:
pass
@abstractmethod
def recon_object(self) -> Any:
pass
@abstractmethod
def recon_probe(self) -> Any:
pass
@abstractmethod
def metric(self) -> float:
pass pass
ptycho/ptycho_example.py
+3 -29
View File
@@ -10,35 +10,9 @@ class ExamplePtychoEngine(PtychoEngine):
def __init__(self, config): def __init__(self, config):
super().__init__(config) super().__init__(config)
self._metric = 0.0
# run single ptychography job based on `config` # run single ptychography job based on `config`
def run(self) -> None: def run(self, run_id="") -> None:
print("[ExamplePtychoEngine] Sleeping for 0.1 second.") print(f"[{run_id}] [ExamplePtychoEngine] Sleeping for 0.1 second.")
time.sleep(0.1) time.sleep(0.1)
# save outputs to self._output
def output(self) -> Any:
if self._output is None:
pass
return self._output
# return reconstructed object (numpy array)
def recon_object(self) -> Any:
if self._recon_object is None:
output = self.output()
pass
return self._recon_object
# return reconstructed probe (numpy array)
def recon_probe(self) -> Any:
if self._recon_probe is None:
output = self.output()
pass
return self._recon_probe
# return metric value for Bayesian optimization
def metric(self) -> float:
if self._metric is None:
output = self.output()
self._metric = random.uniform(0, 1)
return self._metric