| --- |
| license: gpl-3.0 |
| pipeline_tag: graph-ml |
| tags: |
| - code |
| --- |
| --- |
|
|
| import contextlib |
| import os |
| from matplotlib import pyplot as plt |
| import numpy as np |
| import torch |
| import torch.nn as nn |
| import torch.optim as optim |
| import requests |
| from torchvision import datasets, transforms |
| import psutil |
| import time |
| import subprocess |
| import onnxruntime as ort |
| import matplotlib.pyplot as plt |
| import numpy as np |
| import numexpr as ne |
| |
| from transformers import AutoTokenizer, AutoModelForSeq2SeqLM |
|
|
| tokenizer = AutoTokenizer.from_pretrained("janpase97/codeformer-pretrained") |
| |
| model = AutoModelForSeq2SeqLM.from_pretrained("janpase97/codeformer-pretrained") |
|
|
| def check_graphics_api(target_app_name): |
| graphics_api = None |
| |
| with contextlib.suppress(subprocess.CalledProcessError): |
| output = subprocess.check_output(['tasklist', '/FI', f'imagename eq {target_app_name}', '/M']).decode('utf-8') |
| if "opengl32.dll" in output: |
| graphics_api = "OpenGL" |
| elif "d3d11.dll" in output: |
| graphics_api = "DirectX11" |
| elif "d3d12.dll" in output: |
| graphics_api = "DirectX12" |
| elif "vulkan" in output: |
| graphics_api = "VULKAN" |
| return graphics_api |
| |
|
|
| # Get the target application's process object |
| def get_target_app_process(target_app_name): |
| return next( |
| ( |
| process |
| for process in psutil.process_iter(['name']) |
| if process.info['name'] == target_app_name |
| ), |
| None, |
| ) |
| |
| # Attach the AI to the application's process by PID |
| def attach_ai_to_app_pid(target_app_process): |
| if target_app_process is not None: |
| print(f"AI is attached to the application's process with PID: {target_app_process.pid}") |
| return True |
| else: |
| print("Could not find the target application's process to attach the AI.") |
| return False |
| |
| # Check if the targeted application is running |
| def is_target_app_running(target_app_name): |
| return any( |
| process.info['name'] == target_app_name |
| for process in psutil.process_iter(['name']) |
| ) |
| |
| # Create the directory if it doesn't exist |
| directory = r"G:\Epic Games\GTAV\GTA5_AI\trained_models" |
| if not os.path.exists(directory): |
| os.makedirs(directory) |
| |
| # Define the neural network model |
| class NanoCircuit(nn.Module): |
| def __init__(self): |
| super(NanoCircuit, self).__init__() |
| self.fc1 = nn.Linear(784, 128) |
| self.fc2 = nn.Linear(128, 10) |
| |
| def forward(self, x): |
| x = x.view(-1, 784) # Reshape the input from (batch_size, 28, 28) to (batch_size, 784) |
| x = torch.relu(self.fc1(x)) |
| x = self.fc2(x) |
| return x |
| |
| # Set the device to GPU if available |
| device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") |
| |
| # Load the MNIST dataset |
| transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]) |
| train_dataset = datasets.MNIST(root='./data', train=True, download=True, transform=transform) |
| train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=64, shuffle=True) |
| |
| |
| # Initialize the model and move it to the GPU |
| model = NanoCircuit().to(device) |
| criterion = nn.CrossEntropyLoss() |
| optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9) |
| |
| # Train the model on the GPU with a data cap |
| def train_with_data_cap(model, data_loader, criterion, optimizer, device, data_cap_gb): |
| data_processed = 0 |
| data_cap_bytes = data_cap_gb * (1024 ** 3) |
| epoch = 0 |
| |
| while data_processed < data_cap_bytes: |
| running_loss = 0.0 |
| for i, data in enumerate(data_loader, 0): |
| inputs, labels = data |
| inputs, labels = inputs.to(device), labels.to(device) |
| |
| # Update the amount of data processed |
| data_processed += inputs.nelement() * inputs.element_size() |
| if data_processed >= data_cap_bytes: |
| break |
| |
| optimizer.zero_grad() |
| |
| outputs = model(inputs.view(-1, 28 * 28)) |
| loss = criterion(outputs, labels) |
| loss.backward() |
| optimizer.step() |
| |
| running_loss += loss.item() |
| |
| epoch += 1 |
| print(f"Epoch {epoch}, Loss: {running_loss / (i + 1)}") |
| print(f"Data processed: {data_processed / (1024 ** 3):.2f} GB") |
| |
| return model |
| |
|
|
| # Save the updated model as a .onnx file |
| def save_model(model, filepath): |
| dummy_input = torch.randn(1, 1, 28, 28).to(device) |
| torch.onnx.export(model, dummy_input, filepath, input_names=['input'], output_names=['output'], opset_version=11) |
| |
|
|
| # Train the model with a 1 GB data cap |
| trained_model = train_with_data_cap(model, train_loader, criterion, optimizer, device, data_cap_gb=50) |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
|
|
| target_app_name = "GTA5_TRAINED.exe" |
| save_interval_seconds = 5 * 60 |
| application_was_running = False |
| while True: |
| if is_target_app_running(target_app_name): |
| print("Target application is running. Training and updating the model...") |
| trained_model = train_with_data_cap(model, train_loader, criterion, optimizer, device, data_cap_gb=.1) |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| application_was_running = True |
| elif application_was_running: |
| print("Target application has exited. Saving the model...") |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| print("Finished training and saved the model.") |
| break |
| else: |
| print("Target application is not running. Waiting to start training and updating the model...") |
| |
| time.sleep(save_interval_seconds) |
| |
| def train_with_data_cap(model, data_loader, criterion, optimizer, device, data_cap_gb): |
| data_processed = 0 |
| data_cap_bytes = data_cap_gb * (1024 ** 3) |
| epoch = 0 |
| |
| while data_processed < data_cap_bytes: |
| running_loss = 0.0 |
| for i, data in enumerate(data_loader, 0): |
| inputs, labels = data |
| inputs, labels = inputs.to(device), labels.to(device) |
| |
| # Update the amount of data processed |
| data_processed += inputs.nelement() * inputs.element_size() |
| if data_processed >= data_cap_bytes: |
| break |
| |
| optimizer.zero_grad() |
| |
| # Compute the outputs and loss using numexpr |
| outputs = model(inputs.view(-1, 28 * 28)) |
| outputs = outputs.cpu().detach().numpy() |
| labels = labels.cpu().detach().numpy() |
| loss = ne.evaluate("sum(-log(outputs[arange(outputs.shape[0]), labels]))") / len(labels) |
| |
| # Backpropagate and update the model parameters |
| ne.evaluate("loss", out=loss) |
| grad_outputs = np.ones_like(outputs) |
| grad_outputs[np.arange(grad_outputs.shape[0]), labels] = -1 |
| grad_outputs /= len(labels) |
| grad_outputs = ne.evaluate("grad_outputs * loss_grad") |
| grad_outputs = torch.from_numpy(grad_outputs).to(device) |
| outputs = torch.from_numpy(outputs).to(device) |
| loss.backward(grad_outputs) |
| optimizer.step() |
| |
| running_loss += loss.item() |
| |
| epoch += 1 |
| print(f"Epoch {epoch}, Loss: {running_loss / (i + 1)}") |
| print(f"Data processed: {data_processed / (1024 ** 3):.2f} GB") |
| |
| return model |
| |
| # Train the model with a 10 GB data cap |
| trained_model = train_with_data_cap(model, train_loader, criterion, optimizer, os.device_encoding, data_cap_gb=10) |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| |
| target_app_name = "GTA5.exe" |
| save_interval_seconds = 5 * 60 |
| application_was_running = False |
| while True: |
| if is_target_app_running(target_app_name): |
| print("Target application is running. Training and updating the model...") |
| trained_model = train_with_data_cap(model, train_loader, criterion, optimizer, os.device_encoding, data_cap_gb=10) |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| application_was_running = True |
| elif application_was_running: |
| print("Target application has exited. Saving the model...") |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| print("Finished training and saved the model.") |
| break |
| else: |
| print("Target application is not running. Waiting to start training and updating the model...") |
| |
| time.sleep(save_interval_seconds) |
| |
| def train_with_data_cap(model, data_loader, criterion, optimizer, device, data_cap_gb): |
| data_processed = 0 |
| data_cap_bytes = data_cap_gb * (1024 ** 3) |
| epoch = 0 |
| |
| while data_processed < data_cap_bytes: |
| running_loss = 0.0 |
| for i, data in enumerate(data_loader, 0): |
| inputs, labels = data |
| inputs, labels = inputs.to(device), labels.to(device) |
| |
| # Update the amount of data processed |
| data_processed += inputs.nelement() * inputs.element_size() |
| if data_processed >= data_cap_bytes: |
| break |
| |
| optimizer.zero_grad() |
| |
| # Compute the outputs and loss using numexpr |
| outputs = model(inputs.view(-1, 28 * 28)) |
| outputs = outputs.cpu().detach().numpy() |
| labels = labels.cpu().detach().numpy() |
| loss = ne.evaluate("sum(-log(outputs[arange(outputs.shape[0]), labels]))") / len(labels) |
| |
| # Backpropagate and update the model parameters |
| ne.evaluate("loss", out=loss) |
| grad_outputs = np.ones_like(outputs) |
| grad_outputs[np.arange(grad_outputs.shape[0]), labels] = -1 |
| grad_outputs /= len(labels) |
| grad_outputs = ne.evaluate("grad_outputs * loss_grad") |
| grad_outputs = torch.from_numpy(grad_outputs).to(device) |
| outputs = torch.from_numpy(outputs).to(device) |
| loss.backward(grad_outputs) |
| optimizer.step() |
| |
| running_loss += loss.item() |
| |
| epoch += 1 |
| print(f"Epoch {epoch}, Loss: {running_loss / (i + 1)}") |
| print(f"Data processed: {data_processed / (1024 ** 3):.2f} GB") |
| |
| return model |
| |
| target_app_name = "GTA5.exe" |
| save_interval_seconds = 1 * 60 |
| application_was_running = False |
|
|
| while True: |
| if is_target_app_running(target_app_name): |
| print("Target application is running. Training and updating the model...") |
| trained_model = train_with_data_cap(model, train_loader, criterion, optimizer, device, data_cap_gb=10) |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| application_was_running = True |
| elif application_was_running: |
| print("Target application has exited. Saving the model...") |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| print("Finished training and saved the model.") |
| break |
| else: |
| start_time = time.time() |
| print("Target application is not running. Waiting to detect the graphics API...") |
| while (time.time() - start_time) < 5: |
| if is_target_app_running(target_app_name): |
| if graphics_api := check_graphics_api(target_app_name): |
| print(f"Detected {graphics_api} in the target application.") |
| break |
| else: |
| print("Could not detect the graphics API used in the target application.") |
| time.sleep(1) |
| |
| if not is_target_app_running(target_app_name): |
| print("Target application not detected in 5 seconds. Shutting down the AI.") |
| break |
| |
| |
| while True: |
| if is_target_app_running(target_app_name): |
| if graphics_api := check_graphics_api(target_app_name): |
| print(f"Detected {graphics_api} in the target application.") |
| else: |
| print("Could not detect the graphics API used in the target application.") |
| else: |
| start_time = time.time() |
| print("Target application is not running. Waiting to start training and updating the model...") |
| while (time.time() - start_time) < 5: |
| if is_target_app_running(target_app_name): |
| print(f"Detected {graphics_api} in the target application.") |
| break |
| time.sleep(1) |
| |
| if not is_target_app_running(target_app_name): |
| print("Target application not detected in 5 seconds. Shutting down the AI.") |
| break |
| |
| |
| #Generate some random data for the boxplots |
| np.random.seed(0) |
| original_data = np.random.normal(0, 1, 100) |
| trained_data = np.random.normal(0.5, 1, 100) |
| |
| while True: |
| if is_target_app_running(target_app_name): |
| print("Target application is running. Training and updating the model...") |
| trained_model = train_with_data_cap(model, train_loader, criterion, optimizer, device, data_cap_gb=10) |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| |
| # Create a box plot of the original and trained data |
| plt.figure() |
| plt.boxplot([original_data, trained_data], labels=["Original Data", "Trained Data"]) |
| plt.title("Boxplot of Original and Trained Data") |
| plt.ylabel("Values") |
| plt.show() |
| |
| # Save the box plot as an image |
| plt.savefig(r"G:\Epic Games\GTAV\GTA5_AI\Plot Box Comparison\boxplot_comparison.png") |
| |
| application_was_running = True |
| elif application_was_running: |
| print("Target application has exited. Saving the model...") |
| save_model(trained_model, os.path.join(directory, 'GTA5_TRAINED.onnx')) |
| print("Finished training and saved the model.") |
| break |
| else: |
| start_time = time.time() |
| print("Target application is not running. Waiting to detect the graphics API...") |
| while (time.time() - start_time) < 5: |
| if is_target_app_running(target_app_name): |
| if graphics_api := check_graphics_api(target_app_name): |
| print(f"Detected {graphics_api} in the target application.") |
| break |
| else: |
| print("Could not detect the graphics API used in the target application.") |
| time.sleep(1) |
| |
| if not is_target_app_running(target_app_name): |
| print("Target application not detected in 5 seconds. Shutting down the AI.") |
| break |
