Spaces:
Sleeping
Sleeping
| # Copyright 2020 The HuggingFace Datasets Authors and the current dataset script contributor. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| """TODO: Add a description here.""" | |
| import evaluate | |
| import datasets | |
| import ast | |
| import re | |
| import heapq | |
| # TODO: Add BibTeX citation | |
| _CITATION = """\ | |
| @InProceedings{huggingface:module, | |
| title = {A great new module}, | |
| authors={huggingface, Inc.}, | |
| year={2020} | |
| } | |
| """ | |
| # TODO: Add description of the module here | |
| _DESCRIPTION = """\ | |
| This metrics aims to evaluate path planning abilities of LMs. | |
| """ | |
| # TODO: Add description of the arguments of the module here | |
| _KWARGS_DESCRIPTION = """ | |
| Calculates how good the path predicted by the model is. | |
| Args: | |
| predictions: list of predictions to score. Each predictions | |
| should be a string containing coordinates in the format (x, y). | |
| references: list of reference for each prediction. Each | |
| reference should be a string with tokens separated by spaces. | |
| Returns: | |
| compliance_ratio: 1 if the model is able to produce a well formated python list, | |
| feasible_ratio: A path is feasible if it doesn'collide with obstacles, get out of the grid and doesn't move diagonally. This metric is the number of feasible paths over the total number of reachable paths. | |
| success_ratio: A path is successful if it is feasible and goes from the starting point to the end(s) point(s). This metric is the number of successful paths over the total number of reachable paths. | |
| optimal_ratio: A path is optimal if it is successful and its length is equal or shorter to the golden one. This metric is the number of optimal paths over the total number of reachable paths. | |
| distance: Sum of the distance from the last position predicted and the objectives for a feasible path not successful. | |
| unreachable_acc: Number of times the model was able to detect an unreachable path over the total of unreachable paths. -1 if all the goals are reachable. | |
| Examples: | |
| >>> my_new_module = evaluate.load("rfr2003/path_planning_evaluate") | |
| >>> results = my_new_module.compute(generations=['[(0,0), (0,1), (1,1)]', '[(0,0), (1,0), (1,1)]', '[(0,0), (1,0), (1,1), (0,1)]', '(0,0'], golds=[[(0,0), (0,1), (1,1)], [(0,0), (0,1), (1,1)], [(0,0), (0,1)], []], obstacles=[[(1,0)], [(1,0)], [], []], ends=[[(1,1)], [(1,1)], [(0,1)], [(0,1)]], n=[2, 2, 2, 2]) | |
| >>> print(results) | |
| {'compliance_ratio': 0.75, 'success_ratio': 0.6666666666666666,'optimal_ratio': 0.3333333333333333, 'feasible_ratio': 0.6666666666666666, 'distance': 0, 'unreachable_acc': 1.0} | |
| """ | |
| class Path_Planning_evaluate(evaluate.Metric): | |
| """TODO: Short description of my evaluation module.""" | |
| def _info(self): | |
| # TODO: Specifies the evaluate.EvaluationModuleInfo object | |
| return evaluate.MetricInfo( | |
| # This is the description that will appear on the modules page. | |
| module_type="metric", | |
| description=_DESCRIPTION, | |
| citation=_CITATION, | |
| inputs_description=_KWARGS_DESCRIPTION, | |
| # This defines the format of each prediction and reference | |
| features=datasets.Features({ | |
| 'generations': datasets.Value('string'), | |
| 'golds': datasets.Sequence(datasets.Sequence(datasets.Value('int64'))), | |
| 'obstacles': datasets.Sequence(datasets.Sequence(datasets.Value('int64'))), | |
| 'ends': datasets.Sequence(datasets.Sequence(datasets.Value('int64'))), | |
| 'n': datasets.Value('int64'), | |
| }), | |
| ) | |
| def _download_and_prepare(self, dl_manager): | |
| """Optional: download external resources useful to compute the scores""" | |
| # TODO: Download external resources if needed | |
| pass | |
| def _a_star(self, n, obs, start, end): | |
| obs = set(obs) | |
| actions = [(-1, 0), (1, 0), (0, -1), (0, 1)] | |
| def heuristic(position): | |
| return abs(position[0] - end[0]) + abs(position[1] - end[1]) | |
| visited = set() | |
| heap = [] | |
| heapq.heappush(heap, (0, start, [])) | |
| while heap: | |
| cost, current, path = heapq.heappop(heap) | |
| if current == end: | |
| return len(path) | |
| if current in visited: | |
| continue | |
| visited.add(current) | |
| for action in actions: | |
| neighbor = (current[0] + action[0], current[1] + action[1]) | |
| if 0 <= neighbor[0] < n and 0 <= neighbor[1] < n: | |
| if neighbor not in obs: | |
| new_cost = cost + 1 | |
| new_path = path + [current] | |
| heapq.heappush(heap, (new_cost + heuristic(neighbor), neighbor, new_path)) | |
| return 100 | |
| def _calculate_path_dist(self, n, obs, start, unvisited): | |
| dist = 0 | |
| for end in unvisited: | |
| dist += self._a_star(n, obs, start, end) | |
| return dist | |
| def _evaluate_path(self, gen_path, gold_path, ends, obs, n): | |
| metrics = {'correct': False, 'optimal': False, 'detect_unreachable': False, 'feasible': False, 'is_unreachable': False, 'distance': 0} | |
| if len(gold_path) == 0: | |
| metrics['is_unreachable'] = True | |
| if len(gen_path) == 0: | |
| metrics['detect_unreachable'] = True | |
| return metrics | |
| else: | |
| if len(gen_path) == 0: | |
| return metrics | |
| start = gold_path[0] | |
| obst_set = set(obs) | |
| for i, (x, y) in enumerate(gen_path): | |
| if not (0 <= x < n and 0 <= y < n): | |
| return metrics #not feasible, correct or optimal | |
| if (x, y) in obst_set: | |
| return metrics #not feasible, correct or optimal | |
| if i > 0 and (abs(x-gen_path[i][0]) + abs(y-gen_path[i][1]) > 1): | |
| return metrics #not feasible, correct or optimal | |
| metrics['feasible'] = True | |
| unvisited = set() | |
| if gen_path[0] == start and gen_path[-1] in ends: | |
| metrics['correct'] = True | |
| for e in ends: | |
| if e not in gen_path: | |
| metrics['correct'] = False | |
| unvisited.add(e) | |
| if metrics['correct']: | |
| if len(gen_path) <= len(gold_path): | |
| metrics['optimal'] = True | |
| else: | |
| metrics['distance'] = self._calculate_path_dist(n, obs, gen_path[-1], unvisited) | |
| return metrics | |
| def _compute(self, generations, golds, obstacles, ends, n): | |
| assert len(generations) == len(golds) == len(obstacles) == len(ends) == len(n) | |
| assert isinstance(golds, list) | |
| correct, total_reach, total_unreach = 0, 0, 0 | |
| pattern = r"\(\s*(\d*)\s*,\s*(\d*)\s*\)" | |
| metrics = {'compliance': 0, 'correct': 0, 'optimal': 0, 'n_detect_unreachable': 0, 'feasible': 0, 'distance': 0} | |
| for gen, gold, _obs, _ends, _n in zip(generations, golds, obstacles, ends, n): | |
| try: | |
| gen_path = ast.literal_eval(gen) | |
| for i in range(len(gen_path)): | |
| gen_path[i] = (int(gen_path[i][0]), int(gen_path[i][1])) | |
| metrics['compliance'] += 1 | |
| except: | |
| matches = re.findall(pattern, gen) | |
| gen_path = [(int(x), int(y)) for x, y in matches] | |
| gold_path = [(int(p[0]), int(p[1])) for p in gold] | |
| _obs = [(int(p[0]), int(p[1])) for p in _obs] | |
| _ends = [(int(p[0]), int(p[1])) for p in _ends] | |
| m = self._evaluate_path(gen_path, gold_path, _ends, _obs, _n) | |
| if m['is_unreachable']: | |
| total_unreach += 1 | |
| metrics['n_detect_unreachable'] += m['detect_unreachable'] | |
| else: | |
| for k in ['correct', 'optimal', 'feasible']: | |
| metrics[k] += int(m[k]) | |
| if m['feasible'] and not m['correct']: | |
| metrics['distance'] += m['distance'] | |
| total_reach += 1 | |
| diff = metrics['feasible']-metrics['correct'] | |
| r_metrics = {} | |
| r_metrics.update({ | |
| 'compliance_ratio': metrics['compliance']/(total_reach+total_unreach), | |
| 'success_ratio': metrics['correct']/total_reach, | |
| 'optimal_ratio': metrics['optimal']/total_reach, | |
| 'feasible_ratio': metrics['feasible']/total_reach, | |
| 'distance': metrics['distance']/diff if diff > 0 else metrics['distance'], | |
| 'unreachable_acc': metrics['n_detect_unreachable']/total_unreach if total_unreach > 0 else -1, | |
| }) | |
| return r_metrics |