State Management API

Session state and workflow management for ML training pipelines

Overview

The state management API provides comprehensive tools for managing ML workflow sessions, including dataset configurations, model libraries, training configurations, experiments, and session persistence.

Workflow State

init_workflow_state()

Initialize workflow state with default values. Automatically loads persisted sessions if available.

from state.workflow import init_workflow_state

init_workflow_state()

File reference: app/state/workflow.py:54

generate_session_id()

Generate a bird-based session ID for easy identification.

from state.workflow import generate_session_id

session_id = generate_session_id()
# Returns: "swift-golden-falcon"

Returns: str - A unique, human-readable session identifier

File reference: app/state/workflow.py:104

Session Management

get_session_id()

Get the current session ID.

from state.workflow import get_session_id

session_id = get_session_id()

Returns: str - Current session identifier

File reference: app/state/workflow.py:113

clear_workflow_state()

Clear all workflow-related session state and reinitialize with fresh values.

from state.workflow import clear_workflow_state

clear_workflow_state()

File reference: app/state/workflow.py:419

Dataset Configuration

save_dataset_config(config)

Save dataset configuration to session state and persist to disk.

config · dict[str, Any] · required

Dataset configuration dictionary containing paths, splits, and preprocessing settings

from state.workflow import save_dataset_config

config = {
    "dataset_path": "/path/to/dataset",
    "train_split": 70,
    "val_split": 15,
    "test_split": 15,
    "target_size": "224x224",
    "color_mode": "RGB"
}
save_dataset_config(config)

File reference: app/state/workflow.py:119

get_dataset_config()

Retrieve the current dataset configuration.

from state.workflow import get_dataset_config

config = get_dataset_config()

Returns: dict[str, Any] - Dataset configuration dictionary

File reference: app/state/workflow.py:131

has_dataset_config()

Check if dataset is configured.

from state.workflow import has_dataset_config

if has_dataset_config():
    print("Dataset is configured")

Returns: bool - True if dataset configuration exists

File reference: app/state/workflow.py:136

Model Library

add_model_to_library(name, config)

Add a new model to the library.

name · str · required

Human-readable name for the model

config · dict[str, Any] · required

Model configuration dictionary with architecture and hyperparameters

from state.workflow import add_model_to_library

config = {
    "model_type": "Custom CNN",
    "cnn_config": {
        "layers": [
            {"type": "conv", "filters": 32, "kernel_size": 3},
            {"type": "maxpool", "pool_size": 2}
        ]
    }
}

model_id = add_model_to_library("My CNN Model", config)
# Returns: "model_a3f8b2c1"

Returns: str - Unique model ID

File reference: app/state/workflow.py:247

get_model_from_library(model_id)

Get a model by ID from the library.

model_id · str · required

Unique model identifier

from state.workflow import get_model_from_library

model = get_model_from_library("model_a3f8b2c1")
if model:
    print(f"Model: {model['name']}")
    print(f"Type: {model['model_type']}")

Returns: dict[str, Any] | None - Model dictionary or None if not found

File reference: app/state/workflow.py:262

update_model_in_library(model_id, name, config)

Update an existing model in the library.

model_id · str · required

Model identifier to update

name · str · required

New name for the model

config · dict[str, Any] · required

Updated model configuration

from state.workflow import update_model_in_library

success = update_model_in_library(
    "model_a3f8b2c1",
    "Updated CNN Model",
    updated_config
)

Returns: bool - True if model was found and updated

File reference: app/state/workflow.py:270

delete_model_from_library(model_id)

Delete a model from the library.

model_id · str · required

Model identifier to delete

from state.workflow import delete_model_from_library

success = delete_model_from_library("model_a3f8b2c1")

Returns: bool - True if model was found and deleted

File reference: app/state/workflow.py:282

get_model_library()

Get all models in the library.

from state.workflow import get_model_library

models = get_model_library()
for model in models:
    print(f"{model['name']} ({model['model_type']})")

Returns: list[dict[str, Any]] - List of all model dictionaries

File reference: app/state/workflow.py:293

Training Library

add_training_to_library(name, config)

Add a new training configuration to the library.

name · str · required

Human-readable name for the training configuration

config · dict[str, Any] · required

Training configuration with hyperparameters and settings

from state.workflow import add_training_to_library

config = {
    "epochs": 50,
    "batch_size": 32,
    "learning_rate": 0.001,
    "optimizer": "adam"
}

training_id = add_training_to_library("Standard Training", config)
# Returns: "train_d4e9f1a2"

Returns: str - Unique training configuration ID

File reference: app/state/workflow.py:303

get_training_library()

Get all training configurations in the library.

from state.workflow import get_training_library

trainings = get_training_library()
for training in trainings:
    print(f"{training['name']}: {training['config']['epochs']} epochs")

Returns: list[dict[str, Any]] - List of all training configuration dictionaries

File reference: app/state/workflow.py:349

Experiments

create_experiment(name, model_id, training_id)

Create a new experiment by combining a model and training configuration.

name · str · required

Human-readable experiment name

model_id · str · required

Model identifier from library

