Refactored code for launching agents.

bin/Communicator.py

@@ -62,7 +62,7 @@ class Communicator(object):
           raise TimeoutError
         else:
           retryCount -= 1
-          print 'error receiving message, trying again'
+          print '[Trainer] waiting for message'
           time.sleep(0.3)
           #raise ValueError('Error while receiving message')
     (msg,sep,rest) = msg.partition('\0')

bin/Trainer.py

 #!/usr/bin/env python
 # encoding: utf-8
 
-import sys, numpy, time, os, subprocess, re
-from signal import SIGINT
+import sys, numpy, time, os, subprocess
 from Communicator import ClientCommunicator, TimeoutError
 
 class DoneError(Exception):
@@ -12,22 +11,6 @@ class DoneError(Exception):
   def __str__(self):
     return 'Done due to %s' % self.msg
 
-class DummyPopen(object):
-  """ Emulates a Popen object without actually starting a process. """
-  def __init__(self, pid):
-    self.pid = pid
-  def poll(self):
-    try:
-      os.kill(self.pid, 0)
-      return None
-    except OSError:
-      return 0
-  def send_signal(self, sig):
-    try:
-      os.kill(self.pid, sig)
-    except OSError:
-      pass
-
 class Trainer(object):
   """ Trainer is responsible for setting up the players and game.
   """
@@ -68,9 +51,9 @@ class Trainer(object):
     self._numAgents = args.offenseAgents + args.defenseAgents
     self._offenseAgents = args.offenseAgents
     self._defenseAgents = args.defenseAgents
+    self._agentReady = set([]) # Unums of ready agents
     self._agentTeams = [] # Names of the teams the agents are playing for
     self._agentNumInt = [] # List of agents internal team numbers
-    self._agentNumExt = [] # List of agents external team numbers
     self._agentServerPort = args.port # Base Port for agent's server
     self._agentOnBall = args.agent_on_ball # If true, agent starts with the ball
     # =============== MISC =============== #
@@ -85,14 +68,43 @@ class Trainer(object):
     self._teamHoldingBall = None # Team currently in control of the ball
     self._playerHoldingBall = None # Player current in control of ball
     self._agentPopen = [] # Agent's processes
+    self._npcPopen = [] # NPC's processes
+    self._connectedPlayers = []
     self.initMsgHandlers()
 
+  def launch_player(self, player_num, play_offense):
+    """Launches a player using sample_player binary
+
+    Returns a Popen process object
+    """
+    if play_offense:
+      team_name = self._offenseTeamName
+    else:
+      team_name = self._defenseTeamName
+    binary_dir = os.path.dirname(os.path.realpath(__file__))
+    config_dir = os.path.join(binary_dir, '../config/formations-dt')
+    player_conf = os.path.join(binary_dir, '../config/player.conf')
+    player_cmd = os.path.join(binary_dir, 'sample_player')
+    player_cmd += ' -t %s -p %i --config_dir %s ' \
+                  ' --log_dir %s --player-config %s' \
+                  %(team_name, self._serverPort, config_dir,
+                    self._logDir, player_conf)
+    if self._record:
+        player_cmd += ' --record'
+    if player_num == 1:
+        player_cmd += ' -g'
+    kwargs = {'stdout':open('/dev/null', 'w'),
+              'stderr':open('/dev/null', 'w')}
+    p = subprocess.Popen(player_cmd.split(' '), shell = False, **kwargs)
+    return p
+
   def launch_agent(self, agent_num, play_offense, port):
-    """Launch the learning agent using the start_agent.sh script and
-    return a DummyPopen for the process.
+    """Launches a learning agent using the agent binary
 
