Skip to content

module8

Module 8 Callbacks Package.

This package contains callback modules for Module 8 use cases, focusing on metabolic pathway analysis and consortium design.

Modules:

Name Description
uc_8_2_callbacks

Chemical Class Completeness Scorecard

uc_8_3_callbacks

Compound-Specific KO Completeness Scorecard

uc_8_4_callbacks

Pathway Completeness Scorecard for HADEG Pathways

uc_8_5_callbacks

KEGG Pathway Completeness Scorecard

uc_8_6_callbacks

Pathway-Centric Consortium Design by KO Coverage

uc_8_7_callbacks

Intersection of Genes Across Samples

Usage

from src.presentation.callbacks.module8 import register_uc_8_6_callbacks register_uc_8_6_callbacks(app) from src.presentation.callbacks.module8 import register_uc_8_7_callbacks register_uc_8_7_callbacks(app)

Notes

All callback modules follow the pattern: - register_uc_X_Y_callbacks(app) - Main registration function - toggle_uc_X_Y_info_panel() - Collapse toggle - Additional callbacks specific to use case

Author: BioRemPP Development Team Date: 2025-11-20

Functions

register_uc_8_1_callbacks

register_uc_8_1_callbacks(app, plot_service) -> None

Register all UC-8.1 callbacks with Dash app.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers panel toggle, dropdown initialization, and faceted scatter callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_1_callbacks.py
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
def register_uc_8_1_callbacks(app, plot_service) -> None:
    """
    Register all UC-8.1 callbacks with Dash app.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers panel toggle, dropdown initialization, and faceted scatter callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("[UC-8.1] Registering callbacks")

    # ========================================
    # Callback 1: Toggle Informative Panel
    # ========================================
    @app.callback(
        Output("uc-8-1-collapse", "is_open"),
        Input("uc-8-1-collapse-button", "n_clicks"),
        State("uc-8-1-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_1_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """Toggle UC-8.1 informative panel collapse state."""
        if n_clicks:
            logger.debug(f"[UC-8.1] Toggling info panel: {is_open} -> {not is_open}")
            return not is_open
        return is_open

    # ========================================
    # Callback 2: Initialize Compound Class Dropdown
    # ========================================
    @app.callback(
        Output("uc-8-1-compoundclass-dropdown", "options"),
        [
            Input("uc-8-1-accordion-group", "active_item"),
            Input("merged-result-store", "data"),
        ],
        prevent_initial_call=True,
    )
    def initialize_uc_8_1_dropdown(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> list:
        """
        Populate compound class dropdown with available classes.

        Parameters
        ----------
        active_item : str, optional
            Active accordion item ID.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        list
            Dropdown options list.

        Notes
        -----
        - Triggered when accordion opens or data changes
        - Extracts unique compound classes from BioRemPP data
        - Returns empty list if accordion not active or data unavailable
        """
        logger.info(
            f"[UC-8.1] [CALLBACK 2] Dropdown init triggered, "
            f"active_item: {active_item}"
        )
        logger.debug(f"[UC-8.1] [CALLBACK 2] merged_data type: {type(merged_data)}")
        if isinstance(merged_data, dict):
            logger.debug(f"[UC-8.1] [CALLBACK 2] keys: {merged_data.keys()}")

        # Only populate when accordion is open
        if active_item != "uc-8-1-accordion":
            logger.debug("[UC-8.1] [CALLBACK 2] Accordion not active, skip")
            return []

        if not merged_data or "biorempp_df" not in merged_data:
            logger.warning(
                "[UC-8.1] [CALLBACK 2] No BioRemPP data available for dropdown"
            )
            return []

        try:
            df = pd.DataFrame(merged_data["biorempp_df"])
            logger.debug(
                f"[UC-8.1] [CALLBACK 2] DataFrame created: "
                f"{len(df)} rows, columns: {df.columns.tolist()}"
            )

            if df.empty:
                logger.warning("[UC-8.1] [CALLBACK 2] DataFrame is empty")
                return []

            # Find compound class column
            class_col = _find_column(
                df, ["Compound_Class", "compound_class", "CompoundClass", "class"]
            )

            if not class_col:
                logger.warning(
                    f"[UC-8.1] [CALLBACK 2] Compound class column not found. "
                    f"Available: {df.columns.tolist()}"
                )
                return []

            # Get unique compound classes
            classes = sorted(df[class_col].dropna().unique().tolist())
            options = [{"label": c, "value": c} for c in classes]

            logger.info(
                f"[UC-8.1] [CALLBACK 2] Dropdown populated: " f"{len(classes)} classes"
            )
            logger.debug(f"[UC-8.1] [CALLBACK 2] Classes: {classes[:5]}...")

            return options

        except Exception as e:
            logger.error(
                f"[UC-8.1] [CALLBACK 2] Error initializing dropdown: {e}", exc_info=True
            )
            return []

    # ========================================
    # Callback 3: Render Faceted Scatter
    # ========================================
    @app.callback(
        Output("uc-8-1-chart", "children"),
        Input("uc-8-1-compoundclass-dropdown", "value"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_1(
        selected_class: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.1 faceted scatter on compound class dropdown selection.

        Parameters
        ----------
        selected_class : str, optional
            Selected compound class from dropdown.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        html.Div
            Container with chart or error message.

        Notes
        -----
        - Filters BioRemPP data by selected compound class
        - Groups samples by frozenset of compounds (identical profiles)
        - Applies greedy set cover algorithm for minimal group selection
        - Calculates unique KO counts per compound for color scaling
        - Generates faceted scatter with one subplot per minimized group
        """
        logger.info(f"[UC-8.1] [CALLBACK 3] ========== RENDER TRIGGERED ==========")
        logger.info(f"[UC-8.1] [CALLBACK 3] selected_class: '{selected_class}'")
        logger.debug(f"[UC-8.1] [CALLBACK 3] merged_data type: {type(merged_data)}")
        logger.debug(f"[UC-8.1] [CALLBACK 3] is None: {merged_data is None}")

        # Validate compound class selection
        if not selected_class:
            logger.debug("[UC-8.1] No compound class selected")
            return _create_info_message(
                "Please select a Compound Class from the dropdown above.",
                "bi bi-arrow-up-circle",
            )

        try:
            # ========================================
            # Step 1: Validate merged_data structure
            # ========================================
            if not merged_data:
                logger.warning("[UC-8.1] merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please upload and process data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict):
                logger.error("[UC-8.1] merged_data is not a dictionary")
                return _create_error_message(
                    "Invalid data structure. Please reload the application.",
                    "bi bi-x-circle",
                )

            if "biorempp_df" not in merged_data:
                logger.error("[UC-8.1] merged_data does not contain 'biorempp_df' key")
                return _create_error_message(
                    "BioRemPP data not found. This use case requires "
                    "BioRemPP database.",
                    "bi bi-database-x",
                )

            # ========================================
            # Step 2: Extract DataFrame
            # ========================================
            logger.debug("[UC-8.1] Extracting DataFrame from merged_data")
            biorempp_data = merged_data["biorempp_df"]

            if not biorempp_data:
                logger.warning("[UC-8.1] biorempp_df is empty")
                return _create_error_message(
                    "BioRemPP dataset is empty. Please check your input data.",
                    "bi bi-inbox",
                )

            df = pd.DataFrame(biorempp_data)

            if df.empty:
                logger.warning("[UC-8.1] DataFrame is empty after conversion")
                return _create_error_message(
                    "No data available after processing.", "bi bi-inbox"
                )

            logger.info(
                f"[UC-8.1] Processing DataFrame: {len(df)} rows, "
                f"{len(df.columns)} columns"
            )

            # ========================================
            # Step 3: Map column names flexibly
            # ========================================
            col_map = {}

            # Sample column
            sample_col = _find_column(
                df, ["Sample", "sample", "sample_name", "SampleName"]
            )
            if sample_col:
                col_map["sample"] = sample_col

            # Compound name column
            compound_col = _find_column(
                df, ["Compound_Name", "compound_name", "CompoundName", "Compound"]
            )
            if compound_col:
                col_map["compoundname"] = compound_col

            # Compound class column
            class_col = _find_column(
                df, ["Compound_Class", "compound_class", "CompoundClass", "class"]
            )
            if class_col:
                col_map["compoundclass"] = class_col

            # KO column (optional, for color)
            ko_col = _find_column(df, ["KO", "ko", "ko_id", "KO_ID"])
            if ko_col:
                col_map["ko"] = ko_col

            # ========================================
            # Step 4: Validate required columns found
            # ========================================
            required = ["sample", "compoundname", "compoundclass"]
            missing_cols = [col for col in required if col not in col_map]

            if missing_cols:
                logger.error(
                    f"[UC-8.1] Missing columns: {missing_cols}. "
                    f"Available: {df.columns.tolist()}"
                )
                return _create_error_message(
                    f"Required columns not found: {', '.join(missing_cols)}. "
                    f"Available columns: {', '.join(df.columns[:5])}...",
                    "bi bi-exclamation-octagon",
                )

            # ========================================
            # Step 5: Prepare data
            # ========================================
            # Rename columns to standard names
            rename_map = {v: k for k, v in col_map.items()}
            df_work = df.rename(columns=rename_map).copy()

            # Filter by selected compound class
            df_filtered = df_work[df_work["compoundclass"] == selected_class].copy()

            if df_filtered.empty:
                logger.warning(f"[UC-8.1] No data for class '{selected_class}'")
                return _create_error_message(
                    f"No data found for compound class: {selected_class}",
                    "bi bi-search",
                )

            # Clean data
            df_filtered = df_filtered.dropna(subset=["sample", "compoundname"])

            for col in ["sample", "compoundname"]:
                df_filtered[col] = df_filtered[col].astype(str).str.strip()

            # Remove placeholder values
            placeholder_values = ["#N/D", "#N/A", "N/D", "", "nan", "None"]
            for col in ["sample", "compoundname"]:
                df_filtered = df_filtered[~df_filtered[col].isin(placeholder_values)]

            if df_filtered.empty:
                return _create_error_message(
                    f"No valid data for compound class: {selected_class}",
                    "bi bi-funnel",
                )

            logger.info(
                f"[UC-8.1] Filtered to {len(df_filtered)} rows for "
                f"class '{selected_class}'"
            )

            # ========================================
            # Step 6: Group samples by compound profile
            # ========================================
            grouped_df, groups = _group_by_compound_profile(df_filtered)

            if grouped_df.empty or not groups:
                return _create_error_message(
                    "No sample groups found after grouping.", "bi bi-diagram-3"
                )

            logger.info(f"[UC-8.1] Created {len(groups)} groups")

            # ========================================
            # Step 7: Minimize groups with set cover
            # ========================================
            minimized_groups = _minimize_groups(grouped_df)

            if not minimized_groups:
                return _create_error_message(
                    "Could not determine minimal groups.", "bi bi-exclamation-circle"
                )

            # Filter to minimized groups only
            final_df = grouped_df[grouped_df["_group"].isin(minimized_groups)].copy()

            logger.info(f"[UC-8.1] Minimized to {len(minimized_groups)} groups")

            # ========================================
            # Step 8: Calculate KO counts (for color)
            # ========================================
            if "ko" in col_map:
                ko_counts = _calculate_ko_counts(df_filtered)
                if ko_counts is not None:
                    final_df = final_df.merge(
                        ko_counts, left_on="compoundname", right_index=True, how="left"
                    )
                    final_df["_unique_ko_count"] = (
                        final_df["_unique_ko_count"].fillna(0).astype(int)
                    )
                else:
                    final_df["_unique_ko_count"] = 1
            else:
                final_df["_unique_ko_count"] = 1

            # ========================================
            # Step 9: Generate plot
            # ========================================
            fig = _create_frozenset_figure(final_df, minimized_groups, selected_class)

            logger.info("[UC-8.1] Faceted scatter generation successful")

            # ========================================
            # Step 10: Return chart component
            # ========================================
            n_groups = len(minimized_groups)
            n_samples = final_df["sample"].nunique()
            n_compounds = final_df["compoundname"].nunique()

            # Prepare a safe download filename using canonical helper
            selected_class_safe = str(selected_class).replace(" ", "_")
            try:
                suggested = sanitize_filename(
                    "UC-8.1", f"minimal_grouping_{selected_class_safe}", "png"
                )
            except Exception:
                suggested = f"minimal_grouping_{selected_class_safe}.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                [
                    # Statistics summary
                    html.Div(
                        [
                            html.Small(
                                [
                                    html.I(className="bi bi-info-circle me-2"),
                                    f"Minimal Coverage: {n_groups} functional guilds | "
                                    f"{n_samples} samples | {n_compounds} compounds | "
                                    f"Class: {selected_class}",
                                ],
                                className="text-muted",
                            )
                        ],
                        className="mb-2",
                    ),
                    # Graph container with overflow control
                    html.Div(
                        [
                            dcc.Graph(
                                id="uc-8-1-graph",
                                figure=fig,
                                config={
                                    "displayModeBar": True,
                                    "displaylogo": False,
                                    "responsive": True,
                                    "modeBarButtonsToRemove": [
                                        "pan2d",
                                        "lasso2d",
                                        "select2d",
                                    ],
                                    "toImageButtonOptions": {
                                        "format": "svg",
                                        "filename": base_filename,
                                        "height": 600,
                                        "width": 900,
                                        "scale": 6,
                                    },
                                },
                                style={
                                    "height": "600px",
                                    "width": "100%",
                                    "minWidth": "100%",
                                },
                                className="mt-3",
                            )
                        ],
                        style={
                            "width": "100%",
                            "overflowX": "auto",
                            "overflowY": "hidden",
                        },
                    ),
                ]
            )

        except ValueError as ve:
            logger.error(
                f"[UC-8.1] ValueError during processing: {str(ve)}", exc_info=True
            )
            return _create_error_message(
                f"Data validation error: {str(ve)}", "bi bi-exclamation-triangle"
            )

        except Exception as e:
            logger.error(f"[UC-8.1] Unexpected error: {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred: {str(e)}", "bi bi-bug"
            )

    logger.info("[UC-8.1] All callbacks registered successfully")

