Step 2: Create a State Manager
The State Manager extends CellCultureState and defines constants specific to your data.
# my_custom_state.py
from gws_core import Scenario
from gws_plate_reader.cell_culture_app_core.cell_culture_state import CellCultureState
from gws_plate_reader.cell_culture_app_core.cell_culture_recipe import CellCultureRecipe
from .my_custom_recipe import MyCustomRecipe
class MyCustomState(CellCultureState):
"""
Custom state manager for My Custom App.
"""
# Override tags to match your Load Data Task
TAG_FERMENTOR = "my_custom_app"
TAG_FERMENTOR_RECIPE_NAME = "my_custom_recipe_name"
TAG_FERMENTOR_PIPELINE_ID = "my_custom_pipeline_id"
TAG_FERMENTOR_SELECTION_STEP = "my_custom_selection_step"
TAG_FERMENTOR_QUALITY_CHECK_PARENT_SELECTION = "my_custom_qc_parent_selection"
TAG_FERMENTOR_ANALYSES_PARENT_SELECTION = "my_custom_analyses_parent_selection"
TAG_FERMENTOR_ANALYSES_PARENT_QUALITY_CHECK = "my_custom_analyses_parent_qc"
# Process name (used to find scenarios)
PROCESS_NAME_DATA_PROCESSING = 'my_custom_data_processing'
# Column names matching your data structure
BASE_TIME_COLUMN_NAME = "Time (h)" # Your time column name
BATCH_COLUMN_NAME = "Batch" # Your batch column name
SAMPLE_COLUMN_NAME = "Sample" # Your sample column name
def __init__(self, lang_translation_folder_path: str):
super().__init__(lang_translation_folder_path)
def create_recipe_from_scenario(self, scenario: Scenario) -> CellCultureRecipe:
"""
Create a custom recipe instance from a scenario.
"""
return MyCustomRecipe.from_scenario(scenario)
Step 3: Create a Recipe Class
The Recipe class extracts metadata from scenarios and organizes analysis steps. It also provides helper methods to retrieve and manage scenarios.
# my_custom_recipe.py
from dataclasses import dataclass
from typing import Optional, List, Dict
from gws_core import Scenario
from gws_core.tag.entity_tag_list import EntityTagList
from gws_core.tag.tag_entity_type import TagEntityType
from gws_plate_reader.cell_culture_app_core.cell_culture_recipe import CellCultureRecipe
@dataclass
class MyCustomRecipe(CellCultureRecipe):
"""
Recipe class for My Custom App.
Provides methods to access and organize scenarios by analysis steps.
"""
@classmethod
def from_scenario(cls, scenario: Scenario) -> 'MyCustomRecipe':
"""
Create a recipe from a scenario by extracting tags.
"""
entity_tag_list = EntityTagList.find_by_entity(TagEntityType.SCENARIO, scenario.id)
# Extract recipe metadata from tags
recipe_name = cls._extract_tag_value(
entity_tag_list, 'my_custom_recipe_name', default='Unknown Recipe'
)
pipeline_id = cls._extract_tag_value(
entity_tag_list, 'my_custom_pipeline_id'
)
# Extract analysis type (microplate or standard)
microplate_value = cls._extract_tag_value(
entity_tag_list, "microplate_analysis", "false"
)
analysis_type = "microplate" if microplate_value == "true" else "standard"
# Extract file information (optional, for display)
file_tags = [
('my_custom_info_file', 'Info File'),
('my_custom_medium_file', 'Medium File'),
('my_custom_timeseries_file', 'Time Series File')
]
file_info = cls._extract_file_info(entity_tag_list, file_tags)
# Initialize scenarios dictionary
scenarios_by_step = {
"data_processing": [scenario]
}
# Create recipe instance
return cls(
id=scenario.id,
name=recipe_name,
analysis_type=analysis_type,
created_by=scenario.created_by,
created_at=scenario.created_at,
scenarios=scenarios_by_step,
main_scenario=scenario,
pipeline_id=pipeline_id,
file_info=file_info,
has_data_raw=True
)
# Helper methods to access scenarios
def get_load_scenario(self) -> Optional[Scenario]:
"""Get the main data processing scenario."""
return self.scenarios.get('data_processing', [None])[0]
def get_scenarios_for_step(self, step: str) -> List[Scenario]:
"""Get scenarios for a specific step."""
return self.scenarios.get(step, [])
def get_selection_scenarios(self) -> List[Scenario]:
"""Get selection scenarios for this recipe."""
return self.get_scenarios_for_step('selection')
def get_quality_check_scenarios(self) -> List[Scenario]:
"""Get all quality check scenarios for this recipe."""
return self.get_scenarios_for_step('quality_check')
def get_quality_check_scenarios_for_selection(self, selection_id: str) -> List[Scenario]:
"""
Get quality check scenarios linked to a specific selection scenario.
:param selection_id: ID of the parent selection scenario
:return: List of quality check scenarios for this selection
"""
all_qc_scenarios = self.get_quality_check_scenarios()
# Filter by parent selection ID tag
filtered_scenarios = []
for scenario in all_qc_scenarios:
entity_tag_list = EntityTagList.find_by_entity(TagEntityType.SCENARIO, scenario.id)
parent_selection_tags = entity_tag_list.get_tags_by_key(
"my_custom_quality_check_parent_selection"
)
if parent_selection_tags and parent_selection_tags[0].tag_value == selection_id:
filtered_scenarios.append(scenario)
return filtered_scenarios
def add_quality_check_scenario(self, selection_id: str, quality_check_scenario: Scenario) -> None:
"""Add a quality check scenario to this recipe."""
existing_qc_scenarios = self.get_quality_check_scenarios()
updated_qc_scenarios = [quality_check_scenario] + existing_qc_scenarios
# Sort by creation date (most recent last)
updated_qc_scenarios.sort(key=lambda s: s.created_at)
self.add_scenarios_by_step('quality_check', updated_qc_scenarios)
def get_analyses_scenarios(self) -> List[Scenario]:
"""Get all analysis scenarios (PCA, UMAP, Feature Extraction, etc.)."""
return self.get_scenarios_for_step('analyses')
def get_selection_scenarios_organized(self) -> Dict[str, Scenario]:
"""
Get selection scenarios organized by name.
Returns a dict {selection_name: scenario}.
"""
selection_scenarios = self.get_selection_scenarios()
organized = {}
for scenario in selection_scenarios:
# Use scenario title as key, or generate one
key = scenario.title if scenario.title else f"Selection {scenario.id[:8]}"
organized[key] = scenario
return organized
def has_selection_scenarios(self) -> bool:
"""Check if this recipe has any selection scenarios."""
return len(self.get_selection_scenarios()) > 0
Key Methods to Implement:
-
from_scenario(): Extract metadata from scenario tags (REQUIRED) -
get_load_scenario(): Return main data processing scenario -
get_selection_scenarios(): Return all selection scenarios -
get_quality_check_scenarios_for_selection(): Filter QC by parent selection -
get_selection_scenarios_organized(): Organize selections for UI display -
has_selection_scenarios(): Check if selections exist
