fix: merge conflicts

manisha841 · manisha841 · commit c6ea4f9accbe · 2024-05-04T20:01:39.000+05:45
diff --git a/.coveragerc b/.coveragerc
@@ -0,0 +1,2 @@
+[run]
+omit = app.py, test/utils/*
diff --git a/school_center.py b/school_center.py
@@ -40,10 +40,10 @@ def haversine_distance(lat1, lon1, lat2, lon2):
     return distance
 
 
-def centers_within_distance(school: Dict[str, str], centers: Dict[str, str], distance_threshold: float) -> List[Dict[str, any]]:
+def centers_within_distance(school: Dict[str, str], centers: Dict[str, str], distance_threshold: float, relax_threshold: bool) -> List[Dict[str, any]]:
     """
     Return List of centers that are within given distance from school.
-    If there are no centers within given distance return one that is closest
+    relax_threshold: If there are no centers within given distance return one that is closest
     Returned params :
             {'cscode', 'name', 'address', 'capacity', 'lat', 'long', 'distance_km'}
 
@@ -60,33 +60,35 @@ def center_to_dict(c, distance):
     def sort_key(c):
         # intent: sort by preference score DESC then by distance_km ASC
         # leaky abstraction - sorted requires a single numeric value for each element
-        return c['distance_km'] * random.uniform(1, 5) - get_pref(school['scode'], c['cscode'])*100
-
+        return c['distance_km'] * random.uniform(1, 5) - get_pref(school['scode'], c['cscode']) * 100
+    
     school_lat = school.get('lat')
     school_long = school.get('long')
     if len(school_lat) == 0 or len(school_long) == 0:
         return []
 
-    within_distance = []
-    nearest_distance = None
-    nearest_center = None
+    qualifying_centers = []
+    # nearest_distance = None
+    # nearest_center = None
     for c in centers:
+        if school['scode'] == c['cscode'] \
+            or is_allocated(c['cscode'], s['scode']) \
+            or get_pref(school['scode'], c['cscode']) <= PREF_CUTOFF:
+            continue
         distance = haversine_distance(float(school_lat), float(
             school_long), float(c.get('lat')), float(c.get('long')))
-        if school['scode'] == c['cscode']:
-            continue
-        if nearest_center is None or distance < nearest_distance:
-            nearest_center = c
-            nearest_distance = distance
-
-        if distance <= distance_threshold and get_pref(school['scode'], c['cscode']) > PREF_CUTOFF:
-            within_distance.append(center_to_dict(c, distance))
+        # if nearest_center is None or distance < nearest_distance:
+        #     nearest_center = c
+        #     nearest_distance = distance
+        qualifying_centers.append(center_to_dict(c, distance))
 
+    within_distance = [ c for c in qualifying_centers if c['distance_km'] <= distance_threshold ]
     if len(within_distance) > 0:
-        return sorted(within_distance, key=sort_key)
-    else:  # if there are no centers within given  threshold, return one that is closest
-        return [center_to_dict(nearest_center, nearest_distance)]
-
+        return sorted(within_distance, key=sort_key) 
+    elif relax_threshold: # if there are no centers within given threshold, return one that is closest
+        return sorted(qualifying_centers, key=sort_key) 
+    else: 
+        return []
 
 def read_tsv(file_path: str) -> List[Dict[str, str]]:
     """
@@ -271,7 +273,7 @@ def get_output_filename():
 
     for s in schools:
         centers_for_school = centers_within_distance(
-            s, centers, PREF_DISTANCE_THRESHOLD)
+            s, centers, PREF_DISTANCE_THRESHOLD, False)
         to_allot = int(s['count'])
         per_center = calc_per_center(to_allot)
 
@@ -289,8 +291,6 @@ def get_output_filename():
                              c['address'], 
                              c['capacity'], 
                              c['distance_km']])
-            if is_allocated(c['cscode'], s['scode']):
-                continue
             next_allot = min(to_allot, per_center, max(
                 centers_remaining_cap[c['cscode']], MIN_STUDENT_IN_CENTER))
             if to_allot > 0 and next_allot > 0 and centers_remaining_cap[c['cscode']] >= next_allot:
@@ -302,10 +302,8 @@ def get_output_filename():
 
         if to_allot > 0:  # try again with relaxed constraints and more capacity at centers
             expanded_centers = centers_within_distance(
-                s, centers, ABS_DISTANCE_THRESHOLD)
+                s, centers, ABS_DISTANCE_THRESHOLD, True)
             for c in expanded_centers:
-                if is_allocated(c['cscode'], s['scode']):
-                    continue
                 stretched_capacity = math.floor(
                     int(c['capacity']) * STRETCH_CAPACITY_FACTOR + centers_remaining_cap[c['cscode']])
                 next_allot = min(to_allot, max(
@@ -330,7 +328,7 @@ def get_output_filename():
 
         if to_allot > 0:
             remaining += to_allot
-            logger.warn(
+            logger.warning(
                 f"{to_allot}/{s['count']} left for {s['scode']} {s['name-address']} centers: {len(centers_for_school)}")
 
     logger.info("Remaining capacity at each center (remaining_capacity cscode):")
diff --git a/test/__init__.py b/test/__init__.py
diff --git a/test/test_results.py b/test/test_results.py
@@ -0,0 +1,158 @@
+import unittest
+import sys
+import os
+import subprocess
+import warnings
+
+# Add the parent directory to the Python path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from test.utils.custom_tsv_parser import ParseTSVFile
+
+PREF_CUTOFF = -4
+
+
+def get_scode_cscode_id(data):
+    # Create an id with scode and center with sorted values of scode and center
+    school_centers = []
+    for row in data:
+        scode = int(row["scode"])
+        center = int(row["cscode"])
+        sccode_center_id = sorted((scode, center))
+        sccode_center_id = "_".join(map(str, sccode_center_id))
+        school_centers.append(sccode_center_id)
+    return school_centers
+
+
+class TestSchoolCenter(unittest.TestCase):
+    """_Tests to validate the outcome of the output are matching
+    as per the requirements
+
+    Needs the ouptut from the results/school-center.tsv and
+    results/school-center-distance.tsv files to be present
+
+    """
+
+    def setUp(self):
+        self.school_center_file = "results/school-center.tsv"
+        self.school_center_distance_file = "results/school-center-distance.tsv"
+        self.school_center_pref_file = "sample_data/prefs.tsv"
+        schools_tsv = "sample_data/schools_grade12_2081.tsv"
+        centers_tsv = "sample_data/centers_grade12_2081.tsv"
+        cmd = f"python school_center.py {schools_tsv} {centers_tsv} {self.school_center_pref_file}"
+        subprocess.run(cmd, shell=True)
+
+    def tearDown(self):
+        os.remove(self.school_center_file)
+        os.remove(self.school_center_distance_file)
+
+    def test_results_exists(self):
+        """_Test if the application in running which output the results in the
+             results filder_
+
+        Returns:
+            Pass: If the file exists in the results folder
+            Fail: If the file doesnot exists in the results folder
+        """
+        self.assertTrue(os.path.exists(self.school_center_file))
+        self.assertTrue(os.path.exists(self.school_center_distance_file))
+        self.assertTrue(os.path.exists(self.school_center_pref_file))
+
+    def test_scode_student_count_not_more_than_200(self):
+        """_Test if the student count is not more than 200_
+          Test case ID 001:- एक विद्यालयको परिक्षार्थी संख्या हेरी सकभर १००, २०० भन्दा बढी
+                             परीक्षार्थी एकै केन्द्रमा नपर्ने गरी बाँढ्न पर्ने
+
+        Returns:
+            Pass: If the student count is not more than 100
+            Fail: If the student count is more than 100
+        """
+        ptf = ParseTSVFile(self.school_center_file)
+        data = ptf.get_rows()
+        for row in data:
+            student_count = row["allocation"]
+            if int(student_count) > 200:
+                warnings.warn(f"student count is more than 200 for the school {row}")
+
+
+    def test_scode_cscode_not_same(self):
+        """_Test if the output of scode is not equal to cscode_
+          Test case ID :- आफ्नै विद्यालयमा केन्द्र पार्न नहुने
+
+        Returns:
+            Pass: If the scode is not same as cscode
+            Fail: If the scode is same as cscode
+        """
+        scf = ParseTSVFile(self.school_center_file)
+        data = scf.get_rows()
+        failures = []
+        for row in data:
+            scode = row["scode"]
+            cscode = row["cscode"]
+            if scode == cscode:
+                failures.append(f"scode and cscode are same for row {row} {scode}")
+        assert len(failures) == 0, f'{len(failures)} rows failed. {chr(10).join(failures)}'
+
+    def test_no_mutual_centers(self):
+        """_Test if the scode's center is not same as cscode's
+            centre and vice versa_
+          Test case ID :- दुई विद्यालयका परीक्षार्थीको केन्द्र एक अर्कामा पर्न नहुने, अर्थात् कुनै विद्यालयका परीक्षार्थीको केन्द्र परेको विद्यालयका परीक्षार्थीहरूको केन्द्र अघिल्लो विद्यालयमा पार्न नहुने ।
+
+        Returns:
+            Pass: If the scode's center is not same as cscode's center
+            Fail: If the scode's center is same as cscode's center
+        """
+        scf = ParseTSVFile(self.school_center_file)
+        data = scf.get_rows()
+        scodes_centers = []
+
+        scodes_centers = get_scode_cscode_id(data)
+
+        # Check if there are any duplicates id if duplicate then there is a collision between school and center
+        duplicates = [
+            item for item in set(scodes_centers) if scodes_centers.count(item) > 1
+        ]
+
+        self.assertFalse(
+            duplicates,
+            f"Duplicate values found in scode_center_code: {', '.join(duplicates)}",
+        )
+
+    @unittest.skip ("needs review")
+    def test_undesired_cscode_scode_pair(self):
+        """_Test if the schools and the centers are not matched based on the
+        cost preferences defined in the prefs.tsv file_
+          Test case ID :-
+          1 एकै स्वामित्व / व्यवस्थापनको भनी पहिचान भएका केन्द्रमा पार्न नहुने
+          2 विगतमा कुनै विद्यालयको कुनै केन्द्रमा पार्दा समस्या देखिएकोमा केन्द्र दोहोऱ्याउन नहुने
+
+        Returns:
+            Pass: If the schools with undesired scodes are are not paired with its cscodes
+            Fail: If the schools with same management are each other's center
+        """
+
+        scf = ParseTSVFile(self.school_center_file)
+        cpf = ParseTSVFile(self.school_center_pref_file)
+        data_scf = scf.get_rows()
+        data_cpf = cpf.get_rows()
+        for cpf_data in data_cpf:
+            if int(cpf_data["pref"]) < PREF_CUTOFF:
+                data_cpf.remove(cpf_data)
+
+        failures = []
+
+        scodes_centers = get_scode_cscode_id(data_scf)
+
+        undesired_csodes_centers = get_scode_cscode_id(data_cpf)
+        for undesired_cscodes_center in undesired_csodes_centers:
+            if undesired_cscodes_center in scodes_centers:
+                failures.append(
+                    f"Schools with undesired centers {undesired_cscodes_center}"
+                )
+
+        assert len(failures) == 0, f'{len(failures)} rows failed. {chr(10).join(failures)}'
+
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/utils/__init__.py b/test/utils/__init__.py
diff --git a/test/utils/custom_tsv_parser.py b/test/utils/custom_tsv_parser.py
@@ -0,0 +1,50 @@
+import csv
+
+
+class ParseTSVFile:
+    def __init__(self, file_path):
+        self.file_path = file_path
+
+    def read_file(self):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            return reader
+
+    def get_columns(self):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            return reader.fieldnames
+
+    def get_row_count(self):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            return len(list(reader))
+
+    def get_column_count(self):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            return len(reader.fieldnames)
+
+    def get_rows(self):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            data = []
+            for row in reader:
+                data.append(row)
+            return data
+
+    def get_column_data(self, column_name):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            data = []
+            for row in reader:
+                data.append(row[column_name])
+            return data
+
+    def get_row_data(self, row_number):
+        with open(self.file_path, "r", newline="", encoding="utf-8") as file:
+            reader = csv.DictReader(file, delimiter="\t")
+            data = []
+            for row in reader:
+                data.append(row)
+            return data[row_number]

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+[run]`
	`2`	`+omit = app.py, test/utils/*`