This script trains a model using the CIFAR10 dataset. During each epoch, it calculates the average training loss and checks if it exceeds a predefined threshold. If the threshold is exceeded, an alert is sent to a Microsoft Teams channel using a Microsoft Teams card created with the pymsteams
library.
import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from pymsteams import connectorcard
# Microsoft Teams webhook URL
teams_webhook_url = 'https://your_teams_webhook_url'
# Define the model architecture
class CustomModel(nn.Module):
def __init__(self):
super(CustomModel, self).__init__()
# Your model architecture goes here
def forward(self, x):
# Your forward pass logic goes here
return x
# Main script
if __name__ == '__main__':
# Data augmentation and normalization
transform = transforms.Compose([
transforms.RandomHorizontalFlip(),
transforms.RandomCrop(32, padding=4),
transforms.ToTensor(),
transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5])
])
# Load CIFAR10 dataset
train_dataset = datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
test_dataset = datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)
# Data loaders
train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=128, shuffle=True, num_workers=2)
test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=128, shuffle=False, num_workers=2)
# Model initialization
model = CustomModel()
# Loss function and optimizer
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.001, momentum=0.9)
# Training loop
num_epochs = 10
loss_threshold = 1.0
for epoch in range(num_epochs):
# Training
model.train()
train_loss = 0.0
train_correct = 0
train_total = 0
for inputs, labels in train_loader:
optimizer.zero_grad()
# Forward pass
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
# Update training metrics
train_loss += loss.item()
_, predicted = torch.max(outputs.data, 1)
train_total += labels.size(0)
train_correct += (predicted == labels).sum().item()
# Calculate average training loss
train_loss /= len(train_loader)
# Check if the loss exceeds the threshold
if train_loss > loss_threshold:
# Send alert to Microsoft Teams channel
message = f"Training loss exceeded the threshold: {train_loss:.4f}"
# Create a Microsoft Teams card
card = connectorcard.ConnectorCard(webhookurl=teams_webhook_url)
card.title('Training Alert')
card.text(message)
card.send()
print(message)
print(f"Epoch [{epoch+1}/{num_epochs}], Train Loss: {train_loss:.4f}, Train Accuracy: {train_accuracy:.2f}%")