vimgolf_gym.log_parser

  1"""
  2Log parser for VimGolf executor
  3"""
  4
  5import copy
  6import json
  7import os
  8from abc import ABC, abstractmethod
  9from typing import Type
 10
 11from vimgolf_gym.dataclasses import VimGolfEnvResult
 12
 13
 14class AbstractLogParser(ABC):
 15    def __init__(self): ...
 16    @abstractmethod
 17    def feed_line(self, line: str): ...
 18
 19
 20# We keep track of the log file attributes.
 21# Specifically, the file size.
 22# If the file size changes, we will reparse the log.
 23# In more advanced usage, we could seek to given point and parse from there, avoiding reparsing from the beginning.
 24class LogWatcher:
 25    def __init__(self, log_file: str, parser_class: Type[AbstractLogParser]):
 26        """
 27        Initialize the log watcher with a log file and a parser class.
 28
 29        Args:
 30            log_file (str): The path to the log file.
 31            parser_class (Type[AbstractLogParser]): A class that implements the AbstractLogParser interface.
 32
 33        Attributes:
 34            log_file (str): The path to the log file.
 35            parser_class (Type[AbstractLogParser]): The class of the parser.
 36            parser (AbstractLogParser): An instance of the parser class.
 37            last_filesize (int): The size of the log file when it was last checked.
 38            last_position (int): The last read position in the log file.
 39        """
 40        self.log_file = log_file
 41        self.parser_class = parser_class
 42        self.parser = parser_class()
 43        self.last_filesize = 0
 44        self.last_position = 0  # Track the last read position
 45
 46    def update(self, style="advanced"):
 47        """
 48        Update the log watcher using one of three strategies.
 49
 50        The ``advanced`` strategy is the default. It checks the file size
 51        and only reads new content if the file size has changed. If the file
 52        size is smaller than the last recorded size, it resets the parser and
 53        reads the entire file.
 54
 55        The ``simple`` strategy reads the entire log file every time it is
 56        called. This is simple, but slow for large log files.
 57
 58        The ``naive`` strategy reads the entire log file every time it is
 59        called, but it does not reset the parser. This is faster than the
 60        ``simple`` strategy, but it does not handle the case where the log
 61        file is truncated.
 62
 63        Args:
 64            style (str, optional): The update strategy to use. Must be one of 
 65                ``advanced``, ``simple``, or ``naive``. Defaults to ``advanced``.
 66        """
 67        if style == "advanced":
 68            self.advanced_update()
 69        elif style == "simple":
 70            self.simple_update()
 71        elif style == "naive":
 72            self.naive_update()
 73        else:
 74            raise ValueError(
 75                "Unrecognized update option: %s (should be in advanced, simple, naive)"
 76                % style
 77            )
 78
 79    def simple_update(self):
 80        """
 81        Read the entire log file and reset the parser if the file size has changed.
 82
 83        The ``simple`` strategy reads the entire log file every time it is
 84        called. This is simple, but slow for large log files.
 85        """
 86        current_size = (
 87            os.path.getsize(self.log_file) if os.path.exists(self.log_file) else 0
 88        )
 89        if current_size != self.last_filesize:
 90            self.naive_update()
 91            self.last_filesize = current_size
 92
 93    def naive_update(self):
 94        """
 95        Reset the parser and read the entire log file.
 96
 97        This is slow for large log files, but it is simple and handles
 98        the case where the log file is truncated.
 99
100        """
101        self.parser = self.parser_class()
102        if os.path.exists(self.log_file):
103            with open(self.log_file, "r") as f:
104                for line in f.readlines():
105                    self.parser.feed_line(line)
106        self.last_filesize = (
107            os.path.getsize(self.log_file) if os.path.exists(self.log_file) else 0
108        )
109        self.last_position = self.last_filesize
110
111    def advanced_update(self):
112        """
113        Read only the new content from the log file.
114
115        The ``advanced`` strategy reads only the new content from the log file,
116        starting from the last recorded position. This is fast for large log
117        files, but it requires keeping track of the last recorded position.
118
119        If the file size has decreased, it resets the parser and reads the
120        entire file.
121        """
122        if not os.path.exists(self.log_file):
123            return
124
125        current_size = os.path.getsize(self.log_file)
126
127        # If file size decreased (file was truncated), do a full reset
128        if current_size < self.last_filesize:
129            self.naive_update()
130            return
131
132        # If file size increased, read only the new content
133        if current_size > self.last_filesize:
134            with open(self.log_file, "r") as f:
135                f.seek(self.last_position)
136                while True:
137                    line = f.readline()
138                    if not line:  # End of file
139                        break
140                    self.parser.feed_line(line)
141
142                # Update tracking variables
143                self.last_filesize = current_size
144                self.last_position = f.tell()
145
146
147class VimGolfLogWatcher(LogWatcher):
148    def __init__(self, log_file: str, update_style="advanced"):
149        """
150        Initialize a VimGolfLogWatcher with a log file and an update style.
151
152        The VimGolfLogWatcher is a LogWatcher that is specialized for
153        reading VimGolf logs.
154
155        Args:
156            log_file (str): The path to the log file.
157            update_style (str, optional): The update strategy to use. Must be one of 
158                ``advanced``, ``simple``, or ``naive``. Defaults to ``advanced``.
159
160        Returns:
161            [return_type]: [description of return value]
162        """
163        super().__init__(log_file=log_file, parser_class=VimGolfLogParser)
164        self.parser: VimGolfLogParser
165        self.update_style = update_style
166
167    def default_update(self):
168        """
169        Call the update method with the default update style.
170
171        This method is a convenience wrapper around the update method,
172        calling it with the default update style specified in the
173        constructor.
174
175        See the update method for more details.
176        """
177        self.update(style=self.update_style)
178
179    @property
180    def success(self):
181        """
182        Check if the vimgolf challenge has been solved successfully.
183
184        This property checks if the vimgolf challenge has been solved
185        successfully. It updates the log watcher if necessary before
186        checking the success status.
187
188        Returns:
189            bool: True if the challenge has been solved successfully, False otherwise.
190        """
191        self.default_update()
192        return self.parser.success
193
194    def get_best_success_result(self):
195        """
196        Return the best success result in the log watcher.
197
198        This method returns the best success result in the log watcher,
199        updating the log watcher if necessary before returning the result.
200
201        Returns:
202            VimGolfEnvResult: The best success result in the log watcher, or None if 
203            there is no success result.
204        """
205        self.default_update()
206        return self.parser.get_best_success_result()
207
208    def get_last_success_result(self):
209        """
210        Return the last success result in the log watcher.
211
212        This method returns the last success result in the log watcher,
213        updating the log watcher if necessary before returning the result.
214
215        Returns:
216            VimGolfEnvResult: The last success result in the log watcher, or None if 
217            there is no success result.
218        """
219        self.default_update()
220        return self.parser.get_last_success_result()
221
222    @property
223    def results(self):
224        """
225        The results of the vimgolf challenge environment.
226
227        This property returns the results of the vimgolf challenge environment,
228        updating the log watcher if necessary before returning the results.
229
230        Returns:
231            list[VimGolfEnvResult]: The results of the vimgolf challenge environment, or an empty list if
232            there are no results.
233        """
234        self.default_update()
235        return self.parser.results
236
237    @property
238    def success_results(self):
239        """
240        The successful results of the vimgolf challenge environment.
241
242        This property returns the successful results of the vimgolf challenge
243        environment, updating the log watcher if necessary before returning the
244        results.
245
246        Returns:
247            list[VimGolfEnvResult]: The successful results of the vimgolf challenge environment, or an
248            empty list if there are no successful results.
249        """
250        self.default_update()
251        return self.parser.success_results
252
253
254class VimGolfLogParser(AbstractLogParser):
255    def __init__(self):
256        """
257        Initialize the VimGolfLogParser.
258
259        The VimGolfLogParser is initialized with an empty list of results.
260        """
261        self.results: list[VimGolfEnvResult] = []
262
263    def feed_line(self, line: str):
264        """
265        Feed a line to the parser.
266
267        The line should be a JSON-formatted string. The parser will attempt to
268        parse the line as a JSON object, and if it is a dictionary, it will
269        check if the dictionary has an "event_type" key with value
270        "vimgolf_result". If so, it will attempt to parse the value of the
271        "event_data" key as a VimGolfEnvResult object and append it to the
272        results list.
273
274        If the line is not a valid JSON object, or if the JSON object does not
275        have the correct structure, the line will be ignored.
276
277        Args:
278            line (str): The line of text to feed to the parser.
279        """
280        try:
281            data = json.loads(line.strip())
282            if type(data) == dict:
283                if data.get("event_type", None) == "vimgolf_result":
284                    event_data = data.get("event_data", None)
285                    if type(event_data) == dict:
286                        parsed_result = VimGolfEnvResult.parse_obj(event_data)
287                        self.results.append(parsed_result)
288        except json.JSONDecodeError:
289            ...
290
291    @property
292    def success_results(self):
293        """
294        The successful results of the vimgolf challenge environment.
295
296        This property returns the successful results of the vimgolf challenge
297        environment, which are the results in the results list where the
298        correct attribute is True.
299
300        Returns:
301            list[VimGolfEnvResult]: The successful results of the vimgolf challenge environment, or an
302            empty list if there are no successful results.
303        """
304        return [it for it in self.results if it.correct]
305
306    @property
307    def success(self):
308        """
309        Check if the vimgolf challenge has been solved successfully.
310
311        This property checks if the vimgolf challenge has been solved
312        successfully. It returns True if there are any successful results in
313        the results list, and False otherwise.
314
315        Returns:
316            bool: True if the challenge has been solved successfully, False otherwise.
317        """
318        return len(self.success_results) != 0
319
320    def get_last_success_result(self):
321        """
322        Return the last success result in the log watcher.
323
324        This method returns the last success result in the log watcher,
325        which is the last result in the success_results list.
326
327        Returns:
328            VimGolfEnvResult: The last success result in the log watcher, or None if there is no
329            success result.
330        """
331        success_results = self.success_results
332        if success_results:
333            return success_results[-1]
334
335    def get_best_success_result(self):
336        """Return the result with the lowest score"""
337        success_results = copy.deepcopy(self.success_results)
338        if success_results:
339            success_results.sort(key=lambda x: x.score)
340            return success_results[0]