+    Returns a Popen process object
     """
-    print '[Trainer] Launching Agent', str(agent_num)
+    print '[Trainer] Launching %s Agent %d' % \
+      ('offense' if play_offense else 'defense', agent_num)
     if play_offense:
       assert self._numOffense > 0
       team_name = self._offenseTeamName
@@ -111,29 +123,24 @@ class Trainer(object):
       numTeammates = self._numDefense - 1
       numOpponents = self._numOffense
     ext_num = self.convertToExtPlayer(team_name, internal_player_num)
-    self._agentNumExt.append(ext_num)
     binary_dir = os.path.dirname(os.path.realpath(__file__))
-    agentCmd = 'start_agent.sh -t %s -u %i -p %i -P %i --log-dir %s'\
-               ' --numTeammates %i --numOpponents %i'\
-               ' --playingOffense %i --serverPort %i'\
-               %(team_name, ext_num, self._serverPort,
-                 self._coachPort, self._logDir, numTeammates,
-                 numOpponents, play_offense, port)
+    config_dir = os.path.join(binary_dir, '../config/formations-dt')
+    player_conf = os.path.join(binary_dir, '../config/player.conf')
+    agent_cmd =  os.path.join(binary_dir, 'agent')
+    agent_cmd += ' -t %s -p %i --numTeammates %i --numOpponents %i' \
+                 ' --playingOffense %i --serverPort %i --log_dir %s' \
+                 ' --player-config %s --config_dir %s' \
+                 %(team_name, self._serverPort, numTeammates,
+                   numOpponents, play_offense, port, self._logDir,
+                   player_conf, config_dir)
+    if ext_num == 1:
+      agent_cmd += ' -g'
     if self._record:
-      agentCmd += ' --record'
-    agentCmd = os.path.join(binary_dir, agentCmd)
-    agentCmd = agentCmd.split(' ')
-    # Ignore stderr because librcsc continually prints to it
-    kwargs = {'stderr':open('/dev/null','w')}
-    p = subprocess.Popen(agentCmd, **kwargs)
-    p.wait()
-    pid_file = os.path.join(self._logDir, 'start%i'%p.pid)
-    print '[Trainer] Parsing agent\'s pid from file:', pid_file
-    assert os.path.isfile(pid_file)
-    with open(pid_file,'r') as f:
-      output = f.read()
-    pid = int(re.findall('PID: (\d+)',output)[0])
-    return DummyPopen(pid)
+      agent_cmd += ' --record'
+    kwargs = {'stdout':open('/dev/null', 'w'),
+              'stderr':open('/dev/null', 'w')}
+    p = subprocess.Popen(agent_cmd.split(' '), shell = False, **kwargs)
+    return p
 
   def getDefensiveRoster(self, team_name):
     """Returns a list of player numbers on a given team that are thought
@@ -161,6 +168,10 @@ class Trainer(object):
     else:
       return [11,7,8,9,10,6,3,2,4,5]
 
+  def addTeam(self, team_name):
+    """ Adds a team to the team list"""
+    self._teams.append(team_name)
+
   def setTeams(self):
     """ Sets the offensive and defensive teams and player rosters. """
     self._offenseTeamInd = 0
@@ -217,10 +228,8 @@ class Trainer(object):
   def _hear(self, body):
     """ Handle a hear message. """
     timestep,playerInfo,msg = body
-    if len(playerInfo) != 3:
-      return
-    _,team,player = playerInfo
     try:
+      _,team,player = playerInfo[:3]
       length = int(msg[0])
     except:
       return
@@ -232,6 +241,9 @@ class Trainer(object):
         self.startGame()
     elif msg == 'DONE':
       raise DoneError
+    elif msg == 'ready':
+      print '[Trainer] Agent Ready:', team, player
+      self._agentReady.add((team, player))
     else:
       print '[Trainer] Unhandled message from agent: %s' % msg
 
@@ -361,9 +373,9 @@ class Trainer(object):
           val = param[1]
       self._SP[param[0]] = val
 
-  def listenAndProcess(self):
+  def listenAndProcess(self, retry_count=None):
     """ Gather messages and process them. """
-    msg = self.recv()
+    msg = self.recv(retry_count)
     assert((msg[0] == '(') and (msg[-1] == ')')),'|%s|' % msg
     msg = self.parseMsg(msg)
     self.handleMsg(msg)
@@ -389,6 +401,26 @@ class Trainer(object):
     #self.unregisterMsgHandler(*partial)
     self.ignoreMsg(*partial,quiet=True)
 