register_uc_8_2_callbacks

register_uc_8_2_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.2.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_2_callbacks.py
def register_uc_8_2_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.2.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.2 callbacks")

    @app.callback(
        Output("uc-8-2-collapse", "is_open"),
        Input("uc-8-2-collapse-button", "n_clicks"),
        State("uc-8-2-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_2_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.2 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.2 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-2-chart", "children"),
        Input("uc-8-2-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_2(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.2 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts BioRemPP DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / class KOs) × 100%
        - Generates heatmap with samples (rows) × compound classes (columns)
        - Uses PlotService with HeatmapScoredStrategy
        """
        logger.debug(f"UC-8.2 render callback triggered. Active item: {active_item}")

        # Check if UC-8.2 accordion is active
        if not active_item or active_item != "uc-8-2-accordion":
            logger.debug("UC-8.2 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.2: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "biorempp_df" not in merged_data:
                logger.error("UC-8.2: merged_data does not contain 'biorempp_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'biorempp_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.2: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["biorempp_df"])

            if df.empty:
                logger.warning("UC-8.2: DataFrame is empty")
                return _create_error_message(
                    "The dataset is empty. Please load data with KO, Sample, and Compound_Class information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.2: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.2: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.2: Mapped KO to column '{col_name}'")
                    break

            # Try to find Compound_Class column
            class_candidates = [
                "Compound_Class",
                "compound_class",
                "CompoundClass",
                "class",
                "Class",
                "chemical_class",
                "Chemical_Class",
            ]
            for col_name in class_candidates:
                if col_name in df.columns:
                    col_map["Compound_Class"] = col_name
                    logger.debug(
                        f"UC-8.2: Mapped Compound_Class to column '{col_name}'"
                    )
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Compound_Class"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.2: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.2: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Compound_Class"] = (
                df_processed["Compound_Class"].astype(str).str.strip()
            )

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Compound_Class"] != "")
                & (df_processed["Compound_Class"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.2: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.2: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Compound_Class'].nunique()} compound classes"
            )

            # Generate heatmap using PlotService
            # NOTE: This will require a uc_8_2_config.yaml file to be created
            # with HeatmapScoredStrategy configuration
            logger.debug("UC-8.2: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.2",
                filters={},
                customizations={},
            )

            logger.info("UC-8.2: Heatmap generated successfully")

            # Return chart wrapped in loading component
            # Compute canonical filename
            try:
                suggested = sanitize_filename(
                    "UC-8.2", "chemical_class_completeness", "png"
                )
            except Exception:
                suggested = "chemical_class_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.2: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.2: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_2_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.2: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

register_uc_8_3_callbacks

register_uc_8_3_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.3.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_3_callbacks.py
def register_uc_8_3_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.3.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.3 callbacks")

    @app.callback(
        Output("uc-8-3-collapse", "is_open"),
        Input("uc-8-3-collapse-button", "n_clicks"),
        State("uc-8-3-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_3_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.3 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.3 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-3-chart", "children"),
        Input("uc-8-3-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_3(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.3 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts BioRemPP DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / compound KOs) × 100%
        - Generates heatmap with samples (rows) × compound names (columns)
        - Uses PlotService with HeatmapScoredStrategy
        """
        logger.debug(f"UC-8.3 render callback triggered. Active item: {active_item}")

        # Check if UC-8.3 accordion is active
        if not active_item or active_item != "uc-8-3-accordion":
            logger.debug("UC-8.3 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.3: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "biorempp_df" not in merged_data:
                logger.error("UC-8.3: merged_data does not contain 'biorempp_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'biorempp_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.3: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["biorempp_df"])

            if df.empty:
                logger.warning("UC-8.3: DataFrame is empty")
                return _create_error_message(
                    "The dataset is empty. Please load data with KO, Sample, and Compound_Name information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.3: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.3: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.3: Mapped KO to column '{col_name}'")
                    break

            # Try to find Compound_Name column
            compound_candidates = [
                "Compound_Name",
                "compound_name",
                "compoundname",
                "CompoundName",
                "compound",
                "Compound",
            ]
            for col_name in compound_candidates:
                if col_name in df.columns:
                    col_map["Compound_Name"] = col_name
                    logger.debug(f"UC-8.3: Mapped Compound_Name to column '{col_name}'")
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Compound_Name"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.3: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.3: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Compound_Name"] = (
                df_processed["Compound_Name"].astype(str).str.strip()
            )

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Compound_Name"] != "")
                & (df_processed["Compound_Name"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.3: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.3: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Compound_Name'].nunique()} compounds"
            )

            # Generate heatmap using PlotService
            logger.debug("UC-8.3: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.3",
                filters={},
                customizations={},
            )

            logger.info("UC-8.3: Heatmap generated successfully")

            # Use canonical filename
            try:
                suggested = sanitize_filename(
                    "UC-8.3", "compound_specific_completeness", "png"
                )
            except Exception:
                suggested = "compound_specific_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.3: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.3: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_3_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.3: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

register_uc_8_4_callbacks

register_uc_8_4_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.4.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_4_callbacks.py
def register_uc_8_4_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.4.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.4 callbacks")

    @app.callback(
        Output("uc-8-4-collapse", "is_open"),
        Input("uc-8-4-collapse-button", "n_clicks"),
        State("uc-8-4-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_4_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.4 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.4 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-4-chart", "children"),
        Input("uc-8-4-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_4(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.4 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'hadeg_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts HADEG DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / pathway KOs) × 100%
        - Generates heatmap with samples (rows) × pathways (columns)
        - Uses PlotService with HeatmapScoredStrategy
        """
        logger.debug(f"UC-8.4 render callback triggered. Active item: {active_item}")

        # Check if UC-8.4 accordion is active
        if not active_item or active_item != "uc-8-4-accordion":
            logger.debug("UC-8.4 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.4: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "hadeg_df" not in merged_data:
                logger.error("UC-8.4: merged_data does not contain 'hadeg_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'hadeg_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.4: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["hadeg_df"])

            if df.empty:
                logger.warning("UC-8.4: DataFrame is empty")
                return _create_error_message(
                    "The HADEG dataset is empty. Please load data with KO, Sample, and Pathway information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.4: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )
            logger.debug(f"UC-8.4: Available columns: {df.columns.tolist()}")

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.4: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.4: Mapped KO to column '{col_name}'")
                    break

            # Try to find Pathway column
            pathway_candidates = [
                "Pathway",
                "pathway",
                "compound_pathway",
                "Compound_Pathway",
                "PathwayName",
                "pathway_name",
            ]
            for col_name in pathway_candidates:
                if col_name in df.columns:
                    col_map["Pathway"] = col_name
                    logger.debug(f"UC-8.4: Mapped Pathway to column '{col_name}'")
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Pathway"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.4: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.4: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Pathway"] = df_processed["Pathway"].astype(str).str.strip()

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Pathway"] != "")
                & (df_processed["Pathway"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.4: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.4: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Pathway'].nunique()} pathways"
            )

            # Generate heatmap using PlotService
            logger.debug("UC-8.4: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.4",
                filters={},
                customizations={},
            )

            logger.info("UC-8.4: Heatmap generated successfully")

            # Canonical filename for UC-8.4
            try:
                suggested = sanitize_filename("UC-8.4", "pathway_completeness", "png")
            except Exception:
                suggested = "pathway_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.4: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.4: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_4_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.4: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

register_uc_8_5_callbacks

register_uc_8_5_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.5.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Uses color-only visualization for pattern recognition
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_5_callbacks.py
def register_uc_8_5_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.5.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Uses color-only visualization for pattern recognition
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.5 callbacks")

    @app.callback(
        Output("uc-8-5-collapse", "is_open"),
        Input("uc-8-5-collapse-button", "n_clicks"),
        State("uc-8-5-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_5_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.5 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.5 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-5-chart", "children"),
        Input("uc-8-5-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_5(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.5 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'kegg_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts KEGG DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / pathway KOs) × 100%
        - Generates heatmap with samples (rows) × KEGG pathways (columns)
        - Uses color-only display (text_auto: false) for pattern recognition
        - Hover tooltips show exact percentages
        """
        logger.debug(f"UC-8.5 render callback triggered. Active item: {active_item}")

        # Check if UC-8.5 accordion is active
        if not active_item or active_item != "uc-8-5-accordion":
            logger.debug("UC-8.5 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.5: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "kegg_df" not in merged_data:
                logger.error("UC-8.5: merged_data does not contain 'kegg_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'kegg_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.5: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["kegg_df"])

            if df.empty:
                logger.warning("UC-8.5: DataFrame is empty")
                return _create_error_message(
                    "The KEGG dataset is empty. Please load data with KO, Sample, and Pathway information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.5: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.5: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.5: Mapped KO to column '{col_name}'")
                    break

            # Try to find Pathway column (KEGG specific)
            pathway_candidates = [
                "Pathway",
                "pathway",
                "pathname",
                "Pathway_Name",
                "kegg_pathway",
                "KEGG_Pathway",
            ]
            for col_name in pathway_candidates:
                if col_name in df.columns:
                    col_map["Pathway"] = col_name
                    logger.debug(f"UC-8.5: Mapped Pathway to column '{col_name}'")
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Pathway"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.5: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.5: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Pathway"] = df_processed["Pathway"].astype(str).str.strip()

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Pathway"] != "")
                & (df_processed["Pathway"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.5: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.5: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Pathway'].nunique()} KEGG pathways"
            )

            # Generate heatmap using PlotService
            # Uses uc_8_5_config.yaml with HeatmapScoredStrategy and text_auto: false
            logger.debug("UC-8.5: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.5",
                filters={},
                customizations={},
            )

            logger.info("UC-8.5: Heatmap generated successfully (color-only mode)")

            # Return chart wrapped in loading component
            # Canonical filename for UC-8.5
            try:
                suggested = sanitize_filename(
                    "UC-8.5", "kegg_pathway_completeness", "png"
                )
            except Exception:
                suggested = "kegg_pathway_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.5: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.5: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_5_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.5: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

register_uc_8_6_callbacks

register_uc_8_6_callbacks(app, plot_service)

Register all UC-8.6 callbacks with Dash app.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers 5 callbacks: panel toggle, database selection, chart clear, pathway dropdown population, and UpSet plot rendering
  • Supports dual-database mode (HADEG default, KEGG optional)
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_6_callbacks.py
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
def register_uc_8_6_callbacks(app, plot_service):
    """
    Register all UC-8.6 callbacks with Dash app.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers 5 callbacks: panel toggle, database selection, chart clear,
      pathway dropdown population, and UpSet plot rendering
    - Supports dual-database mode (HADEG default, KEGG optional)
    - Refer to official documentation for processing logic details
    """
    logger.info("[UC-8.6] Registering callbacks...")

    @app.callback(
        Output("uc-8-6-collapse", "is_open"),
        Input("uc-8-6-collapse-button", "n_clicks"),
        State("uc-8-6-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_6_info_panel(n_clicks: int, is_open: bool) -> bool:
        """
        Toggle UC-8.6 informative panel collapse state.

        Parameters
        ----------
        n_clicks : int
            Number of button clicks.
        is_open : bool
            Current collapse state.

        Returns
        -------
        bool
            New collapse state (inverted).
        """
        if n_clicks:
            logger.debug(f"[UC-8.6] Toggling info panel: {is_open}{not is_open}")
            return not is_open
        return is_open

    @app.callback(
        [Output("uc-8-6-db-hadeg", "outline"), Output("uc-8-6-db-kegg", "outline")],
        [Input("uc-8-6-db-hadeg", "n_clicks"), Input("uc-8-6-db-kegg", "n_clicks")],
        prevent_initial_call=True,
    )
    def toggle_uc_8_6_database_buttons(
        hadeg_clicks: int, kegg_clicks: int
    ) -> tuple[bool, bool]:
        """
        Toggle database selection buttons for UC-8.6.

        Parameters
        ----------
        hadeg_clicks : int
            Number of clicks on HADEG button.
        kegg_clicks : int
            Number of clicks on KEGG button.

        Returns
        -------
        tuple of (bool, bool)
            Outline states for (HADEG, KEGG).
            False = selected, True = not selected.
        """
        from dash import callback_context

        ctx = callback_context
        if not ctx.triggered:
            return False, True  # HADEG selected by default

        button_id = ctx.triggered[0]["prop_id"].split(".")[0]

        if button_id == "uc-8-6-db-hadeg":
            logger.debug("[UC-8.6] Database switched to HADEG")
            return False, True
        elif button_id == "uc-8-6-db-kegg":
            logger.debug("[UC-8.6] Database switched to KEGG")
            return True, False

        return False, True  # Default to HADEG

    @app.callback(
        Output("uc-8-6-chart", "children", allow_duplicate=True),
        [Input("uc-8-6-db-hadeg", "n_clicks"), Input("uc-8-6-db-kegg", "n_clicks")],
        prevent_initial_call=True,
    )
    def clear_chart_on_database_change(
        hadeg_clicks: Optional[int], kegg_clicks: Optional[int]
    ) -> html.Div:
        """
        Clear chart when database is changed.

        Returns empty div to reset visualization state when user
        switches between HADEG and KEGG databases.
        """
        logger.debug("[UC-8.6] Database changed, clearing chart")
        return html.Div(
            html.P(
                "Please select a pathway from the dropdown to generate "
                "the UpSet plot.",
                className="text-muted text-center p-5",
            ),
            className="border rounded p-3",
        )

    @app.callback(
        [
            Output("uc-8-6-pathway-dropdown", "options"),
            Output("uc-8-6-pathway-dropdown", "value"),
            Output("uc-8-6-pathway-help-text", "children"),
        ],
        [
            Input("uc-8-6-accordion-group", "active_item"),
            Input("uc-8-6-db-hadeg", "n_clicks"),
            Input("uc-8-6-db-kegg", "n_clicks"),
        ],
        [
            State("merged-result-store", "data"),
            State("uc-8-6-db-hadeg", "outline"),
            State("uc-8-6-db-kegg", "outline"),
        ],
        prevent_initial_call=True,
    )
    def populate_uc_8_6_pathway_dropdown(
        active_item: Optional[str],
        hadeg_clicks: Optional[int],
        kegg_clicks: Optional[int],
        merged_data: dict,
        hadeg_outline: bool,
        kegg_outline: bool,
    ) -> tuple[list[dict], Optional[str], str]:
        """
        Populate pathway dropdown when accordion is opened or database is changed.

        Parameters
        ----------
        active_item : str or None
            Active accordion item ID ('uc-8-6-accordion' when opened).
        hadeg_clicks : int or None
            Number of clicks on HADEG button.
        kegg_clicks : int or None
            Number of clicks on KEGG button.
        merged_data : dict
            Dictionary from merged-result-store.
        hadeg_outline : bool
            Whether HADEG button is outlined (not selected).
        kegg_outline : bool
            Whether KEGG button is outlined (not selected).

        Returns
        -------
        tuple of (list of dict, str or None, str)
            - Pathway options: [{'label': 'Name', 'value': 'Name'}, ...]
            - Default value: First pathway or None
            - Help text: Database-specific description

        Notes
        -----
        Triggered when:
        - Accordion is opened (initial load with HADEG default)
        - Database button is clicked (switch between HADEG/KEGG)

        Database Support:
        - HADEG: 58 pathways for degradation analysis (default)
        - KEGG: 19 pathways for general metabolism
        """
        from dash import callback_context

        logger.info("[UC-8.6] populate_pathway_dropdown callback triggered")

        ctx = callback_context
        if not ctx.triggered:
            raise PreventUpdate

        trigger_id = ctx.triggered[0]["prop_id"].split(".")[0]
        logger.debug(f"[UC-8.6] Triggered by: {trigger_id}")

        # Determine selected database based on trigger OR current state
        # Priority: If triggered by button click, use that button
        # Otherwise, use current state (outline values)
        if trigger_id == "uc-8-6-db-hadeg":
            selected_db_key = "hadeg_df"
            db_name = "HADEG"
        elif trigger_id == "uc-8-6-db-kegg":
            selected_db_key = "kegg_df"
            db_name = "KEGG"
        elif not hadeg_outline:
            # HADEG is selected (outline=False)
            selected_db_key = "hadeg_df"
            db_name = "HADEG"
        elif not kegg_outline:
            # KEGG is selected (outline=False)
            selected_db_key = "kegg_df"
            db_name = "KEGG"
        else:
            # Default to HADEG if no button selected
            selected_db_key = "hadeg_df"
            db_name = "HADEG"

        # Set help text based on selected database
        if db_name == "HADEG":
            help_text = (
                "HADEG Database: Pathways for biodegradation and "
                "environmental remediation. Analyze KO distribution across "
                "samples to design consortia for degradation of specific "
                "compounds."
            )
        else:  # KEGG
            help_text = (
                "KEGG Database: General metabolic pathways. Analyze KO "
                "distribution to understand functional complementarity and "
                "redundancy in metabolic processes."
            )

        logger.info(f"[UC-8.6] Selected database: {db_name}")

        # Extract database data
        if selected_db_key == "hadeg_df":
            df = _extract_hadeg_data(merged_data)
        else:  # kegg_df
            df = _extract_kegg_data(merged_data)

        if df is None or len(df) == 0:
            logger.warning(f"[UC-8.6] {db_name} data not available")
            return (
                [],
                None,
                f"⚠️ {db_name} data not available. Please load data " f"first.",
            )

        # Get unique pathways
        unique_pathways = _get_unique_pathways(df)

        if not unique_pathways:
            logger.warning(f"[UC-8.6] No pathways found in {db_name} data")
            return [], None, f"⚠️ No pathways found in {db_name} database."

        # Create dropdown options
        pathway_options = [
            {"label": pathway, "value": pathway} for pathway in unique_pathways
        ]

        # Don't set default value - let user choose explicitly
        default_value = None

        logger.info(
            f"[UC-8.6] Dropdown populated with {len(pathway_options)} "
            f"{db_name} pathways, no default selection (user must choose)"
        )

        return pathway_options, default_value, help_text

    @app.callback(
        Output("uc-8-6-chart", "children"),
        Input("uc-8-6-pathway-dropdown", "value"),
        [
            State("merged-result-store", "data"),
            State("uc-8-6-db-hadeg", "outline"),
            State("uc-8-6-db-kegg", "outline"),
        ],
        prevent_initial_call=True,
    )
    def render_uc_8_6(
        selected_pathway: Optional[str],
        merged_data: dict,
        hadeg_outline: bool,
        kegg_outline: bool,
    ) -> html.Div:
        """
        Generate UpSet plot for selected pathway from selected database.

        Parameters
        ----------
        selected_pathway : str or None
            Selected pathway name from dropdown.
        merged_data : dict
            Dictionary from merged-result-store.
        hadeg_outline : bool
            Whether HADEG button is outlined (not selected).
        kegg_outline : bool
            Whether KEGG button is outlined (not selected).

        Returns
        -------
        html.Div
            Container with UpSet plot or error message.

        Notes
        -----
        Rendering logic:
        1. Triggered by pathway dropdown selection
        2. Determines selected database (HADEG or KEGG)
        3. Extracts data from appropriate database
        4. Filters data for selected pathway
        5. Builds KO sets grouped by Sample
        6. Creates UpSet DataFrame format
        7. Passes to PlotService for visualization

        Database Support:
        - HADEG: Biodegradation pathways
        - KEGG: General metabolic pathways

        Error Handling:
        - No pathway selected → Prevent update
        - No data → Error message
        - Missing columns → Error message
        - No KOs for pathway → Warning message
        - Plot generation error → Error message with details
        """
        logger.info("[UC-8.6] render_uc_8_6 callback triggered")
        logger.debug(f"[UC-8.6] Selected pathway: '{selected_pathway}'")

        # Validate pathway selection
        if not selected_pathway:
            logger.debug("[UC-8.6] No pathway selected, preventing update")
            raise PreventUpdate

        # Determine selected database
        if not hadeg_outline:
            selected_db_key = "hadeg_df"
            db_name = "HADEG"
        elif not kegg_outline:
            selected_db_key = "kegg_df"
            db_name = "KEGG"
        else:
            # Default to HADEG
            selected_db_key = "hadeg_df"
            db_name = "HADEG"

        logger.info(
            f"[UC-8.6] Generating UpSet plot for pathway '{selected_pathway}' "
            f"from {db_name} database"
        )

        # Extract data from selected database
        if selected_db_key == "hadeg_df":
            df = _extract_hadeg_data(merged_data)
        else:  # kegg_df
            df = _extract_kegg_data(merged_data)

        if df is None or len(df) == 0:
            logger.error(f"[UC-8.6] {db_name} data not available")
            return _create_error_message(
                f"❌ **Error**: {db_name} data not available. Please upload "
                f"and process data first.",
                "danger",
            )

        # Filter data for selected pathway
        pathway_df = _filter_pathway_data(df, selected_pathway)

        if pathway_df is None or len(pathway_df) == 0:
            logger.warning(
                f"[UC-8.6] No data found for pathway '{selected_pathway}' "
                f"in {db_name} database"
            )
            return _create_error_message(
                f"⚠️ **No data available** for pathway "
                f"**{selected_pathway}** in {db_name} database.\n\n"
                f"This pathway may not be present in the {db_name} dataset, "
                f"or no KOs are associated with it. "
                f"Please select a different pathway.",
                "warning",
            )

        # Build KO sets grouped by Sample
        sample_ko_sets = _build_gene_ko_sets(pathway_df)

        if not sample_ko_sets:
            logger.error("[UC-8.6] Failed to build Sample-KO sets")
            return _create_error_message(
                "❌ **Error**: Failed to process KO data. Please check the "
                "dataset format and try again.",
                "danger",
            )

        # Create UpSet DataFrame from sets
        # Format: [{'category': sample_name, 'identifier': ko_id}, ...]
        upset_data = []
        for sample_name, ko_set in sample_ko_sets.items():
            for ko_id in ko_set:
                upset_data.append({"category": sample_name, "identifier": ko_id})

        upset_df = pd.DataFrame(upset_data)

        logger.info(
            f"[UC-8.6] Created UpSet DataFrame: {len(upset_df)} rows, "
            f"{len(sample_ko_sets)} samples, "
            f"{len(set(upset_df['identifier']))} unique KOs"
        )
        logger.debug(
            f"[UC-8.6] UpSet DataFrame sample (first 10 rows):\n" f"{upset_df.head(10)}"
        )

        # Generate plot using PlotService
        try:
            logger.info("[UC-8.6] Calling PlotService.generate_plot...")

            # Instantiate PlotService

            # Generate plot
            fig = plot_service.generate_plot(data=upset_df, use_case_id="UC-8.6")

            logger.info("[UC-8.6] Plot generated successfully")

            # Build a short, safe basename for the exported filename
            pathway_safe = str(selected_pathway).replace(" ", "_")
            db_short = db_name.lower()
            try:
                suggested = sanitize_filename(
                    "UC-8.6", f"upset_{db_short}_{pathway_safe}", "png"
                )
            except Exception:
                suggested = f"upset_{db_short}_{pathway_safe}.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                [
                    dcc.Graph(
                        figure=fig,
                        config={
                            "displayModeBar": True,
                            "displaylogo": False,
                            "toImageButtonOptions": {
                                "format": "svg",
                                "filename": base_filename,
                                "height": 800,
                                "width": 1000,
                                "scale": 2,
                            },
                        },
                    )
                ]
            )

        except Exception as e:
            logger.error(f"[UC-8.6] Error generating plot: {str(e)}", exc_info=True)
            return _create_error_message(
                f"❌ **Error generating plot**: {str(e)}\n\n"
                f"Please check the logs for more details.",
                "danger",
            )

    logger.info("[UC-8.6] Callbacks registered successfully")

register_uc_8_7_callbacks

register_uc_8_7_callbacks(app, plot_service)

Register all UC-8.7 callbacks with Dash app.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers 3 callbacks: panel toggle, sample dropdown population, and UpSet plot rendering
  • Requires minimum 2 samples for UpSet plot generation
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_7_callbacks.py
def register_uc_8_7_callbacks(app, plot_service):
    """
    Register all UC-8.7 callbacks with Dash app.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers 3 callbacks: panel toggle, sample dropdown population,
      and UpSet plot rendering
    - Requires minimum 2 samples for UpSet plot generation
    - Refer to official documentation for processing logic details
    """
    logger.info("[UC-8.7] Registering callbacks...")

    @app.callback(
        Output("uc-8-7-collapse", "is_open"),
        Input("uc-8-7-collapse-button", "n_clicks"),
        State("uc-8-7-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_7_info_panel(n_clicks: int, is_open: bool) -> bool:
        """
        Toggle UC-8.7 informative panel collapse state.

        Parameters
        ----------
        n_clicks : int
            Number of button clicks.
        is_open : bool
            Current collapse state.

        Returns
        -------
        bool
            New collapse state (inverted).
        """
        if n_clicks:
            logger.debug(f"[UC-8.7] Toggling info panel: {is_open}{not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-7-sample-dropdown", "options"),
        Input("uc-8-7-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def populate_uc_8_7_sample_dropdown(
        active_item: Optional[str], merged_data: dict
    ) -> list[dict]:
        """
        Populate sample dropdown when accordion is opened.

        Parameters
        ----------
        active_item : str or None
            Active accordion item ID ('uc-8-7-accordion' when opened).
        merged_data : dict
            Dictionary from merged-result-store.

        Returns
        -------
        list of dict
            Sample options: [{'label': 'Name', 'value': 'Name'}, ...]

        Notes
        -----
        Triggered when accordion is opened.
        Extracts unique sample identifiers from BioRemPP database.
        """
        logger.info("[UC-8.7] populate_sample_dropdown callback triggered")
        logger.debug(f"[UC-8.7] active_item: {active_item}")

        # Only populate when accordion is opened
        if active_item != "uc-8-7-accordion":
            logger.debug("[UC-8.7] Accordion not opened, preventing update")
            raise PreventUpdate

        logger.info("[UC-8.7] Accordion opened, populating sample dropdown...")

        # Extract BioRemPP data
        biorempp_df = _extract_biorempp_data(merged_data)
        if biorempp_df is None or biorempp_df.empty:
            logger.error("[UC-8.7] No BioRemPP data available")
            return []

        # Verify required column exists
        if "Sample" not in biorempp_df.columns:
            logger.error(
                f"[UC-8.7] 'Sample' column not found. "
                f"Available columns: {list(biorempp_df.columns)}"
            )
            return []

        # Extract unique sample names
        samples = biorempp_df["Sample"].dropna().unique()
        samples = sorted([s for s in samples if s and str(s).strip()])

        logger.info(f"[UC-8.7] Found {len(samples)} unique samples: {samples}")

        # Create dropdown options
        options = [{"label": sample, "value": sample} for sample in samples]

        logger.info(f"[UC-8.7] Returning {len(options)} dropdown options")

        return options

    @app.callback(
        Output("uc-8-7-chart", "children"),
        Input("uc-8-7-sample-dropdown", "value"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_7(selected_samples: Optional[list], merged_data: dict) -> html.Div:
        """
        Generate UpSet plot for selected samples.

        Parameters
        ----------
        selected_samples : list of str or None
            Selected sample identifiers from dropdown.
        merged_data : dict
            Dictionary from merged-result-store.

        Returns
        -------
        html.Div
            Container with UpSet plot or error message.

        Notes
        -----
        Rendering logic:
        1. Triggered by sample dropdown selection
        2. Validates minimum 2 samples selected
        3. Extracts sample-KO associations from BioRemPP database
        4. Builds KO sets for each selected sample
        5. Creates UpSet plot data structure
        6. Passes to PlotService for visualization

        Error Handling:
        - Less than 2 samples → Error message
        - No data → Error message
        - Missing columns → Error message
        - Plot generation error → Error message with details

        Based on CLI reference: docs/CLI_UC/8.7/plot.py
        """
        logger.info("[UC-8.7] render_uc_8_7 callback triggered")
        logger.debug(f"[UC-8.7] Selected samples: {selected_samples}")

        # Validate input
        if not selected_samples or len(selected_samples) < 2:
            logger.warning(
                f"[UC-8.7] Insufficient samples selected: "
                f"{len(selected_samples) if selected_samples else 0}"
            )
            return _create_error_message(
                "❌ **Error**: Please select at least **2 samples** to "
                "generate the UpSet plot.\n\n"
                "The UpSet plot requires multiple samples to show "
                "meaningful intersections. Select 2 or more samples from "
                "the dropdown above.",
                "warning",
            )

        logger.info(
            f"[UC-8.7] Generating UpSet plot for {len(selected_samples)} "
            f"samples: {selected_samples}"
        )

        try:
            # Extract BioRemPP data
            biorempp_df = _extract_biorempp_data(merged_data)
            if biorempp_df is None or biorempp_df.empty:
                logger.error("[UC-8.7] No BioRemPP data available")
                return _create_error_message(
                    "❌ **Error**: No data available.\n\n"
                    "The merged result store does not contain BioRemPP data. "
                    "Please ensure data has been properly loaded.",
                    "danger",
                )

            # Prepare data for UpSet plot
            filtered_df = _prepare_upsetplot_data(biorempp_df, selected_samples)

            if filtered_df.empty:
                logger.warning(
                    f"[UC-8.7] No data found for selected samples: "
                    f"{selected_samples}"
                )
                return _create_error_message(
                    "⚠️ **Warning**: No data found for selected samples.\n\n"
                    f"The selected samples ({', '.join(selected_samples)}) "
                    "do not have any KO associations in the database.",
                    "warning",
                )

            logger.info(
                f"[UC-8.7] Filtered data: {len(filtered_df)} unique " "sample-KO pairs"
            )

            # Build KO memberships (which samples have which KOs)
            # Format: {KO_identifier: [sample1, sample2, ...]}
            logger.info("[UC-8.7] Generating KO to sample memberships...")
            memberships = (
                filtered_df.groupby("KO")["Sample"]
                .apply(lambda x: list(set(x)))
                .to_dict()
            )

            if not memberships:
                logger.error("[UC-8.7] No valid KO/sample memberships found")
                return _create_error_message(
                    "❌ **Error**: No valid gene associations found.\n\n"
                    "Unable to build KO memberships for the selected samples.",
                    "danger",
                )

            logger.info(f"[UC-8.7] Generated memberships for {len(memberships)} KOs")
            logger.debug(
                f"[UC-8.7] Sample memberships (first 3): "
                f"{dict(list(memberships.items())[:3])}"
            )

            # Prepare data for UpSet plot strategy
            # Format: [{category: sample, identifier: ko}, ...]
            upset_data = []
            for ko, samples in memberships.items():
                for sample in samples:
                    upset_data.append({"category": sample, "identifier": ko})

            # Convert to DataFrame (required by PlotService)
            upset_df = pd.DataFrame(upset_data)

            logger.info(
                f"[UC-8.7] Created UpSet DataFrame: {len(upset_df)} rows, "
                f"{len(selected_samples)} samples, "
                f"{len(memberships)} unique KOs"
            )
            logger.debug(
                f"[UC-8.7] UpSet DataFrame sample (first 10 rows):\n"
                f"{upset_df.head(10)}"
            )

            # Generate plot using PlotService
            logger.info("[UC-8.7] Calling PlotService to generate UpSet plot...")

            fig = plot_service.generate_plot(data=upset_df, use_case_id="UC-8.7")

            if fig is None:
                logger.error("[UC-8.7] PlotService returned None")
                return _create_error_message(
                    "❌ **Error**: Plot generation failed.\n\n"
                    "PlotService returned None. Check logs for details.",
                    "danger",
                )

            logger.info("[UC-8.7] [OK] UpSet plot generated successfully")

            # Prepare safe filename based on selected samples (use first 3 names)
            sample_label_parts = [
                str(s).replace(" ", "_") for s in selected_samples[:3]
            ]
            sample_label = "-".join(sample_label_parts)
            try:
                suggested = sanitize_filename(
                    "UC-8.7", f"samples_{len(selected_samples)}_{sample_label}", "png"
                )
            except Exception:
                suggested = f"samples_{len(selected_samples)}_{sample_label}.png"

            base_filename = os.path.splitext(suggested)[0]

            # Return plot in a Graph component with canonical filename
            return html.Div(
                dcc.Graph(
                    id="uc-8-7-upset-plot",
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "modeBarButtonsToRemove": ["lasso2d", "select2d"],
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                className="mt-3",
            )

        except ValueError as ve:
            logger.error(f"[UC-8.7] ValueError: {ve}", exc_info=True)
            return _create_error_message(
                f"❌ **Validation Error**: {str(ve)}\n\n"
                "Please check your data and try again.",
                "danger",
            )

        except Exception as e:
            logger.error(
                f"[UC-8.7] Unexpected error generating plot: {e}", exc_info=True
            )
            return _create_error_message(
                f"❌ **Error**: Plot generation failed.\n\n"
                f"**Details**: {str(e)}\n\n"
                "Please check the console logs for more information.",
                "danger",
            )

    logger.info("[UC-8.7] Callbacks registered successfully")

:docstring: :show-root-toc: false :show-source: false :heading-level: 2 :::

Module 8 Callbacks

This page documents the callback functions for Module 8 (Use Cases 8.1 to 8.7).


uc_8_1_callbacks

UC-8.1 Callbacks - Minimal Sample Grouping for Complete Compound Coverage.

This module implements callback functions for visualizing minimal sample groupings through faceted scatter analysis using set cover optimization.

Functions:

Name Description
register_uc_8_1_callbacks

Register all UC-8.1 callbacks with Dash app.

Notes
  • Refer to official documentation for use case details
  • Uses Frozenset set Strategy for minimal group visualization
  • Applies greedy set cover algorithm for optimization
  • BioRemPP database REQUIRED

Version: 1.0.0

Functions

register_uc_8_1_callbacks
register_uc_8_1_callbacks(app, plot_service) -> None

Register all UC-8.1 callbacks with Dash app.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers panel toggle, dropdown initialization, and faceted scatter callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_1_callbacks.py
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
def register_uc_8_1_callbacks(app, plot_service) -> None:
    """
    Register all UC-8.1 callbacks with Dash app.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers panel toggle, dropdown initialization, and faceted scatter callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("[UC-8.1] Registering callbacks")

    # ========================================
    # Callback 1: Toggle Informative Panel
    # ========================================
    @app.callback(
        Output("uc-8-1-collapse", "is_open"),
        Input("uc-8-1-collapse-button", "n_clicks"),
        State("uc-8-1-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_1_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """Toggle UC-8.1 informative panel collapse state."""
        if n_clicks:
            logger.debug(f"[UC-8.1] Toggling info panel: {is_open} -> {not is_open}")
            return not is_open
        return is_open

    # ========================================
    # Callback 2: Initialize Compound Class Dropdown
    # ========================================
    @app.callback(
        Output("uc-8-1-compoundclass-dropdown", "options"),
        [
            Input("uc-8-1-accordion-group", "active_item"),
            Input("merged-result-store", "data"),
        ],
        prevent_initial_call=True,
    )
    def initialize_uc_8_1_dropdown(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> list:
        """
        Populate compound class dropdown with available classes.

        Parameters
        ----------
        active_item : str, optional
            Active accordion item ID.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        list
            Dropdown options list.

        Notes
        -----
        - Triggered when accordion opens or data changes
        - Extracts unique compound classes from BioRemPP data
        - Returns empty list if accordion not active or data unavailable
        """
        logger.info(
            f"[UC-8.1] [CALLBACK 2] Dropdown init triggered, "
            f"active_item: {active_item}"
        )
        logger.debug(f"[UC-8.1] [CALLBACK 2] merged_data type: {type(merged_data)}")
        if isinstance(merged_data, dict):
            logger.debug(f"[UC-8.1] [CALLBACK 2] keys: {merged_data.keys()}")

        # Only populate when accordion is open
        if active_item != "uc-8-1-accordion":
            logger.debug("[UC-8.1] [CALLBACK 2] Accordion not active, skip")
            return []

        if not merged_data or "biorempp_df" not in merged_data:
            logger.warning(
                "[UC-8.1] [CALLBACK 2] No BioRemPP data available for dropdown"
            )
            return []

        try:
            df = pd.DataFrame(merged_data["biorempp_df"])
            logger.debug(
                f"[UC-8.1] [CALLBACK 2] DataFrame created: "
                f"{len(df)} rows, columns: {df.columns.tolist()}"
            )

            if df.empty:
                logger.warning("[UC-8.1] [CALLBACK 2] DataFrame is empty")
                return []

            # Find compound class column
            class_col = _find_column(
                df, ["Compound_Class", "compound_class", "CompoundClass", "class"]
            )

            if not class_col:
                logger.warning(
                    f"[UC-8.1] [CALLBACK 2] Compound class column not found. "
                    f"Available: {df.columns.tolist()}"
                )
                return []

            # Get unique compound classes
            classes = sorted(df[class_col].dropna().unique().tolist())
            options = [{"label": c, "value": c} for c in classes]

            logger.info(
                f"[UC-8.1] [CALLBACK 2] Dropdown populated: " f"{len(classes)} classes"
            )
            logger.debug(f"[UC-8.1] [CALLBACK 2] Classes: {classes[:5]}...")

            return options

        except Exception as e:
            logger.error(
                f"[UC-8.1] [CALLBACK 2] Error initializing dropdown: {e}", exc_info=True
            )
            return []

    # ========================================
    # Callback 3: Render Faceted Scatter
    # ========================================
    @app.callback(
        Output("uc-8-1-chart", "children"),
        Input("uc-8-1-compoundclass-dropdown", "value"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_1(
        selected_class: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.1 faceted scatter on compound class dropdown selection.

        Parameters
        ----------
        selected_class : str, optional
            Selected compound class from dropdown.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        html.Div
            Container with chart or error message.

        Notes
        -----
        - Filters BioRemPP data by selected compound class
        - Groups samples by frozenset of compounds (identical profiles)
        - Applies greedy set cover algorithm for minimal group selection
        - Calculates unique KO counts per compound for color scaling
        - Generates faceted scatter with one subplot per minimized group
        """
        logger.info(f"[UC-8.1] [CALLBACK 3] ========== RENDER TRIGGERED ==========")
        logger.info(f"[UC-8.1] [CALLBACK 3] selected_class: '{selected_class}'")
        logger.debug(f"[UC-8.1] [CALLBACK 3] merged_data type: {type(merged_data)}")
        logger.debug(f"[UC-8.1] [CALLBACK 3] is None: {merged_data is None}")

        # Validate compound class selection
        if not selected_class:
            logger.debug("[UC-8.1] No compound class selected")
            return _create_info_message(
                "Please select a Compound Class from the dropdown above.",
                "bi bi-arrow-up-circle",
            )

        try:
            # ========================================
            # Step 1: Validate merged_data structure
            # ========================================
            if not merged_data:
                logger.warning("[UC-8.1] merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please upload and process data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict):
                logger.error("[UC-8.1] merged_data is not a dictionary")
                return _create_error_message(
                    "Invalid data structure. Please reload the application.",
                    "bi bi-x-circle",
                )

            if "biorempp_df" not in merged_data:
                logger.error("[UC-8.1] merged_data does not contain 'biorempp_df' key")
                return _create_error_message(
                    "BioRemPP data not found. This use case requires "
                    "BioRemPP database.",
                    "bi bi-database-x",
                )

            # ========================================
            # Step 2: Extract DataFrame
            # ========================================
            logger.debug("[UC-8.1] Extracting DataFrame from merged_data")
            biorempp_data = merged_data["biorempp_df"]

            if not biorempp_data:
                logger.warning("[UC-8.1] biorempp_df is empty")
                return _create_error_message(
                    "BioRemPP dataset is empty. Please check your input data.",
                    "bi bi-inbox",
                )

            df = pd.DataFrame(biorempp_data)

            if df.empty:
                logger.warning("[UC-8.1] DataFrame is empty after conversion")
                return _create_error_message(
                    "No data available after processing.", "bi bi-inbox"
                )

            logger.info(
                f"[UC-8.1] Processing DataFrame: {len(df)} rows, "
                f"{len(df.columns)} columns"
            )

            # ========================================
            # Step 3: Map column names flexibly
            # ========================================
            col_map = {}

            # Sample column
            sample_col = _find_column(
                df, ["Sample", "sample", "sample_name", "SampleName"]
            )
            if sample_col:
                col_map["sample"] = sample_col

            # Compound name column
            compound_col = _find_column(
                df, ["Compound_Name", "compound_name", "CompoundName", "Compound"]
            )
            if compound_col:
                col_map["compoundname"] = compound_col

            # Compound class column
            class_col = _find_column(
                df, ["Compound_Class", "compound_class", "CompoundClass", "class"]
            )
            if class_col:
                col_map["compoundclass"] = class_col

            # KO column (optional, for color)
            ko_col = _find_column(df, ["KO", "ko", "ko_id", "KO_ID"])
            if ko_col:
                col_map["ko"] = ko_col

            # ========================================
            # Step 4: Validate required columns found
            # ========================================
            required = ["sample", "compoundname", "compoundclass"]
            missing_cols = [col for col in required if col not in col_map]

            if missing_cols:
                logger.error(
                    f"[UC-8.1] Missing columns: {missing_cols}. "
                    f"Available: {df.columns.tolist()}"
                )
                return _create_error_message(
                    f"Required columns not found: {', '.join(missing_cols)}. "
                    f"Available columns: {', '.join(df.columns[:5])}...",
                    "bi bi-exclamation-octagon",
                )

            # ========================================
            # Step 5: Prepare data
            # ========================================
            # Rename columns to standard names
            rename_map = {v: k for k, v in col_map.items()}
            df_work = df.rename(columns=rename_map).copy()

            # Filter by selected compound class
            df_filtered = df_work[df_work["compoundclass"] == selected_class].copy()

            if df_filtered.empty:
                logger.warning(f"[UC-8.1] No data for class '{selected_class}'")
                return _create_error_message(
                    f"No data found for compound class: {selected_class}",
                    "bi bi-search",
                )

            # Clean data
            df_filtered = df_filtered.dropna(subset=["sample", "compoundname"])

            for col in ["sample", "compoundname"]:
                df_filtered[col] = df_filtered[col].astype(str).str.strip()

            # Remove placeholder values
            placeholder_values = ["#N/D", "#N/A", "N/D", "", "nan", "None"]
            for col in ["sample", "compoundname"]:
                df_filtered = df_filtered[~df_filtered[col].isin(placeholder_values)]

            if df_filtered.empty:
                return _create_error_message(
                    f"No valid data for compound class: {selected_class}",
                    "bi bi-funnel",
                )

            logger.info(
                f"[UC-8.1] Filtered to {len(df_filtered)} rows for "
                f"class '{selected_class}'"
            )

            # ========================================
            # Step 6: Group samples by compound profile
            # ========================================
            grouped_df, groups = _group_by_compound_profile(df_filtered)

            if grouped_df.empty or not groups:
                return _create_error_message(
                    "No sample groups found after grouping.", "bi bi-diagram-3"
                )

            logger.info(f"[UC-8.1] Created {len(groups)} groups")

            # ========================================
            # Step 7: Minimize groups with set cover
            # ========================================
            minimized_groups = _minimize_groups(grouped_df)

            if not minimized_groups:
                return _create_error_message(
                    "Could not determine minimal groups.", "bi bi-exclamation-circle"
                )

            # Filter to minimized groups only
            final_df = grouped_df[grouped_df["_group"].isin(minimized_groups)].copy()

            logger.info(f"[UC-8.1] Minimized to {len(minimized_groups)} groups")

            # ========================================
            # Step 8: Calculate KO counts (for color)
            # ========================================
            if "ko" in col_map:
                ko_counts = _calculate_ko_counts(df_filtered)
                if ko_counts is not None:
                    final_df = final_df.merge(
                        ko_counts, left_on="compoundname", right_index=True, how="left"
                    )
                    final_df["_unique_ko_count"] = (
                        final_df["_unique_ko_count"].fillna(0).astype(int)
                    )
                else:
                    final_df["_unique_ko_count"] = 1
            else:
                final_df["_unique_ko_count"] = 1

            # ========================================
            # Step 9: Generate plot
            # ========================================
            fig = _create_frozenset_figure(final_df, minimized_groups, selected_class)

            logger.info("[UC-8.1] Faceted scatter generation successful")

            # ========================================
            # Step 10: Return chart component
            # ========================================
            n_groups = len(minimized_groups)
            n_samples = final_df["sample"].nunique()
            n_compounds = final_df["compoundname"].nunique()

            # Prepare a safe download filename using canonical helper
            selected_class_safe = str(selected_class).replace(" ", "_")
            try:
                suggested = sanitize_filename(
                    "UC-8.1", f"minimal_grouping_{selected_class_safe}", "png"
                )
            except Exception:
                suggested = f"minimal_grouping_{selected_class_safe}.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                [
                    # Statistics summary
                    html.Div(
                        [
                            html.Small(
                                [
                                    html.I(className="bi bi-info-circle me-2"),
                                    f"Minimal Coverage: {n_groups} functional guilds | "
                                    f"{n_samples} samples | {n_compounds} compounds | "
                                    f"Class: {selected_class}",
                                ],
                                className="text-muted",
                            )
                        ],
                        className="mb-2",
                    ),
                    # Graph container with overflow control
                    html.Div(
                        [
                            dcc.Graph(
                                id="uc-8-1-graph",
                                figure=fig,
                                config={
                                    "displayModeBar": True,
                                    "displaylogo": False,
                                    "responsive": True,
                                    "modeBarButtonsToRemove": [
                                        "pan2d",
                                        "lasso2d",
                                        "select2d",
                                    ],
                                    "toImageButtonOptions": {
                                        "format": "svg",
                                        "filename": base_filename,
                                        "height": 600,
                                        "width": 900,
                                        "scale": 6,
                                    },
                                },
                                style={
                                    "height": "600px",
                                    "width": "100%",
                                    "minWidth": "100%",
                                },
                                className="mt-3",
                            )
                        ],
                        style={
                            "width": "100%",
                            "overflowX": "auto",
                            "overflowY": "hidden",
                        },
                    ),
                ]
            )

        except ValueError as ve:
            logger.error(
                f"[UC-8.1] ValueError during processing: {str(ve)}", exc_info=True
            )
            return _create_error_message(
                f"Data validation error: {str(ve)}", "bi bi-exclamation-triangle"
            )

        except Exception as e:
            logger.error(f"[UC-8.1] Unexpected error: {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred: {str(e)}", "bi bi-bug"
            )

    logger.info("[UC-8.1] All callbacks registered successfully")

uc_8_2_callbacks

UC-8.2 Callbacks - Chemical Class Completeness Scorecard.

This module implements callback functions for visualizing chemical class completeness through heatmap scorecard analysis.

Functions:

Name Description
register_uc_8_2_callbacks

Register all UC-8.2 callbacks with Dash app.

Notes
  • Refer to official documentation for use case details
  • Uses HeatmapScoredStrategy for completeness scorecard visualization
  • BioRemPP database REQUIRED

Version: 1.0.0

Functions

register_uc_8_2_callbacks
register_uc_8_2_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.2.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_2_callbacks.py
def register_uc_8_2_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.2.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.2 callbacks")

    @app.callback(
        Output("uc-8-2-collapse", "is_open"),
        Input("uc-8-2-collapse-button", "n_clicks"),
        State("uc-8-2-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_2_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.2 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.2 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-2-chart", "children"),
        Input("uc-8-2-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_2(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.2 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts BioRemPP DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / class KOs) × 100%
        - Generates heatmap with samples (rows) × compound classes (columns)
        - Uses PlotService with HeatmapScoredStrategy
        """
        logger.debug(f"UC-8.2 render callback triggered. Active item: {active_item}")

        # Check if UC-8.2 accordion is active
        if not active_item or active_item != "uc-8-2-accordion":
            logger.debug("UC-8.2 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.2: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "biorempp_df" not in merged_data:
                logger.error("UC-8.2: merged_data does not contain 'biorempp_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'biorempp_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.2: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["biorempp_df"])

            if df.empty:
                logger.warning("UC-8.2: DataFrame is empty")
                return _create_error_message(
                    "The dataset is empty. Please load data with KO, Sample, and Compound_Class information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.2: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.2: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.2: Mapped KO to column '{col_name}'")
                    break

            # Try to find Compound_Class column
            class_candidates = [
                "Compound_Class",
                "compound_class",
                "CompoundClass",
                "class",
                "Class",
                "chemical_class",
                "Chemical_Class",
            ]
            for col_name in class_candidates:
                if col_name in df.columns:
                    col_map["Compound_Class"] = col_name
                    logger.debug(
                        f"UC-8.2: Mapped Compound_Class to column '{col_name}'"
                    )
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Compound_Class"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.2: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.2: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Compound_Class"] = (
                df_processed["Compound_Class"].astype(str).str.strip()
            )

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Compound_Class"] != "")
                & (df_processed["Compound_Class"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.2: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.2: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Compound_Class'].nunique()} compound classes"
            )

            # Generate heatmap using PlotService
            # NOTE: This will require a uc_8_2_config.yaml file to be created
            # with HeatmapScoredStrategy configuration
            logger.debug("UC-8.2: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.2",
                filters={},
                customizations={},
            )

            logger.info("UC-8.2: Heatmap generated successfully")

            # Return chart wrapped in loading component
            # Compute canonical filename
            try:
                suggested = sanitize_filename(
                    "UC-8.2", "chemical_class_completeness", "png"
                )
            except Exception:
                suggested = "chemical_class_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.2: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.2: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_2_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.2: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

uc_8_3_callbacks

UC-8.3 Callbacks - Compound-Specific KO Completeness Scorecard.

This module implements callback functions for visualizing compound-specific KO completeness through heatmap scorecard analysis.

Functions:

Name Description
register_uc_8_3_callbacks

Register all UC-8.3 callbacks with Dash app.

Notes
  • Refer to official documentation for use case details
  • Uses HeatmapScoredStrategy for compound-level completeness visualization
  • BioRemPP database REQUIRED

Version: 1.0.0

Functions

register_uc_8_3_callbacks
register_uc_8_3_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.3.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_3_callbacks.py
def register_uc_8_3_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.3.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.3 callbacks")

    @app.callback(
        Output("uc-8-3-collapse", "is_open"),
        Input("uc-8-3-collapse-button", "n_clicks"),
        State("uc-8-3-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_3_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.3 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.3 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-3-chart", "children"),
        Input("uc-8-3-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_3(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.3 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'biorempp_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts BioRemPP DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / compound KOs) × 100%
        - Generates heatmap with samples (rows) × compound names (columns)
        - Uses PlotService with HeatmapScoredStrategy
        """
        logger.debug(f"UC-8.3 render callback triggered. Active item: {active_item}")

        # Check if UC-8.3 accordion is active
        if not active_item or active_item != "uc-8-3-accordion":
            logger.debug("UC-8.3 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.3: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "biorempp_df" not in merged_data:
                logger.error("UC-8.3: merged_data does not contain 'biorempp_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'biorempp_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.3: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["biorempp_df"])

            if df.empty:
                logger.warning("UC-8.3: DataFrame is empty")
                return _create_error_message(
                    "The dataset is empty. Please load data with KO, Sample, and Compound_Name information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.3: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.3: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.3: Mapped KO to column '{col_name}'")
                    break

            # Try to find Compound_Name column
            compound_candidates = [
                "Compound_Name",
                "compound_name",
                "compoundname",
                "CompoundName",
                "compound",
                "Compound",
            ]
            for col_name in compound_candidates:
                if col_name in df.columns:
                    col_map["Compound_Name"] = col_name
                    logger.debug(f"UC-8.3: Mapped Compound_Name to column '{col_name}'")
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Compound_Name"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.3: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.3: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Compound_Name"] = (
                df_processed["Compound_Name"].astype(str).str.strip()
            )

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Compound_Name"] != "")
                & (df_processed["Compound_Name"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.3: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.3: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Compound_Name'].nunique()} compounds"
            )

            # Generate heatmap using PlotService
            logger.debug("UC-8.3: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.3",
                filters={},
                customizations={},
            )

            logger.info("UC-8.3: Heatmap generated successfully")

            # Use canonical filename
            try:
                suggested = sanitize_filename(
                    "UC-8.3", "compound_specific_completeness", "png"
                )
            except Exception:
                suggested = "compound_specific_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.3: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.3: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_3_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.3: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

uc_8_4_callbacks

UC-8.4 Callbacks - Pathway Completeness Scorecard for HADEG Pathways.

This module implements callback functions for visualizing HADEG pathway completeness through heatmap scorecard analysis.

Functions:

Name Description
register_uc_8_4_callbacks

Register all UC-8.4 callbacks with Dash app.

Notes
  • Refer to official documentation for use case details
  • Uses HeatmapScoredStrategy for pathway completeness visualization
  • HADEG database REQUIRED

Version: 1.0.0

Functions

register_uc_8_4_callbacks
register_uc_8_4_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.4.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_4_callbacks.py
def register_uc_8_4_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.4.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.4 callbacks")

    @app.callback(
        Output("uc-8-4-collapse", "is_open"),
        Input("uc-8-4-collapse-button", "n_clicks"),
        State("uc-8-4-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_4_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.4 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.4 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-4-chart", "children"),
        Input("uc-8-4-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_4(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.4 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'hadeg_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts HADEG DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / pathway KOs) × 100%
        - Generates heatmap with samples (rows) × pathways (columns)
        - Uses PlotService with HeatmapScoredStrategy
        """
        logger.debug(f"UC-8.4 render callback triggered. Active item: {active_item}")

        # Check if UC-8.4 accordion is active
        if not active_item or active_item != "uc-8-4-accordion":
            logger.debug("UC-8.4 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.4: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "hadeg_df" not in merged_data:
                logger.error("UC-8.4: merged_data does not contain 'hadeg_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'hadeg_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.4: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["hadeg_df"])

            if df.empty:
                logger.warning("UC-8.4: DataFrame is empty")
                return _create_error_message(
                    "The HADEG dataset is empty. Please load data with KO, Sample, and Pathway information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.4: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )
            logger.debug(f"UC-8.4: Available columns: {df.columns.tolist()}")

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.4: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.4: Mapped KO to column '{col_name}'")
                    break

            # Try to find Pathway column
            pathway_candidates = [
                "Pathway",
                "pathway",
                "compound_pathway",
                "Compound_Pathway",
                "PathwayName",
                "pathway_name",
            ]
            for col_name in pathway_candidates:
                if col_name in df.columns:
                    col_map["Pathway"] = col_name
                    logger.debug(f"UC-8.4: Mapped Pathway to column '{col_name}'")
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Pathway"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.4: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.4: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Pathway"] = df_processed["Pathway"].astype(str).str.strip()

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Pathway"] != "")
                & (df_processed["Pathway"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.4: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.4: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Pathway'].nunique()} pathways"
            )

            # Generate heatmap using PlotService
            logger.debug("UC-8.4: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.4",
                filters={},
                customizations={},
            )

            logger.info("UC-8.4: Heatmap generated successfully")

            # Canonical filename for UC-8.4
            try:
                suggested = sanitize_filename("UC-8.4", "pathway_completeness", "png")
            except Exception:
                suggested = "pathway_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.4: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.4: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_4_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.4: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

uc_8_5_callbacks

UC-8.5 Callbacks - KEGG Pathway Completeness Scorecard.

This module implements callback functions for visualizing KEGG pathway completeness through color-only heatmap scorecard analysis.

Functions:

Name Description
register_uc_8_5_callbacks

Register all UC-8.5 callbacks with Dash app.

Notes
  • Refer to official documentation foruse case details
  • Uses HeatmapScoredStrategy with color-only visualization (text_auto: false)
  • KEGG database REQUIRED

Version: 1.0.0

Functions

register_uc_8_5_callbacks
register_uc_8_5_callbacks(app, plot_service) -> None

Register all callbacks for UC-8.5.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers information panel toggle and heatmap rendering callbacks
  • Uses color-only visualization for pattern recognition
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_5_callbacks.py
def register_uc_8_5_callbacks(app, plot_service) -> None:
    """
    Register all callbacks for UC-8.5.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers information panel toggle and heatmap rendering callbacks
    - Uses color-only visualization for pattern recognition
    - Refer to official documentation for processing logic details
    """
    logger.info("Registering UC-8.5 callbacks")

    @app.callback(
        Output("uc-8-5-collapse", "is_open"),
        Input("uc-8-5-collapse-button", "n_clicks"),
        State("uc-8-5-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_5_info_panel(n_clicks: Optional[int], is_open: bool) -> bool:
        """
        Toggle UC-8.5 information panel visibility.

        Parameters
        ----------
        n_clicks : int, optional
            Number of times collapse button has been clicked.
        is_open : bool
            Current state of collapse component.

        Returns
        -------
        bool
            New state of collapse component (True = open, False = closed).
        """
        if n_clicks:
            logger.debug(f"UC-8.5 info panel toggled. New state: {not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-5-chart", "children"),
        Input("uc-8-5-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_5(
        active_item: Optional[str], merged_data: Optional[Dict[str, Any]]
    ) -> html.Div:
        """
        Render UC-8.5 heatmap scorecard when accordion is activated.

        Parameters
        ----------
        active_item : str, optional
            ID of currently active accordion item.
        merged_data : dict, optional
            Dictionary containing 'kegg_df' key.

        Returns
        -------
        html.Div
            Container with loading spinner and heatmap or error message.

        Raises
        ------
        PreventUpdate
            If accordion not active or data not ready.
        ValueError
            If required columns are missing or data invalid.

        Notes
        -----
        - Extracts KEGG DataFrame and maps column names flexibly
        - Calculates completeness scores: (sample KOs / pathway KOs) × 100%
        - Generates heatmap with samples (rows) × KEGG pathways (columns)
        - Uses color-only display (text_auto: false) for pattern recognition
        - Hover tooltips show exact percentages
        """
        logger.debug(f"UC-8.5 render callback triggered. Active item: {active_item}")

        # Check if UC-8.5 accordion is active
        if not active_item or active_item != "uc-8-5-accordion":
            logger.debug("UC-8.5 accordion not active. Preventing update.")
            raise PreventUpdate

        try:
            # Validate merged_data structure
            if not merged_data:
                logger.warning("UC-8.5: merged_data is None or empty")
                return _create_error_message(
                    "No data available. Please load or merge data first.",
                    "bi bi-exclamation-triangle",
                )

            if not isinstance(merged_data, dict) or "kegg_df" not in merged_data:
                logger.error("UC-8.5: merged_data does not contain 'kegg_df' key")
                return _create_error_message(
                    "Invalid data structure. Expected 'kegg_df' in merged data.",
                    "bi bi-x-circle",
                )

            # Extract DataFrame
            logger.debug("UC-8.5: Extracting DataFrame from merged_data")
            df = pd.DataFrame(merged_data["kegg_df"])

            if df.empty:
                logger.warning("UC-8.5: DataFrame is empty")
                return _create_error_message(
                    "The KEGG dataset is empty. Please load data with KO, Sample, and Pathway information.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.5: Processing DataFrame with {len(df)} rows and {len(df.columns)} columns"
            )

            # Map column names flexibly to handle different naming conventions
            col_map = {}

            # Try to find sample column
            sample_candidates = [
                "Sample",
                "sample",
                "sample_id",
                "Sample_ID",
                "sampleID",
                "genome",
                "Genome",
                "organism",
            ]
            for col_name in sample_candidates:
                if col_name in df.columns:
                    col_map["Sample"] = col_name
                    logger.debug(f"UC-8.5: Mapped Sample to column '{col_name}'")
                    break

            # Try to find KO column
            ko_candidates = ["KO", "ko", "ko_id", "KO_ID", "ko_number"]
            for col_name in ko_candidates:
                if col_name in df.columns:
                    col_map["KO"] = col_name
                    logger.debug(f"UC-8.5: Mapped KO to column '{col_name}'")
                    break

            # Try to find Pathway column (KEGG specific)
            pathway_candidates = [
                "Pathway",
                "pathway",
                "pathname",
                "Pathway_Name",
                "kegg_pathway",
                "KEGG_Pathway",
            ]
            for col_name in pathway_candidates:
                if col_name in df.columns:
                    col_map["Pathway"] = col_name
                    logger.debug(f"UC-8.5: Mapped Pathway to column '{col_name}'")
                    break

            # Validate required columns were found
            required_fields = ["Sample", "KO", "Pathway"]
            missing_fields = [
                field for field in required_fields if field not in col_map
            ]

            if missing_fields:
                logger.error(f"UC-8.5: Missing required columns: {missing_fields}")
                return _create_error_message(
                    f"Missing required columns: {', '.join(missing_fields)}. "
                    f"Available columns: {', '.join(df.columns.tolist())}",
                    "bi bi-x-circle",
                )

            # Rename columns to standard names for processing
            df_processed = df.rename(columns={v: k for k, v in col_map.items()})

            # Clean data: strip whitespace, handle nulls
            logger.debug("UC-8.5: Cleaning data")
            df_processed = df_processed.copy()
            df_processed["Sample"] = df_processed["Sample"].astype(str).str.strip()
            df_processed["KO"] = df_processed["KO"].astype(str).str.strip().str.upper()
            df_processed["Pathway"] = df_processed["Pathway"].astype(str).str.strip()

            # Remove null/empty entries
            df_processed = df_processed[
                (df_processed["Sample"] != "")
                & (df_processed["Sample"] != "nan")
                & (df_processed["KO"] != "")
                & (df_processed["KO"] != "NAN")
                & (df_processed["Pathway"] != "")
                & (df_processed["Pathway"] != "nan")
            ]

            if df_processed.empty:
                logger.warning("UC-8.5: No valid data after cleaning")
                return _create_error_message(
                    "No valid data available after cleaning. Check for null values.",
                    "bi bi-inbox",
                )

            logger.info(
                f"UC-8.5: Cleaned data - {len(df_processed)} records, "
                f"{df_processed['Sample'].nunique()} samples, "
                f"{df_processed['Pathway'].nunique()} KEGG pathways"
            )

            # Generate heatmap using PlotService
            # Uses uc_8_5_config.yaml with HeatmapScoredStrategy and text_auto: false
            logger.debug("UC-8.5: Generating heatmap via PlotService")

            fig = plot_service.generate_plot(
                data=df_processed,
                use_case_id="UC-8.5",
                filters={},
                customizations={},
            )

            logger.info("UC-8.5: Heatmap generated successfully (color-only mode)")

            # Return chart wrapped in loading component
            # Canonical filename for UC-8.5
            try:
                suggested = sanitize_filename(
                    "UC-8.5", "kegg_pathway_completeness", "png"
                )
            except Exception:
                suggested = "kegg_pathway_completeness.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                dcc.Graph(
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                style={"minHeight": "650px"},
            )

        except ValueError as ve:
            logger.error(f"UC-8.5: Validation error - {str(ve)}", exc_info=True)
            return _create_error_message(
                f"Data validation error: {str(ve)}",
                "bi bi-exclamation-triangle",
            )

        except KeyError as ke:
            logger.error(f"UC-8.5: Missing key error - {str(ke)}", exc_info=True)
            return _create_error_message(
                f"Configuration error: Missing key '{str(ke)}'. "
                "Please ensure uc_8_5_config.yaml exists and is properly configured.",
                "bi bi-gear",
            )

        except Exception as e:
            logger.error(f"UC-8.5: Unexpected error - {str(e)}", exc_info=True)
            return _create_error_message(
                f"An unexpected error occurred while rendering the heatmap: {str(e)}",
                "bi bi-bug",
            )

uc_8_6_callbacks

UC-8.6 Callbacks - Pathway-Centric Consortium Design by KO Coverage.

This module implements callback functions for visualizing KO distribution across samples for consortium design using UpSet plot analysis.

Functions:

Name Description
register_uc_8_6_callbacks

Register all UC-8.6 callbacks with Dash app.

Notes
  • Refer to official documentation for use case details
  • Uses UpSetStrategy for KO distribution visualization
  • Supports HADEG and KEGG databases

Version: 1.0.0

Functions

register_uc_8_6_callbacks
register_uc_8_6_callbacks(app, plot_service)

Register all UC-8.6 callbacks with Dash app.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers 5 callbacks: panel toggle, database selection, chart clear, pathway dropdown population, and UpSet plot rendering
  • Supports dual-database mode (HADEG default, KEGG optional)
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_6_callbacks.py
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
def register_uc_8_6_callbacks(app, plot_service):
    """
    Register all UC-8.6 callbacks with Dash app.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers 5 callbacks: panel toggle, database selection, chart clear,
      pathway dropdown population, and UpSet plot rendering
    - Supports dual-database mode (HADEG default, KEGG optional)
    - Refer to official documentation for processing logic details
    """
    logger.info("[UC-8.6] Registering callbacks...")

    @app.callback(
        Output("uc-8-6-collapse", "is_open"),
        Input("uc-8-6-collapse-button", "n_clicks"),
        State("uc-8-6-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_6_info_panel(n_clicks: int, is_open: bool) -> bool:
        """
        Toggle UC-8.6 informative panel collapse state.

        Parameters
        ----------
        n_clicks : int
            Number of button clicks.
        is_open : bool
            Current collapse state.

        Returns
        -------
        bool
            New collapse state (inverted).
        """
        if n_clicks:
            logger.debug(f"[UC-8.6] Toggling info panel: {is_open}{not is_open}")
            return not is_open
        return is_open

    @app.callback(
        [Output("uc-8-6-db-hadeg", "outline"), Output("uc-8-6-db-kegg", "outline")],
        [Input("uc-8-6-db-hadeg", "n_clicks"), Input("uc-8-6-db-kegg", "n_clicks")],
        prevent_initial_call=True,
    )
    def toggle_uc_8_6_database_buttons(
        hadeg_clicks: int, kegg_clicks: int
    ) -> tuple[bool, bool]:
        """
        Toggle database selection buttons for UC-8.6.

        Parameters
        ----------
        hadeg_clicks : int
            Number of clicks on HADEG button.
        kegg_clicks : int
            Number of clicks on KEGG button.

        Returns
        -------
        tuple of (bool, bool)
            Outline states for (HADEG, KEGG).
            False = selected, True = not selected.
        """
        from dash import callback_context

        ctx = callback_context
        if not ctx.triggered:
            return False, True  # HADEG selected by default

        button_id = ctx.triggered[0]["prop_id"].split(".")[0]

        if button_id == "uc-8-6-db-hadeg":
            logger.debug("[UC-8.6] Database switched to HADEG")
            return False, True
        elif button_id == "uc-8-6-db-kegg":
            logger.debug("[UC-8.6] Database switched to KEGG")
            return True, False

        return False, True  # Default to HADEG

    @app.callback(
        Output("uc-8-6-chart", "children", allow_duplicate=True),
        [Input("uc-8-6-db-hadeg", "n_clicks"), Input("uc-8-6-db-kegg", "n_clicks")],
        prevent_initial_call=True,
    )
    def clear_chart_on_database_change(
        hadeg_clicks: Optional[int], kegg_clicks: Optional[int]
    ) -> html.Div:
        """
        Clear chart when database is changed.

        Returns empty div to reset visualization state when user
        switches between HADEG and KEGG databases.
        """
        logger.debug("[UC-8.6] Database changed, clearing chart")
        return html.Div(
            html.P(
                "Please select a pathway from the dropdown to generate "
                "the UpSet plot.",
                className="text-muted text-center p-5",
            ),
            className="border rounded p-3",
        )

    @app.callback(
        [
            Output("uc-8-6-pathway-dropdown", "options"),
            Output("uc-8-6-pathway-dropdown", "value"),
            Output("uc-8-6-pathway-help-text", "children"),
        ],
        [
            Input("uc-8-6-accordion-group", "active_item"),
            Input("uc-8-6-db-hadeg", "n_clicks"),
            Input("uc-8-6-db-kegg", "n_clicks"),
        ],
        [
            State("merged-result-store", "data"),
            State("uc-8-6-db-hadeg", "outline"),
            State("uc-8-6-db-kegg", "outline"),
        ],
        prevent_initial_call=True,
    )
    def populate_uc_8_6_pathway_dropdown(
        active_item: Optional[str],
        hadeg_clicks: Optional[int],
        kegg_clicks: Optional[int],
        merged_data: dict,
        hadeg_outline: bool,
        kegg_outline: bool,
    ) -> tuple[list[dict], Optional[str], str]:
        """
        Populate pathway dropdown when accordion is opened or database is changed.

        Parameters
        ----------
        active_item : str or None
            Active accordion item ID ('uc-8-6-accordion' when opened).
        hadeg_clicks : int or None
            Number of clicks on HADEG button.
        kegg_clicks : int or None
            Number of clicks on KEGG button.
        merged_data : dict
            Dictionary from merged-result-store.
        hadeg_outline : bool
            Whether HADEG button is outlined (not selected).
        kegg_outline : bool
            Whether KEGG button is outlined (not selected).

        Returns
        -------
        tuple of (list of dict, str or None, str)
            - Pathway options: [{'label': 'Name', 'value': 'Name'}, ...]
            - Default value: First pathway or None
            - Help text: Database-specific description

        Notes
        -----
        Triggered when:
        - Accordion is opened (initial load with HADEG default)
        - Database button is clicked (switch between HADEG/KEGG)

        Database Support:
        - HADEG: 58 pathways for degradation analysis (default)
        - KEGG: 19 pathways for general metabolism
        """
        from dash import callback_context

        logger.info("[UC-8.6] populate_pathway_dropdown callback triggered")

        ctx = callback_context
        if not ctx.triggered:
            raise PreventUpdate

        trigger_id = ctx.triggered[0]["prop_id"].split(".")[0]
        logger.debug(f"[UC-8.6] Triggered by: {trigger_id}")

        # Determine selected database based on trigger OR current state
        # Priority: If triggered by button click, use that button
        # Otherwise, use current state (outline values)
        if trigger_id == "uc-8-6-db-hadeg":
            selected_db_key = "hadeg_df"
            db_name = "HADEG"
        elif trigger_id == "uc-8-6-db-kegg":
            selected_db_key = "kegg_df"
            db_name = "KEGG"
        elif not hadeg_outline:
            # HADEG is selected (outline=False)
            selected_db_key = "hadeg_df"
            db_name = "HADEG"
        elif not kegg_outline:
            # KEGG is selected (outline=False)
            selected_db_key = "kegg_df"
            db_name = "KEGG"
        else:
            # Default to HADEG if no button selected
            selected_db_key = "hadeg_df"
            db_name = "HADEG"

        # Set help text based on selected database
        if db_name == "HADEG":
            help_text = (
                "HADEG Database: Pathways for biodegradation and "
                "environmental remediation. Analyze KO distribution across "
                "samples to design consortia for degradation of specific "
                "compounds."
            )
        else:  # KEGG
            help_text = (
                "KEGG Database: General metabolic pathways. Analyze KO "
                "distribution to understand functional complementarity and "
                "redundancy in metabolic processes."
            )

        logger.info(f"[UC-8.6] Selected database: {db_name}")

        # Extract database data
        if selected_db_key == "hadeg_df":
            df = _extract_hadeg_data(merged_data)
        else:  # kegg_df
            df = _extract_kegg_data(merged_data)

        if df is None or len(df) == 0:
            logger.warning(f"[UC-8.6] {db_name} data not available")
            return (
                [],
                None,
                f"⚠️ {db_name} data not available. Please load data " f"first.",
            )

        # Get unique pathways
        unique_pathways = _get_unique_pathways(df)

        if not unique_pathways:
            logger.warning(f"[UC-8.6] No pathways found in {db_name} data")
            return [], None, f"⚠️ No pathways found in {db_name} database."

        # Create dropdown options
        pathway_options = [
            {"label": pathway, "value": pathway} for pathway in unique_pathways
        ]

        # Don't set default value - let user choose explicitly
        default_value = None

        logger.info(
            f"[UC-8.6] Dropdown populated with {len(pathway_options)} "
            f"{db_name} pathways, no default selection (user must choose)"
        )

        return pathway_options, default_value, help_text

    @app.callback(
        Output("uc-8-6-chart", "children"),
        Input("uc-8-6-pathway-dropdown", "value"),
        [
            State("merged-result-store", "data"),
            State("uc-8-6-db-hadeg", "outline"),
            State("uc-8-6-db-kegg", "outline"),
        ],
        prevent_initial_call=True,
    )
    def render_uc_8_6(
        selected_pathway: Optional[str],
        merged_data: dict,
        hadeg_outline: bool,
        kegg_outline: bool,
    ) -> html.Div:
        """
        Generate UpSet plot for selected pathway from selected database.

        Parameters
        ----------
        selected_pathway : str or None
            Selected pathway name from dropdown.
        merged_data : dict
            Dictionary from merged-result-store.
        hadeg_outline : bool
            Whether HADEG button is outlined (not selected).
        kegg_outline : bool
            Whether KEGG button is outlined (not selected).

        Returns
        -------
        html.Div
            Container with UpSet plot or error message.

        Notes
        -----
        Rendering logic:
        1. Triggered by pathway dropdown selection
        2. Determines selected database (HADEG or KEGG)
        3. Extracts data from appropriate database
        4. Filters data for selected pathway
        5. Builds KO sets grouped by Sample
        6. Creates UpSet DataFrame format
        7. Passes to PlotService for visualization

        Database Support:
        - HADEG: Biodegradation pathways
        - KEGG: General metabolic pathways

        Error Handling:
        - No pathway selected → Prevent update
        - No data → Error message
        - Missing columns → Error message
        - No KOs for pathway → Warning message
        - Plot generation error → Error message with details
        """
        logger.info("[UC-8.6] render_uc_8_6 callback triggered")
        logger.debug(f"[UC-8.6] Selected pathway: '{selected_pathway}'")

        # Validate pathway selection
        if not selected_pathway:
            logger.debug("[UC-8.6] No pathway selected, preventing update")
            raise PreventUpdate

        # Determine selected database
        if not hadeg_outline:
            selected_db_key = "hadeg_df"
            db_name = "HADEG"
        elif not kegg_outline:
            selected_db_key = "kegg_df"
            db_name = "KEGG"
        else:
            # Default to HADEG
            selected_db_key = "hadeg_df"
            db_name = "HADEG"

        logger.info(
            f"[UC-8.6] Generating UpSet plot for pathway '{selected_pathway}' "
            f"from {db_name} database"
        )

        # Extract data from selected database
        if selected_db_key == "hadeg_df":
            df = _extract_hadeg_data(merged_data)
        else:  # kegg_df
            df = _extract_kegg_data(merged_data)

        if df is None or len(df) == 0:
            logger.error(f"[UC-8.6] {db_name} data not available")
            return _create_error_message(
                f"❌ **Error**: {db_name} data not available. Please upload "
                f"and process data first.",
                "danger",
            )

        # Filter data for selected pathway
        pathway_df = _filter_pathway_data(df, selected_pathway)

        if pathway_df is None or len(pathway_df) == 0:
            logger.warning(
                f"[UC-8.6] No data found for pathway '{selected_pathway}' "
                f"in {db_name} database"
            )
            return _create_error_message(
                f"⚠️ **No data available** for pathway "
                f"**{selected_pathway}** in {db_name} database.\n\n"
                f"This pathway may not be present in the {db_name} dataset, "
                f"or no KOs are associated with it. "
                f"Please select a different pathway.",
                "warning",
            )

        # Build KO sets grouped by Sample
        sample_ko_sets = _build_gene_ko_sets(pathway_df)

        if not sample_ko_sets:
            logger.error("[UC-8.6] Failed to build Sample-KO sets")
            return _create_error_message(
                "❌ **Error**: Failed to process KO data. Please check the "
                "dataset format and try again.",
                "danger",
            )

        # Create UpSet DataFrame from sets
        # Format: [{'category': sample_name, 'identifier': ko_id}, ...]
        upset_data = []
        for sample_name, ko_set in sample_ko_sets.items():
            for ko_id in ko_set:
                upset_data.append({"category": sample_name, "identifier": ko_id})

        upset_df = pd.DataFrame(upset_data)

        logger.info(
            f"[UC-8.6] Created UpSet DataFrame: {len(upset_df)} rows, "
            f"{len(sample_ko_sets)} samples, "
            f"{len(set(upset_df['identifier']))} unique KOs"
        )
        logger.debug(
            f"[UC-8.6] UpSet DataFrame sample (first 10 rows):\n" f"{upset_df.head(10)}"
        )

        # Generate plot using PlotService
        try:
            logger.info("[UC-8.6] Calling PlotService.generate_plot...")

            # Instantiate PlotService

            # Generate plot
            fig = plot_service.generate_plot(data=upset_df, use_case_id="UC-8.6")

            logger.info("[UC-8.6] Plot generated successfully")

            # Build a short, safe basename for the exported filename
            pathway_safe = str(selected_pathway).replace(" ", "_")
            db_short = db_name.lower()
            try:
                suggested = sanitize_filename(
                    "UC-8.6", f"upset_{db_short}_{pathway_safe}", "png"
                )
            except Exception:
                suggested = f"upset_{db_short}_{pathway_safe}.png"

            base_filename = os.path.splitext(suggested)[0]

            return html.Div(
                [
                    dcc.Graph(
                        figure=fig,
                        config={
                            "displayModeBar": True,
                            "displaylogo": False,
                            "toImageButtonOptions": {
                                "format": "svg",
                                "filename": base_filename,
                                "height": 800,
                                "width": 1000,
                                "scale": 2,
                            },
                        },
                    )
                ]
            )

        except Exception as e:
            logger.error(f"[UC-8.6] Error generating plot: {str(e)}", exc_info=True)
            return _create_error_message(
                f"❌ **Error generating plot**: {str(e)}\n\n"
                f"Please check the logs for more details.",
                "danger",
            )

    logger.info("[UC-8.6] Callbacks registered successfully")

uc_8_7_callbacks

UC-8.7 Callbacks - Intersection of Genes Across Samples.

This module implements callback functions for visualizing multi-sample gene set intersections using UpSet plot analysis.

Functions:

Name Description
register_uc_8_7_callbacks

Register all UC-8.7 callbacks with Dash app.

Notes
  • Refer to official documentation for use case details
  • Uses UpSetStrategy for gene set intersection visualization
  • Supports BioRemPP, HADEG, and KEGG databases

Version: 1.0.0

Functions

register_uc_8_7_callbacks
register_uc_8_7_callbacks(app, plot_service)

Register all UC-8.7 callbacks with Dash app.

Parameters:

Name Type Description Default
app Dash

Dash application instance.

required
plot_service PlotService

Singleton PlotService instance (shared across all callbacks).

required
Notes
  • Registers 3 callbacks: panel toggle, sample dropdown population, and UpSet plot rendering
  • Requires minimum 2 samples for UpSet plot generation
  • Refer to official documentation for processing logic details
Source code in src/presentation/callbacks/module8/uc_8_7_callbacks.py
def register_uc_8_7_callbacks(app, plot_service):
    """
    Register all UC-8.7 callbacks with Dash app.

    Parameters
    ----------
    app : Dash
        Dash application instance.
    plot_service : PlotService
        Singleton PlotService instance (shared across all callbacks).

    Notes
    -----
    - Registers 3 callbacks: panel toggle, sample dropdown population,
      and UpSet plot rendering
    - Requires minimum 2 samples for UpSet plot generation
    - Refer to official documentation for processing logic details
    """
    logger.info("[UC-8.7] Registering callbacks...")

    @app.callback(
        Output("uc-8-7-collapse", "is_open"),
        Input("uc-8-7-collapse-button", "n_clicks"),
        State("uc-8-7-collapse", "is_open"),
        prevent_initial_call=True,
    )
    def toggle_uc_8_7_info_panel(n_clicks: int, is_open: bool) -> bool:
        """
        Toggle UC-8.7 informative panel collapse state.

        Parameters
        ----------
        n_clicks : int
            Number of button clicks.
        is_open : bool
            Current collapse state.

        Returns
        -------
        bool
            New collapse state (inverted).
        """
        if n_clicks:
            logger.debug(f"[UC-8.7] Toggling info panel: {is_open}{not is_open}")
            return not is_open
        return is_open

    @app.callback(
        Output("uc-8-7-sample-dropdown", "options"),
        Input("uc-8-7-accordion-group", "active_item"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def populate_uc_8_7_sample_dropdown(
        active_item: Optional[str], merged_data: dict
    ) -> list[dict]:
        """
        Populate sample dropdown when accordion is opened.

        Parameters
        ----------
        active_item : str or None
            Active accordion item ID ('uc-8-7-accordion' when opened).
        merged_data : dict
            Dictionary from merged-result-store.

        Returns
        -------
        list of dict
            Sample options: [{'label': 'Name', 'value': 'Name'}, ...]

        Notes
        -----
        Triggered when accordion is opened.
        Extracts unique sample identifiers from BioRemPP database.
        """
        logger.info("[UC-8.7] populate_sample_dropdown callback triggered")
        logger.debug(f"[UC-8.7] active_item: {active_item}")

        # Only populate when accordion is opened
        if active_item != "uc-8-7-accordion":
            logger.debug("[UC-8.7] Accordion not opened, preventing update")
            raise PreventUpdate

        logger.info("[UC-8.7] Accordion opened, populating sample dropdown...")

        # Extract BioRemPP data
        biorempp_df = _extract_biorempp_data(merged_data)
        if biorempp_df is None or biorempp_df.empty:
            logger.error("[UC-8.7] No BioRemPP data available")
            return []

        # Verify required column exists
        if "Sample" not in biorempp_df.columns:
            logger.error(
                f"[UC-8.7] 'Sample' column not found. "
                f"Available columns: {list(biorempp_df.columns)}"
            )
            return []

        # Extract unique sample names
        samples = biorempp_df["Sample"].dropna().unique()
        samples = sorted([s for s in samples if s and str(s).strip()])

        logger.info(f"[UC-8.7] Found {len(samples)} unique samples: {samples}")

        # Create dropdown options
        options = [{"label": sample, "value": sample} for sample in samples]

        logger.info(f"[UC-8.7] Returning {len(options)} dropdown options")

        return options

    @app.callback(
        Output("uc-8-7-chart", "children"),
        Input("uc-8-7-sample-dropdown", "value"),
        State("merged-result-store", "data"),
        prevent_initial_call=True,
    )
    def render_uc_8_7(selected_samples: Optional[list], merged_data: dict) -> html.Div:
        """
        Generate UpSet plot for selected samples.

        Parameters
        ----------
        selected_samples : list of str or None
            Selected sample identifiers from dropdown.
        merged_data : dict
            Dictionary from merged-result-store.

        Returns
        -------
        html.Div
            Container with UpSet plot or error message.

        Notes
        -----
        Rendering logic:
        1. Triggered by sample dropdown selection
        2. Validates minimum 2 samples selected
        3. Extracts sample-KO associations from BioRemPP database
        4. Builds KO sets for each selected sample
        5. Creates UpSet plot data structure
        6. Passes to PlotService for visualization

        Error Handling:
        - Less than 2 samples → Error message
        - No data → Error message
        - Missing columns → Error message
        - Plot generation error → Error message with details

        Based on CLI reference: docs/CLI_UC/8.7/plot.py
        """
        logger.info("[UC-8.7] render_uc_8_7 callback triggered")
        logger.debug(f"[UC-8.7] Selected samples: {selected_samples}")

        # Validate input
        if not selected_samples or len(selected_samples) < 2:
            logger.warning(
                f"[UC-8.7] Insufficient samples selected: "
                f"{len(selected_samples) if selected_samples else 0}"
            )
            return _create_error_message(
                "❌ **Error**: Please select at least **2 samples** to "
                "generate the UpSet plot.\n\n"
                "The UpSet plot requires multiple samples to show "
                "meaningful intersections. Select 2 or more samples from "
                "the dropdown above.",
                "warning",
            )

        logger.info(
            f"[UC-8.7] Generating UpSet plot for {len(selected_samples)} "
            f"samples: {selected_samples}"
        )

        try:
            # Extract BioRemPP data
            biorempp_df = _extract_biorempp_data(merged_data)
            if biorempp_df is None or biorempp_df.empty:
                logger.error("[UC-8.7] No BioRemPP data available")
                return _create_error_message(
                    "❌ **Error**: No data available.\n\n"
                    "The merged result store does not contain BioRemPP data. "
                    "Please ensure data has been properly loaded.",
                    "danger",
                )

            # Prepare data for UpSet plot
            filtered_df = _prepare_upsetplot_data(biorempp_df, selected_samples)

            if filtered_df.empty:
                logger.warning(
                    f"[UC-8.7] No data found for selected samples: "
                    f"{selected_samples}"
                )
                return _create_error_message(
                    "⚠️ **Warning**: No data found for selected samples.\n\n"
                    f"The selected samples ({', '.join(selected_samples)}) "
                    "do not have any KO associations in the database.",
                    "warning",
                )

            logger.info(
                f"[UC-8.7] Filtered data: {len(filtered_df)} unique " "sample-KO pairs"
            )

            # Build KO memberships (which samples have which KOs)
            # Format: {KO_identifier: [sample1, sample2, ...]}
            logger.info("[UC-8.7] Generating KO to sample memberships...")
            memberships = (
                filtered_df.groupby("KO")["Sample"]
                .apply(lambda x: list(set(x)))
                .to_dict()
            )

            if not memberships:
                logger.error("[UC-8.7] No valid KO/sample memberships found")
                return _create_error_message(
                    "❌ **Error**: No valid gene associations found.\n\n"
                    "Unable to build KO memberships for the selected samples.",
                    "danger",
                )

            logger.info(f"[UC-8.7] Generated memberships for {len(memberships)} KOs")
            logger.debug(
                f"[UC-8.7] Sample memberships (first 3): "
                f"{dict(list(memberships.items())[:3])}"
            )

            # Prepare data for UpSet plot strategy
            # Format: [{category: sample, identifier: ko}, ...]
            upset_data = []
            for ko, samples in memberships.items():
                for sample in samples:
                    upset_data.append({"category": sample, "identifier": ko})

            # Convert to DataFrame (required by PlotService)
            upset_df = pd.DataFrame(upset_data)

            logger.info(
                f"[UC-8.7] Created UpSet DataFrame: {len(upset_df)} rows, "
                f"{len(selected_samples)} samples, "
                f"{len(memberships)} unique KOs"
            )
            logger.debug(
                f"[UC-8.7] UpSet DataFrame sample (first 10 rows):\n"
                f"{upset_df.head(10)}"
            )

            # Generate plot using PlotService
            logger.info("[UC-8.7] Calling PlotService to generate UpSet plot...")

            fig = plot_service.generate_plot(data=upset_df, use_case_id="UC-8.7")

            if fig is None:
                logger.error("[UC-8.7] PlotService returned None")
                return _create_error_message(
                    "❌ **Error**: Plot generation failed.\n\n"
                    "PlotService returned None. Check logs for details.",
                    "danger",
                )

            logger.info("[UC-8.7] [OK] UpSet plot generated successfully")

            # Prepare safe filename based on selected samples (use first 3 names)
            sample_label_parts = [
                str(s).replace(" ", "_") for s in selected_samples[:3]
            ]
            sample_label = "-".join(sample_label_parts)
            try:
                suggested = sanitize_filename(
                    "UC-8.7", f"samples_{len(selected_samples)}_{sample_label}", "png"
                )
            except Exception:
                suggested = f"samples_{len(selected_samples)}_{sample_label}.png"

            base_filename = os.path.splitext(suggested)[0]

            # Return plot in a Graph component with canonical filename
            return html.Div(
                dcc.Graph(
                    id="uc-8-7-upset-plot",
                    figure=fig,
                    config={
                        "displayModeBar": True,
                        "displaylogo": False,
                        "modeBarButtonsToRemove": ["lasso2d", "select2d"],
                        "toImageButtonOptions": {
                            "format": "svg",
                            "filename": base_filename,
                            "height": 800,
                            "width": 1000,
                            "scale": 2,
                        },
                    },
                ),
                className="mt-3",
            )

        except ValueError as ve:
            logger.error(f"[UC-8.7] ValueError: {ve}", exc_info=True)
            return _create_error_message(
                f"❌ **Validation Error**: {str(ve)}\n\n"
                "Please check your data and try again.",
                "danger",
            )

        except Exception as e:
            logger.error(
                f"[UC-8.7] Unexpected error generating plot: {e}", exc_info=True
            )
            return _create_error_message(
                f"❌ **Error**: Plot generation failed.\n\n"
                f"**Details**: {str(e)}\n\n"
                "Please check the console logs for more information.",
                "danger",
            )

    logger.info("[UC-8.7] Callbacks registered successfully")