gws_plate_reader

This file contains helper functions for recipe management, UI display, and file information.

# functions_steps.py
from typing import List, Dict, Optional
import streamlit as st
from gws_core import Scenario, ScenarioStatus, EntityTagList, TagEntityType
from gws_core.streamlit import StreamlitTranslateService
from .my_custom_state import MyCustomState


def get_status_emoji(status: ScenarioStatus) -> str:
    """Return appropriate emoji for scenario status."""
    emoji_map = {
        ScenarioStatus.DRAFT: "📝",
        ScenarioStatus.IN_QUEUE: "⏳",
        ScenarioStatus.WAITING_FOR_CLI_PROCESS: "⏸️",
        ScenarioStatus.RUNNING: "🔄",
        ScenarioStatus.SUCCESS: "✅",
        ScenarioStatus.ERROR: "❌",
        ScenarioStatus.PARTIALLY_RUN: "✔️"
    }
    return emoji_map.get(status, "❓")


def get_status_material_icon(status: ScenarioStatus) -> str:
    """Return material icon for scenario status."""
    icon_map = {
        ScenarioStatus.SUCCESS: "check_circle",
        ScenarioStatus.ERROR: "error",
        ScenarioStatus.RUNNING: "pending",
        ScenarioStatus.IN_QUEUE: "schedule",
    }
    return icon_map.get(status, "help")


def create_custom_recipe_table_data(
    scenarios: List[Scenario],
    custom_state: MyCustomState,
    translate_service: StreamlitTranslateService
) -> List[Dict]:
    """
    Create table data from recipe scenarios, grouped by recipe.
    This is used to display recipes in a table on the first page.
    """
    table_data = []
    recipes_dict = {}
    
    for scenario in scenarios:
        # Get tags for this scenario
        entity_tag_list = EntityTagList.find_by_entity(TagEntityType.SCENARIO, scenario.id)
        
        # Extract recipe name
        recipe_name_tags = entity_tag_list.get_tags_by_key(
            custom_state.TAG_FERMENTOR_RECIPE_NAME
        )
        recipe_name = (recipe_name_tags[0].tag_value if recipe_name_tags 
                      else scenario.title or f"Scenario {scenario.id[:8]}")
        
        # Initialize recipe entry if not exists
        if recipe_name not in recipes_dict:
            recipes_dict[recipe_name] = {
                'recipe_name': recipe_name,
                'load_scenario': None,
                'selection_scenarios': [],
                'pipeline_id': '',
                'is_microplate': False,
                'created_at': None,
                'created_by': ''
            }
        
        # Check if this is a load scenario (data processing)
        custom_tags = entity_tag_list.get_tags_by_key(custom_state.TAG_FERMENTOR)
        is_load_scenario = any(
            tag.tag_value == custom_state.TAG_DATA_PROCESSING for tag in custom_tags
        )
        
        if is_load_scenario:
            recipes_dict[recipe_name]['load_scenario'] = scenario
            recipes_dict[recipe_name]['created_at'] = scenario.created_at
            recipes_dict[recipe_name]['created_by'] = scenario.created_by.full_name
            
            # Extract pipeline ID
            pipeline_id_tags = entity_tag_list.get_tags_by_key(
                custom_state.TAG_FERMENTOR_PIPELINE_ID
            )
            if pipeline_id_tags:
                recipes_dict[recipe_name]['pipeline_id'] = pipeline_id_tags[0].tag_value
            
            # Extract microplate flag
            microplate_tags = entity_tag_list.get_tags_by_key(
                custom_state.TAG_MICROPLATE_ANALYSIS
            )
            if microplate_tags and microplate_tags[0].tag_value == "true":
                recipes_dict[recipe_name]['is_microplate'] = True
    
    # Convert dict to list for table display
    for recipe_name, recipe_data in recipes_dict.items():
        if recipe_data['load_scenario']:  # Only include recipes with load scenario
            table_data.append({
                'recipe_name': recipe_name,
                'status': get_status_emoji(recipe_data['load_scenario'].status),
                'scenario_id': recipe_data['load_scenario'].id,
                'created_at': recipe_data['created_at'].strftime('%Y-%m-%d %H:%M'),
                'created_by': recipe_data['created_by'],
                'type': '🧪' if recipe_data['is_microplate'] else '🔬'
            })
    
    return table_data


def get_custom_input_files_info(
    scenario_id: str,
    custom_state: MyCustomState,
    translate_service: StreamlitTranslateService
) -> List[Dict]:
    """
    Get information about input files for a scenario.
    Returns a list of dicts with file metadata.
    """
    entity_tag_list = EntityTagList.find_by_entity(TagEntityType.SCENARIO, scenario_id)
    
    files_info = []
    
    # Define file tags to look for
    file_tag_keys = [
        ('my_custom_info_file', 'Info File'),
        ('my_custom_medium_file', 'Medium File'),
        ('my_custom_timeseries_file', 'Time Series File')
    ]
    
    for tag_key, display_name in file_tag_keys:
        file_tags = entity_tag_list.get_tags_by_key(tag_key)
        if file_tags:
            files_info.append({
                'name': display_name,
                'filename': file_tags[0].tag_value,
                'type': 'CSV' if 'csv' in tag_key else 'ZIP'
            })
    
    return files_info


def get_analysis_summary(
    scenario: Scenario,
    custom_state: MyCustomState,
    translate_service: StreamlitTranslateService
) -> Dict:
    """
    Get summary information about an analysis scenario.
    Used to display overview information.
    """
    entity_tag_list = EntityTagList.find_by_entity(TagEntityType.SCENARIO, scenario.id)
    
    # Extract recipe info
    recipe_name = custom_state._extract_tag_value(
        entity_tag_list, custom_state.TAG_FERMENTOR_RECIPE_NAME, scenario.title
    )
    
    pipeline_id = custom_state._extract_tag_value(
        entity_tag_list, custom_state.TAG_FERMENTOR_PIPELINE_ID, scenario.id
    )
    
    microplate_value = custom_state._extract_tag_value(
        entity_tag_list, custom_state.TAG_MICROPLATE_ANALYSIS, "false"
    )
    
    return {
        'recipe_name': recipe_name,
        'pipeline_id': pipeline_id,
        'is_microplate': microplate_value == "true",
        'status': scenario.status,
        'created_at': scenario.created_at,
        'created_by': scenario.created_by.full_name
    }

Key Functions to Implement:

• get_status_emoji(): Status visualization
• create_custom_recipe_table_data(): Recipe list for first page
• get_custom_input_files_info(): Display file information
• get_analysis_summary(): Overview page data