+  def waitOnPlayer(self, player_num, on_offense):
+    """Wait on a launched player to connect and be reported by the
+    server.
+
+    """
+    self.send('(look)')
+    partial = ['ok','look']
+    self._numPlayers = 0
+    def f(body):
+      for i in xrange(4, len(body)):
+        _,team,num = body[i][0][:3]
+        if (team, num) not in self._connectedPlayers:
+          self._connectedPlayers.append((team,num))
+    self.registerMsgHandler(f,*partial,quiet=True)
+    team_name = self._offenseTeamName if on_offense else self._defenseTeamName
+    while (team_name, str(player_num)) not in self._connectedPlayers:
+      self.listenAndProcess()
+      self.send('(look)')
+    self.ignoreMsg(*partial,quiet=True)
+
   def checkIfAllPlayersConnected(self):
     """ Returns true if all players are connected. """
     self.send('(look)')
@@ -397,8 +429,8 @@ class Trainer(object):
     def f(x):
       self._numPlayers = len(x) - 4 # -4 for time, ball, goal_l, and goal_r
       self.send('(look)')
-    self.registerMsgHandler(f,*partial)
-    while self._numPlayers != 2 * 11:
+    self.registerMsgHandler(f,*partial,quiet=True)
+    while self._numPlayers != 2 * 11: # self._numOffense + self._numDefense:
       self.listenAndProcess()
     self.ignoreMsg(*partial,quiet=True)
 
