implement line_score functionality

app/data_processing/data_provider.py

@@ -1,8 +1,8 @@
 from typing import List, Tuple
-
+import math
 from traitlets import Float
 from data_processing.get_smallAreaInfo import get_smallAreas
-from data_processing.point_scoring import score_current
+from data_processing.point_scoring import score_current, calc_score_line
 from data_processing.get_station_coverage import get_station_coverage
 from data_processing.aggregate_data import get_feature_df
 import os
@@ -17,7 +17,7 @@ class Data_provider():
         self.df = get_feature_df()
         self.transformer_to_3057 = Transformer.from_crs("EPSG:4326", "EPSG:3057", always_xy=True)
         self.transformer_to_4326 = Transformer.from_crs("EPSG:3057", "EPSG:4326", always_xy=True)
-
+        self.PENALTY_SCALE = 100
     
 
     def get_station_score(self, station_coord, w_density=1, w_income=1, w_age=1, radius=400, EPSG_4326=True):
@@ -83,21 +83,51 @@ class Data_provider():
         """
         return self.transformer_to_4326.transform(*station_coord)
 
-    def total_score(self, stations_coordinates: List[Tuple[Float]], w_density=1, w_income=1, w_age=1, radius=400, EPSG_4326=True):
-        scores = []
+    def line_score(self, stations_coordinates: List[Tuple[Float]], w_density=1, w_income=1, w_age=1, radius=400, EPSG_4326=True):
+        coords_formatted = []
+        coords_scores = []
         for coord in stations_coordinates:
-            score = self.get_station_scores(coord) - get_penalty()
-
-
-
-if __name__ == '__main__':
-    dummy_coord1 = (356250.0, 408250.0)  # EPSG:3057 coordinates
-    dummy_coord2 = (-21.910388, 64.144947)  # EPSG:4326 coordinates
-    dummy_coord3 = (358374.26032876654, 407938.72289760906) # ISN93/Lambert
-    # dummy_coord3 = 
-    backend = Data_provider()
-    print("dummy_coord1: ", backend.get_station_score(dummy_coord1, EPSG_4326=False))
-    print("dummy_coord2: ", backend.get_station_score(dummy_coord2))
-    print("dummy_coord3: ", backend.get_station_score(dummy_coord3, EPSG_4326=False))
-
-
+            coord_formatted = self.__convert_to_3057(coord) if EPSG_4326 else coord
+            coords_formatted.append(coord_formatted)
+            coords_scores.append(self.get_station_score(coord_formatted, w_density, w_income, w_age, radius, EPSG_4326=False))
+        result = calc_score_line(coords_formatted, coords_scores, w_density, w_income, w_age, radius)
+        return result
+
+
+
+
+# dummy_coord1 = (356250.0, 408250.0)  # EPSG:3057 coordinates
+# dummy_coord2 = (-21.910388, 64.144947)  # EPSG:4326 coordinates
+# dummy_coord3 = (358374.26032876654, 407938.72289760906) # ISN93/Lambert
+
+# lane = [
+#     (-2427839.8601560914, 9371544.591676729),
+#     (-2431634.4660833703, 9372554.640815599),
+#     (-2434649.232041, 9374913.131190613),
+#     (-2436330.9382214467, 9375856.527340621),
+#     (-2438828.8871577685, 9378030.440208027),
+#     (-2439956.860815385, 9378055.050542375),
+#     (-2442307.1477456186, 9379548.07749282),
+#     (-2441441.6843210473, 9380909.849326741),
+#     (-2441400.6670971345, 9384909.028658295),
+#     (-2441962.6030647475, 9385105.911333079),
+#     (-2443122.365070897, 9385984.705355423),
+#     (-2441898.000937084, 9387355.706064725),
+#     (-2423704.8112703268, 9387223.425517604),
+#     (-2425747.4690212114, 9387959.684686847),
+#     (-2433768.8998727645, 9383004.29132282),
+#     (-2429552.329254474, 9382532.593247818),
+#     (-2436008.4402984325, 9385453.019590447),
+#     (-2438288.9979480137, 9386006.752113278),
+#     (-2439449.7853847607, 9386252.855456758)]
+
+# transformer = Transformer.from_crs("EPSG:3857", "EPSG:3057", always_xy=True)
+# for i in range(len(lane)):
+#     lane[i] = transformer.transform(*lane[i])
+
+
+# backend = Data_provider()
+# # print("dummy_coord1: ", backend.get_station_score(dummy_coord1, EPSG_4326=False))
+# # print("dummy_coord2: ", backend.get_station_score(dummy_coord2))
+# # print("dummy_coord3: ", backend.get_station_score(dummy_coord3, EPSG_4326=False))
+# print("Line score", backend.line_score(lane, EPSG_4326=False))

app/data_processing/point_scoring.py

@@ -1,5 +1,7 @@
 import csv
 import os 
+from typing import List, Tuple
+import math
 
 
 def score_current(station_coord, df_features, cov_smsv, w_density, w_income, w_age) -> float:
@@ -103,7 +105,6 @@ def get_age_score(age_distribution):
 
     return weighted_sum / total_population
 
-
 def get_income_score(income_distribution):
     """
     Calculate a score based on income distribution.
@@ -142,6 +143,54 @@ def get_income_score(income_distribution):
 
     return weighted_sum / total_population
 
+def calculate_distance(coord1, coord2):
+        """Calculate Euclidean distance between two coordinates in EPSG:3057 format."""
+        x1, y1 = coord1
+        x2, y2 = coord2
+        return math.sqrt((x2 - x1)**2 + (y2 - y1)**2)
+
+def calc_score_line(stations_coordinates: List[Tuple[float]], station_scores: dict[str, list[float]], w_density, w_income, w_age, radius):
+    # print("stations_coords1 ", stations_coordinates)
+    PENALTY_SCALE = 1.0
+    individual_total_scores = [station["total_score"] for station in station_scores]
+    overlap_factors = [0] * len(stations_coordinates)  # Initialize overlap factors for each station
+    total_penalty = 0
+
+    # Aggregate individual scores
+    total_individual_score = sum(individual_total_scores) / len(individual_total_scores) if individual_total_scores else 0
+
+    # Calculate overlap factors
+    for i, coord1 in enumerate(stations_coordinates):
+        for j, coord2 in enumerate(stations_coordinates):
+            if i != j:  # Avoid self-comparison
+                distance = calculate_distance(coord1, coord2)
+                if distance < radius:
+                    # Calculate overlap fraction (inverse of distance within the radius)
+                    overlap_factor = (radius - distance) / radius  # Normalize overlap to [0, 1]
+                    overlap_factors[i] += overlap_factor
+                    overlap_factors[j] += overlap_factor
+
+    # Scale down individual scores based on overlap factors
+    adjusted_scores = [
+        max(0, score * (1 - PENALTY_SCALE * min(1, overlap_factors[i])))  # Cap scaling at 100%
+        for i, score in enumerate(individual_total_scores)
+    ]
+
+    # Final aggregated score
+    final_score = sum(adjusted_scores) / len(adjusted_scores) if adjusted_scores else 0
+
+    # Return detailed results
+    result = {
+        "individual_scores": individual_total_scores,
+        "adjusted_scores": adjusted_scores,
+        "overlap_factors": overlap_factors,
+        "total_individual_score": total_individual_score,
+        "final_score": final_score
+    }
+    return result
+
+
+
 
 if __name__ == '__main__':
     filename = os.path.join('output.csv')