HomeDocumentation
Python SDK
The Telesoft Healthcare AI Python SDK makes it easy to integrate our medical AI capabilities into your Python applications. This SDK simplifies authentication, request handling, and response parsing so you can focus on building your healthcare solutions.
Installation
Install the SDK using pip:
pip install telesoft-healthcare-aiThe SDK requires Python 3.7 or later.
Authentication
Initialize the SDK with your API key:
import os
from telesoft import TelesoftAI
# Initialize the client with your API key
telesoft = TelesoftAI(api_key="YOUR_API_KEY")
# Or use an environment variable (recommended)
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))⚠️ API Key Security
Never hardcode your API key directly in your source code. Use environment variables or a secure configuration management system.
Basic Usage
Here's a simple example of using the SDK to analyze patient symptoms:
import os
from telesoft import TelesoftAI
# Initialize the client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
# Analyze patient symptoms
def analyze_patient_symptoms():
analysis = telesoft.diagnostics.analyze(
patient_data={
"age": 45,
"sex": "female",
"symptoms": ["cough", "fever", "shortness of breath"],
"duration": "5 days",
"medical_history": ["hypertension", "type 2 diabetes"],
"medications": ["lisinopril", "metformin"]
},
options={
"include_confidence_scores": True,
"include_differential_diagnosis": True,
"include_recommendations": True,
"include_evidence": True
}
)
# Access the results
print(f"Primary diagnosis: {analysis.primary_diagnosis.condition}")
print(f"Confidence: {analysis.primary_diagnosis.confidence_score:.2f}")
print("
Differential diagnoses:")
for diagnosis in analysis.differential_diagnoses:
print(f"- {diagnosis.condition} ({diagnosis.confidence_score:.2f})")
print("
Recommended tests:")
for test in analysis.recommendations.diagnostic_tests:
print(f"- {test.name}: {test.rationale}")
return analysis
# Run the analysis
if __name__ == "__main__":
try:
analyze_patient_symptoms()
except Exception as e:
print(f"Error: {e}")Advanced Features
Analyzing Medical Images
Use the SDK to analyze medical images:
import os
from telesoft import TelesoftAI
# Initialize the client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
# Analyze a chest X-ray image
def analyze_chest_xray(image_path):
with open(image_path, "rb") as image_file:
analysis = telesoft.imaging.analyze(
imaging_type="chest-xray",
image=image_file,
options={
"include_annotations": True,
"include_heatmap": True,
"confidence_threshold": 0.5
}
)
print(f"Found {len(analysis.findings)} findings:")
for finding in analysis.findings:
print(f"- {finding.description} (confidence: {finding.confidence:.2f})")
print(f" Location: {finding.location}")
# Save the annotated image if provided
if analysis.annotated_image:
with open("annotated_xray.png", "wb") as f:
f.write(analysis.annotated_image)
print("Annotated image saved as annotated_xray.png")
return analysis
# Example usage
if __name__ == "__main__":
analyze_chest_xray("path/to/chest_xray.dcm")Using with Async/Await
The SDK provides async versions of all methods for use with asyncio:
import os
import asyncio
from telesoft import TelesoftAI
# Initialize the client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
# Async function to analyze multiple patients concurrently
async def analyze_multiple_patients(patient_data_list):
# Create a list of tasks
tasks = [
telesoft.diagnostics.analyze_async(
patient_data=patient_data,
options={"include_differential_diagnosis": True}
)
for patient_data in patient_data_list
]
# Execute all tasks concurrently
results = await asyncio.gather(*tasks)
# Process results
for i, analysis in enumerate(results):
print(f"Patient {i+1}:")
print(f"Primary diagnosis: {analysis.primary_diagnosis.condition}")
print(f"Confidence: {analysis.primary_diagnosis.confidence_score:.2f}")
print("")
return results
# Example usage
async def main():
# Sample patient data
patients = [
{
"age": 45,
"sex": "female",
"symptoms": ["cough", "fever", "fatigue"]
},
{
"age": 60,
"sex": "male",
"symptoms": ["chest pain", "shortness of breath"]
},
{
"age": 7,
"sex": "male",
"symptoms": ["sore throat", "rash", "fever"]
}
]
await analyze_multiple_patients(patients)
# Run the async function
if __name__ == "__main__":
asyncio.run(main())Custom Configuration
Configure the SDK with advanced options:
import os
from telesoft import TelesoftAI, RetryConfig
# Create a custom retry configuration
retry_config = RetryConfig(
max_retries=3, # Maximum number of retries
initial_delay_ms=500, # Initial retry delay in milliseconds
max_delay_ms=5000, # Maximum retry delay
backoff_factor=2.0, # Exponential backoff multiplier
retry_status_codes=[429, 503] # Status codes to retry on
)
# Initialize the client with custom configuration
telesoft = TelesoftAI(
api_key=os.environ.get("TELESOFT_API_KEY"),
base_url="https://api.telesoft.us/v1", # Custom API endpoint
timeout=30.0, # Timeout in seconds
retry_config=retry_config, # Custom retry behavior
debug=True # Enable debug logging
)Using with Machine Learning Frameworks
Integration with Pandas
Process patient data from DataFrames:
import os
import pandas as pd
from telesoft import TelesoftAI
# Initialize the client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
# Load patient data from a CSV file
df = pd.read_csv("patient_data.csv")
# Process each patient record
def process_patient_records(df):
results = []
for index, row in df.iterrows():
# Convert row to patient data format
patient_data = {
"age": int(row["age"]),
"sex": row["sex"],
"symptoms": row["symptoms"].split(","),
"medical_history": row.get("medical_history", "").split(",") if not pd.isna(row.get("medical_history", "")) else []
}
# Analyze patient data
try:
analysis = telesoft.diagnostics.analyze(
patient_data=patient_data,
options={"include_differential_diagnosis": True}
)
# Append results
results.append({
"patient_id": row.get("id", index),
"primary_diagnosis": analysis.primary_diagnosis.condition,
"confidence": analysis.primary_diagnosis.confidence_score,
"differential_diagnoses": [d.condition for d in analysis.differential_diagnoses[:3]]
})
except Exception as e:
print(f"Error processing patient {row.get('id', index)}: {e}")
results.append({
"patient_id": row.get("id", index),
"error": str(e)
})
# Convert results to DataFrame
results_df = pd.DataFrame(results)
return results_df
# Process patient records and save results
results_df = process_patient_records(df)
results_df.to_csv("analysis_results.csv", index=False)
print(f"Processed {len(results_df)} patient records.")Integration with PyTorch
Use Telesoft's analysis as part of a PyTorch pipeline:
import os
import torch
import torch.nn as nn
import numpy as np
from telesoft import TelesoftAI
# Initialize Telesoft client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
# Define a simple PyTorch model that uses Telesoft API for feature enhancement
class EnhancedDiagnosisModel(nn.Module):
def __init__(self, input_size, hidden_size, output_size):
super(EnhancedDiagnosisModel, self).__init__()
self.model = nn.Sequential(
nn.Linear(input_size, hidden_size),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(hidden_size, output_size),
nn.Softmax(dim=1)
)
def forward(self, x):
return self.model(x)
def preprocess_with_telesoft(self, patient_data_list):
"""Enhance input features using Telesoft API"""
enhanced_features = []
for patient_data in patient_data_list:
# Get analysis from Telesoft API
analysis = telesoft.diagnostics.analyze(
patient_data=patient_data,
options={"include_evidence": True}
)
# Extract features from Telesoft analysis
telesoft_features = self._extract_features_from_analysis(analysis)
# Combine with original features
original_features = self._encode_patient_data(patient_data)
combined_features = np.concatenate([original_features, telesoft_features])
enhanced_features.append(combined_features)
# Convert to PyTorch tensor
return torch.tensor(np.array(enhanced_features), dtype=torch.float32)
def _extract_features_from_analysis(self, analysis):
# Extract useful features from Telesoft analysis
features = []
# Add primary diagnosis confidence
features.append(analysis.primary_diagnosis.confidence_score)
# Add top 3 differential diagnosis confidences
diff_scores = [d.confidence_score for d in analysis.differential_diagnoses[:3]]
# Pad with zeros if fewer than 3
diff_scores.extend([0.0] * (3 - len(diff_scores)))
features.extend(diff_scores)
# Add number of evidence factors
features.append(len(analysis.primary_diagnosis.evidence) / 10.0) # Normalize
return np.array(features)
def _encode_patient_data(self, patient_data):
# Convert patient data to feature vector (simplified example)
features = []
# Age (normalized)
features.append(patient_data.get("age", 0) / 100.0)
# Sex (binary encoding)
features.append(1.0 if patient_data.get("sex") == "male" else 0.0)
# Number of symptoms (normalized)
features.append(len(patient_data.get("symptoms", [])) / 10.0)
# Number of medical history items (normalized)
features.append(len(patient_data.get("medical_history", [])) / 10.0)
return np.array(features)
# Example usage
model = EnhancedDiagnosisModel(input_size=8, hidden_size=16, output_size=5)
patient_data = [
{
"age": 45,
"sex": "female",
"symptoms": ["cough", "fever", "fatigue"],
"medical_history": ["hypertension"]
},
# More patient records...
]
# Preprocess with Telesoft
enhanced_input = model.preprocess_with_telesoft(patient_data)
# Make prediction
with torch.no_grad():
predictions = model(enhanced_input)
print(predictions)Handling Errors
The SDK provides structured error handling:
import os
from telesoft import TelesoftAI
from telesoft.exceptions import (
TelesoftError,
AuthenticationError,
RateLimitError,
ValidationError,
ServiceError
)
# Initialize the client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
def safe_analyze(patient_data):
try:
analysis = telesoft.diagnostics.analyze(
patient_data=patient_data,
options={"include_differential_diagnosis": True}
)
return analysis
except AuthenticationError as e:
# Handle authentication issues
print(f"Authentication error: {e}")
# Check API key validity
except RateLimitError as e:
# Handle rate limiting
print(f"Rate limit exceeded. Retry after {e.retry_after} seconds")
# Implement retry with exponential backoff
except ValidationError as e:
# Handle input validation errors
print(f"Validation error: {e}")
for error in e.errors:
print(f"- {error.field}: {error.message}")
# Fix input data issues
except ServiceError as e:
# Handle server-side errors
print(f"Service error: {e}. Status code: {e.status_code}")
# Possibly retry or report to Telesoft support
except TelesoftError as e:
# Handle any other Telesoft API errors
print(f"API error: {e}")
except Exception as e:
# Handle unexpected errors
print(f"Unexpected error: {e}")
return None
# Example usage
patient_data = {
"age": 45,
"sex": "female",
"symptoms": ["cough", "fever", "fatigue"]
}
analysis = safe_analyze(patient_data)
if analysis:
print(f"Diagnosis: {analysis.primary_diagnosis.condition}")ℹ️ Error Handling Best Practices
For production applications, implement exponential backoff with jitter for retries on rate limit errors and temporary service issues. Consider using a circuit breaker pattern for handling persistent service outages.
Using with Flask/Django
Here's an example of integrating the SDK with Flask:
import os
from flask import Flask, request, jsonify
from telesoft import TelesoftAI
from telesoft.exceptions import TelesoftError
app = Flask(__name__)
# Initialize the Telesoft client
telesoft = TelesoftAI(api_key=os.environ.get("TELESOFT_API_KEY"))
@app.route('/api/analyze', methods=['POST'])
def analyze_symptoms():
try:
# Get patient data from request
patient_data = request.json.get('patient_data')
if not patient_data:
return jsonify({"error": "Missing patient_data"}), 400
# Validate required fields
if not all(k in patient_data for k in ["age", "sex", "symptoms"]):
return jsonify({"error": "Missing required fields in patient_data"}), 400
# Call Telesoft API
analysis = telesoft.diagnostics.analyze(
patient_data=patient_data,
options={
"include_confidence_scores": True,
"include_differential_diagnosis": True,
"include_recommendations": True
}
)
# Convert to serializable format
result = {
"primary_diagnosis": {
"condition": analysis.primary_diagnosis.condition,
"confidence_score": analysis.primary_diagnosis.confidence_score
},
"differential_diagnoses": [
{
"condition": d.condition,
"confidence_score": d.confidence_score
}
for d in analysis.differential_diagnoses
],
"recommendations": {
"diagnostic_tests": [
{
"name": t.name,
"rationale": t.rationale
}
for t in analysis.recommendations.diagnostic_tests
]
}
}
return jsonify(result)
except TelesoftError as e:
# Handle Telesoft API errors
return jsonify({"error": str(e)}), 400
except Exception as e:
# Handle unexpected errors
app.logger.error(f"Unexpected error: {e}")
return jsonify({"error": "Internal server error"}), 500
@app.route('/api/analyze-image', methods=['POST'])
def analyze_medical_image():
try:
# Check if the post request has the file part
if 'image' not in request.files:
return jsonify({"error": "No image provided"}), 400
image_file = request.files['image']
imaging_type = request.form.get('imaging_type', 'chest-xray')
# Call Telesoft API
analysis = telesoft.imaging.analyze(
imaging_type=imaging_type,
image=image_file,
options={
"include_annotations": True,
"confidence_threshold": 0.5
}
)
# Convert to serializable format
result = {
"findings": [
{
"description": f.description,
"confidence": f.confidence,
"location": f.location
}
for f in analysis.findings
]
}
return jsonify(result)
except TelesoftError as e:
# Handle Telesoft API errors
return jsonify({"error": str(e)}), 400
except Exception as e:
# Handle unexpected errors
app.logger.error(f"Unexpected error: {e}")
return jsonify({"error": "Internal server error"}), 500
if __name__ == '__main__':
app.run(debug=True)