#### Note

The process required to implement an optimizable algorithm will always be very similar to what we did\n here.\n It doesn't matter, if the optimization only optimizes a threshold or trains a neuronal network.\n The structure will be very similar.

\n\nFrom a scientific perspective, we optimize our parameter by trying to find all R-peaks without a height restriction\nfirst.\nBased on the detected R-peaks, we determine, which of them are actually correctly detected, by checking if they are\nwithin the threshold `r_peak_match_tolerance_s` of a reference R-peak.\nThen we find the best height threshold to maximise our predictive power within these preliminary detected peaks.\n\nAgain, there are probably better ways to do it... But this is just an example, and we already have way too much code\nthat is not relevant for you to understand the basics of Algorithms.\n\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"from sklearn.metrics import roc_curve\nfrom tpcp import HyperParameter, OptimizableParameter, make_optimize_safe\n\nfrom examples.algorithms.algorithms_qrs_detection_final import (\n match_events_with_reference,\n)\n\n\nclass OptimizableQrsDetector(QRSDetector):\n min_r_peak_height_over_baseline: OptimizableParameter[float]\n r_peak_match_tolerance_s: HyperParameter[float]\n\n def __init__(\n self,\n max_heart_rate_bpm: float = 200.0,\n min_r_peak_height_over_baseline: float = 1.0,\n r_peak_match_tolerance_s: float = 0.01,\n high_pass_filter_cutoff_hz: float = 1,\n ):\n self.r_peak_match_tolerance_s = r_peak_match_tolerance_s\n super().__init__(\n max_heart_rate_bpm=max_heart_rate_bpm,\n min_r_peak_height_over_baseline=min_r_peak_height_over_baseline,\n high_pass_filter_cutoff_hz=high_pass_filter_cutoff_hz,\n )\n\n @make_optimize_safe\n def self_optimize(\n self,\n ecg_data: list[pd.Series],\n r_peaks: list[pd.Series],\n sampling_rate_hz: float,\n ):\n all_labels = []\n all_peak_heights = []\n for d, p in zip(ecg_data, r_peaks):\n filtered = self._filter(d.to_numpy().flatten(), sampling_rate_hz)\n # Find all potential peaks without the height threshold\n potential_peaks = self._search_strategy(\n filtered, sampling_rate_hz, use_height=False\n )\n # Determine the label for each peak, by matching them with our ground truth\n labels = np.zeros(potential_peaks.shape)\n matches = match_events_with_reference(\n events=potential_peaks,\n reference=p.to_numpy().astype(int),\n tolerance=self.r_peak_match_tolerance_s * sampling_rate_hz,\n )\n tp_matches = matches[(~np.isnan(matches)).all(axis=1), 0].astype(\n int\n )\n labels[tp_matches] = 1\n labels = labels.astype(bool)\n all_labels.append(labels)\n all_peak_heights.append(filtered[potential_peaks])\n all_labels = np.hstack(all_labels)\n all_peak_heights = np.hstack(all_peak_heights)\n # We \"brute-force\" a good cutoff by testing a bunch of thresholds and then calculating the Youden Index for\n # each.\n fpr, tpr, thresholds = roc_curve(all_labels, all_peak_heights)\n youden_index = tpr - fpr\n # The best Youden index gives us a balance between sensitivity and specificity.\n self.min_r_peak_height_over_baseline = thresholds[\n np.argmax(youden_index)\n ]\n return self"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Testing the implementation\nTo test the trainable implementation, we need a train and a test set.\nIn this case we simply use the first two recordings as train set and a third recording as test set.\n\nThen we first call `self_optimize` with the train data.\n\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"train_data = example_data[:2]\ntrain_ecg_data = [d.data[\"ecg\"] for d in train_data]\ntrain_r_peaks = [d.r_peak_positions_[\"r_peak_position\"] for d in train_data]\n\nalgorithm = OptimizableQrsDetector()\nalgorithm = algorithm.self_optimize(\n train_ecg_data, train_r_peaks, train_data.sampling_rate_hz\n)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"After the optimization, we can access the modified parameters.\n\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"print(\n \"The optimized value of the threshold `min_r_peak_height_over_baseline` is:\",\n algorithm.min_r_peak_height_over_baseline,\n)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Then we can apply the algorithm to our test set.\nAnd again, we can see that the algorithm works fine on the piece of data we are inspecting here.\n\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"test_data = example_data[3]\ntest_ecg_data = test_data.data[\"ecg\"]\n\nalgorithm = algorithm.detect(test_ecg_data, test_data.sampling_rate_hz)\n\n# Visualize the results\nplt.figure()\nplt.plot(test_ecg_data[:5000])\nsubset_peaks = algorithm.r_peak_positions_[algorithm.r_peak_positions_ < 5000.0]\nplt.plot(subset_peaks, test_ecg_data[subset_peaks], \"s\")\nplt.show()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.14"
}
},
"nbformat": 4,
"nbformat_minor": 0
}