vimgolf_gym.lib
Library of vimgolf-gym
1""" 2Library of vimgolf-gym 3""" 4 5# TODO: implement a openai-gym style interface 6# reference link: https://github.com/Farama-Foundation/Gymnasium 7 8# TODO: implement a gradual scoring system by comparing the buffer with the target output, extracting the vim edit buffer in the middle of execution 9 10# Vim API reference: 11# https://github.com/LachlanGray/vim-agent 12# https://github.com/nsbradford/VimGPT 13 14# TODO: dockerize the executor part, push the image, offer an option or environment variable to use dockerized executor 15 16import atexit 17import os 18import pathlib 19import shutil 20import sys 21import tempfile 22import typing 23import time 24import zipfile 25 26import PIL.Image 27import requests 28import uuid 29 30import vimgolf.vimgolf as vimgolf 31import vimgolf_gym.dataclasses as dataclasses 32import vimgolf_gym.log_parser as log_parser 33import vimgolf_gym.terminal_executor as terminal_executor 34from vimgolf_gym._vimrc import ( 35 _prepare_cybergod_vimrc_with_buffer_file, 36 _CYBERGOD_VIMGOLF_VIMRC_FILEPATH, 37) 38import subprocess 39 40_HOMEDIR = os.path.expanduser("~") 41 42_CYBERGOD_VIMGOLF_DATASET_BASEDIR = os.path.join( 43 _HOMEDIR, ".cache", "cybergod-vimgolf-dataset" 44) 45 46_CYBERGOD_VIMGOLF_GYM_DATASET_DOWNLOADED = os.path.join( 47 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, "DATASET_DOWNLOADED" 48) # a flag to indicate whether the dataset has been downloaded 49 50os.makedirs(_CYBERGOD_VIMGOLF_DATASET_BASEDIR, exist_ok=True) 51 52__all__ = [ 53 "make", 54 "list_local_challenge_ids", 55 "get_local_challenge_definition", 56 "get_local_challenge_metadata", 57 "get_local_challenge_worst_solution", 58 "get_local_challenge_worst_solution_header", 59 "VimGolfEnv", 60] 61 62 63class DatasetInitError(Exception): 64 pass 65 66 67def assert_challenge_id_length(challenge_id: str) -> bool: 68 """Assert the challenge_id length to be 24""" 69 assert len(challenge_id) == 24 70 71 72def make( 73 env_name: str, 74 custom_challenge: typing.Optional[dataclasses.VimGolfCustomChallenge] = None, 75 use_docker: bool = False, 76 log_buffer: bool = False, 77) -> "VimGolfEnv": 78 """ 79 Create a VimGolf environment. 80 81 The env_name can be one of the following: 82 - `vimgolf-test`: A simple test environment. 83 - `vimgolf-local-<challenge_id>`: A local environment for a specific challenge. 84 - `vimgolf-online-<challenge_id>`: An online environment for a specific challenge. 85 - `vimgolf-custom`: A custom environment. The `custom_challenge` parameter will be used. 86 87 Args: 88 env_name (str): The name of the environment to create. 89 use_docker (bool, optional): Whether to use a dockerized executor. Defaults to False. 90 log_buffer (bool, optional): Whether to log the editor buffer or not. Defaults to False. 91 custom_challenge (Optional[VimGolfCustomChallenge], optional): The custom challenge to use. Defaults to None. 92 93 Raises: 94 NotImplementedError: If the environment name is not recognized. 95 96 Returns: 97 VimGolfEnv: The created environment 98 """ 99 if use_docker: 100 os.environ["VIMGOLF_GYM_USE_DOCKER"] = "1" 101 else: 102 os.environ["VIMGOLF_GYM_USE_DOCKER"] = "0" 103 104 if log_buffer: 105 os.environ["VIMGOLF_GYM_LOG_BUFFER"] = "1" 106 else: 107 os.environ["VIMGOLF_GYM_LOG_BUFFER"] = "0" 108 109 if env_name == "vimgolf-test": 110 env = make_test() 111 elif env_name == "vimgolf-custom": 112 assert ( 113 custom_challenge 114 ), "custom_challenge must be provided for vimgolf-custom environment" 115 env = make_env_with_text( 116 input=custom_challenge.input, output=custom_challenge.output 117 ) 118 elif env_name.startswith("vimgolf-local-"): 119 challenge_id = env_name[len("vimgolf-local-") :] 120 assert_challenge_id_length(challenge_id) 121 env = make_offline_with_cybergod_dataset(challenge_id) 122 elif env_name.startswith("vimgolf-online-"): 123 challenge_id = env_name[len("vimgolf-online-") :] 124 assert_challenge_id_length(challenge_id) 125 env = make_online(challenge_id) 126 else: 127 raise NotImplementedError 128 return env 129 130 131def make_test() -> "VimGolfEnv": 132 """ 133 Create an environment for a simple test challenge. 134 135 The test challenge is about typing "hello world" into the buffer and then saving and quitting vim. 136 The expected solution is "hello world\nhello world\n". 137 """ 138 input_text = "" 139 output_text = "hello world\nhello world\n" 140 return make_env_with_text(input_text, output_text) 141 142 143def make_env_with_text(input_text: str, output_text: str) -> "VimGolfEnv": 144 """ 145 Create a VimGolfEnv with given input and output text. 146 147 Creates a temporary directory and two files, one for the input and one for the output. 148 Then it calls make_offline with the paths of the two files. 149 150 Args: 151 input_text (str): The starting code. 152 output_text (str): The expected solution code. 153 154 Returns: 155 VimGolfEnv: The environment object. 156 """ 157 tempdir = tempfile.TemporaryDirectory() 158 atexit.register(tempdir.cleanup) 159 input_file = os.path.join(tempdir.name, "input.txt") 160 output_file = os.path.join(tempdir.name, "output.txt") 161 with open(input_file, "w") as f: 162 f.write(input_text) 163 with open(output_file, "w") as f: 164 f.write(output_text) 165 return make_offline(input_file, output_file) 166 167 168def make_offline(input_file: str, output_file: str) -> "VimGolfEnv": 169 """ 170 Create an environment from a VimGolf challenge given as a local file. 171 172 Args: 173 input_file (str): Path to the file containing the starting code. 174 output_file (str): Path to the file containing the expected solution code. 175 176 Returns: 177 VimGolfEnv: An environment for the given challenge. 178 """ 179 use_docker = os.environ.get("VIMGOLF_GYM_USE_DOCKER", None) == "1" 180 log_buffer = os.environ.get("VIMGOLF_GYM_LOG_BUFFER", None) == "1" 181 return VimGolfEnv( 182 input_file=input_file, 183 output_file=output_file, 184 use_docker=use_docker, 185 log_buffer=log_buffer, 186 ) 187 188 189def make_online(challenge_id: str) -> "VimGolfEnv": 190 """ 191 Create an environment from a VimGolf challenge online. 192 193 Given a challenge_id, obtain the challenge definition from the VimGolf website 194 and create an environment out of it. 195 196 Args: 197 challenge_id (str): Unique identifier for the challenge. 198 199 Returns: 200 VimGolfEnv: An environment for the given challenge. 201 """ 202 challenge_url = vimgolf.get_challenge_url(challenge_id) 203 challenge_data = requests.get(challenge_url).content 204 challenge = dataclasses.VimGolfChallengeDefinition.parse_raw(challenge_data) 205 return make_env_with_challenge(challenge) 206 207 208def make_env_with_challenge( 209 challenge: dataclasses.VimGolfChallengeDefinition, 210) -> "VimGolfEnv": 211 """ 212 Create an environment from a VimGolfChallengeDefinition object. 213 214 This function simply passes the input and output text to make_env_with_text, 215 which will create a VimGolfEnv with the given text. 216 217 Args: 218 challenge: VimGolfChallengeDefinition object 219 220 Returns: 221 VimGolfEnv: An environment for the given challenge 222 """ 223 return make_env_with_text( 224 input_text=challenge.input.data, output_text=challenge.output.data 225 ) 226 227 228def init_cybergod_vimgolf_dataset() -> None: 229 """ 230 Initialize the local dataset by downloading it if it does not exist yet. 231 232 After this function is called, the local dataset should be downloaded and 233 ready to use. 234 235 This function is called by `list_local_challenge_ids` and `make_offline_with_cybergod_dataset`. 236 """ 237 if not os.path.exists(_CYBERGOD_VIMGOLF_GYM_DATASET_DOWNLOADED): 238 download_cybergod_vimgolf_dataset() 239 240 241def list_local_challenge_ids() -> list[str]: 242 """ 243 List all challenge ids in the local dataset. 244 245 This function will download the local dataset if it does not exist yet. 246 247 Returns: 248 list[str]: a list of all challenge ids in the local dataset. 249 """ 250 251 init_cybergod_vimgolf_dataset() 252 challenges_dir = os.path.join( 253 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, 254 "challenges", 255 ) 256 challenge_ids = os.listdir(challenges_dir) 257 return challenge_ids 258 259 260def make_offline_with_cybergod_dataset(challenge_id: str) -> "VimGolfEnv": 261 """ 262 Load a VimGolf challenge from the local dataset and make an environment 263 out of it. 264 265 Given a challenge_id, find the corresponding challenge definition in the 266 dataset, parse it into a VimGolfChallengeDefinition object, and create a 267 VimGolfEnv out of it. 268 269 Args: 270 challenge_id (str): Unique identifier for the challenge. 271 272 Returns: 273 VimGolfEnv: An environment for the given challenge. 274 """ 275 init_cybergod_vimgolf_dataset() 276 challenge = get_local_challenge_definition(challenge_id) 277 return make_env_with_challenge(challenge) 278 279 280def get_local_challenge_metadata(challenge_id: str): 281 """ 282 Load a VimGolf challenge's metadata from the local dataset. 283 284 Given a challenge_id, find the corresponding JSON file in the dataset 285 and parse it into a VimGolfChallengeMetadata object. 286 287 Args: 288 challenge_id (str): Unique identifier for the challenge. 289 290 Returns: 291 VimGolfChallengeMetadata: Parsed challenge metadata. 292 293 Raises: 294 AssertionError: If the metadata file does not exist. 295 """ 296 metadata_file = os.path.join( 297 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, "challenges", challenge_id, "metadata.json" 298 ) 299 assert os.path.exists(metadata_file), ( 300 "Metadata file '%s' does not exist" % metadata_file 301 ) 302 with open(metadata_file, "r") as f: 303 metadata = dataclasses.VimGolfChallengeMetadata.parse_raw(f.read()) 304 return metadata 305 306 307def get_local_challenge_worst_solution(challenge_id: str): 308 """ 309 Load the worst solution for a challenge from the local dataset. 310 311 Given a challenge_id, find the corresponding JSON file in the dataset 312 and parse it into a VimGolfPublicSolution object. 313 314 Args: 315 challenge_id (str): Unique identifier for the challenge. 316 317 Returns: 318 VimGolfPublicSolution: Parsed worst solution. 319 320 Raises: 321 AssertionError: If the worst solution file does not exist. 322 """ 323 metadata_file = os.path.join( 324 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, 325 "challenges", 326 challenge_id, 327 "worst_solution.json", 328 ) 329 assert os.path.exists(metadata_file), ( 330 "Worst solution file '%s' does not exist" % metadata_file 331 ) 332 with open(metadata_file, "r") as f: 333 Worst_solution = dataclasses.VimGolfPublicSolution.parse_raw(f.read()) 334 return Worst_solution 335 336 337def get_local_challenge_worst_solution_header(challenge_id: str): 338 """ 339 Parse the worst solution's header string from the local dataset. 340 341 Given a challenge_id, find the corresponding worst solution file in the dataset 342 and parse its header string into a VimGolfParsedPublicSolutionHeader object. 343 344 Args: 345 challenge_id (str): Unique identifier for the challenge. 346 347 Returns: 348 VimGolfParsedPublicSolutionHeader: Parsed header string. 349 350 Raises: 351 AssertionError: If the worst solution file does not exist. 352 """ 353 solution = get_local_challenge_worst_solution(challenge_id) 354 header = solution.header 355 ret = dataclasses.parse_public_solution_header(header) 356 return ret 357 358 359def get_local_challenge_definition(challenge_id: str): 360 """ 361 Load a VimGolf challenge definition from the local dataset. 362 363 Given a challenge_id, find the corresponding JSON file in the dataset 364 and parse it into a VimGolfChallengeDefinition object. 365 366 Args: 367 challenge_id (str): Unique identifier for the challenge. 368 369 Returns: 370 VimGolfChallengeDefinition: Parsed challenge definition. 371 372 Raises: 373 AssertionError: If the challenge file does not exist. 374 """ 375 challenge_file = os.path.join( 376 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, "challenges", challenge_id, "challenge.json" 377 ) 378 assert os.path.exists(challenge_file), ( 379 "Challenge file '%s' does not exist" % challenge_file 380 ) 381 with open(challenge_file, "r") as f: 382 challenge = dataclasses.VimGolfChallengeDefinition.parse_raw(f.read()) 383 return challenge 384 385 386def download_cybergod_vimgolf_dataset(): 387 """ 388 Download the CyberGod VimGolf dataset from various sources. 389 390 This function is called when the dataset is not initialized yet. It downloads the dataset 391 from various sources (Kaggle, Hugging Face, GitHub Releases, GitHub Mirror) and extracts it 392 to the dataset directory. After the download is finished, it touches the flag file 393 _CYBERGOD_VIMGOLF_GYM_DATASET_DOWNLOADED to indicate that the dataset is initialized. 394 395 If the download fails, it raises an exception and cleans up the dataset directory. 396 397 :raises DatasetInitError: If the dataset download fails. 398 """ 399 print( 400 "Initializing CyberGod VimGolf dataset at:", _CYBERGOD_VIMGOLF_DATASET_BASEDIR 401 ) 402 try: 403 # TODO: add huggingface, hf-mirror.com, github releases and github mirror links 404 download_urls = [ 405 "https://www.kaggle.com/api/v1/datasets/download/jessysisca/vimgolf-challenges-and-solutions", 406 "https://hf-mirror.com/datasets/James4Ever0/vimgolf_challenges_and_solutions/resolve/main/challenges.zip?download=true", 407 "https://huggingface.co/datasets/James4Ever0/vimgolf_challenges_and_solutions/resolve/main/challenges.zip?download=true", 408 "https://github.com/James4Ever0/vimgolf-gym/releases/download/dataset-release/challenges.zip", 409 "https://bgithub.xyz/James4Ever0/vimgolf-gym/releases/download/dataset-release/challenges.zip", 410 ] 411 with tempfile.TemporaryDirectory() as tempdir: 412 zip_file_path = os.path.join( 413 tempdir, "vimgolf-challenges-and-solutions.zip" 414 ) 415 with open(zip_file_path, "wb") as f: 416 content = None 417 for url in download_urls: 418 try: 419 print("Downloading:", url) 420 content = requests.get( 421 url, allow_redirects=True, timeout=10 422 ).content 423 break 424 except requests.Timeout: 425 print("Timeout, trying next URL") 426 if content: 427 f.write(content) 428 else: 429 raise DatasetInitError("Failed to download the dataset") 430 with zipfile.ZipFile(zip_file_path, "r") as zip_ref: 431 # extract to CYBERGOD_VIMGOLF_GYM_DATASET_DIR 432 zip_ref.extractall(_CYBERGOD_VIMGOLF_DATASET_BASEDIR) 433 # after all, touch the flag _CYBERGOD_VIMGOLF_GYM_DATASET_DOWNLOADED 434 pathlib.Path(_CYBERGOD_VIMGOLF_GYM_DATASET_DOWNLOADED).touch() 435 finally: 436 if not os.path.exists(_CYBERGOD_VIMGOLF_GYM_DATASET_DOWNLOADED): 437 # cleanup the dataset basedir, if the dataset is not downloaded successfully 438 shutil.rmtree(_CYBERGOD_VIMGOLF_DATASET_BASEDIR) 439 440 441class VimGolfEnv: 442 def __init__( 443 self, 444 input_file: str, 445 output_file: str, 446 width: int = 80, 447 height: int = 24, 448 init_keys: str = "", 449 use_docker: bool = False, 450 log_buffer: bool = False, 451 ): 452 """Initialize the environment with the given input and output files. 453 454 Args: 455 input_file (str): the input file path 456 output_file (str): the output file path 457 width (int): the width of the terminal 458 height (int): the height of the terminal 459 use_docker (bool): whether use dockerized executor or local (requiring vim installed) 460 log_buffer (bool): whether to log the editor buffer or not 461 init_keys (str): initial input keys in Vimgolf solution style 462 """ 463 464 self.use_docker = use_docker 465 """whether use dockerized executor or local (requiring vim installed)""" 466 467 self.input_file = input_file 468 """the input file path""" 469 470 self.output_file = output_file 471 """the output file path""" 472 473 self.log_buffer = log_buffer 474 """whether to log the editor buffer or not""" 475 476 self.init_keys = init_keys 477 """initial input keys in Vimgolf solution style""" 478 479 assert os.path.isfile( 480 self.input_file 481 ), f"Input file {self.input_file} does not exist." 482 assert os.path.isfile( 483 self.output_file 484 ), f"Output file {self.output_file} does not exist." 485 486 # TODO: run a modified version of vimgolf local python script writing progress to a jsonl file, which embeds in this script, for easy state inspection and data collection (we can create a temporary directory for cleanup) 487 self.log_directory = tempfile.TemporaryDirectory() 488 """the log directory, where tempfiles stored""" 489 490 self.log_file = os.path.join(self.log_directory.name, "vimgolf.log") 491 """the log file path, used to retrieve progress info of the vimgolf process""" 492 493 self.buffer_file = os.path.join( 494 self.log_directory.name, "vimgolf_editor_buffer" 495 ) 496 """the editor buffer, used to track granual progress""" 497 498 self._container_name = str(uuid.uuid4()) 499 500 if self.use_docker: 501 mountpoint = "/vimgolf_gym_workdir" 502 docker_output_file = os.path.join(mountpoint, "out") 503 docker_input_file = os.path.join(mountpoint, "in") 504 docker_buffer_file = os.path.join(mountpoint, "vimgolf_editor_buffer") 505 docker_log_file = os.path.join(mountpoint, "vimgolf.log") 506 shutil.copy(self.input_file, os.path.join(self.log_directory.name, "in")) 507 shutil.copy(self.output_file, os.path.join(self.log_directory.name, "out")) 508 extra_docker_run_params = [] 509 if self.log_buffer: 510 _prepare_cybergod_vimrc_with_buffer_file(docker_buffer_file) 511 extra_docker_run_params += [ 512 "-v", 513 "%s:%s:ro" 514 % ( 515 _CYBERGOD_VIMGOLF_VIMRC_FILEPATH, 516 "/usr/local/lib/python3.10/dist-packages/vimgolf/vimgolf.vimrc", 517 ), 518 ] 519 self.command = [ 520 "docker", 521 "run", 522 "--rm", 523 "-it", 524 "--name", 525 self._container_name, 526 "-v", 527 "%s:%s" % (self.log_directory.name, mountpoint), 528 *extra_docker_run_params, 529 "--entrypoint", 530 "python3", 531 "agile4im/cybergod_vimgolf_gym", 532 "-m", 533 "vimgolf_gym.vimgolf", 534 "--input_file", 535 docker_input_file, 536 "--output_file", 537 docker_output_file, 538 "--log_file", 539 docker_log_file, 540 ] 541 if self.init_keys: 542 self.command += ["--init_keys", self.init_keys] 543 else: 544 extra_flags = [] 545 if self.log_buffer: 546 extra_flags += ["--buffer_file", self.buffer_file] 547 if self.init_keys: 548 extra_flags += ["--init_keys", self.init_keys] 549 self.command = [ 550 sys.executable, 551 "-m", 552 "vimgolf_gym.vimgolf", 553 "--input_file", 554 self.input_file, 555 "--output_file", 556 self.output_file, 557 "--log_file", 558 self.log_file, 559 *extra_flags, 560 ] 561 562 self._closing = False 563 564 self.command: list[str] 565 """the command passed to underlying terminal executor""" 566 567 self.width = width 568 """terminal width""" 569 self.height = height 570 """terminal height""" 571 self.create_executor_and_log_watcher() 572 573 self.executor: terminal_executor.TerminalExecutor 574 """terminal executor for running vimgolf process""" 575 self.log_watcher: log_parser.VimGolfLogWatcher 576 """log watcher for tracking vimgolf log output""" 577 atexit.register(self.log_directory.cleanup) 578 atexit.register(self._kill_docker_container) 579 580 @property 581 def buffer(self) -> typing.Optional[bytes]: 582 """The editor buffer""" 583 if not self.log_buffer: 584 return None 585 else: 586 if os.path.isfile(self.buffer_file): 587 with open(self.buffer_file, "rb") as f: 588 return f.read() 589 590 def act(self, action: str): 591 """Take an action 592 593 Args: 594 action (str): the action to take 595 """ 596 self.executor.input(action) 597 598 @property 599 def success(self): 600 """Check if the vimgolf challenge has been solved successfully""" 601 return self.log_watcher.success 602 603 def get_best_success_result(self): 604 """ 605 Return the best success result in the log watcher. 606 607 Returns: 608 Optional[VimGolfEnvResult]: The best success result. 609 """ 610 return self.log_watcher.get_best_success_result() 611 612 def get_last_success_result(self): 613 """ 614 Return the last success result in the log watcher. 615 616 Returns: 617 Optional[VimGolfEnvResult]: The last success result 618 """ 619 return self.log_watcher.get_last_success_result() 620 621 @property 622 def results(self): 623 """The results of the vimgolf challenge environment 624 625 Returns: 626 list[VimGolfEnvResult]: The results of the vimgolf challenge environment 627 """ 628 return self.log_watcher.results 629 630 def __enter__(self): 631 return self 632 633 def __exit__(self, exc, value, tb): 634 self.close() 635 636 @property 637 def success_results(self): 638 """The success results of the vimgolf challenge environment 639 640 Returns: 641 list[VimGolfEnvResult]: The success results of the vimgolf challenge environment 642 """ 643 return self.log_watcher.success_results 644 645 def create_executor_and_log_watcher(self): 646 """Create the executor and log watcher""" 647 self.executor = terminal_executor.TerminalExecutor( 648 command=self.command, width=self.width, height=self.height 649 ) 650 if not hasattr(self, "log_watcher"): 651 self.log_watcher = log_parser.VimGolfLogWatcher(self.log_file) 652 # shall we wait the executor be ready 653 # we wait for the 'play(...)' indicator to appear in the log file. 654 while True: 655 if os.path.exists(self.log_file): 656 with open(self.log_file, "r") as f: 657 log_content = f.read() 658 if "play(...)" in log_content: 659 break 660 time.sleep(0.5) 661 662 def reset(self): 663 """Reset the environment""" 664 self.close() 665 self._closing = False 666 self.create_executor_and_log_watcher() 667 668 def render(self): 669 """Render the environment""" 670 screenshot = self.screenshot() 671 # display the screenshot 672 screenshot.show() 673 674 def screenshot(self): 675 """Take a screenshot of the environment 676 677 Returns: 678 PIL.Image.Image: The screenshot 679 """ 680 with tempfile.TemporaryDirectory() as tmpdir: 681 png_tmpfile_path = os.path.join(tmpdir, "screenshot.png") 682 self.executor.screenshot(png_tmpfile_path) 683 image = PIL.Image.open(png_tmpfile_path) 684 return image 685 686 def _kill_docker_container(self): 687 if self.use_docker: 688 subprocess.run( 689 ["docker", "kill", self._container_name], capture_output=True 690 ) 691 692 def verify_keys(self, keys: str): 693 """Verify a solution by its keys 694 695 Args: 696 keys (str): the keys to verify, in Vimgolf style 697 """ 698 assert keys, "Keys cannot be empty" 699 with VimGolfEnv( 700 input_file=self.input_file, 701 output_file=self.output_file, 702 init_keys=keys, 703 use_docker=self.use_docker, 704 log_buffer=True, 705 ) as env: 706 for _ in range(3): 707 success = env.success 708 if success: 709 break 710 time.sleep(1) 711 if not success: 712 buffer = env.buffer 713 with open(self.output_file, "rb") as f: 714 expected_output = f.read() 715 success = buffer == expected_output 716 return success 717 718 def close(self): 719 """Close the environment""" 720 if not self._closing: 721 self._closing = True 722 self.executor.close() 723 self._kill_docker_container() 724 del self.executor 725 if os.path.exists(self.log_file): 726 os.remove(self.log_file) 727 setattr(self, "executor", None) 728 729 def calculate_relative_inverse_score(self, score:int, worst_score:typing.Optional[int] =None): 730 """Calculate the relative inverse score of the given score 731 732 Args: 733 score (int): The score to calculate the relative inverse score of. 734 worst_score (int, optional): The worst score to use. Defaults to None. 735 736 Returns: 737 float: The relative inverse score. 738 """ 739 assert score >= 0, "Score must be non-negative" 740 if worst_score is None: 741 with open(self.output_file, "r") as f: 742 worst_score = len(f.read()) + 10 743 ret = worst_score / score 744 return ret
73def make( 74 env_name: str, 75 custom_challenge: typing.Optional[dataclasses.VimGolfCustomChallenge] = None, 76 use_docker: bool = False, 77 log_buffer: bool = False, 78) -> "VimGolfEnv": 79 """ 80 Create a VimGolf environment. 81 82 The env_name can be one of the following: 83 - `vimgolf-test`: A simple test environment. 84 - `vimgolf-local-<challenge_id>`: A local environment for a specific challenge. 85 - `vimgolf-online-<challenge_id>`: An online environment for a specific challenge. 86 - `vimgolf-custom`: A custom environment. The `custom_challenge` parameter will be used. 87 88 Args: 89 env_name (str): The name of the environment to create. 90 use_docker (bool, optional): Whether to use a dockerized executor. Defaults to False. 91 log_buffer (bool, optional): Whether to log the editor buffer or not. Defaults to False. 92 custom_challenge (Optional[VimGolfCustomChallenge], optional): The custom challenge to use. Defaults to None. 93 94 Raises: 95 NotImplementedError: If the environment name is not recognized. 96 97 Returns: 98 VimGolfEnv: The created environment 99 """ 100 if use_docker: 101 os.environ["VIMGOLF_GYM_USE_DOCKER"] = "1" 102 else: 103 os.environ["VIMGOLF_GYM_USE_DOCKER"] = "0" 104 105 if log_buffer: 106 os.environ["VIMGOLF_GYM_LOG_BUFFER"] = "1" 107 else: 108 os.environ["VIMGOLF_GYM_LOG_BUFFER"] = "0" 109 110 if env_name == "vimgolf-test": 111 env = make_test() 112 elif env_name == "vimgolf-custom": 113 assert ( 114 custom_challenge 115 ), "custom_challenge must be provided for vimgolf-custom environment" 116 env = make_env_with_text( 117 input=custom_challenge.input, output=custom_challenge.output 118 ) 119 elif env_name.startswith("vimgolf-local-"): 120 challenge_id = env_name[len("vimgolf-local-") :] 121 assert_challenge_id_length(challenge_id) 122 env = make_offline_with_cybergod_dataset(challenge_id) 123 elif env_name.startswith("vimgolf-online-"): 124 challenge_id = env_name[len("vimgolf-online-") :] 125 assert_challenge_id_length(challenge_id) 126 env = make_online(challenge_id) 127 else: 128 raise NotImplementedError 129 return env
Create a VimGolf environment.
The env_name can be one of the following:
vimgolf-test
: A simple test environment.vimgolf-local-<challenge_id>
: A local environment for a specific challenge.vimgolf-online-<challenge_id>
: An online environment for a specific challenge.vimgolf-custom
: A custom environment. Thecustom_challenge
parameter will be used.
Arguments:
- env_name (str): The name of the environment to create.
- use_docker (bool, optional): Whether to use a dockerized executor. Defaults to False.
- log_buffer (bool, optional): Whether to log the editor buffer or not. Defaults to False.
- custom_challenge (Optional[VimGolfCustomChallenge], optional): The custom challenge to use. Defaults to None.
Raises:
- NotImplementedError: If the environment name is not recognized.
Returns:
VimGolfEnv: The created environment
242def list_local_challenge_ids() -> list[str]: 243 """ 244 List all challenge ids in the local dataset. 245 246 This function will download the local dataset if it does not exist yet. 247 248 Returns: 249 list[str]: a list of all challenge ids in the local dataset. 250 """ 251 252 init_cybergod_vimgolf_dataset() 253 challenges_dir = os.path.join( 254 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, 255 "challenges", 256 ) 257 challenge_ids = os.listdir(challenges_dir) 258 return challenge_ids
List all challenge ids in the local dataset.
This function will download the local dataset if it does not exist yet.
Returns:
list[str]: a list of all challenge ids in the local dataset.
360def get_local_challenge_definition(challenge_id: str): 361 """ 362 Load a VimGolf challenge definition from the local dataset. 363 364 Given a challenge_id, find the corresponding JSON file in the dataset 365 and parse it into a VimGolfChallengeDefinition object. 366 367 Args: 368 challenge_id (str): Unique identifier for the challenge. 369 370 Returns: 371 VimGolfChallengeDefinition: Parsed challenge definition. 372 373 Raises: 374 AssertionError: If the challenge file does not exist. 375 """ 376 challenge_file = os.path.join( 377 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, "challenges", challenge_id, "challenge.json" 378 ) 379 assert os.path.exists(challenge_file), ( 380 "Challenge file '%s' does not exist" % challenge_file 381 ) 382 with open(challenge_file, "r") as f: 383 challenge = dataclasses.VimGolfChallengeDefinition.parse_raw(f.read()) 384 return challenge
Load a VimGolf challenge definition from the local dataset.
Given a challenge_id, find the corresponding JSON file in the dataset and parse it into a VimGolfChallengeDefinition object.
Arguments:
- challenge_id (str): Unique identifier for the challenge.
Returns:
VimGolfChallengeDefinition: Parsed challenge definition.
Raises:
- AssertionError: If the challenge file does not exist.
281def get_local_challenge_metadata(challenge_id: str): 282 """ 283 Load a VimGolf challenge's metadata from the local dataset. 284 285 Given a challenge_id, find the corresponding JSON file in the dataset 286 and parse it into a VimGolfChallengeMetadata object. 287 288 Args: 289 challenge_id (str): Unique identifier for the challenge. 290 291 Returns: 292 VimGolfChallengeMetadata: Parsed challenge metadata. 293 294 Raises: 295 AssertionError: If the metadata file does not exist. 296 """ 297 metadata_file = os.path.join( 298 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, "challenges", challenge_id, "metadata.json" 299 ) 300 assert os.path.exists(metadata_file), ( 301 "Metadata file '%s' does not exist" % metadata_file 302 ) 303 with open(metadata_file, "r") as f: 304 metadata = dataclasses.VimGolfChallengeMetadata.parse_raw(f.read()) 305 return metadata
Load a VimGolf challenge's metadata from the local dataset.
Given a challenge_id, find the corresponding JSON file in the dataset and parse it into a VimGolfChallengeMetadata object.
Arguments:
- challenge_id (str): Unique identifier for the challenge.
Returns:
VimGolfChallengeMetadata: Parsed challenge metadata.
Raises:
- AssertionError: If the metadata file does not exist.
308def get_local_challenge_worst_solution(challenge_id: str): 309 """ 310 Load the worst solution for a challenge from the local dataset. 311 312 Given a challenge_id, find the corresponding JSON file in the dataset 313 and parse it into a VimGolfPublicSolution object. 314 315 Args: 316 challenge_id (str): Unique identifier for the challenge. 317 318 Returns: 319 VimGolfPublicSolution: Parsed worst solution. 320 321 Raises: 322 AssertionError: If the worst solution file does not exist. 323 """ 324 metadata_file = os.path.join( 325 _CYBERGOD_VIMGOLF_DATASET_BASEDIR, 326 "challenges", 327 challenge_id, 328 "worst_solution.json", 329 ) 330 assert os.path.exists(metadata_file), ( 331 "Worst solution file '%s' does not exist" % metadata_file 332 ) 333 with open(metadata_file, "r") as f: 334 Worst_solution = dataclasses.VimGolfPublicSolution.parse_raw(f.read()) 335 return Worst_solution
Load the worst solution for a challenge from the local dataset.
Given a challenge_id, find the corresponding JSON file in the dataset and parse it into a VimGolfPublicSolution object.
Arguments:
- challenge_id (str): Unique identifier for the challenge.
Returns:
VimGolfPublicSolution: Parsed worst solution.
Raises:
- AssertionError: If the worst solution file does not exist.
338def get_local_challenge_worst_solution_header(challenge_id: str): 339 """ 340 Parse the worst solution's header string from the local dataset. 341 342 Given a challenge_id, find the corresponding worst solution file in the dataset 343 and parse its header string into a VimGolfParsedPublicSolutionHeader object. 344 345 Args: 346 challenge_id (str): Unique identifier for the challenge. 347 348 Returns: 349 VimGolfParsedPublicSolutionHeader: Parsed header string. 350 351 Raises: 352 AssertionError: If the worst solution file does not exist. 353 """ 354 solution = get_local_challenge_worst_solution(challenge_id) 355 header = solution.header 356 ret = dataclasses.parse_public_solution_header(header) 357 return ret
Parse the worst solution's header string from the local dataset.
Given a challenge_id, find the corresponding worst solution file in the dataset and parse its header string into a VimGolfParsedPublicSolutionHeader object.
Arguments:
- challenge_id (str): Unique identifier for the challenge.
Returns:
VimGolfParsedPublicSolutionHeader: Parsed header string.
Raises:
- AssertionError: If the worst solution file does not exist.
442class VimGolfEnv: 443 def __init__( 444 self, 445 input_file: str, 446 output_file: str, 447 width: int = 80, 448 height: int = 24, 449 init_keys: str = "", 450 use_docker: bool = False, 451 log_buffer: bool = False, 452 ): 453 """Initialize the environment with the given input and output files. 454 455 Args: 456 input_file (str): the input file path 457 output_file (str): the output file path 458 width (int): the width of the terminal 459 height (int): the height of the terminal 460 use_docker (bool): whether use dockerized executor or local (requiring vim installed) 461 log_buffer (bool): whether to log the editor buffer or not 462 init_keys (str): initial input keys in Vimgolf solution style 463 """ 464 465 self.use_docker = use_docker 466 """whether use dockerized executor or local (requiring vim installed)""" 467 468 self.input_file = input_file 469 """the input file path""" 470 471 self.output_file = output_file 472 """the output file path""" 473 474 self.log_buffer = log_buffer 475 """whether to log the editor buffer or not""" 476 477 self.init_keys = init_keys 478 """initial input keys in Vimgolf solution style""" 479 480 assert os.path.isfile( 481 self.input_file 482 ), f"Input file {self.input_file} does not exist." 483 assert os.path.isfile( 484 self.output_file 485 ), f"Output file {self.output_file} does not exist." 486 487 # TODO: run a modified version of vimgolf local python script writing progress to a jsonl file, which embeds in this script, for easy state inspection and data collection (we can create a temporary directory for cleanup) 488 self.log_directory = tempfile.TemporaryDirectory() 489 """the log directory, where tempfiles stored""" 490 491 self.log_file = os.path.join(self.log_directory.name, "vimgolf.log") 492 """the log file path, used to retrieve progress info of the vimgolf process""" 493 494 self.buffer_file = os.path.join( 495 self.log_directory.name, "vimgolf_editor_buffer" 496 ) 497 """the editor buffer, used to track granual progress""" 498 499 self._container_name = str(uuid.uuid4()) 500 501 if self.use_docker: 502 mountpoint = "/vimgolf_gym_workdir" 503 docker_output_file = os.path.join(mountpoint, "out") 504 docker_input_file = os.path.join(mountpoint, "in") 505 docker_buffer_file = os.path.join(mountpoint, "vimgolf_editor_buffer") 506 docker_log_file = os.path.join(mountpoint, "vimgolf.log") 507 shutil.copy(self.input_file, os.path.join(self.log_directory.name, "in")) 508 shutil.copy(self.output_file, os.path.join(self.log_directory.name, "out")) 509 extra_docker_run_params = [] 510 if self.log_buffer: 511 _prepare_cybergod_vimrc_with_buffer_file(docker_buffer_file) 512 extra_docker_run_params += [ 513 "-v", 514 "%s:%s:ro" 515 % ( 516 _CYBERGOD_VIMGOLF_VIMRC_FILEPATH, 517 "/usr/local/lib/python3.10/dist-packages/vimgolf/vimgolf.vimrc", 518 ), 519 ] 520 self.command = [ 521 "docker", 522 "run", 523 "--rm", 524 "-it", 525 "--name", 526 self._container_name, 527 "-v", 528 "%s:%s" % (self.log_directory.name, mountpoint), 529 *extra_docker_run_params, 530 "--entrypoint", 531 "python3", 532 "agile4im/cybergod_vimgolf_gym", 533 "-m", 534 "vimgolf_gym.vimgolf", 535 "--input_file", 536 docker_input_file, 537 "--output_file", 538 docker_output_file, 539 "--log_file", 540 docker_log_file, 541 ] 542 if self.init_keys: 543 self.command += ["--init_keys", self.init_keys] 544 else: 545 extra_flags = [] 546 if self.log_buffer: 547 extra_flags += ["--buffer_file", self.buffer_file] 548 if self.init_keys: 549 extra_flags += ["--init_keys", self.init_keys] 550 self.command = [ 551 sys.executable, 552 "-m", 553 "vimgolf_gym.vimgolf", 554 "--input_file", 555 self.input_file, 556 "--output_file", 557 self.output_file, 558 "--log_file", 559 self.log_file, 560 *extra_flags, 561 ] 562 563 self._closing = False 564 565 self.command: list[str] 566 """the command passed to underlying terminal executor""" 567 568 self.width = width 569 """terminal width""" 570 self.height = height 571 """terminal height""" 572 self.create_executor_and_log_watcher() 573 574 self.executor: terminal_executor.TerminalExecutor 575 """terminal executor for running vimgolf process""" 576 self.log_watcher: log_parser.VimGolfLogWatcher 577 """log watcher for tracking vimgolf log output""" 578 atexit.register(self.log_directory.cleanup) 579 atexit.register(self._kill_docker_container) 580 581 @property 582 def buffer(self) -> typing.Optional[bytes]: 583 """The editor buffer""" 584 if not self.log_buffer: 585 return None 586 else: 587 if os.path.isfile(self.buffer_file): 588 with open(self.buffer_file, "rb") as f: 589 return f.read() 590 591 def act(self, action: str): 592 """Take an action 593 594 Args: 595 action (str): the action to take 596 """ 597 self.executor.input(action) 598 599 @property 600 def success(self): 601 """Check if the vimgolf challenge has been solved successfully""" 602 return self.log_watcher.success 603 604 def get_best_success_result(self): 605 """ 606 Return the best success result in the log watcher. 607 608 Returns: 609 Optional[VimGolfEnvResult]: The best success result. 610 """ 611 return self.log_watcher.get_best_success_result() 612 613 def get_last_success_result(self): 614 """ 615 Return the last success result in the log watcher. 616 617 Returns: 618 Optional[VimGolfEnvResult]: The last success result 619 """ 620 return self.log_watcher.get_last_success_result() 621 622 @property 623 def results(self): 624 """The results of the vimgolf challenge environment 625 626 Returns: 627 list[VimGolfEnvResult]: The results of the vimgolf challenge environment 628 """ 629 return self.log_watcher.results 630 631 def __enter__(self): 632 return self 633 634 def __exit__(self, exc, value, tb): 635 self.close() 636 637 @property 638 def success_results(self): 639 """The success results of the vimgolf challenge environment 640 641 Returns: 642 list[VimGolfEnvResult]: The success results of the vimgolf challenge environment 643 """ 644 return self.log_watcher.success_results 645 646 def create_executor_and_log_watcher(self): 647 """Create the executor and log watcher""" 648 self.executor = terminal_executor.TerminalExecutor( 649 command=self.command, width=self.width, height=self.height 650 ) 651 if not hasattr(self, "log_watcher"): 652 self.log_watcher = log_parser.VimGolfLogWatcher(self.log_file) 653 # shall we wait the executor be ready 654 # we wait for the 'play(...)' indicator to appear in the log file. 655 while True: 656 if os.path.exists(self.log_file): 657 with open(self.log_file, "r") as f: 658 log_content = f.read() 659 if "play(...)" in log_content: 660 break 661 time.sleep(0.5) 662 663 def reset(self): 664 """Reset the environment""" 665 self.close() 666 self._closing = False 667 self.create_executor_and_log_watcher() 668 669 def render(self): 670 """Render the environment""" 671 screenshot = self.screenshot() 672 # display the screenshot 673 screenshot.show() 674 675 def screenshot(self): 676 """Take a screenshot of the environment 677 678 Returns: 679 PIL.Image.Image: The screenshot 680 """ 681 with tempfile.TemporaryDirectory() as tmpdir: 682 png_tmpfile_path = os.path.join(tmpdir, "screenshot.png") 683 self.executor.screenshot(png_tmpfile_path) 684 image = PIL.Image.open(png_tmpfile_path) 685 return image 686 687 def _kill_docker_container(self): 688 if self.use_docker: 689 subprocess.run( 690 ["docker", "kill", self._container_name], capture_output=True 691 ) 692 693 def verify_keys(self, keys: str): 694 """Verify a solution by its keys 695 696 Args: 697 keys (str): the keys to verify, in Vimgolf style 698 """ 699 assert keys, "Keys cannot be empty" 700 with VimGolfEnv( 701 input_file=self.input_file, 702 output_file=self.output_file, 703 init_keys=keys, 704 use_docker=self.use_docker, 705 log_buffer=True, 706 ) as env: 707 for _ in range(3): 708 success = env.success 709 if success: 710 break 711 time.sleep(1) 712 if not success: 713 buffer = env.buffer 714 with open(self.output_file, "rb") as f: 715 expected_output = f.read() 716 success = buffer == expected_output 717 return success 718 719 def close(self): 720 """Close the environment""" 721 if not self._closing: 722 self._closing = True 723 self.executor.close() 724 self._kill_docker_container() 725 del self.executor 726 if os.path.exists(self.log_file): 727 os.remove(self.log_file) 728 setattr(self, "executor", None) 729 730 def calculate_relative_inverse_score(self, score:int, worst_score:typing.Optional[int] =None): 731 """Calculate the relative inverse score of the given score 732 733 Args: 734 score (int): The score to calculate the relative inverse score of. 735 worst_score (int, optional): The worst score to use. Defaults to None. 736 737 Returns: 738 float: The relative inverse score. 739 """ 740 assert score >= 0, "Score must be non-negative" 741 if worst_score is None: 742 with open(self.output_file, "r") as f: 743 worst_score = len(f.read()) + 10 744 ret = worst_score / score 745 return ret
443 def __init__( 444 self, 445 input_file: str, 446 output_file: str, 447 width: int = 80, 448 height: int = 24, 449 init_keys: str = "", 450 use_docker: bool = False, 451 log_buffer: bool = False, 452 ): 453 """Initialize the environment with the given input and output files. 454 455 Args: 456 input_file (str): the input file path 457 output_file (str): the output file path 458 width (int): the width of the terminal 459 height (int): the height of the terminal 460 use_docker (bool): whether use dockerized executor or local (requiring vim installed) 461 log_buffer (bool): whether to log the editor buffer or not 462 init_keys (str): initial input keys in Vimgolf solution style 463 """ 464 465 self.use_docker = use_docker 466 """whether use dockerized executor or local (requiring vim installed)""" 467 468 self.input_file = input_file 469 """the input file path""" 470 471 self.output_file = output_file 472 """the output file path""" 473 474 self.log_buffer = log_buffer 475 """whether to log the editor buffer or not""" 476 477 self.init_keys = init_keys 478 """initial input keys in Vimgolf solution style""" 479 480 assert os.path.isfile( 481 self.input_file 482 ), f"Input file {self.input_file} does not exist." 483 assert os.path.isfile( 484 self.output_file 485 ), f"Output file {self.output_file} does not exist." 486 487 # TODO: run a modified version of vimgolf local python script writing progress to a jsonl file, which embeds in this script, for easy state inspection and data collection (we can create a temporary directory for cleanup) 488 self.log_directory = tempfile.TemporaryDirectory() 489 """the log directory, where tempfiles stored""" 490 491 self.log_file = os.path.join(self.log_directory.name, "vimgolf.log") 492 """the log file path, used to retrieve progress info of the vimgolf process""" 493 494 self.buffer_file = os.path.join( 495 self.log_directory.name, "vimgolf_editor_buffer" 496 ) 497 """the editor buffer, used to track granual progress""" 498 499 self._container_name = str(uuid.uuid4()) 500 501 if self.use_docker: 502 mountpoint = "/vimgolf_gym_workdir" 503 docker_output_file = os.path.join(mountpoint, "out") 504 docker_input_file = os.path.join(mountpoint, "in") 505 docker_buffer_file = os.path.join(mountpoint, "vimgolf_editor_buffer") 506 docker_log_file = os.path.join(mountpoint, "vimgolf.log") 507 shutil.copy(self.input_file, os.path.join(self.log_directory.name, "in")) 508 shutil.copy(self.output_file, os.path.join(self.log_directory.name, "out")) 509 extra_docker_run_params = [] 510 if self.log_buffer: 511 _prepare_cybergod_vimrc_with_buffer_file(docker_buffer_file) 512 extra_docker_run_params += [ 513 "-v", 514 "%s:%s:ro" 515 % ( 516 _CYBERGOD_VIMGOLF_VIMRC_FILEPATH, 517 "/usr/local/lib/python3.10/dist-packages/vimgolf/vimgolf.vimrc", 518 ), 519 ] 520 self.command = [ 521 "docker", 522 "run", 523 "--rm", 524 "-it", 525 "--name", 526 self._container_name, 527 "-v", 528 "%s:%s" % (self.log_directory.name, mountpoint), 529 *extra_docker_run_params, 530 "--entrypoint", 531 "python3", 532 "agile4im/cybergod_vimgolf_gym", 533 "-m", 534 "vimgolf_gym.vimgolf", 535 "--input_file", 536 docker_input_file, 537 "--output_file", 538 docker_output_file, 539 "--log_file", 540 docker_log_file, 541 ] 542 if self.init_keys: 543 self.command += ["--init_keys", self.init_keys] 544 else: 545 extra_flags = [] 546 if self.log_buffer: 547 extra_flags += ["--buffer_file", self.buffer_file] 548 if self.init_keys: 549 extra_flags += ["--init_keys", self.init_keys] 550 self.command = [ 551 sys.executable, 552 "-m", 553 "vimgolf_gym.vimgolf", 554 "--input_file", 555 self.input_file, 556 "--output_file", 557 self.output_file, 558 "--log_file", 559 self.log_file, 560 *extra_flags, 561 ] 562 563 self._closing = False 564 565 self.command: list[str] 566 """the command passed to underlying terminal executor""" 567 568 self.width = width 569 """terminal width""" 570 self.height = height 571 """terminal height""" 572 self.create_executor_and_log_watcher() 573 574 self.executor: terminal_executor.TerminalExecutor 575 """terminal executor for running vimgolf process""" 576 self.log_watcher: log_parser.VimGolfLogWatcher 577 """log watcher for tracking vimgolf log output""" 578 atexit.register(self.log_directory.cleanup) 579 atexit.register(self._kill_docker_container)
Initialize the environment with the given input and output files.
Arguments:
- input_file (str): the input file path
- output_file (str): the output file path
- width (int): the width of the terminal
- height (int): the height of the terminal
- use_docker (bool): whether use dockerized executor or local (requiring vim installed)
- log_buffer (bool): whether to log the editor buffer or not
- init_keys (str): initial input keys in Vimgolf solution style
581 @property 582 def buffer(self) -> typing.Optional[bytes]: 583 """The editor buffer""" 584 if not self.log_buffer: 585 return None 586 else: 587 if os.path.isfile(self.buffer_file): 588 with open(self.buffer_file, "rb") as f: 589 return f.read()
The editor buffer
591 def act(self, action: str): 592 """Take an action 593 594 Args: 595 action (str): the action to take 596 """ 597 self.executor.input(action)
Take an action
Arguments:
- action (str): the action to take
599 @property 600 def success(self): 601 """Check if the vimgolf challenge has been solved successfully""" 602 return self.log_watcher.success
Check if the vimgolf challenge has been solved successfully
604 def get_best_success_result(self): 605 """ 606 Return the best success result in the log watcher. 607 608 Returns: 609 Optional[VimGolfEnvResult]: The best success result. 610 """ 611 return self.log_watcher.get_best_success_result()
Return the best success result in the log watcher.
Returns:
Optional[VimGolfEnvResult]: The best success result.
613 def get_last_success_result(self): 614 """ 615 Return the last success result in the log watcher. 616 617 Returns: 618 Optional[VimGolfEnvResult]: The last success result 619 """ 620 return self.log_watcher.get_last_success_result()
Return the last success result in the log watcher.
Returns:
Optional[VimGolfEnvResult]: The last success result
622 @property 623 def results(self): 624 """The results of the vimgolf challenge environment 625 626 Returns: 627 list[VimGolfEnvResult]: The results of the vimgolf challenge environment 628 """ 629 return self.log_watcher.results
The results of the vimgolf challenge environment
Returns:
list[VimGolfEnvResult]: The results of the vimgolf challenge environment
637 @property 638 def success_results(self): 639 """The success results of the vimgolf challenge environment 640 641 Returns: 642 list[VimGolfEnvResult]: The success results of the vimgolf challenge environment 643 """ 644 return self.log_watcher.success_results
The success results of the vimgolf challenge environment
Returns:
list[VimGolfEnvResult]: The success results of the vimgolf challenge environment
646 def create_executor_and_log_watcher(self): 647 """Create the executor and log watcher""" 648 self.executor = terminal_executor.TerminalExecutor( 649 command=self.command, width=self.width, height=self.height 650 ) 651 if not hasattr(self, "log_watcher"): 652 self.log_watcher = log_parser.VimGolfLogWatcher(self.log_file) 653 # shall we wait the executor be ready 654 # we wait for the 'play(...)' indicator to appear in the log file. 655 while True: 656 if os.path.exists(self.log_file): 657 with open(self.log_file, "r") as f: 658 log_content = f.read() 659 if "play(...)" in log_content: 660 break 661 time.sleep(0.5)
Create the executor and log watcher
663 def reset(self): 664 """Reset the environment""" 665 self.close() 666 self._closing = False 667 self.create_executor_and_log_watcher()
Reset the environment
669 def render(self): 670 """Render the environment""" 671 screenshot = self.screenshot() 672 # display the screenshot 673 screenshot.show()
Render the environment
675 def screenshot(self): 676 """Take a screenshot of the environment 677 678 Returns: 679 PIL.Image.Image: The screenshot 680 """ 681 with tempfile.TemporaryDirectory() as tmpdir: 682 png_tmpfile_path = os.path.join(tmpdir, "screenshot.png") 683 self.executor.screenshot(png_tmpfile_path) 684 image = PIL.Image.open(png_tmpfile_path) 685 return image
Take a screenshot of the environment
Returns:
PIL.Image.Image: The screenshot
693 def verify_keys(self, keys: str): 694 """Verify a solution by its keys 695 696 Args: 697 keys (str): the keys to verify, in Vimgolf style 698 """ 699 assert keys, "Keys cannot be empty" 700 with VimGolfEnv( 701 input_file=self.input_file, 702 output_file=self.output_file, 703 init_keys=keys, 704 use_docker=self.use_docker, 705 log_buffer=True, 706 ) as env: 707 for _ in range(3): 708 success = env.success 709 if success: 710 break 711 time.sleep(1) 712 if not success: 713 buffer = env.buffer 714 with open(self.output_file, "rb") as f: 715 expected_output = f.read() 716 success = buffer == expected_output 717 return success
Verify a solution by its keys
Arguments:
- keys (str): the keys to verify, in Vimgolf style
719 def close(self): 720 """Close the environment""" 721 if not self._closing: 722 self._closing = True 723 self.executor.close() 724 self._kill_docker_container() 725 del self.executor 726 if os.path.exists(self.log_file): 727 os.remove(self.log_file) 728 setattr(self, "executor", None)
Close the environment
730 def calculate_relative_inverse_score(self, score:int, worst_score:typing.Optional[int] =None): 731 """Calculate the relative inverse score of the given score 732 733 Args: 734 score (int): The score to calculate the relative inverse score of. 735 worst_score (int, optional): The worst score to use. Defaults to None. 736 737 Returns: 738 float: The relative inverse score. 739 """ 740 assert score >= 0, "Score must be non-negative" 741 if worst_score is None: 742 with open(self.output_file, "r") as f: 743 worst_score = len(f.read()) + 10 744 ret = worst_score / score 745 return ret
Calculate the relative inverse score of the given score
Arguments:
- score (int): The score to calculate the relative inverse score of.
- worst_score (int, optional): The worst score to use. Defaults to None.
Returns:
float: The relative inverse score.