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
def make( env_name: str, custom_challenge: Optional[vimgolf_gym.dataclasses.VimGolfCustomChallenge] = None, use_docker: bool = False, log_buffer: bool = False) -> VimGolfEnv:
 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. The custom_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

def list_local_challenge_ids() -> list[str]:
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.

def get_local_challenge_definition(challenge_id: str):
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.
def get_local_challenge_metadata(challenge_id: str):
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.
def get_local_challenge_worst_solution(challenge_id: str):
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.
def get_local_challenge_worst_solution_header(challenge_id: str):
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.
class VimGolfEnv:
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
VimGolfEnv( input_file: str, output_file: str, width: int = 80, height: int = 24, init_keys: str = '', use_docker: bool = False, log_buffer: bool = False)
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
use_docker

whether use dockerized executor or local (requiring vim installed)

input_file

the input file path

output_file

the output file path

log_buffer

whether to log the editor buffer or not

init_keys

initial input keys in Vimgolf solution style

log_directory

the log directory, where tempfiles stored

log_file

the log file path, used to retrieve progress info of the vimgolf process

buffer_file

the editor buffer, used to track granual progress

command: list[str]

the command passed to underlying terminal executor

width

terminal width

height

terminal height

terminal executor for running vimgolf process

log watcher for tracking vimgolf log output

buffer: Optional[bytes]
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

def act(self, action: str):
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
success
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

def get_best_success_result(self):
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.

def get_last_success_result(self):
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

results
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

success_results
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

def create_executor_and_log_watcher(self):
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

def reset(self):
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

def render(self):
669    def render(self):
670        """Render the environment"""
671        screenshot = self.screenshot()
672        # display the screenshot
673        screenshot.show()

Render the environment

def screenshot(self):
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

def verify_keys(self, keys: str):
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
def close(self):
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

def calculate_relative_inverse_score(self, score: int, worst_score: Optional[int] = None):
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.