Coverage for graphica/gui/datasets/operations/fitting.py: 85.3%
240 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 16:38 +0000
1"""曲線フィット(単発・一括・多峰分離)と、フィット結果の再表示・注釈への焼き込みの本体。"""
2import numpy as np
3import pandas as pd
4from PySide6.QtCore import Qt
5from PySide6.QtWidgets import QMessageBox, QProgressDialog
7from graphica.gui import notify
8from graphica.core.analysis import (calculate_confidence_band, calculate_curve_fit, fit_curve_task,
9 get_fit_model_id, multi_peak_fit_task)
10from graphica.core.dataset import Dataset
11from graphica.core.fit_models import fit_type_label
12from graphica.core.provenance import build_provenance
13from graphica.gui.datasets.operations.runner import Operation
14from graphica.gui.dialogs import FitDialog, MultiPeakFitDialog, ResultDialog
15from graphica.gui.task_runner import TaskRunner
18def build_fit_result_dict(fit_type, custom_formula, fit, weighted, x_range, source_dataset,
19 p0_overrides=None, fixed_params=None, bounds=None, loss='linear'):
20 """Dataset.fit_result に保存する形。pickle と JSON の両方で往復できるよう素の Python 型にする。"""
21 popt, pcov, perr = fit['popt'], fit['pcov'], fit['perr']
22 return {
23 'fit_type': fit_type,
24 'fit_model_id': get_fit_model_id(fit_type),
25 'custom_formula': custom_formula,
26 'param_names': list(fit['param_names']),
27 'params': [float(v) for v in popt],
28 'param_errors': [float(v) for v in perr],
29 'covariance': [[float(v) for v in row] for row in pcov],
30 'r_squared': float(fit['r_squared']),
31 'residuals': [float(v) for v in fit['residuals']],
32 'residual_x': [float(v) for v in fit['x_data_used']],
33 'weighted': bool(weighted),
34 'x_range': [float(x_range[0]), float(x_range[1])] if x_range is not None else None,
35 'source_dataset_id': source_dataset.dataset_id,
36 'source_dataset_name': source_dataset.name,
37 'p0_overrides': dict(p0_overrides) if p0_overrides else {},
38 'fixed_params': dict(fixed_params) if fixed_params else {},
39 'bounds': {k: [float(v[0]), float(v[1])] for k, v in bounds.items()} if bounds else {},
40 'loss': loss,
41 }
44def build_multi_peak_fit_result_dict(fit, source_dataset):
45 """多峰分離版。component_type / n_components は core/methods_text.py が参照するので名前を変えない。"""
46 popt, pcov, perr = fit['popt'], fit['pcov'], fit['perr']
47 return {
48 'fit_type': 'multi_peak',
49 'component_type': fit['component_type'],
50 'n_components': fit['n_components'],
51 'baseline_type': fit['baseline_type'],
52 'components': fit['components'],
53 'param_names': list(fit['param_names']),
54 'params': [float(v) for v in popt],
55 'param_errors': [float(v) for v in perr],
56 'covariance': [[float(v) for v in row] for row in pcov],
57 'r_squared': float(fit['r_squared']),
58 'residuals': [float(v) for v in fit['residuals']],
59 'residual_x': [float(v) for v in fit['x_data_used']],
60 'source_dataset_id': source_dataset.dataset_id,
61 'source_dataset_name': source_dataset.name,
62 }
65def add_band_columns_to_fit_df(fit_df, fit, band_type):
66 """
67 信頼帯・予測帯の y_lower / y_upper 列を足し、足せた band_type を返す(Dataset.fit_band_display 用)。
68 自由度不足などで計算できなければ列を足さず None。フィット自体は成功なので失敗にしない。
69 """
70 if band_type is None:
71 return None
72 try:
73 band = calculate_confidence_band(
74 fit['x_fit'], fit['fit_func'], fit['popt'], fit['pcov'], fit['residuals'], band_type=band_type,
75 )
76 except ValueError:
77 return None
78 fit_df['y_lower'] = band['y_lower']
79 fit_df['y_upper'] = band['y_upper']
80 return band_type
83def _param_lines(param_names, params, errors, fixed_params=None):
84 """「名前 = 値 ± 標準誤差」。固定したパラメータと、誤差を持たない古い結果は値だけ。"""
85 fixed = fixed_params or {}
86 if errors is None:
87 errors = [None] * len(params)
88 text = ""
89 for name, value, err in zip(param_names, params, errors):
90 if err is None or name in fixed:
91 text += f" {name} = {value: .4e}\n"
92 else:
93 text += f" {name} = {value: .4e} ± {err:.1e}\n"
94 return text
97def _fit_summary_text(fit_type, custom_formula, param_names, params, param_errors, r_squared,
98 weighted, x_range, fixed_params, bounds, loss):
99 fit_label = fit_type if custom_formula is None else f"{fit_type} {custom_formula}"
100 text = f"[{fit_label}] のフィッティング結果:\n"
101 text += _param_lines(param_names, params, param_errors, fixed_params)
102 text += f" R^2 = {r_squared: .5f}\n"
103 if weighted:
104 text += " (Y誤差列を重みとして使用)\n"
105 if x_range is not None:
106 text += f" (フィット範囲: {x_range[0]: .4g} 〜 {x_range[1]: .4g})\n"
107 if fixed_params:
108 text += f" (固定: {fixed_params})\n"
109 if bounds:
110 text += f" (範囲拘束: {bounds})\n"
111 if loss != 'linear':
112 text += f" (ロバストフィット: {loss})\n"
113 return text
116def multi_peak_summary_text(component_label, n_components, param_names, params, param_errors, r_squared):
117 text = f"[多峰分離({component_label} x{n_components})] のフィッティング結果:\n"
118 text += _param_lines(param_names, params, param_errors)
119 return text + f" R^2 = {r_squared: .5f}\n"
122def format_fit_result_text(fit_result):
123 """保存済みの fit_result だけから、フィット直後と同じ体裁の結果文を作る(再フィットしない)。"""
124 return _fit_summary_text(
125 fit_type_label(fit_result), fit_result.get('custom_formula'),
126 fit_result.get('param_names', []), fit_result.get('params', []), fit_result.get('param_errors'),
127 fit_result.get('r_squared', float('nan')), fit_result.get('weighted'), fit_result.get('x_range'),
128 fit_result.get('fixed_params'), fit_result.get('bounds'), fit_result.get('loss', 'linear'),
129 )
132def _fit_dataset(source, fit, fit_type, custom_formula, sigma, x_range,
133 p0_overrides, fixed_params, bounds, band_type, loss, operation):
134 """フィット結果から「Fit (元の名前)」のデータセットを作る。GUI に触れないのでワーカーからも呼べる。"""
135 fit_result = build_fit_result_dict(
136 fit_type=fit_type, custom_formula=custom_formula, fit=fit,
137 weighted=sigma is not None, x_range=x_range, source_dataset=source,
138 p0_overrides=p0_overrides, fixed_params=fixed_params, bounds=bounds, loss=loss,
139 )
140 text = _fit_summary_text(fit_type, custom_formula, fit['param_names'], fit['popt'], fit['perr'], fit['r_squared'],
141 sigma is not None, x_range, fixed_params, bounds, loss)
142 fit_df = pd.DataFrame({'x_fit': fit['x_fit'], 'y_fit': fit['y_fit']})
143 applied_band_type = add_band_columns_to_fit_df(fit_df, fit, band_type)
144 dataset = Dataset(
145 name=f"Fit ({source.name})", df=fit_df, x_col_name='x_fit', y_col_name='y_fit',
146 color=source.color, linestyle='--', marker='None', linewidth=source.linewidth,
147 use_secondary_y=source.use_secondary_y, subplot_target=source.subplot_target,
148 fit_info=text, fit_result=fit_result, fit_band_display=applied_band_type,
149 provenance=build_provenance(operation, fit_result, [source]),
150 )
151 return dataset, text
154def batch_fit_worker(datasets, fit_type, custom_formula, use_weighted, x_range,
155 p0_overrides, fixed_params, bounds, band_type, loss='linear',
156 report_progress=None, is_cancelled=None):
157 """
158 一括フィットの計算(バックグラウンドのスレッドで動く。Qt には触れない)。
159 1件のフィットは中断できないので、キャンセルは「残りを飛ばす」単位。
160 """
161 results = []
162 total = len(datasets)
163 for i, dataset in enumerate(datasets):
164 if is_cancelled is not None and is_cancelled():
165 break
166 if report_progress is not None:
167 report_progress(i, total, dataset.name)
168 sigma = dataset.y_err_data if use_weighted else None
169 try:
170 fit = calculate_curve_fit(
171 dataset.x_data, dataset.y_data, fit_type, custom_formula=custom_formula,
172 sigma=sigma, x_range=x_range,
173 p0_overrides=p0_overrides, fixed_params=fixed_params, bounds=bounds, loss=loss,
174 )
175 except Exception as e:
176 results.append({'source_name': dataset.name, 'fit_dataset': None, 'error': str(e)})
177 continue
178 fit_dataset, _ = _fit_dataset(dataset, fit, fit_type, custom_formula, sigma, x_range,
179 p0_overrides, fixed_params, bounds, band_type, loss, 'batch_curve_fit')
180 results.append({'source_name': dataset.name, 'fit_dataset': fit_dataset, 'error': None})
181 return results
184def stop_runner(runner):
185 """実行中の QThread を破棄すると Qt がプロセスを落とすので、切断してから終わるまで待つ。"""
186 try:
187 runner.succeeded.disconnect()
188 runner.failed.disconnect()
189 except (RuntimeError, TypeError):
190 pass
191 runner.requestInterruption()
192 runner.wait()
193 runner.deleteLater()
196FIT_RESULT_WINDOW = "fit"
197FIT_RUNNER = "fit_runner"
198BATCH_FIT_RUNNER = "batch_fit_runner"
199MULTI_PEAK_FIT_RUNNER = "multi_peak_fit_runner"
200RUNNER_NAMES = (FIT_RUNNER, BATCH_FIT_RUNNER, MULTI_PEAK_FIT_RUNNER)
202_BUSY_TITLE = "実行中"
203_FIT_BUSY_TEXT = "別のフィット処理が実行中です。完了までお待ちください。"
204_EXPORT_TITLE = "フィット結果のエクスポート"
206# 計算の終わりを受ける関数は run_operation の外(Qt のシグナル)で呼ばれるので、流れを止める例外は投げない。
209def _show_result(op, title, text, csv_data, residual_x, residual_y):
210 op.show_result_window(FIT_RESULT_WINDOW, lambda: ResultDialog(
211 title, text, op.parent, csv_data=csv_data, residual_x=residual_x, residual_y=residual_y))
214def _ensure_idle(op, runner_name, text):
215 if op.state.runners.get(runner_name) is not None:
216 op.stop_with_information(text, title=_BUSY_TITLE)
219def _finish_runner(op, runner_name):
220 runner = op.state.runners.get(runner_name)
221 if runner is not None: 221 ↛ exitline 221 didn't return from function '_finish_runner' because the condition on line 221 was always true
222 runner.wait()
223 runner.deleteLater()
224 op.state.runners[runner_name] = None
227# --- 単発フィット ---
229def fit_current_dataset(op):
230 """設定はダイアログで集め、計算はバックグラウンドで行う。"""
231 source = op.current_dataset()
232 _ensure_idle(op, FIT_RUNNER, _FIT_BUSY_TEXT)
233 x_data, y_data = source.x_data, source.y_data
234 numeric_x = op.as_numbers(source, "x")
235 x_min = float(np.min(numeric_x)) if len(numeric_x) else None
236 x_max = float(np.max(numeric_x)) if len(numeric_x) else None
237 (fit_type, custom_formula, use_weighted, x_range,
238 p0_overrides, fixed_params, bounds, band_type, loss) = FitDialog.get_fit_type(op.parent, x_min=x_min, x_max=x_max)
239 if fit_type is None:
240 op.stop()
241 sigma = source.y_err_data if use_weighted else None
243 runner = TaskRunner(
244 fit_curve_task, x_data, y_data, fit_type, custom_formula=custom_formula,
245 sigma=sigma, x_range=x_range,
246 p0_overrides=p0_overrides, fixed_params=fixed_params, bounds=bounds, loss=loss,
247 )
248 runner.succeeded.connect(lambda fit: _on_fit_succeeded(
249 op, source, fit_type, custom_formula, sigma, x_range, p0_overrides, fixed_params, bounds, band_type, loss, fit,
250 ))
251 runner.failed.connect(lambda error_message: _on_fit_failed(op, error_message))
252 op.state.runners[FIT_RUNNER] = runner
253 op.host.set_fit_button_enabled(False)
254 runner.start()
257def _finish_fit_runner(op):
258 _finish_runner(op, FIT_RUNNER)
259 op.host.set_fit_button_enabled(True)
262def _on_fit_failed(op, error_message):
263 _finish_fit_runner(op)
264 notify.warning(op.parent, "フィットエラー", f"フィッティングに失敗しました:\n{error_message}")
267def _on_fit_succeeded(op, source, fit_type, custom_formula, sigma, x_range,
268 p0_overrides, fixed_params, bounds, band_type, loss, fit):
269 _finish_fit_runner(op)
270 dataset, text = _fit_dataset(source, fit, fit_type, custom_formula, sigma, x_range,
271 p0_overrides, fixed_params, bounds, band_type, loss, 'curve_fit')
272 op.host.add_derived_dataset(dataset, source, op.title)
273 csv_data = pd.DataFrame({
274 'パラメータ': list(fit['param_names']) + ['R^2'],
275 '値': list(fit['popt']) + [fit['r_squared']],
276 })
277 # 残差は NaN の除外とフィット範囲の適用後の点なので、x もそれに合わせる
278 _show_result(op, "フィッティング完了", text, csv_data, fit['x_data_used'], fit['residuals'])
281# --- 一括フィット ---
283def batch_fit_selected(op):
284 """選択中の各データセットに同じ設定でフィットする。結果はまとめて足し、描き直しは1回だけ。"""
285 parent = op.parent
286 selected = op.selected_datasets(at_least=2, message="2つ以上のデータセットを選択してください。")
287 _ensure_idle(op, BATCH_FIT_RUNNER, "別のバッチフィット処理が実行中です。完了までお待ちください。")
288 (fit_type, custom_formula, use_weighted, x_range,
289 p0_overrides, fixed_params, bounds, band_type, loss) = FitDialog.get_fit_type(parent)
290 if fit_type is None:
291 op.stop()
292 target_folder = op.host.target_folder_for_new_dataset()
294 progress = QProgressDialog("バッチカーブフィットを実行中...", "キャンセル", 0, len(selected), parent)
295 progress.setWindowModality(Qt.WindowModality.WindowModal)
296 progress.setMinimumDuration(0)
297 progress.setValue(0)
299 runner = TaskRunner(
300 batch_fit_worker, selected, fit_type, custom_formula, use_weighted, x_range,
301 p0_overrides, fixed_params, bounds, band_type, loss,
302 )
303 runner.progress.connect(lambda done, total, message: progress.setValue(done))
304 progress.canceled.connect(runner.requestInterruption)
305 runner.succeeded.connect(lambda results: _on_batch_succeeded(op, results, target_folder, progress))
306 runner.failed.connect(lambda msg: _on_batch_failed(op, msg, progress))
307 op.state.runners[BATCH_FIT_RUNNER] = runner
308 runner.start()
311def _finish_batch_runner(op, progress):
312 _finish_runner(op, BATCH_FIT_RUNNER)
313 progress.close()
316def _on_batch_failed(op, error_message, progress):
317 _finish_batch_runner(op, progress)
318 notify.warning(op.parent, op.title, f"バッチフィット処理に失敗しました:\n{error_message}")
321def _on_batch_succeeded(op, results, target_folder, progress):
322 """キャンセルされたときは、終わった分だけが results に入っている。"""
323 _finish_batch_runner(op, progress)
324 added = [r['fit_dataset'] for r in results if r['fit_dataset'] is not None]
325 succeeded = [r['source_name'] for r in results if r['fit_dataset'] is not None]
326 failed = [f"{r['source_name']}: {r['error']}" for r in results
327 if r['fit_dataset'] is None and r['error'] is not None]
328 if added: 328 ↛ 330line 328 didn't jump to line 330 because the condition on line 328 was always true
329 op.host.add_datasets_to_folder(added, target_folder, op.title)
330 if not succeeded and not failed: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 return
332 message = f"{len(succeeded)}件のフィットに成功しました。"
333 if failed:
334 message += "\n\n失敗:\n" + "\n".join(failed)
335 notify.information(op.parent, op.title, message)
338# --- 多峰分離フィット ---
340def multi_peak_fit_current_dataset(op):
341 """
342 グラフ上のクリックで置いたピークの初期値をダイアログに引き継ぐ。ダイアログを閉じたら
343 (OK でもキャンセルでも)置いた初期値は消し、配置モードも抜ける。
344 """
345 source = op.current_dataset()
346 _ensure_idle(op, MULTI_PEAK_FIT_RUNNER, _FIT_BUSY_TEXT)
347 x_data, y_data = source.x_data, source.y_data
348 component_type, baseline_type, initial_guesses = MultiPeakFitDialog.get_multi_peak_fit_settings(
349 op.parent, x_data=x_data, y_data=y_data, initial_guesses=op.host.pending_peak_guesses(),
350 )
351 op.host.finish_peak_placement()
352 if component_type is None:
353 op.stop()
355 runner = TaskRunner(multi_peak_fit_task, x_data, y_data, component_type, initial_guesses,
356 baseline_type=baseline_type)
357 runner.succeeded.connect(lambda fit: _on_multi_peak_fit_succeeded(op, source, fit))
358 runner.failed.connect(lambda error_message: _on_multi_peak_fit_failed(op, error_message))
359 op.state.runners[MULTI_PEAK_FIT_RUNNER] = runner
360 op.host.set_multi_peak_fit_button_enabled(False)
361 runner.start()
364def _finish_multi_peak_runner(op):
365 _finish_runner(op, MULTI_PEAK_FIT_RUNNER)
366 op.host.set_multi_peak_fit_button_enabled(True)
369def _on_multi_peak_fit_failed(op, error_message):
370 _finish_multi_peak_runner(op)
371 notify.warning(op.parent, "多峰分離フィットエラー", f"フィッティングに失敗しました:\n{error_message}")
374def _on_multi_peak_fit_succeeded(op, source, fit):
375 _finish_multi_peak_runner(op)
376 popt, param_names, r_squared = fit['popt'], fit['param_names'], fit['r_squared']
377 component_label = dict(MultiPeakFitDialog.COMPONENT_TYPES).get(fit['component_type'], fit['component_type'])
378 text = multi_peak_summary_text(component_label, fit['n_components'], param_names, popt, fit['perr'], r_squared)
380 fit_result = build_multi_peak_fit_result_dict(fit, source_dataset=source)
381 op.host.add_derived_dataset(Dataset(
382 name=f"MultiPeakFit ({source.name})",
383 df=pd.DataFrame({'x_fit': fit['x_fit'], 'y_fit': fit['y_fit']}),
384 x_col_name='x_fit', y_col_name='y_fit',
385 color=source.color, linestyle='--', marker='None', linewidth=source.linewidth,
386 use_secondary_y=source.use_secondary_y, subplot_target=source.subplot_target,
387 fit_info=text, fit_result=fit_result,
388 provenance=build_provenance('multi_peak_fit', fit_result, [source]),
389 ), source, op.title)
390 csv_data = pd.DataFrame({'パラメータ': list(param_names) + ['R^2'], '値': list(popt) + [r_squared]})
391 _show_result(op, "多峰分離フィット完了", text, csv_data, fit['x_data_used'], fit['residuals'])
394# --- 保存済みのフィット結果 ---
396def show_fit_result(op):
397 """
398 現在のデータセットに保存されたフィット結果(fit_result)を、再フィットせずに表と残差で出し直す。
399 続けて、要約をグラフの注釈として焼き込むかを尋ねる。
400 """
401 dataset = op.current_dataset()
402 fit_result = dataset.fit_result
403 if fit_result is None:
404 # メニューは無効化しているが、別の経路から呼ばれたときのため
405 op.stop_with_information(
406 "このデータセットは曲線フィットの結果を持っていません。\n"
407 "曲線フィットで生成されたデータセット(名前が「Fit (...)」の\n"
408 "もの、またはfit_resultを保持しているもの)を選択してください。"
409 )
411 csv_data = pd.DataFrame({
412 'パラメータ': list(fit_result.get('param_names', [])) + ['R^2'],
413 '値': list(fit_result.get('params', [])) + [fit_result.get('r_squared')],
414 })
415 _show_result(op, op.title, format_fit_result_text(fit_result), csv_data,
416 fit_result.get('residual_x'), fit_result.get('residuals'))
418 reply = notify.question(
419 op.parent, op.title,
420 "このフィット結果の要約(パラメータ値±誤差・R^2)を、\n"
421 "グラフ上のテキスト注釈として焼き込みますか?\n"
422 "(焼き込み後は注釈モードで通常の注釈と同様に移動・編集・削除できます)",
423 QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
424 QMessageBox.StandardButton.No
425 )
426 if reply == QMessageBox.StandardButton.Yes:
427 burn_fit_result_annotation(op, dataset, fit_result)
430def burn_fit_result_annotation(op, dataset, fit_result):
431 """
432 フィット結果の要約を、手で置く注釈と同じテキスト注釈として足す(Undo でき、あとから編集できる)。
433 置き場所はフィット曲線の中央の点。端やピークより、どの形の曲線でも曲線上の無難な位置になる。
434 """
435 axis_index = dataset.subplot_target
436 if axis_index is None or axis_index >= op.host.axis_count(): 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true
437 op.stop_with_warning("注釈を追加する対象のプロットが見つかりませんでした。")
438 x_data, y_data = dataset.x_data, dataset.y_data
439 if len(x_data) == 0: 439 ↛ 440line 439 didn't jump to line 440 because the condition on line 439 was never true
440 op.stop_with_warning("フィット曲線にデータ点が無いため、注釈を追加できませんでした。")
441 mid = len(x_data) // 2
442 anchor = (float(x_data[mid]), float(y_data[mid]))
444 lines = [f"フィット: {fit_type_label(fit_result, '')}"]
445 params = fit_result.get('params', [])
446 errors = fit_result.get('param_errors', [None] * len(params))
447 for name, value, err in zip(fit_result.get('param_names', []), params, errors):
448 lines.append(f"{name} = {value:.4g} ± {err:.4g}" if err is not None else f"{name} = {value:.4g}")
449 r_squared = fit_result.get('r_squared')
450 if r_squared is not None: 450 ↛ 453line 450 didn't jump to line 453 because the condition on line 450 was always true
451 lines.append(f"R^2 = {r_squared:.5f}")
453 op.host.add_annotation(axis_index, {
454 'type': 'text', 'text': "\n".join(lines), 'xy': anchor, 'xytext': anchor, 'color': '#000000',
455 }, description="フィット結果の注釈焼き込み")
456 op.host.show_status("フィット結果を注釈として焼き込みました")
459FITTING_OPERATIONS = {
460 "fit_current_dataset": Operation("カーブフィット", fit_current_dataset),
461 "batch_fit_selected": Operation("バッチカーブフィット", batch_fit_selected),
462 "multi_peak_fit_current_dataset": Operation("多峰分離フィット", multi_peak_fit_current_dataset),
463 "show_fit_result": Operation(_EXPORT_TITLE, show_fit_result),
464}
467def burn_operation(dataset, fit_result):
468 return Operation(_EXPORT_TITLE, lambda op: burn_fit_result_annotation(op, dataset, fit_result))