Models

This document provides comprehensive documentation for model-related classes in the ViewAI Python SDK.

Overview

The ViewAI SDK provides several model-related classes:

• Model: Deploy and manage trained ML models

• Prediction: Encapsulate prediction results

• DataPoint: Validate individual data points for prediction

• ModelSchemaManager: Manage model schemas and validation

---

Model

The Model class handles the complete lifecycle of machine learning model deployment to the ViewAI platform.

Key Features

• Automatic ONNX export from sklearn models for version-agnostic deployment

• Comprehensive schema generation and validation

• Model deployment to ViewAI backend with S3 upload

• Schema management with type validation and category constraints

• Backward compatibility with pickle-based deployment

Initialization

Model(model, df, target, schema, name, workspace, project, client, export_format="onnx")

Parameters:

• model (Any): Trained sklearn model or pipeline (must be compatible with sklearn's API)

• df (DataFrame): Training dataframe used for schema generation (should contain all feature columns and target)

• target (str): Name of the target column in the dataframe

• schema (dict or ModelSchemaManager, optional): Model schema specification

  • None: Auto-generate schema from model and data

  • dict: Pre-defined schema dictionary

  • ModelSchemaManager: Existing schema manager instance

• name (str): Human-readable name for the model deployment

• workspace (Workspace): Workspace object with workspace_id attribute

• project (Project, optional): Optional Project object with project_id attribute

• client (ViewAIClient): ViewAIClient instance for API communication

• export_format (str): Export format for the model (default: "onnx")

  • "onnx" (recommended): Cross-platform, version-agnostic format

  • "pickle" (legacy): Python-specific serialization

Example:

from sklearn.ensemble import RandomForestClassifier
from viewai_client import ViewAIClient

# Train your model
model = RandomForestClassifier()
model.fit(X_train, y_train)

# Deploy to ViewAI
client = ViewAIClient(api_key="your-api-key")
workspace = client.retrieve_default_workspace()

viewai_model = Model(
    model=model,
    df=train_df,
    target="target_column",
    schema=None,  # Auto-generate schema
    name="My RF Model",
    workspace=workspace,
    project=None,
    client=client,
    export_format="onnx"
)

---

deploy()

Deploy pre-trained model to ViewAI platform.

Returns: Dashboard ID (str) for the deployed model, or None if deployment fails

Example:

dashboard_id = model.deploy()
if dashboard_id:
    print(f"Model deployed: {dashboard_id}")
    print(f"View at: https://app.viewai.ca/dashboard/{dashboard_id}")

Process:

1

Request pre-signed S3 URLs

Request presigned S3 URLs from ViewAI backend.

2

Save dataframe to CSV

Save the training/dataframe artifact to CSV for upload.

3

Export model

Export the model to ONNX or pickle format depending on export_format.

4

Generate and save schema

Generate the schema JSON and save locally for upload.

5

Upload artifacts to S3

Upload the CSV, model artifact, and schema JSON to the presigned S3 URLs.

6

Poll for deployment completion

Poll the backend for deployment completion and return dashboard ID.

---

Schema Management Methods

get_schema()

Get the current model schema.

Returns: Complete schema dictionary with all field specifications

Example:

schema = model.get_schema()
print(schema.keys())  # dict_keys(['fields', 'onnx_metadata'])
print(schema['fields']['age'])  # {'type': 'int', 'min': 0, 'max': 120}

---

update_schema_column_categories(column_name, categories=None, allow_new_categories=None)

Update categories for a categorical column in the schema.

Parameters:

• column_name (str): Name of the categorical column to update

• categories (List, optional): New list of allowed category values

• allow_new_categories (bool, optional): Whether to allow new categories at inference time

Returns: self (for method chaining)

Example:

model.update_schema_column_categories(
    column_name="country",
    categories=["USA", "Canada", "Mexico"],
    allow_new_categories=False
)

---

update_schema_column_range(column_name, min_value=None, max_value=None)

Update value range for a numerical column in the schema.

Parameters:

• column_name (str): Name of the numerical column to update

• min_value (float, optional): Minimum allowed value (inclusive)

• max_value (float, optional): Maximum allowed value (inclusive)

Returns: self (for method chaining)

Example:

model.update_schema_column_range(
    column_name="age",
    min_value=0,
    max_value=120
)

model.update_schema_column_range(
    column_name="temperature",
    min_value=-50.0,
    max_value=50.0
)

---

update_schema_column_type(column_name, data_type)

Update data type for a column in the schema.

Parameters:

• column_name (str): Name of the column to update

• data_type (str): New data type. Must be one of:

  • "int": Integer values

  • "float": Floating point values

  • "category": Categorical/discrete values

  • "str": String values

Returns: self (for method chaining)

Example:

model.update_schema_column_type(
    column_name="user_id",
    data_type="str"
)

model.update_schema_column_type(
    column_name="score",
    data_type="float"
)

---

save_schema_locally(file_path="schema.json")

Save schema to a local JSON file.

Parameters:

• file_path (str): Path where schema JSON will be saved (default: "schema.json")

Returns: self (for method chaining)

Example:

model.save_schema_locally("./schemas/model_v1_schema.json")

# Or use default path
model.save_schema_locally()

---

preview_schema()

Print a formatted preview of the schema.

Returns: self (for method chaining)

Example:

model.preview_schema()

Output:

============================================================
📋 MODEL SCHEMA PREVIEW
============================================================

🔹 age
   Type: int
   Range: [0, 120]

🔹 country
   Type: category
   Choices: ['USA', 'Canada', 'Mexico', 'UK', 'Germany']
   Allow new: False

🔹 income
   Type: float
   Range: [0.0, 1000000.0]

============================================================

---

validate_data(df)

Validate a dataframe against the current schema.

Parameters:

• df (DataFrame): DataFrame to validate against the schema

Returns: Dictionary containing validation results:

• 'valid' (bool): True if validation passed

• 'errors' (list): List of error messages

Example:

import pandas as pd

# Valid data
valid_df = pd.DataFrame({
    'age': [25, 30, 45],
    'country': ['USA', 'Canada', 'Mexico']
})
result = model.validate_data(valid_df)
print(result)  # {'valid': True, 'errors': []}

# Invalid data
invalid_df = pd.DataFrame({
    'age': [25, 30, 150],  # 150 exceeds max
    'country': ['USA', 'Canada', 'Unknown']  # Unknown not allowed
})
result = model.validate_data(invalid_df)
print(result)

# {
#     'valid': False,
#     'errors': [
#         'age: Value above maximum 120',
#         'country: Invalid categories {\'Unknown\'}. Allowed: [...]'
#     ]
# }

---

Complete Model Example

from sklearn.ensemble import RandomForestClassifier
from viewai_client import ViewAIClient
import pandas as pd

# Prepare data
df = pd.read_csv("training_data.csv")
X = df.drop("target", axis=1)
y = df["target"]

# Train model
rf_model = RandomForestClassifier(n_estimators=100)
rf_model.fit(X, y)

# Initialize client
client = ViewAIClient(api_key="your-api-key")
workspace = client.retrieve_default_workspace()

# Create deployable model
model = client.create_deployable_model(
    model=rf_model,
    dataset=df,
    target_column="target",
    model_name="RF Classifier v1.0"
)

# Customize schema
model.update_schema_column_range("age", min_value=0, max_value=120)
model.update_schema_column_categories(
    "region",
    categories=["North", "South", "East", "West"],
    allow_new_categories=False
)

# Preview and validate
model.preview_schema()

test_df = pd.DataFrame({...})
validation = model.validate_data(test_df)
if validation['valid']:
    # Deploy model
    dashboard_id = model.deploy()
    print(f"Model deployed: {dashboard_id}")
else:
    print("Validation errors:", validation['errors'])

---

Prediction

The Prediction class encapsulates prediction results and provides methods to extract specific information.

Initialization

Prediction(data)

Parameters:

• data (dict): Dictionary containing prediction results and associated data

Example:

prediction = Prediction({
    "predicted_churn": 1,
    "probability_0": 0.3,
    "probability_1": 0.7,
    "date": "2024-01-15"
})

---

get_probabilities()

Extract probability values or predicted categories from the prediction.

Returns: Dictionary containing relevant probability or prediction data

Example:

prediction = Prediction({
    "predicted_class": 1,
    "probability_0": 0.3,
    "probability_1": 0.7
})

probs = prediction.get_probabilities()
print(probs)

# {'predicted_class': 1, 'probability_0': 0.3, 'probability_1': 0.7}

---

get_timestamp()

Retrieve the timestamp associated with the prediction.

Returns: Timestamp string or empty string if not available

Example:

prediction = Prediction({"date": "2024-01-15", "value": 42})
timestamp = prediction.get_timestamp()
print(timestamp)  # '2024-01-15'

---

get_raw_prediction()

Return the complete raw prediction data.

Returns: The complete prediction data dictionary

Example:

prediction = Prediction({"predicted_value": 42, "confidence": 0.95})
raw = prediction.get_raw_prediction()
print(raw)  # {'predicted_value': 42, 'confidence': 0.95}

---

get(key, default=None)

Get a specific value from the prediction data.

Parameters:

• key (str): The key to retrieve from prediction data

• default (Any): Default value to return if key is not found

Returns: Value associated with the key, or default if not found

Example:

prediction = Prediction({"predicted_value": 42})
value = prediction.get("predicted_value")
print(value)  # 42

missing = prediction.get("nonexistent", "N/A")
print(missing)  # 'N/A'

---

Dictionary-like Access

The Prediction class supports dictionary-style access:

Example:

prediction = Prediction({"predicted_value": 42, "confidence": 0.95})

# Dictionary-style access
value = prediction["predicted_value"]
print(value)  # 42

# Check if key exists
if "confidence" in prediction:
    print(prediction["confidence"])

# String representation
print(prediction)  # Prediction(probabilities={...})

---

DataPoint

The DataPoint class validates individual data points against model schemas before sending them for prediction.

Initialization

DataPoint(data, model_id, api_key)

Parameters:

• data (dict): The data point structured as a dictionary

• model_id (str): The identifier for the model to score against

• api_key (str): API key used to authenticate and retrieve the model schema

Example:

data_point = DataPoint(
    data={"age": 25, "income": 50000.0, "region": "west"},
    model_id="model_123",
    api_key="your-api-key"
)

---

get_data()

Validate and return the data point.

Returns: The validated data dictionary

Raises:

• ValueError: If validation fails

Example:

data_point = DataPoint({...}, "model_123", "api-key")
try:
    validated_data = data_point.get_data()
    print("Data is valid:", validated_data)
except ValueError as e:
    print(f"Validation failed: {e}")

---

validate_data()

Validate the data point against the schema's validation rules.

Raises:

• ValueError: If any validation rule is violated

Example:

data_point = DataPoint({...}, "model_123", "api-key")
try:
    data_point.validate_data()
    print("Validation passed")
except ValueError as e:
    print(f"Validation failed: {e}")

---

ModelSchemaManager

The ModelSchemaManager class manages loading and applying model-specific schemas for validation and configuration.

Initialization

ModelSchemaManager(model_id=None, api_key=None)

Parameters:

• model_id (str, optional): Unique identifier for the model

• api_key (str, optional): API key for backend service authentication

Example:

# With model ID and API key (loads schema from API)
manager = ModelSchemaManager(model_id="production_model", api_key="abc123")

# Without API key (for local-only operations)
manager = ModelSchemaManager()

---

Schema Loading and Saving

fetch_schema_from_api(model_id)

Fetch a schema from the backend service via an API call.

Parameters:

• model_id (str): Model ID for which the schema needs to be fetched

Returns: Loaded schema dictionary or empty dictionary if request fails

Example:

manager = ModelSchemaManager(api_key="secret_key")
schema = manager.fetch_schema_from_api("production_model")
print(schema)

---

upload_schema_to_api(model_id, schema)

Upload a schema to the backend service via an API call.

Parameters:

• model_id (str): Model ID for which the schema should be uploaded

• schema (dict): Schema dictionary to be uploaded

Returns: Response dictionary from API or empty dictionary if request fails

Example:

manager = ModelSchemaManager(api_key="secret_key")
schema = {
    "model_id": "my_model",
    "fields": {
        "age": {"type": "int", "min": 0, "max": 120},
        "income": {"type": "float", "min": 0.0}
    }
}
response = manager.upload_schema_to_api("my_model", schema)

---

load_schema_from_file(file_path)

Load a schema from a JSON file.

Parameters:

• file_path (str): Path to the JSON file containing the schema

Returns: Loaded schema dictionary or empty dictionary if file cannot be read

Example:

manager = ModelSchemaManager()
schema = manager.load_schema_from_file('/path/to/schema.json')
manager.schema = schema

---

save_schema_to_file(file_path)

Save the current schema to a JSON file.

Parameters:

• file_path (str): Path to the JSON file where the schema should be saved

Example:

manager = ModelSchemaManager()
manager.schema = {"fields": {"age": {"type": "int"}}}
manager.save_schema_to_file('/path/to/schema.json')

---

Schema Generation

generate_schema_from_dataframe(model, df, allowed_categories=None, always_allow_categories=False)

Generate a comprehensive schema based on model features and DataFrame.

Parameters:

• model (Any): The machine learning model or pipeline

• df (DataFrame): The DataFrame used for training the model

• allowed_categories (List[str], optional): List of features where new categories are allowed

• always_allow_categories (bool): If True, allow new categories in all categorical features

Returns: Generated model schema with comprehensive field definitions

Example:

from sklearn.ensemble import RandomForestClassifier
import pandas as pd

# Create sample data
df = pd.DataFrame({
    'age': [25, 30, 35, 40],
    'income': [50000.0, 60000.0, 55000.0, 70000.0],
    'region': ['west', 'east', 'west', 'south'],
    'status': [1, 2, 1, 2]
})

# Train model
X = df[['age', 'income', 'region', 'status']]
y = [0, 1, 0, 1]
model = RandomForestClassifier()
model.fit(X, y)

# Generate schema
manager = ModelSchemaManager()
schema = manager.generate_schema_from_dataframe(
    model,
    df,
    allowed_categories=['region']
)

# Access schema details
print(schema['fields']['age'])

# {'type': 'int', 'min': 25, 'max': 40}

print(schema['fields']['region'])

# {'type': 'category', 'choices': ['west', 'east', 'south'],
#  'allow_new_categories': True, 'n_unique': 3, 'missing_count': 0}

---

Schema Modification

update_column_categories(column_name, categories=None, allow_new_categories=None)

Update the possible values and handling of new categories for a categorical column.

Parameters:

• column_name (str): The name of the column to update

• categories (List, optional): A list of new categories

• allow_new_categories (bool, optional): Whether to allow new categories

Example:

manager.update_column_categories(
    'status',
    categories=['active', 'inactive', 'pending']
)

manager.update_column_categories(
    'status',
    allow_new_categories=True
)

---

update_column_range(column_name, min_value, max_value)

Update the value range for a specific numeric column.

Parameters:

• column_name (str): The name of the column to update

• min_value (int or float): The new minimum allowed value

• max_value (int or float): The new maximum allowed value

Example:

manager.update_column_range('age', 0, 120)

---

update_column_data_type(column_name, data_type)

Update the data type of a column in the schema.

Parameters:

• column_name (str): The name of the column to update

• data_type (str): The new data type ('int', 'float', 'str', 'category')

Example:

manager.update_column_data_type('user_id', 'str')

---

Validation

get_validation_rules()

Extract validation rules from the loaded schema.

Returns: Dictionary of validation rules with validator functions and error messages

Example:

manager = ModelSchemaManager()
manager.schema = {
    'fields': {
        'age': {'type': 'int'},
        'name': {'type': 'str', 'allow_new_categories': True},
        'status': {'type': 'str', 'choices': ['active', 'inactive']}
    }
}

rules = manager.get_validation_rules()

# Validate integer field
print(rules['age']['validator'](25))  # True
print(rules['age']['validator']("25"))  # False

# Validate categorical field with choices
print(rules['status']['validator']('active'))  # True
print(rules['status']['validator']('pending'))  # False

---

handle_missing_values(df, missing_value_rules=None)

Handle missing values in the DataFrame according to custom or default rules.

Parameters:

• df (DataFrame): The DataFrame to process

• missing_value_rules (dict, optional): Dictionary specifying how to handle missing values

Returns: DataFrame with missing values handled

Example:

import pandas as pd
import numpy as np

manager = ModelSchemaManager()
df = pd.DataFrame({
    'age': [25, np.nan, 35],
    'income': [50000, 60000, np.nan],
    'name': ['Alice', np.nan, 'Charlie']
})

# Default handling - fill all NaN with "NaN" string
df_filled = manager.handle_missing_values(df.copy())
print(df_filled['age'].tolist())  # [25, 'NaN', 35]

# Custom rules
rules = {
    'age': 0,  # Fill age NaN with 0
    'income': lambda x: 50000,  # Fill income NaN with 50000
    'name': 'Unknown'  # Fill name NaN with 'Unknown'
}
df_custom = manager.handle_missing_values(df.copy(), rules)
print(df_custom['age'].tolist())  # [25, 0, 35]
print(df_custom['name'].tolist())  # ['Alice', 'Unknown', 'Charlie']

---

Complete Schema Example

from viewai_client.models.schema import ModelSchemaManager
import pandas as pd
from sklearn.ensemble import RandomForestClassifier

# Prepare data
df = pd.DataFrame({
    'age': [25, 30, 35, 40, 45],
    'income': [50000.0, 60000.0, 55000.0, 70000.0, 65000.0],
    'region': ['west', 'east', 'west', 'south', 'north'],
    'target': [0, 1, 0, 1, 0]
})

# Train model
X = df.drop('target', axis=1)
y = df['target']
model = RandomForestClassifier()
model.fit(X, y)

# Create schema manager and generate schema
manager = ModelSchemaManager()
schema = manager.generate_schema_from_dataframe(
    model=model,
    df=df,
    allowed_categories=['region']
)

# Customize schema
manager.update_column_range('age', 18, 100)
manager.update_column_range('income', 0.0, 1000000.0)

# Save schema
manager.save_schema_to_file('model_schema.json')

# Validate data
rules = manager.get_validation_rules()
test_data = {'age': 30, 'income': 60000.0, 'region': 'east'}

for field, rule in rules.items():
    if field in test_data:
        if not rule['validator'](test_data[field]):
            print(f"Validation failed: {rule['error']}")

---

See Also

• ViewAIClient - Main client class

• Service Classes - Detailed service class documentation

• Configuration - Configuration options

• Exceptions - Exception handling