training_id · str · required

Training configuration identifier from library

from state.workflow import create_experiment

exp_id = create_experiment(
    "CNN Baseline",
    "model_a3f8b2c1",
    "train_d4e9f1a2"
)
# Returns: "exp_b7c4d2e5"

Returns: str - Unique experiment ID

File reference: app/state/workflow.py:359

get_experiment(exp_id)

Get an experiment by ID.

exp_id · str · required

Experiment identifier

from state.workflow import get_experiment

experiment = get_experiment("exp_b7c4d2e5")
if experiment:
    print(f"Status: {experiment['status']}")
    print(f"Current epoch: {experiment['current_epoch']}")

Returns: dict[str, Any] | None - Experiment dictionary or None if not found

Experiment Status Values:

• ready - Ready to start training
• training - Currently training
• paused - Training paused
• completed - Training finished successfully
• failed - Training failed with error

File reference: app/state/workflow.py:380

update_experiment(exp_id, updates)

Update experiment fields.

exp_id · str · required

Experiment identifier

updates · dict[str, Any] · required

Dictionary of fields to update

from state.workflow import update_experiment

success = update_experiment("exp_b7c4d2e5", {
    "status": "training",
    "current_epoch": 10,
    "metrics": {
        "train_loss": 0.234,
        "val_acc": 0.892
    }
})

Returns: bool - True if experiment was found and updated

File reference: app/state/workflow.py:388

delete_experiment(exp_id)

Delete an experiment.

exp_id · str · required

Experiment identifier to delete

from state.workflow import delete_experiment

success = delete_experiment("exp_b7c4d2e5")

Returns: bool - True if experiment was found and deleted

File reference: app/state/workflow.py:398

get_experiments()

Get all experiments.

from state.workflow import get_experiments

experiments = get_experiments()
for exp in experiments:
    print(f"{exp['name']}: {exp['status']}")

Returns: list[dict[str, Any]] - List of all experiment dictionaries

File reference: app/state/workflow.py:409

Session Persistence

save_session(session_id)

Save current session state to disk.

session_id · str · required

Session identifier to save

from state.persistence import save_session
from state.workflow import get_session_id

session_id = get_session_id()
if save_session(session_id):
    print("Session saved successfully")

Returns: bool - True if saved successfully

File reference: app/state/persistence.py:23

load_session(session_id)

Load session state from disk into st.session_state.

session_id · str · required

Session identifier to load

from state.persistence import load_session

if load_session("swift-golden-falcon"):
    print("Session loaded successfully")

Returns: bool - True if loaded successfully

File reference: app/state/persistence.py:66

list_saved_sessions()

Get list of all saved session IDs, sorted by modification time (newest first).

from state.persistence import list_saved_sessions

sessions = list_saved_sessions()
for session in sessions:
    print(f"Session: {session}")

Returns: list[str] - List of session IDs

File reference: app/state/persistence.py:110

delete_session(session_id)

Delete a saved session file.

session_id · str · required

Session identifier to delete

from state.persistence import delete_session

if delete_session("old-session-id"):
    print("Session deleted successfully")

Returns: bool - True if deleted successfully

File reference: app/state/persistence.py:136

Cache Management

get_dataset_info()

Get cached dataset scan information.

from state.cache import get_dataset_info

info = get_dataset_info()
if info:
    print(f"Total training samples: {info['total_train']}")
    print(f"Total validation samples: {info['total_val']}")
    print(f"Classes: {', '.join(info['classes'])}")

Returns: DatasetInfo | None - Cached dataset information or None

DatasetInfo Structure:

• total_train: int - Total training samples
• total_val: int - Total validation samples
• classes: list[str] - List of class names
• train_samples: dict[str, int] - Samples per class in training set
• val_samples: dict[str, int] - Samples per class in validation set
• sample_paths: dict[str, list[Path]] - Sample image paths per class

File reference: app/state/cache.py:49

set_dataset_info(info)

Cache dataset scan information.

info · DatasetInfo · required

Dataset information dictionary to cache

from state.cache import set_dataset_info

info = {
    "total_train": 1000,
    "total_val": 200,
    "classes": ["malware_a", "malware_b"],
    "train_samples": {"malware_a": 600, "malware_b": 400},
    "val_samples": {"malware_a": 120, "malware_b": 80}
}
set_dataset_info(info)

File reference: app/state/cache.py:54

clear_cache_state()

Clear all cached data and reinitialize with defaults.

from state.cache import clear_cache_state

clear_cache_state()

File reference: app/state/cache.py:109

WorkflowState Type

Type definition for workflow state fields:

class WorkflowState(TypedDict, total=False):
    session_id: str
    dataset_config: dict[str, Any]
    model_library: list[dict[str, Any]]
    training_library: list[dict[str, Any]]
    experiments: list[dict[str, Any]]
    model_config: dict[str, Any]  # Legacy
    training_config: dict[str, Any]  # Legacy
    training_active: bool  # Legacy
    monitor_config: dict[str, Any]  # Legacy
    results: dict[str, Any] | None  # Legacy

File reference: app/state/workflow.py:33