Utilities¶
Utility functions and helpers for common tasks in Virtualitics SDK apps.
Image Utilities¶
Helper method to return a base64 encoding of an image. Supported filetypes include jpg, jpeg, png, and gif.
:meta private:
Parameters:
-
filepath–Relative path to image file.
Returns:
-
–
str, either empty string or the image string
This method allows an image to be downloaded asynchronously from S3 using the download Image endpoint.
Parameters:
-
image_path–Relative path to image file.
Returns:
-
–
The url to download the image.
Image Conversion¶
from virtualitics_sdk.utils.image_utils import (
image_to_base64,
base64_to_image,
resize_image
)
# Convert image to base64 for embedding
with open('logo.png', 'rb') as f:
base64_str = image_to_base64(f.read())
# Use in Image element
from virtualitics_sdk import Image
image = Image(
image_path=f"data:image/png;base64,{base64_str}",
size=ImageSize.MEDIUM
)
# Resize image
resized = resize_image(
image_data=image_bytes,
width=800,
height=600,
maintain_aspect=True
)
LLM Utilities¶
df_to_compact_markdown(df, title: Optional[str] = None, max_rows: Optional[int] = 10, max_cols: Optional[int] = None)
Convert a pandas DataFrame to a compact markdown representation with statistics.
Parameters:
-
df–(pd.DataFrame): The DataFrame to convert
-
title(Optional[str], default:None) –Title for the table
-
max_rows(Optional[int], default:10) –Maximum number of sample rows to include (default: 5)
-
max_cols(Optional[int], default:None) –Maximum number of columns to show (None = all) Returns: str: Markdown representation of the DataFrame
Prompt Templates¶
from virtualitics_sdk.utils.llm import format_prompt
# Format a prompt with context
prompt = format_prompt(
template="Analyze this dataset: {dataset_info}",
dataset_info=df.describe().to_string()
)
Type Utilities¶
Type Validation¶
from virtualitics_sdk.utils.types import (
validate_dataframe,
validate_numeric,
validate_string
)
# Validate DataFrame structure
is_valid = validate_dataframe(
df=data,
required_columns=["id", "name", "value"],
dtypes={"id": "int64", "value": "float64"}
)
# Validate numeric input
if validate_numeric(user_input, min_value=0, max_value=100):
process_value(user_input)
Progress Utilities¶
TQDM Integration¶
StepProgressTqdm(flow_metadata: FlowMetadata, total: int, starting_progress: Union[int, float] = 0, target_progress: Union[int, float] = 100, step_size: int = 10, callback: Optional[Callable] = None, call_super_update: bool = True, init_update: Optional[str] = None, **kwargs)
It is a tqdm wrapper that accept some parameters to update the front-end progress bar accordingly to the progress made in tqdm.
In addition to the classic tqdm init parameters it accepts a reference to the store interface and apply the update_progress call.
Parameters:
-
flow_metadata(FlowMetadata) –The app metadata necessary to create a store interface.
-
total(int) –Number of elements.
-
starting_progress(Union[int, float], default:0) –Progress starting point, defaults to 0.
-
target_progress(Union[int, float], default:100) –It is the target progress the tqdm engine will reach, defaults to 100.
-
callback(Optional[Callable], default:None) –Function that changes the default StepProgressTqdm manual update.
-
call_super_update(bool, default:True) –Whether by default the update method should call the default tqdm update method, defaults to True.
-
init_update(Optional[str], default:None) –An optional first update string to display before the first iteration of the loop has completed, defaults to None.
-
**kwargs–All the parameters the tqdm init exposes. EXAMPLE .. code-block:: python # Imports from virtualitics_sdk.utils.tqdm import StepProgressTqdm . . . # Example usage . . . class ExStep(Step): def run(self, flow_metadata): . . . progress_bar = StepProgressTqdm(flow_metadata, starting_progress=0, target_progress=100, total=100, step_size=10, desc="My progress bar", init_update="Starting my progress bar...") . . . for i in range(100): progress_bar.update(1)
Override the base update method. Useful for manual updates. If no store is provided, it will only call the super.update implementation otherwise it will perform an update to the front-end, calling the update_progress function.
Parameters:
from virtualitics_sdk.utils.tqdm import tqdm
class ProcessingStep(Step):
def run(self, flow_metadata):
data = self._inLink.large_dataset.data
results = []
for i, row in tqdm(data.iterrows(), total=len(data), desc="Processing"):
# Update progress
self._progress = int((i / len(data)) * 100)
self._message = f"Processing row {i+1} of {len(data)}"
# Process row
result = process_row(row)
results.append(result)
return Page(...)
Data Helpers¶
DataFrame Utilities¶
def clean_dataframe(df):
"""Clean and prepare DataFrame for display."""
# Remove unnamed index columns
df = df.loc[:, ~df.columns.str.contains('^Unnamed')]
# Convert datetime columns
for col in df.select_dtypes(include=['datetime64']).columns:
df[col] = df[col].dt.strftime('%Y-%m-%d %H:%M:%S')
# Round float columns
for col in df.select_dtypes(include=['float64']).columns:
df[col] = df[col].round(2)
# Replace NaN with empty string
df = df.fillna('')
return df
Data Validation¶
def validate_dataset(df, required_columns, min_rows=1):
"""Validate dataset meets requirements."""
errors = []
# Check required columns
missing = set(required_columns) - set(df.columns)
if missing:
errors.append(f"Missing columns: {missing}")
# Check minimum rows
if len(df) < min_rows:
errors.append(f"Dataset has {len(df)} rows, minimum {min_rows} required")
# Check for empty values in required columns
for col in required_columns:
if col in df.columns and df[col].isna().any():
errors.append(f"Column '{col}' contains null values")
return errors if errors else None
Serialization Helpers¶
import pickle
import dill
def safe_pickle(obj):
"""Safely pickle objects, falling back to dill if needed."""
try:
return pickle.dumps(obj)
except (pickle.PicklingError, TypeError):
return dill.dumps(obj)
def safe_unpickle(data):
"""Safely unpickle data."""
try:
return pickle.loads(data)
except (pickle.UnpicklingError, TypeError):
return dill.loads(data)
String Utilities¶
def truncate_string(text, max_length=100, suffix="..."):
"""Truncate string to maximum length."""
if len(text) <= max_length:
return text
return text[:max_length - len(suffix)] + suffix
def sanitize_filename(filename):
"""Remove invalid characters from filename."""
import re
return re.sub(r'[<>:"/\\|?*]', '_', filename)
Date/Time Utilities¶
from datetime import datetime, timedelta
def format_timestamp(dt, format='%Y-%m-%d %H:%M:%S'):
"""Format datetime for display."""
if isinstance(dt, str):
dt = datetime.fromisoformat(dt)
return dt.strftime(format)
def get_date_range(days_back=30):
"""Get date range for filtering."""
end_date = datetime.now()
start_date = end_date - timedelta(days=days_back)
return start_date, end_date
Error Handling Utilities¶
def create_error_page(error_message, title="Error"):
"""Create standardized error page."""
from virtualitics_sdk import Page, Section, Card, RichText
return Page(
title=title,
sections=[
Section(
title="Error",
cards=[
Card(
title=title,
content=[
RichText(f"❌ **Error:** {error_message}")
]
)
]
)
]
)
def create_success_page(message, title="Success"):
"""Create standardized success page."""
from virtualitics_sdk import Page, Section, Card, RichText
return Page(
title=title,
sections=[
Section(
title="Success",
cards=[
Card(
title=title,
content=[
RichText(f"✅ **Success:** {message}")
]
)
]
)
]
)
Best Practices¶
- Reusability: Create utility functions for repeated operations
- Error Handling: Include validation and error handling
- Documentation: Document utility functions clearly
- Testing: Test utilities with edge cases
- Performance: Optimize for common use cases