ensure attractiveness is always an int #25
| @@ -63,7 +63,7 @@ def new_trip(preferences: Preferences, start: tuple[float, float], end: tuple[fl | ||||
|     refined_tour = refiner.refine_optimization(landmarks, base_tour, preferences.max_time_minute, preferences.detour_tolerance_minute) | ||||
|  | ||||
|     linked_tour = LinkedLandmarks(refined_tour) | ||||
|     # upon creation of the trip, persistence of both the trip and its landmarks is ensured. Ca | ||||
|     # upon creation of the trip, persistence of both the trip and its landmarks is ensured | ||||
|     trip = Trip.from_linked_landmarks(linked_tour, cache_client) | ||||
|     return trip | ||||
|  | ||||
| @@ -84,4 +84,4 @@ def get_landmark(landmark_uuid: str) -> Landmark: | ||||
|         landmark = cache_client.get(f"landmark_{landmark_uuid}") | ||||
|         return landmark | ||||
|     except KeyError: | ||||
|         raise HTTPException(status_code=404, detail="Landmark not found") | ||||
|         raise HTTPException(status_code=404, detail="Landmark not found") | ||||
|   | ||||
| @@ -1,6 +1,6 @@ | ||||
| city_bbox_side: 7500 #m | ||||
| radius_close_to: 50 | ||||
| church_coeff: 0.75 | ||||
| church_coeff: 0.5 | ||||
| nature_coeff: 1.25 | ||||
| overall_coeff: 10 | ||||
| tag_exponent: 1.15 | ||||
|   | ||||
| @@ -14,26 +14,22 @@ class Landmark(BaseModel) : | ||||
|     osm_id : int | ||||
|     attractiveness : int | ||||
|     n_tags : int | ||||
|     image_url : Optional[str] = None                            # TODO future | ||||
|     image_url : Optional[str] = None | ||||
|     website_url : Optional[str] = None | ||||
|     wikipedia_url : Optional[str] = None | ||||
|     description : Optional[str] = None                          # TODO future | ||||
|     duration : Optional[int] = 0                                # TODO future | ||||
|     duration : Optional[int] = 0 | ||||
|     name_en : Optional[str] = None | ||||
|  | ||||
|     # Unique ID of a given landmark | ||||
|     uuid: str = Field(default_factory=uuid4)                    # TODO implement this ASAP | ||||
|     uuid: str = Field(default_factory=uuid4) | ||||
|      | ||||
|     # Additional properties depending on specific tour | ||||
|     must_do : Optional[bool] = False | ||||
|     must_avoid : Optional[bool] = False | ||||
|     is_secondary : Optional[bool] = False                       # TODO future    | ||||
|      | ||||
|     time_to_reach_next : Optional[int] = 0                      # TODO fix this in existing code | ||||
|     next_uuid : Optional[str] = None                            # TODO implement this ASAP | ||||
|  | ||||
|     def __hash__(self) -> int: | ||||
|         return self.uuid.int | ||||
|     time_to_reach_next : Optional[int] = 0 | ||||
|     next_uuid : Optional[str] = None | ||||
|      | ||||
|     def __str__(self) -> str: | ||||
|         time_to_next_str = f", time_to_next={self.time_to_reach_next}" if self.time_to_reach_next else "" | ||||
| @@ -42,3 +38,15 @@ class Landmark(BaseModel) : | ||||
|         if self.type in ["start", "finish", "nature", "shopping"] : type_str += '\t ' | ||||
|         return f'Landmark{type_str}: [{self.name} @{self.location}, score={self.attractiveness}{time_to_next_str}{is_secondary_str}]' | ||||
|      | ||||
|     def distance(self, value: 'Landmark') -> float: | ||||
|         return (self.location[0] - value.location[0])**2 + (self.location[1] - value.location[1])**2 | ||||
|  | ||||
|     def __hash__(self) -> int: | ||||
|         return hash(self.name) | ||||
|  | ||||
|     def __eq__(self, value: 'Landmark') -> bool: | ||||
|         # eq and hash must be consistent | ||||
|         # in particular, if two objects are equal, their hash must be equal | ||||
|         # uuid and osm_id are just shortcuts to avoid comparing all the properties | ||||
|         # if they are equal, we know that the name is also equal and in turn the hash is equal | ||||
|         return self.uuid == value.uuid or self.osm_id == value.osm_id or (self.name == value.name and self.distance(value) < 0.001) | ||||
|   | ||||
| @@ -27,4 +27,4 @@ class Trip(BaseModel): | ||||
|         # for landmark in landmarks: | ||||
|         #     cache_client.set(f"landmark_{landmark.uuid}", landmark, expire=3600) | ||||
|  | ||||
|         return trip | ||||
|         return trip | ||||
|   | ||||
| @@ -16,10 +16,8 @@ def get_time(p1: tuple[float, float], p2: tuple[float, float]) -> int: | ||||
|     Args: | ||||
|         p1 (Tuple[float, float]): Coordinates of the starting location. | ||||
|         p2 (Tuple[float, float]): Coordinates of the destination. | ||||
|         detour (float): Detour factor affecting the distance. | ||||
|         speed (float): Walking speed in kilometers per hour. | ||||
|  | ||||
|     Returns: | ||||
|         Returns: | ||||
|         int: Time to travel from p1 to p2 in minutes. | ||||
|     """ | ||||
|  | ||||
|   | ||||
| @@ -1,15 +1,11 @@ | ||||
| import math as m | ||||
| import math | ||||
| import yaml | ||||
| import logging | ||||
|  | ||||
| from OSMPythonTools.overpass import Overpass, overpassQueryBuilder | ||||
| from OSMPythonTools.cachingStrategy import CachingStrategy, JSON | ||||
| from pywikibot import ItemPage, Site | ||||
| from pywikibot import config | ||||
| config.put_throttle = 0 | ||||
| config.maxlag = 0 | ||||
|  | ||||
| from structs.preferences import Preferences, Preference | ||||
| from structs.preferences import Preferences | ||||
| from structs.landmark import Landmark | ||||
| from .take_most_important import take_most_important | ||||
| import constants | ||||
| @@ -46,7 +42,7 @@ class LandmarkManager: | ||||
|             self.viewpoint_bonus = parameters['viewpoint_bonus'] | ||||
|             self.pay_bonus = parameters['pay_bonus'] | ||||
|             self.N_important = parameters['N_important'] | ||||
|              | ||||
|  | ||||
|         with constants.OPTIMIZER_PARAMETERS_PATH.open('r') as f: | ||||
|             parameters = yaml.safe_load(f) | ||||
|             self.walking_speed = parameters['average_walking_speed'] | ||||
| @@ -69,87 +65,42 @@ class LandmarkManager: | ||||
|         preferences (Preferences): The user's preference settings that influence the landmark selection. | ||||
|  | ||||
|         Returns: | ||||
|             tuple[list[Landmark], list[Landmark]]: | ||||
|                 - A list of all existing landmarks. | ||||
|                 - A list of the most important landmarks based on the user's preferences. | ||||
|         tuple[list[Landmark], list[Landmark]]: | ||||
|         - A list of all existing landmarks. | ||||
|         - A list of the most important landmarks based on the user's preferences. | ||||
|         """ | ||||
|  | ||||
|         max_walk_dist = (preferences.max_time_minute/2)/60*self.walking_speed*1000/self.detour_factor | ||||
|         reachable_bbox_side = min(max_walk_dist, self.max_bbox_side) | ||||
|  | ||||
|         L = [] | ||||
|         # use set to avoid duplicates, this requires some __methods__ to be set in Landmark | ||||
|         all_landmarks = set() | ||||
|  | ||||
|         bbox = self.create_bbox(center_coordinates, reachable_bbox_side) | ||||
|         # list for sightseeing | ||||
|         if preferences.sightseeing.score != 0: | ||||
|             score_function = lambda score: int(score*10*preferences.sightseeing.score/5)   # self.count_elements_close_to(loc) + | ||||
|             L1 = self.fetch_landmarks(bbox, self.amenity_selectors['sightseeing'], preferences.sightseeing.type, score_function) | ||||
|             L += L1 | ||||
|             score_function = lambda score: score * 10 * preferences.sightseeing.score / 5 | ||||
|             current_landmarks = self.fetch_landmarks(bbox, self.amenity_selectors['sightseeing'], preferences.sightseeing.type, score_function) | ||||
|             all_landmarks.update(current_landmarks) | ||||
|  | ||||
|         # list for nature | ||||
|         if preferences.nature.score != 0: | ||||
|             score_function = lambda score: int(score*10*self.nature_coeff*preferences.nature.score/5)   # self.count_elements_close_to(loc) + | ||||
|             L2 = self.fetch_landmarks(bbox, self.amenity_selectors['nature'], preferences.nature.type, score_function) | ||||
|             L += L2 | ||||
|             score_function = lambda score: score * 10 * self.nature_coeff * preferences.nature.score / 5 | ||||
|             current_landmarks = self.fetch_landmarks(bbox, self.amenity_selectors['nature'], preferences.nature.type, score_function) | ||||
|             all_landmarks.update(current_landmarks) | ||||
|  | ||||
|         # list for shopping | ||||
|         if preferences.shopping.score != 0: | ||||
|             score_function = lambda score: int(score*10*preferences.shopping.score/5)   # self.count_elements_close_to(loc) + | ||||
|             L3 = self.fetch_landmarks(bbox, self.amenity_selectors['shopping'], preferences.shopping.type, score_function) | ||||
|             L += L3 | ||||
|             score_function = lambda score: score * 10 * preferences.shopping.score / 5 | ||||
|             current_landmarks = self.fetch_landmarks(bbox, self.amenity_selectors['shopping'], preferences.shopping.type, score_function) | ||||
|             all_landmarks.update(current_landmarks) | ||||
|  | ||||
|  | ||||
|         L = self.remove_duplicates(L) | ||||
|         # self.correct_score(L, preferences) | ||||
|         landmarks_constrained = take_most_important(all_landmarks, self.N_important) | ||||
|         self.logger.info(f'Generated {len(all_landmarks)} landmarks around {center_coordinates}, and constrained to {len(landmarks_constrained)} most important ones.') | ||||
|  | ||||
|         L_constrained = take_most_important(L, self.N_important) | ||||
|         self.logger.info(f'Generated {len(L)} landmarks around {center_coordinates}, and constrained to {len(L_constrained)} most important ones.') | ||||
|         return all_landmarks, landmarks_constrained | ||||
|  | ||||
|         return L, L_constrained | ||||
|  | ||||
|  | ||||
|     def remove_duplicates(self, landmarks: list[Landmark]) -> list[Landmark]: | ||||
|         """ | ||||
|         Removes duplicate landmarks based on their names from the given list. Only retains the landmark with highest score | ||||
|  | ||||
|         Parameters: | ||||
|         landmarks (list[Landmark]): A list of Landmark objects. | ||||
|  | ||||
|         Returns: | ||||
|         list[Landmark]: A list of unique Landmark objects based on their names. | ||||
|         """ | ||||
|  | ||||
|         L_clean = [] | ||||
|         names = [] | ||||
|  | ||||
|         for landmark in landmarks: | ||||
|             if landmark.name in names:  | ||||
|                 continue   | ||||
|             else: | ||||
|                 names.append(landmark.name) | ||||
|                 L_clean.append(landmark) | ||||
|          | ||||
|         return L_clean | ||||
|          | ||||
|  | ||||
|     def correct_score(self, landmarks: list[Landmark], preferences: Preferences) -> None: | ||||
|         """ | ||||
|         Adjust the attractiveness score of each landmark in the list based on user preferences. | ||||
|  | ||||
|         This method updates the attractiveness of each landmark by scaling it according to the user's preference score. | ||||
|         The score adjustment is computed using a simple linear transformation based on the preference score. | ||||
|  | ||||
|         Args: | ||||
|             landmarks (list[Landmark]): A list of landmarks whose scores need to be corrected. | ||||
|             preferences (Preferences): The user's preference settings that influence the attractiveness score adjustment. | ||||
|         """ | ||||
|  | ||||
|         score_dict = { | ||||
|             preferences.sightseeing.type: preferences.sightseeing.score, | ||||
|             preferences.nature.type: preferences.nature.score, | ||||
|             preferences.shopping.type: preferences.shopping.score | ||||
|         } | ||||
|         for landmark in landmarks: | ||||
|             landmark.attractiveness = int(landmark.attractiveness * score_dict[landmark.type] / 5)         | ||||
|  | ||||
|  | ||||
|     def count_elements_close_to(self, coordinates: tuple[float, float]) -> int: | ||||
| @@ -172,7 +123,7 @@ class LandmarkManager: | ||||
|  | ||||
|         radius = self.radius_close_to | ||||
|  | ||||
|         alpha = (180*radius) / (6371000*m.pi) | ||||
|         alpha = (180 * radius) / (6371000 * math.pi) | ||||
|         bbox = {'latLower':lat-alpha,'lonLower':lon-alpha,'latHigher':lat+alpha,'lonHigher': lon+alpha} | ||||
|  | ||||
|         # Build the query to find elements within the radius | ||||
| @@ -216,7 +167,7 @@ class LandmarkManager: | ||||
|  | ||||
|         # Convert distance to degrees | ||||
|         lat_diff = half_side_length_km / 111  # 1 degree latitude is approximately 111 km | ||||
|         lon_diff = half_side_length_km / (111 * m.cos(m.radians(lat)))  # Adjust for longitude based on latitude | ||||
|         lon_diff = half_side_length_km / (111 * math.cos(math.radians(lat)))  # Adjust for longitude based on latitude | ||||
|  | ||||
|         # Calculate bbox | ||||
|         min_lat = lat - lat_diff | ||||
| @@ -265,22 +216,18 @@ class LandmarkManager: | ||||
|                 result = self.overpass.query(query) | ||||
|             except Exception as e: | ||||
|                 self.logger.error(f"Error fetching landmarks: {e}") | ||||
|                 return | ||||
|              | ||||
|                 continue | ||||
|  | ||||
|             for elem in result.elements(): | ||||
|  | ||||
|                 name = elem.tag('name')                             # Add name | ||||
|                 location = (elem.centerLat(), elem.centerLon())     # Add coordinates (lat, lon) | ||||
|                 name = elem.tag('name') | ||||
|                 location = (elem.centerLat(), elem.centerLon()) | ||||
|  | ||||
|                 # TODO: exclude these from the get go | ||||
|                 # skip if unprecise location | ||||
|                 if name is None or location[0] is None: | ||||
|                     continue | ||||
|  | ||||
|                 # skip if unused | ||||
|                 # if 'disused:leisure' in elem.tags().keys(): | ||||
|                 #     continue | ||||
|                  | ||||
|                 # skip if part of another building | ||||
|                 if 'building:part' in elem.tags().keys() and elem.tag('building:part') == 'yes': | ||||
|                     continue | ||||
| @@ -291,7 +238,6 @@ class LandmarkManager: | ||||
|                 n_tags = len(elem.tags().keys())    # Add number of tags | ||||
|                 score = n_tags**self.tag_exponent   # Add score | ||||
|                 website_url = None | ||||
|                 wikpedia_url = None | ||||
|                 image_url = None | ||||
|                 name_en = None | ||||
|  | ||||
| @@ -299,22 +245,17 @@ class LandmarkManager: | ||||
|                 skip = False | ||||
|                 for tag in elem.tags().keys(): | ||||
|                     if "pay" in tag: | ||||
|                         score += self.pay_bonus             # discard payment options for tags | ||||
|                         # payment options are a good sign | ||||
|                         score += self.pay_bonus | ||||
|  | ||||
|                     if "disused" in tag: | ||||
|                         skip = True             # skip disused amenities | ||||
|                         # skip disused amenities | ||||
|                         skip = True | ||||
|                         break | ||||
|  | ||||
|                     if "wiki" in tag: | ||||
|                         score += self.wikipedia_bonus             # wikipedia entries count more | ||||
|                          | ||||
|                     # if tag == "wikidata": | ||||
|                     #     Q = elem.tag('wikidata') | ||||
|                     #     site = Site("wikidata", "wikidata") | ||||
|                     #     item = ItemPage(site, Q) | ||||
|                     #     item.get() | ||||
|                     #     n_languages = len(item.labels) | ||||
|                     #     n_tags += n_languages/10 | ||||
|                         # wikipedia entries count more | ||||
|                         score += self.wikipedia_bonus | ||||
|  | ||||
|                     if "viewpoint" in tag: | ||||
|                         score += self.viewpoint_bonus | ||||
| @@ -335,47 +276,43 @@ class LandmarkManager: | ||||
|                         if tag == "building" and elem.tag('building') in ['retail', 'supermarket', 'parking']: | ||||
|                             skip = True | ||||
|                             break | ||||
|                      | ||||
|                     # Get additional information | ||||
|                     # if tag == 'wikipedia' : | ||||
|                     #     wikpedia_url = elem.tag('wikipedia') | ||||
|                     if tag in ['website', 'contact:website'] : | ||||
|  | ||||
|                     if tag in ['website', 'contact:website']: | ||||
|                         website_url = elem.tag(tag) | ||||
|                     if tag == 'image' : | ||||
|                     if tag == 'image': | ||||
|                         image_url = elem.tag('image') | ||||
|                     if tag =='name:en' : | ||||
|                     if tag =='name:en': | ||||
|                         name_en = elem.tag('name:en') | ||||
|  | ||||
|                 if skip: | ||||
|                     continue | ||||
|  | ||||
|                 score = score_function(score) | ||||
|                 if "place_of_worship" in elem.tags().values() : | ||||
|                     score = int(score*self.church_coeff) | ||||
|                 if "place_of_worship" in elem.tags().values(): | ||||
|                     score = score * self.church_coeff | ||||
|                     duration = 15 | ||||
|                  | ||||
|                 elif "museum" in elem.tags().values() : | ||||
|                     score = int(score*self.church_coeff) | ||||
|                 elif "museum" in elem.tags().values(): | ||||
|                     score = score * self.church_coeff | ||||
|                     duration = 60 | ||||
|                  | ||||
|                 else :  | ||||
|                 else: | ||||
|                     duration = 5 | ||||
|  | ||||
|                 # Generate the landmark and append it to the list | ||||
|                 # finally create our own landmark object | ||||
|                 landmark = Landmark( | ||||
|                     name=name, | ||||
|                     type=elem_type, | ||||
|                     location=location, | ||||
|                     osm_type=osm_type, | ||||
|                     osm_id=osm_id, | ||||
|                     attractiveness=score, | ||||
|                     must_do=False, | ||||
|                     n_tags=int(n_tags), | ||||
|                     duration = duration,  | ||||
|                     name_en=name_en, | ||||
|                     image_url=image_url, | ||||
|                     # wikipedia_url=wikpedia_url, | ||||
|                     website_url=website_url | ||||
|                     name = name, | ||||
|                     type = elem_type, | ||||
|                     location = location, | ||||
|                     osm_type = osm_type, | ||||
|                     osm_id = osm_id, | ||||
|                     attractiveness = int(score), | ||||
|                     must_do = False, | ||||
|                     n_tags = int(n_tags), | ||||
|                     duration = int(duration), | ||||
|                     name_en = name_en, | ||||
|                     image_url = image_url, | ||||
|                     website_url = website_url | ||||
|                 ) | ||||
|                 return_list.append(landmark) | ||||
|          | ||||
|   | ||||
| @@ -1,38 +1,16 @@ | ||||
| from structs.landmark import Landmark | ||||
|  | ||||
| def take_most_important(landmarks: list[Landmark], N_important) -> list[Landmark] : | ||||
|     L = len(landmarks) | ||||
|     L_copy = [] | ||||
|     L_clean = [] | ||||
|     scores = [0]*len(landmarks) | ||||
|     names = [] | ||||
|     name_id = {} | ||||
| def take_most_important(landmarks: list[Landmark], n_important) -> list[Landmark]: | ||||
|     """ | ||||
|     Given a list of landmarks, return the n_important most important landmarks | ||||
|     Parameters: | ||||
|     landmarks: list[Landmark] - list of landmarks | ||||
|     n_important: int - number of most important landmarks to return | ||||
|     Returns: | ||||
|     list[Landmark] - list of the n_important most important landmarks | ||||
|     """ | ||||
|  | ||||
|     for i, elem in enumerate(landmarks) : | ||||
|         if elem.name not in names : | ||||
|             names.append(elem.name) | ||||
|             name_id[elem.name] = [i] | ||||
|             L_copy.append(elem) | ||||
|         else : | ||||
|             name_id[elem.name] += [i] | ||||
|             scores = [] | ||||
|             for j in name_id[elem.name] : | ||||
|                 scores.append(L[j].attractiveness) | ||||
|             best_id = max(range(len(scores)), key=scores.__getitem__) | ||||
|             t = name_id[elem.name][best_id] | ||||
|             if t == i : | ||||
|                 for old in L_copy : | ||||
|                     if old.name == elem.name : | ||||
|                         old.attractiveness = L[t].attractiveness | ||||
|      | ||||
|     scores = [0]*len(L_copy) | ||||
|     for i, elem in enumerate(L_copy) : | ||||
|         scores[i] = elem.attractiveness | ||||
|     # Sort landmarks by attractiveness (descending) | ||||
|     landmarks.sort(key=lambda x: x.attractiveness, reverse=True) | ||||
|  | ||||
|     res = sorted(range(len(scores)), key = lambda sub: scores[sub])[-(N_important-L):] | ||||
|  | ||||
|     for i, elem in enumerate(L_copy) : | ||||
|         if i in res : | ||||
|             L_clean.append(elem) | ||||
|  | ||||
|     return L_clean | ||||
|     return landmarks[:n_important] | ||||
|   | ||||
		Reference in New Issue
	
	Block a user