@@ -543,9 +575,7 @@ class Trainer(object):
     ends to setup for the next trial.
 
     """
-    # Always Move the offensive goalie to the left goal
-    self.movePlayer(self._offenseTeamName, 0, [-0.5 * self.PITCH_LENGTH, 0])
-    # Move the rest of the offense
+    # Move the offense
     for i in xrange(1, self._numOffense + 1):
       self.movePlayer(self._offenseTeamName, i, self.getOffensiveResetPosition())
     # Move the agent to the ball
@@ -558,26 +588,8 @@ class Trainer(object):
     for i in xrange(1, self._numDefense):
       self.movePlayer(self._defenseTeamName, i, self.getDefensiveResetPosition())
 
-  def removeNonHFOPlayers(self):
-    """Removes players that aren't involved in HFO game.
-
-    The players whose numbers are greater than numOffense/numDefense
-    are sent to left-field.
-
-    """
-    offensive_agent_numbers = self._agentNumInt[:self._offenseAgents]
-    defensive_agent_numbers = self._agentNumInt[self._offenseAgents:]
-    for i in xrange(self._numOffense + 1, 11):
-      if i not in offensive_agent_numbers:
-        self.movePlayer(self._offenseTeamName, i, [-0.25 * self.PITCH_LENGTH, 0])
-    for i in xrange(self._numDefense, 11):
-      if i not in defensive_agent_numbers:
-        self.movePlayer(self._defenseTeamName, i, [-0.25 * self.PITCH_LENGTH, 0])
-
   def step(self):
     """ Takes a simulated step. """
-    # self.send('(check_ball)')
-    self.removeNonHFOPlayers()
     self._teamHoldingBall, self._playerHoldingBall = self.calcBallHolder()
     if self._teamHoldingBall is not None:
       self._lastFrameBallTouched = self._frame
@@ -654,16 +666,66 @@ class Trainer(object):
   def run(self, necProcesses):
     """ Run the trainer """
     try:
-      for agent_num in xrange(self._offenseAgents):
-        port = self._agentServerPort + agent_num
-        agent = self.launch_agent(agent_num, play_offense=True, port=port)
-        self._agentPopen.append(agent)
-        necProcesses.append([agent, 'offense_agent_' + str(agent_num)])
-      for agent_num in xrange(self._defenseAgents):
-        port = self._agentServerPort + agent_num + self._offenseAgents
-        agent = self.launch_agent(agent_num, play_offense=False, port=port)
-        self._agentPopen.append(agent)
-        necProcesses.append([agent, 'defense_agent_' + str(agent_num)])
+      self.setTeams()
+      offense_unums = self._offenseOrder[1: self._numOffense + 1]
+      sorted_offense_agent_unums = sorted(self._offenseOrder[1:self._offenseAgents+1])
+      defense_unums = self._defenseOrder[: self._numDefense]
+      sorted_defense_agent_unums = sorted(self._defenseOrder[:self._defenseAgents])
+      unnecessary_players = []
+
+      # Launch offense
+      agent_num = 0
+      for player_num in xrange(1, 12):
+        if agent_num < self._offenseAgents and \
+           sorted_offense_agent_unums[agent_num] == player_num:
+          port = self._agentServerPort + agent_num
+          agent = self.launch_agent(agent_num, play_offense=True, port=port)
+          self._agentPopen.append(agent)
+          necProcesses.append([agent, 'offense_agent_' + str(agent_num)])
+          agent_num += 1
+        else:
+          player = self.launch_player(player_num, play_offense = True)
+          if player_num in offense_unums:
+            self._npcPopen.append(player)
+            necProcesses.append([player, 'offense_npc_' + str(player_num)])
+          else:
+            unnecessary_players.append(player)
+        self.waitOnPlayer(player_num, on_offense=True)
+      self.waitOnTeam(first = False)
+
+      # Launch defense
+      agent_num = 0
+      for player_num in xrange(1, 12):
+        if agent_num < self._defenseAgents and \
+           sorted_defense_agent_unums[agent_num] == player_num:
+          port = self._agentServerPort + agent_num + self._offenseAgents
+          agent = self.launch_agent(agent_num, play_offense=False, port=port)
+          self._agentPopen.append(agent)
+          necProcesses.append([agent, 'defense_agent_' + str(agent_num)])
+          agent_num += 1
+        else:
+          player = self.launch_player(player_num, play_offense = False)
+          if player_num in defense_unums:
+            self._npcPopen.append(player)
+            necProcesses.append([player, 'defense_npc_' + str(player_num)])
+          else:
+            unnecessary_players.append(player)
+        self.waitOnPlayer(player_num, on_offense=False)
+      self.waitOnTeam(first = False)
+      self.checkIfAllPlayersConnected()
+
+      print '[Trainer] Agents awaiting your connections'
+      necOff = set([(self._offenseTeamName,str(x)) for x in sorted_offense_agent_unums])
+      necDef = set([(self._defenseTeamName,str(x)) for x in sorted_defense_agent_unums])
+      necAgents = necOff.union(necDef)
+      while self.checkLive(necProcesses) and self._agentReady != necAgents:
+        self.listenAndProcess(1000)
+
+      # Terminate unnecessary players
+      print '[Trainer] Removing unnecessary players'
+      for player in unnecessary_players:
+        player.terminate()
+
       # Broadcast the HFO configuration
       offense_nums = ' '.join([str(self.convertToExtPlayer(self._offenseTeamName, i))
                                for i in xrange(1, self._numOffense + 1)])
@@ -674,6 +736,7 @@ class Trainer(object):
                 %(self._offenseTeamName, self._defenseTeamName,
                   self._numOffense, self._numDefense,
                   offense_nums, defense_nums))
+      print '[Trainer] Starting game'
       self.startGame()
       while self.checkLive(necProcesses):
         prevFrame = self._frame
@@ -685,11 +748,23 @@ class Trainer(object):
     except (KeyboardInterrupt, DoneError):
       print '[Trainer] Finished'
     finally:
-      for p in self._agentPopen:
-        p.send_signal(SIGINT)
       try:
         self._comm.sendMsg('(bye)')
       except:
         pass
+      for p in self._agentPopen:
+        try:
+          p.terminate()
+          time.sleep(0.1)
+          p.kill()
+        except:
+          pass
+      for p in self._npcPopen:
+        try:
+          p.terminate()
+          time.sleep(0.1)
+          p.kill()
+        except:
+          pass
       self._comm.close()
       self.printStats()

src/agent.cpp

@@ -86,6 +86,7 @@
 #include <rcsc/player/say_message_builder.h>
 #include <rcsc/player/audio_sensor.h>
 #include <rcsc/player/freeform_parser.h>
+#include <rcsc/player/free_message.h>
 
 #include <rcsc/common/basic_client.h>
 #include <rcsc/common/logger.h>
@@ -108,6 +109,7 @@
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
+#include <poll.h>
 
 using namespace rcsc;
 using namespace hfo;
@@ -119,7 +121,7 @@ Agent::Agent()
       M_action_generator(createActionGenerator()),
       lastTrainerMessageTime(-1),
       server_port(6008),
-      server_running(false),
+      client_connected(false),
       num_teammates(-1),
       num_opponents(-1),
       playing_offense(false)
@@ -230,12 +232,14 @@ bool Agent::initImpl(CmdLineParser & cmd_parser) {
 
     assert(num_teammates >= 0);
     assert(num_opponents >= 0);
+
+    startServer(server_port);
     return true;
 }
 
 void Agent::startServer(int server_port) {
   std::cout << "[Agent Server] Starting Server on Port " << server_port << std::endl;
-  struct sockaddr_in serv_addr, cli_addr;
+  struct sockaddr_in serv_addr;
   sockfd = socket(AF_INET, SOCK_STREAM, 0);
   if (sockfd < 0) {
     perror("[Agent Server] ERROR opening socket");
@@ -250,16 +254,32 @@ void Agent::startServer(int server_port) {
     exit(1);
   }
   listen(sockfd, 5);
-  socklen_t clilen = sizeof(cli_addr);
-  std::cout << "[Agent Server] Waiting for client to connect... " << std::endl;
-  newsockfd = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
-  if (newsockfd < 0) {
-    perror("[Agent Server] ERROR on accept");
-    close(sockfd);
-    exit(1);
+}
+
+void Agent::listenForConnection() {
+  int rv;
+  struct pollfd ufd;
+  ufd.fd = sockfd;
+  ufd.events = POLLIN;
+  rv = poll(&ufd, 1, 1000);
+  if (rv == -1) {
+    perror("poll"); // error occurred in poll()
+  } else if (rv == 0) {
+    std::cout << "[Agent Server] Waiting for client to connect... " << std::endl;
+  } else {
+    if (ufd.revents & POLLIN) {
+      struct sockaddr_in cli_addr;
+      socklen_t clilen = sizeof(cli_addr);
+      newsockfd = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
+      if (newsockfd < 0) {
+        perror("[Agent Server] ERROR on accept");
+        close(sockfd);
+        exit(1);
+      }
+      std::cout << "[Agent Server] Connected" << std::endl;
+      clientHandshake();
+    }
   }
-  std::cout << "[Agent Server] Connected" << std::endl;
-  server_running = true;
 }
 
 void Agent::clientHandshake() {
@@ -316,6 +336,9 @@ void Agent::clientHandshake() {
     exit(1);
   }
   std::cout << "[Agent Server] Handshake complete" << std::endl;
+  client_connected = true;
+  rcsc::FreeMessage<5> *free_msg = new FreeMessage<5>("ready");
+  addSayMessage(free_msg);
 }
 
 FeatureExtractor* Agent::getFeatureExtractor(feature_set_t feature_set_indx,
@@ -367,9 +390,13 @@ status_t Agent::getGameStatus(const rcsc::AudioSensor& audio_sensor,
   virtual method in super class
 */
 void Agent::actionImpl() {
-  if (!server_running) {
-    startServer(server_port);
-    clientHandshake();
+  // For now let's not worry about turning the neck or setting the vision.
+  this->setViewAction(new View_Tactical());
+  this->setNeckAction(new Neck_TurnToBallOrScan());
+
+  if (!client_connected) {
+    listenForConnection();
+    return;
   }
 
   // Update and send the game status
@@ -451,10 +478,6 @@ void Agent::actionImpl() {
       close(sockfd);
       exit(1);
   }
-
-  // For now let's not worry about turning the neck or setting the vision.
-  this->setViewAction(new View_Tactical());
-  this->setNeckAction(new Neck_TurnToBallOrScan());
 }
 
 /*-------------------------------------------------------------------*/

src/agent.h

@@ -72,7 +72,7 @@ protected:
 
   // Start the server and listen for a connection.
   void startServer(int server_port=6008);
-
+  void listenForConnection();
   // Transmit information to the client and ensure it can recieve.
   void clientHandshake();
 
@@ -80,7 +80,7 @@ protected:
   FeatureExtractor* feature_extractor;
   long lastTrainerMessageTime; // Last time the trainer sent a message
   int server_port; // Port to start the server on
-  bool server_running; // Is the server running?
+  bool client_connected; // Has the client connected and handshake?
   int sockfd, newsockfd; // Server sockets
   int num_teammates, num_opponents;
   bool playing_offense;