#!/usr/bin/env python """Gomill integration for CLOP. Designed for use with CLOP 0.0.8, available from http://remi.coulom.free.fr/CLOP/ """ # The 'connection script' interface is as follows: # - the command-line arguments are: # - any arguments specified in the Script line # - processor # - seed # - then pairs of arguments (parameter name, parameter value) # - the connection script prints a single character to stdout: # 'W' for candidate win, 'D' for draw, 'L' for loss # any further output is ignored # - if the output doesn't start with 'W', 'D', or 'L', it's treated as an error, # and the complete stdout and stderr are reported # (by convention, print "Error:" with a description) # - the connection script's exit status is ignored import os import sys from optparse import OptionParser from gomill import compact_tracebacks from gomill import competitions from gomill import game_jobs from gomill.competitions import ( Competition, CompetitionError, ControlFileError, Player_config) from gomill.job_manager import JobFailed from gomill.ringmasters import ( Ringmaster, RingmasterError, RingmasterInternalError) from gomill.settings import * from gomill.common import opponent_of PARAMETER_TYPES = [ "LinearParameter", "IntegerParameter", "GammaParameter", "IntegerGammaParameter", ] parameter_settings = [ Setting('code', interpret_identifier), Setting('type', interpret_enum(*PARAMETER_TYPES)), Setting('min', interpret_float), Setting('max', interpret_float), ] class Parameter_config(Quiet_config): """Parameter (ie, dimension) description for use in control files.""" # positional or keyword positional_arguments = ('code',) # keyword-only keyword_arguments = tuple(setting.name for setting in parameter_settings if setting.name != 'code') class Parameter_spec(object): """Internal description of a parameter spec from the configuration file. Public attributes: code -- identifier """ def format_for_clop(self): """Return the parameter configuration string for the .clop file.""" if self.is_integer: fmt = "%s %s %d %d" else: fmt = "%s %s %f %f" return fmt % (self.type, self.code, self.min, self.max) def interpret_value(self, s): """Convert CLOP command-line parameter to an engine parameter. Returns an int or float. """ if self.is_integer: return int(s) else: return float(s) def format_for_display(self, v): return str(v) class Clop_tuner(Competition): def control_file_globals(self): result = Competition.control_file_globals(self) result.update({ 'Parameter' : Parameter_config, }) return result global_settings = (Competition.global_settings + competitions.game_settings + [ Setting('candidate_colour', interpret_colour), Setting('parallel', interpret_int, default=1), Setting('clop_H', interpret_float, default=3), Setting('correlations', interpret_enum('all', 'none'), default='all'), Setting('stop_on_error', interpret_bool, default=True), ]) special_settings = [ Setting('opponent', interpret_identifier), Setting('parameters', interpret_sequence_of_quiet_configs(Parameter_config)), Setting('make_candidate', interpret_callable), ] def parameter_spec_from_config(self, parameter_config): """Make a Parameter_spec from a Parameter_config. Raises ControlFileError if there is an error in the configuration. Returns a Parameter_spec with all attributes set. """ arguments = parameter_config.resolve_arguments() interpreted = load_settings(parameter_settings, arguments) pspec = Parameter_spec() for name, value in interpreted.iteritems(): setattr(pspec, name, value) pspec.is_integer = ("Integer" in pspec.type) if pspec.is_integer: if pspec.min != int(pspec.min): raise ControlFileError("'min': should be an integer") if pspec.max != int(pspec.max): raise ControlFileError("'max': should be an integer") return pspec def initialise_from_control_file(self, config): Competition.initialise_from_control_file(self, config) competitions.validate_handicap( self.handicap, self.handicap_style, self.board_size) try: specials = load_settings(self.special_settings, config) except ValueError, e: raise ControlFileError(str(e)) try: self.opponent = self.players[specials['opponent']] except KeyError: raise ControlFileError( "opponent: unknown player %s" % specials['opponent']) self.parameter_specs = [] if not specials['parameters']: raise ControlFileError("parameters: empty list") seen_codes = set() for i, parameter_spec in enumerate(specials['parameters']): try: pspec = self.parameter_spec_from_config(parameter_spec) except StandardError, e: code = parameter_spec.get_key() if code is None: code = i raise ControlFileError("parameter %s: %s" % (code, e)) if pspec.code in seen_codes: raise ControlFileError( "duplicate parameter code: %s" % pspec.code) seen_codes.add(pspec.code) self.parameter_specs.append(pspec) self.candidate_maker_fn = specials['make_candidate'] def get_clop_parameter_specs(self): """Describe the parameters in the format used in the .clop file. Returns a list of strings. """ return [pspec.format_for_clop() for pspec in self.parameter_specs] def interpret_clop_parameters(self, clop_parameters): """Convert the CLOP command-line parameters to engine parameters. clop_parameters -- list of pairs of strings (parameter name, parameter value) Returns a list of engine parameters, suitable for passing to make_candidate(). """ engine_parameters = [] try: if len(clop_parameters) != len(self.parameter_specs): raise ValueError for pspec, (name, value) in \ zip(self.parameter_specs, clop_parameters): if name != pspec.code: raise ValueError engine_parameters.append(pspec.interpret_value(value)) return engine_parameters except ValueError: raise CompetitionError( "couldn't interpret parameters: %s" % repr(clop_parameters)) def format_engine_parameters(self, engine_parameters): return "; ".join( "%s %s" % (pspec.code, pspec.format_for_display(v)) for pspec, v in zip(self.parameter_specs, engine_parameters)) def make_candidate(self, player_code, engine_parameters): """Make a player using the specified engine parameters. Returns a game_jobs.Player. """ try: candidate_config = self.candidate_maker_fn(*engine_parameters) except Exception: raise CompetitionError( "error from make_candidate()\n%s" % compact_tracebacks.format_traceback(skip=1)) if not isinstance(candidate_config, Player_config): raise CompetitionError( "make_candidate() returned %r, not Player" % candidate_config) try: candidate = self.game_jobs_player_from_config( player_code, candidate_config) except Exception, e: raise CompetitionError( "bad player spec from make_candidate():\n" "%s\nparameters were: %s" % (e, self.format_engine_parameters(engine_parameters))) return candidate def get_game_for_parameters(self, clop_seed, clop_parameters): """Return the details of the next game to play. clop_seed -- second command-line parameter passed by clop clop_parameters -- remaining command-line parameters passed by clop. This is like Competition.get_game(), but it never returns NoGameAvailable. """ engine_parameters = self.interpret_clop_parameters(clop_parameters) candidate = self.make_candidate("#%s" % clop_seed, engine_parameters) job = game_jobs.Game_job() job.game_id = clop_seed if self.candidate_colour == 'b': job.player_b = candidate job.player_w = self.opponent else: job.player_b = self.opponent job.player_w = candidate job.board_size = self.board_size job.komi = self.komi job.move_limit = self.move_limit job.handicap = self.handicap job.handicap_is_free = (self.handicap_style == 'free') job.use_internal_scorer = (self.scorer == 'internal') job.internal_scorer_handicap_compensation = \ self.internal_scorer_handicap_compensation job.sgf_event = self.competition_code job.sgf_note = ("Candidate parameters: %s" % self.format_engine_parameters(engine_parameters)) return job class Clop_ringmaster(Ringmaster): def __init__(self, *args, **kwargs): Ringmaster.__init__(self, *args, **kwargs) # clop uses .log, so we need something different self.log_pathname = os.path.join( self.base_directory, self.competition_code) + ".elog" @staticmethod def _get_competition_class(competition_type): if competition_type == "clop_tuner": return Clop_tuner else: raise ValueError def ensure_output_directories(self): if self.record_games: try: if not os.path.exists(self.sgf_dir_pathname): os.mkdir(self.sgf_dir_pathname) except EnvironmentError: raise RingmasterError("failed to create SGF directory:\n%s" % e) if self.write_gtp_logs: try: if not os.path.exists(self.gtplog_dir_pathname): os.mkdir(self.gtplog_dir_pathname) except EnvironmentError: raise RingmasterError( "failed to create GTP log directory:\n%s" % e) def open_logfile(self): try: self.logfile = open(self.log_pathname, "a") except EnvironmentError, e: raise RingmasterError("failed to open log file:\n%s" % e) def run_game_for_clop(self, seed, parameters): """Act as a CLOP connection script. seed -- seed string passed by clop parameters -- list of pairs of strings (parameter name, parameter value) Returns the message to print. """ self._initialise_presenter() try: job = self.competition.get_game_for_parameters(seed, parameters) self._prepare_job(job) start_msg = "starting game %s: %s (b) vs %s (w)" % ( job.game_id, job.player_b.code, job.player_w.code) self.log(start_msg) response = job.run() self.log("response from game %s" % response.game_id) for warning in response.warnings: self.warn(warning) for log_entry in response.log_entries: self.log(log_entry) except (CompetitionError, JobFailed), e: raise RingmasterError(e) result = response.game_result candidate_colour = self.competition.candidate_colour if result.winning_colour == candidate_colour: message = "W" elif result.winning_colour == opponent_of(candidate_colour): message = "L" elif result.is_jigo: message = "D" else: if self.competition.stop_on_error: # Don't want the experiment to stop just because a single game # failed (eg, went over the move limit), so treat it as a draw. message = "D" else: raise RingmasterError("unexpected game result: %s" % result.sgf_result) return message clop_template = """\ Name %(experiment_name)s Script %(connection_pathname)s %(control_filename)s run-game %(parameter_specs)s %(processor_specs)s Replications 1 DrawElo %(drawelo)s H %(clop_H)s Correlations %(correlations)s StopOnError %(stop_on_error)s """ def do_setup(ringmaster, arguments, options): """Create the .clop file, and any needed directories.""" connection_pathname = os.path.abspath(__file__) control_filename = os.path.basename(ringmaster.control_pathname) clop_pathname = os.path.join(ringmaster.base_directory, "%s.clop" % ringmaster.competition_code) competition = ringmaster.competition experiment_name = ringmaster.competition_code parameter_specs = "\n".join(competition.get_clop_parameter_specs()) processor_specs = "\n".join( "Processor par%d" % i for i in xrange(competition.parallel)) if competition.komi == int(competition.komi) or competition.stop_on_error: drawelo = "100" else: drawelo = "0" clop_H = competition.clop_H correlations = competition.correlations stop_on_error = "true" if competition.stop_on_error else "false" with open(clop_pathname, "w") as f: f.write(clop_template % locals()) ringmaster.ensure_output_directories() def do_run_game(ringmaster, arguments, options): """Act as a CLOP connection script.""" try: processor, seed = arguments[:2] except ValueError: raise RingmasterError("not enough connection script arguments") parameter_args = arguments[2:] parameters = [] i = 0 try: while i < len(parameter_args): parameters.append((parameter_args[i], parameter_args[i+1])) i += 2 except LookupError: raise RingmasterError("parameter without value: %s" % parameter_args[i]) ringmaster.set_display_mode('quiet') ringmaster.open_logfile() message = ringmaster.run_game_for_clop(seed, parameters) print message _actions = { "setup" : (do_setup, False), "run-game" : (do_run_game, True), } def main(argv): usage = ("%prog [connection script arguments]\n\n" "commands: setup, run-game") description = ( "`setup` generates a .clop file for use with clop-gui or clop-console. " "Then `run-game` is used (behind the scenes) as the connection script.") parser = OptionParser(usage=usage, description=description) (options, args) = parser.parse_args(argv) if len(args) == 0: parser.error("no control file specified") if len(args) == 1: parser.error("no command specified") command = args[1] command_args = args[2:] try: action, takes_arguments = _actions[command] except KeyError: parser.error("no such command: %s" % command) if command_args and not takes_arguments: parser.error("too many arguments for %s" % command) ctl_pathname = args[0] try: if not os.path.exists(ctl_pathname): raise RingmasterError("control file %s not found" % ctl_pathname) ringmaster = Clop_ringmaster(ctl_pathname) action(ringmaster, command_args, options) exit_status = 0 except RingmasterError, e: print >>sys.stderr, "gomill-clop:", e exit_status = 1 except KeyboardInterrupt: exit_status = 3 except RingmasterInternalError, e: print >>sys.stderr, "gomill-clop: internal error" print >>sys.stderr, e exit_status = 4 except: print >>sys.stderr, "gomill-clop: internal error" compact_tracebacks.log_traceback() exit_status = 4 if exit_status != 0 and command == 'run-game': # Make sure the problem runner sees an error response print "Error" sys.exit(exit_status) if __name__ == "__main__": main(sys.argv[1:])