From 19f3cff188b6cb5cbe22e66776f70841e8b80f7e Mon Sep 17 00:00:00 2001 From: Sooyoung Cheong <64125280+c-sooyoung@users.noreply.github.com> Date: Thu, 30 Jul 2026 20:56:58 +0900 Subject: [PATCH] removed PtychoEngine unnecessary methods --- ptycho/fold_slice.py | 106 ++++++++++++++------------------------- ptycho/ptycho_base.py | 22 +------- ptycho/ptycho_example.py | 32 ++---------- 3 files changed, 42 insertions(+), 118 deletions(-) diff --git a/ptycho/fold_slice.py b/ptycho/fold_slice.py index d0242f6..394ca81 100644 --- a/ptycho/fold_slice.py +++ b/ptycho/fold_slice.py @@ -2,6 +2,7 @@ import os import sys import shutil import subprocess +import re from typing import Any from scipy.io import loadmat import numpy as np @@ -12,102 +13,71 @@ class FoldSlicePtychoEngine(PtychoEngine): def __init__(self, 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: - config = self.config - fold_slice_output_dir = self.fold_slice_output_dir - + def run(self, run_id="") -> None: fold_slice_dict = {} - fold_slice_dict['raw_data'] = config['io']['input_data_path'] - fold_slice_dict['result_dir'] = os.path.join(fold_slice_output_dir, '') - fold_slice_dict.update(config['ptycho']['params']) + fold_slice_dict['raw_data'] = self.config['io']['input_data_path'] + fold_slice_dict['result_dir'] = os.path.join(self._output_dir, '') + 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)): - shutil.rmtree(os.path.join(fold_slice_output_dir)) - os.makedirs(os.path.join(fold_slice_output_dir)) + if os.path.exists(os.path.join(self._output_dir)): + shutil.rmtree(os.path.join(self._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') for key, value in fold_slice_dict.items(): 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 = [ - f"cd('{fold_slice_path}');", + f"cd('{self._fold_slice_path}');", "cd('ptycho');", - f"prepare_data('{setup_txt}');", - f"run_multislice_new('{setup_txt}');" + f"prepare_data('{self._setup_txt_path}');", + f"run_multislice_new('{self._setup_txt_path}');" ] p = subprocess.Popen( ['matlab', '-batch', ' '.join(matlab_commands)], - stdout=subprocess.PIPE if verbosity > 0 else subprocess.DEVNULL, - stderr=subprocess.STDOUT if verbosity > 0 else subprocess.DEVNULL, + stdout=subprocess.PIPE if self._verbosity > 0 else subprocess.DEVNULL, + stderr=subprocess.STDOUT if self._verbosity > 0 else subprocess.DEVNULL, text=True ) - if not verbosity == 0: - header = header + '[fold slice]' + if self._verbosity > 0: 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 else: print(f"fold_slice running. Set verbosity > 0 for full fold_slice output.") 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( - fold_slice_output_dir, - f"{config['ptycho']['params']['scan_number']}", - f"roi{config['ptycho']['params']['roi_label']}" + self._output_dir, + f"{self.config['ptycho']['params']['scan_number']}", + f"roi{self.config['ptycho']['params']['roi_label']}" ) 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]) - output_mat_path = os.path.join(output_dir, f"Niter{config['ptycho']['params']['Niter']}.mat") - return output_mat_path - + mat_path = os.path.join(output_dir, f"Niter{self.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") - def output(self) -> Any: - if self._output is None: - output_mat_path = self.get_output_mat_path() - if not os.path.exists(output_mat_path): - 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 - + if not os.path.exists(mat_path): + raise FileNotFoundError(f"{mat_path} does not exist. Please check the fold_slice output.") + if not os.path.exists(image_path): + raise FileNotFoundError(f"{image_path} does not exist. Please check the fold_slice output.") - def recon_object(self) -> np.ndarray: - if self._recon_object is None: - output = self.output() - self._recon_object = output['object'] # type: ignore - return self._recon_object + self._output = loadmat(mat_path) + self._metric = -np.log(float(self._output['outputs']['fourier_error_out'][0][0].squeeze()[-1])) - - def recon_probe(self) -> np.ndarray: - if self._recon_probe is None: - output = self.output() - 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 + os.makedirs(os.path.join(self.config['io']['result_dir'], "mat"), exist_ok=True) + os.makedirs(os.path.join(self.config['io']['result_dir'], "tiff"), exist_ok=True) + shutil.copy(mat_path, os.path.join(self.config['io']['result_dir'], "mat", f"{self._metric:.4f}_{run_id}.mat")) + shutil.copy(image_path, os.path.join(self.config['io']['result_dir'], "tiff", f"{self._metric:.4f}_{run_id}.tiff")) diff --git a/ptycho/ptycho_base.py b/ptycho/ptycho_base.py index 5da347a..3186a2c 100644 --- a/ptycho/ptycho_base.py +++ b/ptycho/ptycho_base.py @@ -7,28 +7,8 @@ class PtychoEngine(ABC): def __init__(self, config): self.config = config - self._output = None - self._recon_object = None - self._recon_probe = None - self._metric = None @abstractmethod - def run(self) -> 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: + def run(self, run_id="") -> None: pass diff --git a/ptycho/ptycho_example.py b/ptycho/ptycho_example.py index 7193f3b..3b39ca6 100644 --- a/ptycho/ptycho_example.py +++ b/ptycho/ptycho_example.py @@ -10,35 +10,9 @@ class ExamplePtychoEngine(PtychoEngine): def __init__(self, config): super().__init__(config) + self._metric = 0.0 # run single ptychography job based on `config` - def run(self) -> None: - print("[ExamplePtychoEngine] Sleeping for 0.1 second.") + def run(self, run_id="") -> None: + print(f"[{run_id}] [ExamplePtychoEngine] Sleeping for 0.1 second.") 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