Qubicon SDK

Overview

The Qubicon SDK is a Python client library designed for interacting with the Qubicon API. It provides methods for authentication, process management, model creation, data extraction, and more.

This SDK wraps the OpenAPI-generated client and adds custom functionalities to streamline interactions with the Qubicon platform.

Usage Modes

The Qubicon SDK can be used in two ways:

1. As a Python Library – Import and use the SDK programmatically in your Python applications.
2. As an Interactive CLI – Run an interactive command-line interface to execute commands without writing code.

To start the interactive menu, run:

qubicon-client --interactive

All function outputs are returned in JSON format, making it easy to parse and integrate with other tools.

Features

• Authentication & Token Management
• Process Management (Create, List, Delete Processes & Groups)
• Model Management (Create, Fetch, Export Computable Models)
• Physical Quantities Handling
• Process Data Extraction & Export
• File Upload & Advanced Model Management
• Event Streaming

Installation

Install the SDK using pip:

pip install qubicon

Alternatively, install from a local package:

pip install /path/to/qubicon_package

Quick Start

1. Initialize the SDK

from qubicon.core import QubiconCore, AuthenticatedClient

# Define base URL (e.g., production or staging server)
BASE_URL = "https://qubicon-hostname"

# Create a client instance
client = AuthenticatedClient(base_url=BASE_URL, token=None)

# Initialize QubiconCore with the authenticated client
qubicon = QubiconCore(client)

2. Login to Qubicon

username = "your_username"
password = "your_password"
groupid = 1234
token = qubicon.login_user(username, password, group_id=groupid)
print("Token:", token)

3. List Processes

processes = qubicon.list_processes()
print(processes)

4. Create a Process

new_process = qubicon.create_process(
    name="Test Process",
    recipe_id=1822,
    description="Automated process creation"
)
print(new_process)

5. Delete a Process

response = qubicon.delete_process(process_id=1234)
print(response)

API Reference

Authentication

def login_user(username: str, password: str) -> Optional[str]

• Logs in a user and retrieves an authentication token.

def refresh_token(refresh_token: str) -> Optional[str]

• Refreshes an expired authentication token.

Process Management

def list_processes()

• Lists all processes with optional filters.

def create_process(name: str, recipe_id: int, description: str = "") -> Optional[Dict[str, Any]]

• Creates a new process using a specified recipe.

def start_process(process_id: int) -> bool

• Start a process by ID.

def delete_process(process_id: int) -> Optional[Dict[str, Any]]

• Deletes a specific process by ID.

def delete_multiple_processes(process_ids: List[int]) -> Dict[int, Optional[Dict[str, Any]]]

• Deletes multiple processes in a batch operation.

def create_process_group(name: str) -> Optional[Dict[str, Any]]

• Creates a new process group.

def add_processes_to_group(group_id: int, process_ids: List[int]) -> Optional[Dict[str, Any]]

• Adds multiple processes to an existing group.

def start_processgroup(group_id: int) -> bool

• Start a process group.

def stop_processgroup(group_id: int) -> bool

• Stop a process group.

def get_process_state(process_id: int) -> Dict[str, Any]

• Fetch the current state for a specific process ID.

def get_active_processes_list_state(
        statuses: List[str] = ["READY", "RUNNING", "WARMING_UP"],
    ) -> List[Dict[str, Any]]

• Fetch status data for processes.

Computable Models

def list_computable_models()

• Lists all available computable models.

def fetch_model_details(model_id: int)

• Fetches detailed information about a computable model.

def create_computable_model(model_data: dict) -> Optional[Dict[str, Any]]

• Creates a new computable model with the given specifications.

def export_model_to_json(model_id: int, filename: str)

• Exports a computable model into an importable JSON format.

Physical Quantities

def list_physical_quantities()

• Retrieves all available physical quantities.

def create_physical_quantity(pq_dict: Dict[str, Any]) -> Optional[Dict[str, Any]]

• Creates a new physical quantity in the system.

def check_existing_physical_quantity(pq_name: str) -> Optional[dict]

• Checks if a physical quantity already exists.

Process Data & Channels

def extract_channels(process_id: int)

• Extracts available channels for a given process.

def extract_process_data(process_id: int, output_format: str = "csv", output_file: str = "process_data.csv")

• Extracts and exports process data in a file.

def extract_offline_channels(process_id: int)

• Extracts available offline channels for a given process.

def extract_offline_process_data(process_id: int, selected_channels: List[Dict[str, Any]], start_date: [int], end_date: [int])

• Extracts and exports offline process data for selected channels and a given period.

def get_process_samplings(self, process_id: int) -> List[Dict[str, Any]]

• Fetch detailed sampling data for a given process ID.

def discard_sampling_data(self, external_sampling_id: str, process_id: int) -> bool

• Discard a specific external sampling for a given process.

def get_offline_equipment_data(equipment_id: int, process_id: int,)

• Get the list of offline data for an offline equipment and a process.

def get_process_tag_values(process_id: int) -> List[Dict[str, Any]]

• Fetches process tag values and extracts TagDto with corresponding values.

def set_tag_value(self, process_id: int, tag_value_id: int, new_value: Any) -> Optional[Dict[str, Any]]:

• Updates a tag value for a specific process and tag value ID.

File Upload & Advanced Models

def upload_file(file_path: str) -> Optional[Dict[str, Any]]

• Uploads an external file (e.g., ZIP archive) to Qubicon.

def create_advanced_model(model_data: dict, file_path: str) -> Optional[Dict[str, Any]]

• Creates an advanced model that references an external file.

Sampling

def generate_and_upload_sampling_plan(process_group_id: int, number_of_samples: int, sample_prefix: str = "Day")

• Generates and uploads a sampling plan to the specified process group ID.

def generate_and_upload_sampling_plan_single_process(process_id: int, number_of_samples: int, sample_prefix: str = "Day")

• Generates and uploads a sampling plan to the specified process ID.

Event Streaming

def stream_events(event_types: List[str], nonce: str)

• Streams real-time events from the API.