forked from moest-np/center-randomize
-
Notifications
You must be signed in to change notification settings - Fork 0
/
school_center.py
300 lines (255 loc) · 11.7 KB
/
school_center.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
from utils.custom_logger import configure_logging
from typing import Dict, List
import os
import argparse
import logging
import random
import csv
import math
# Parameters
PREF_DISTANCE_THRESHOLD = 2 # Preferred threshold distance in km
ABS_DISTANCE_THRESHOLD = 7 # Absolute threshold distance in km
MIN_STUDENT_IN_CENTER = 10 # Min. no of students from a school to be assigned to a center in normal circumstances
STRETCH_CAPACITY_FACTOR = 0.02 # How much can center capacity be streched if need arises
PREF_CUTOFF = -4 # Do not allocate students with pref score less than cutoff
configure_logging()
logger = logging.getLogger(__name__)
def create_dir(dirPath: str):
"""
Create the given directory if it doesn't exists
- Creates all the directories needed to resolve to the provided directory path
"""
if not os.path.exists(dirPath):
os.makedirs(dirPath)
def haversine_distance(lat1, lon1, lat2, lon2):
"""
Calculate the great circle distance between two points
on the earth specified in decimal degrees
- Reference: https://en.wikipedia.org/wiki/Haversine_formula
"""
# Convert decimal degrees to radians
lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2])
# Haversine formula
dlon = lon2 - lon1
dlat = lat2 - lat1
a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
radius_earth = 6371 # Average Radius of Earth in km
distance = radius_earth * c
return distance
def centers_within_distance(school: Dict[str, str], centers: Dict[str, str], distance_threshold: float) -> 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
Returned params :
{'cscode', 'name', 'address', 'capacity', 'lat', 'long', 'distance_km'}
"""
def center_to_dict(c, distance):
return {'cscode': c['cscode'],
'name': c['name'],
'address': c['address'],
'capacity': c['capacity'],
'lat': c['lat'],
'long': c['long'],
'distance_km': 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
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
for c in centers:
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 == 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 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)]
def read_tsv(file_path: str) -> List[Dict[str, str]]:
"""
Function to read the tsv file for school.tsv and centers.tsv
Return a list of schools/centers as dicts.
"""
data = []
with open(file_path, 'r', newline='', encoding='utf-8') as file:
reader = csv.DictReader(file, delimiter='\t')
for row in reader:
data.append(dict(row))
return data
def read_prefs(file_path: str) -> Dict[str, Dict[str, int]]:
"""
Read the tsv file for pref.tsv
Return a dict of dicts key scode and then cscode
"""
prefs = {}
with open(file_path, 'r', newline='', encoding='utf-8') as file:
reader = csv.DictReader(file, delimiter='\t')
for row in reader:
if prefs.get(row['scode']):
if prefs[row['scode']].get(row['cscode']):
prefs[row['scode']][row['cscode']] += int(row['pref'])
else:
prefs[row['scode']][row['cscode']] = int(row['pref'])
else:
prefs[row['scode']] = {row['cscode']: int(row['pref'])}
return prefs
def get_pref(scode, cscode) -> int:
"""
Return the preference score for the given school and center.
If the school has no preference for the center return 0.
"""
if prefs.get(scode):
if prefs[scode].get(cscode):
return prefs[scode][cscode]
else:
return 0
else:
return 0
def calc_per_center(count: int) -> int:
"""
Return the number of students that can be allocated to a center based on student count.
"""
if count <= 400:
return 100
# elif count <= 900:
# return 200
else:
return 200
def school_sort_key(s):
# intent: allocate students from schools with large students count first
# to avoid excessive fragmentation
return (-1 if int(s['count']) > 500 else 1) * random.uniform(1, 100)
def allocate(scode: str, cscode: str, count: int):
"""
Allocate the given number of students to the given center.
"""
if scode not in allocations:
allocations[scode] = {cscode: count}
elif cscode not in allocations[scode]:
allocations[scode][cscode] = count
else:
allocations[scode][cscode] += count
def is_allocated(scode1: str, scode2: str) -> bool:
"""
Return true if the given school has been allocated to the given center.
"""
return allocations.get(scode1, {}).get(scode2) is not None
parser = argparse.ArgumentParser(
prog='center randomizer',
description='Assigns centers to exam centers to students')
parser.add_argument('schools_tsv', default='schools.tsv',
help="Tab separated (TSV) file containing school details")
parser.add_argument('centers_tsv', default='centers.tsv',
help="Tab separated (TSV) file containing center details")
parser.add_argument('prefs_tsv', default='prefs.tsv',
help="Tab separated (TSV) file containing preference scores")
parser.add_argument(
'-o', '--output', default='school-center.tsv', help='Output file')
parser.add_argument('-s', '--seed', action='store', metavar='SEEDVALUE',
default=None, type=float,
help='Initialization seed for Random Number Generator')
args = parser.parse_args()
random = random.Random(args.seed) #overwrites the random module to use seeded rng
schools = sorted(read_tsv(args.schools_tsv), key= school_sort_key)
centers = read_tsv(args.centers_tsv)
centers_remaining_cap = {c['cscode']: int(c['capacity']) for c in centers}
prefs = read_prefs(args.prefs_tsv)
remaining = 0 # stores count of non allocated students
allocations = {} # to track mutual allocations
OUTPUT_DIR = 'results/'
create_dir(OUTPUT_DIR) # Create the output directory if not exists
with open('{}school-center-distance.tsv'.format(OUTPUT_DIR), 'w', encoding='utf-8') as intermediate_file, \
open(OUTPUT_DIR + args.output, 'w', encoding='utf-8') as a_file:
writer = csv.writer(intermediate_file, delimiter="\t")
writer.writerow(["scode",
"s_count",
"school_name",
"school_lat",
"school_long",
"cscode",
"center_name",
"center_address",
"center_capacity",
"distance_km"])
allocation_file = csv.writer(a_file, delimiter='\t')
allocation_file.writerow(["scode",
"school",
"cscode",
"center",
"center_address",
"allocation",
"distance_km"])
for s in schools:
centers_for_school = centers_within_distance(
s, centers, PREF_DISTANCE_THRESHOLD)
to_allot = int(s['count'])
per_center = calc_per_center(to_allot)
allocated_centers = {}
# per_center = math.ceil(to_allot / min(calc_num_centers(to_allot), len(centers_for_school)))
for c in centers_for_school:
writer.writerow([s['scode'],
s['count'],
s['name-address'],
s['lat'],
s['long'],
c['cscode'],
c['name'],
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:
allocated_centers[c['cscode']] = c
allocate(s['scode'], c['cscode'], next_allot)
# allocation.writerow([s['scode'], s['name-address'], c['cscode'], c['name'], c['address'], next_allot, c['distance_km']])
to_allot -= next_allot
centers_remaining_cap[c['cscode']] -= next_allot
if to_allot > 0: # try again with relaxed constraints and more capacity at centers
expanded_centers = centers_within_distance(
s, centers, ABS_DISTANCE_THRESHOLD)
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(
stretched_capacity, MIN_STUDENT_IN_CENTER))
if to_allot > 0 and next_allot > 0 and stretched_capacity >= next_allot:
allocated_centers[c['cscode']] = c
allocate(s['scode'], c['cscode'], next_allot)
# allocation.writerow([s['scode'], s['name-address'], c['cscode'], c['name'], c['address'], next_allot, c['distance_km']])
to_allot -= next_allot
centers_remaining_cap[c['cscode']] -= next_allot
for c in allocated_centers.values():
allocation_file.writerow([s['scode'],
s['name-address'],
c['cscode'],
c['name'],
c['address'],
allocations[s['scode']][c['cscode']],
c['distance_km']])
if to_allot > 0:
remaining += to_allot
logger.warn(
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):")
logger.info(sorted([(v, k)
for k, v in centers_remaining_cap.items() if v != 0]))
logger.info(
f"Total remaining capacity across all centers: {sum({k:v for k, v in centers_remaining_cap.items() if v != 0}.values())}")
logger.info(f"Students not assigned: {remaining}")