Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from rlbot.botmanager.bot_helper_process import BotHelperProcess
from rlbot.utils.logging_utils import get_logger
import sys
from time import sleep
import psutil
from tensorboardX import SummaryWriter
name = 'online'
load_model = False
load_exp = False
class TorchManager(BotHelperProcess):
def __init__(self, agent_metadata_queue, quit_event):
super().__init__(agent_metadata_queue, quit_event)
from multiprocessing.managers import BaseManager
from leviathan.torch_memory import RewardMemory
BaseManager.register('Memory', RewardMemory)
self.logger = get_logger('torch_mgr')
self.metadata_list = [None, None]
self.manager = BaseManager()
self.manager.start()
self.game_memory = self.manager.Memory()
self.writer = SummaryWriter()
self.n_iter = 0
if load_exp:
import os
import sys
import psutil
from rlbot.botmanager.bot_helper_process import BotHelperProcess
from rlbot.utils.logging_utils import get_logger
import time
from framework.utils import get_repo_directory
class BaseHiveManager(BotHelperProcess):
batch_size = 2000
memory_size = 100000
def __init__(self, agent_metadata_queue, quit_event):
super().__init__(agent_metadata_queue, quit_event)
sys.path.insert(0, get_repo_directory()) # this is for separate process imports
self.logger = get_logger('base_hive_mgr')
self.actor_model = self.get_model()
self.shared_model_handle = self.get_shared_model_handle()
self.manager = self.setup_manager()
self.game_memory = self.manager.Memory(self.memory_size, self.actor_model.get_input_state_dimension(),
self.actor_model.get_model_output_dimension())
self.model_path = None
self.load_model = None