# Solving an algorithm to develop a redundancy index

This solves an algorithm that we have. The script is used by an ArcGIS tool as well as another tool that takes this script and cycles through roughly 6,000 bridges.

Note: I didn't separate the code intentionally into configs and testing code for you to review the whole thing in one screen.

"""
This file solves the algorithm to develop a redundancy index
"""
import doctest
import unittest as unittest

# CONSTANTS

# DEBUG VARIABLE
# Runs testing if set to True
DEBUG = False

MAJOR_EXPECTED_LIFE = 75
STANDARD_EXPECETD_LIFE = 50

UNIT_COST_BRIDGE_OVER_WATER_WITHOUT_PIER = 4500
UNIT_COST_BRIDGE_OVER_WATER_WITH_PIER = 5500

BRIDGE_TYPOGRAPHY_ALWAYS_REDUNDANT = ('NOT IN USE')
BRIDGE_TYPOGRAPHY_SHOW_BLOCKER = ('COVERED',
'HERITAGE',
'NO DETOUR',
'LIFELINE',
'LANDMARK')

'WATER WITH PIER': 5500,
'WATER WITHOUT PIER': 4500}

CURRENT_YEAR = 2013

'COLLECTOR': 4,
'LOCAL NUMBERED': 2,
'LOCAL NAMED': 1}

YEARS_SINCE_LAST_REHAB = 10

# TODO change the following to include all CS that buses go on
BUS_ROUTES = (1., 2., 7., 15., 16.,
8.001, 8.002,
11.001, 11.002, 11.003, 11.004, 11.005,
11.006, 11.007, 11.008, 11.009)

# The following are Constant Assumptions that are made by the initial model:
MILEAGE_COST_PER_KM = 0.5
ASSUMED_HOURLY_WAGE_PER_HOUR = 25
DETOUR_DEFAULT_SPEED = 60

# the following are the weighting for each subindex of the redundancy index
B_OVER_C_WEIGHTING = 30
BCI_WEIGHTING = 25
AGE_INDEX_WEIGHTING = 20
BUS_ROUTE_INDEX_WEIGHTING = 10
RECENT_REHAB_INDEX_WEIGHTING = 5

# End of Constants Section

def route_class(route_number):
"""
returns the road classification based on its number and control section
roads >= 200 --> Local numbered
named --> local named
roads >= 100 and < 200: collector
"""
if type(route_number) is str:
return 'local named'
elif type(route_number) is int:
if 1 <= route_number < 100:
return 'arterial'
elif 100 <= route_number < 200:
return 'collector'
elif route_number >= 200:
return 'local numbered'

def structure_class(route_classification):
"""
if the road that the bridge is on is a 'collector' or 'arterial'
then it's considered major.
The bridge is considered 'standard' in all other cases
"""
return 'major'
else:
return 'standard'

def expected_life(structure_class):
"""
if structure class of the bridge is major, then the expected life is 75
else, the expected life is 50
"""
if structure_class.upper() == 'MAJOR':
return MAJOR_EXPECTED_LIFE
else:
return STANDARD_EXPECETD_LIFE

def potentially_redundant(bridge_typography):
"""
NOTE: In the algorithm, this is called "initial_criteria
if the bridge is 'NOT IN USE' then it's redundant
If the bridge is covered, heritage, no detour, lifeline, or landmark then
it is NOT REDUNDANT
All Other bridges should be considered for redundancy
"""
if bridge_typography.upper() in BRIDGE_TYPOGRAPHY_ALWAYS_REDUNDANT:
return 'redundant'
elif bridge_typography.upper() in BRIDGE_TYPOGRAPHY_SHOW_BLOCKER:
return 'not redundant'
else:
return "potentially redundant"

def unit_cost(below_bridge):
"""
if below the bridge is a roadway, then unit cost is 3000
if below the bridge is a waterway with no piers, then the unit cost is
4500
if below the bridge is waterway with piers, then the unit cost is
500
"""
return BRIDGE_BELOW_TYPE[below_bridge.upper()]

assumed_hourly_wage, detour_speed, deck_area, unit_cost):
"""
Equation developed by the algorithm writers to return the benefit
over cost of a bridge
"""
return detour_distance * AADT * 365 * life * \
(float(mileage_cost) +
(assumed_hourly_wage / float(detour_speed))) / \
(deck_area * float(unit_cost))

def age(year_built):
"""
return how old the bridge is based on current year
"""
# TODO what happens if current year is < year built?
# TODO how does M& T define bridges with no ages?
return CURRENT_YEAR - year_built

"""
return True if there is a Load Rating
"""
# TODO test for load_rating == None

def bus_route(route, control_section=0):
"""
if route 1, 2, 7, 15, 16, 11 (CS 001->009), 8 (CS 001-002)
"""
if control_section and route + (float(control_section) / 1000) in BUS_ROUTES:
return True
elif control_section == 0 and route in BUS_ROUTES:
return True
else:
return False

def BCR_index(BCR, max_BCR, weighting):
"""
returns the first index that the algorithm needs.
BCR = Benefit over cost ratio of bridge
max_BCR = max BCR in the system (highest BCR)

"""
return float(BCR) / (float(max_BCR) * float(weighting))

def BCI_index(BCI, weighting):
"""
Returns the second index that the algorithm needs
"""
return BCI * (float(weighting) / 100)

def age_index(expected_life, age, weighting):
"""
Third index that the algorithm requires
age_index = [()expected life - age) / expected_life] * weighting
max_age_index = weighting and min_age_index = 0
"""
# NOTE I can use numpy for age_index MAX and MIN but I don't want to
#      import a whole library jut for one small equation
age_index = ((float(expected_life) - float(age))
/ float(expected_life)) * float(weighting)
if 0 < age_index >= weighting:
return weighting
elif age_index < 0:
return 0
else:
return age_index

"""
fourth index that the algorithm requires
Road class index, arterial =5, collector = 5, local named = 2,
local numbered = 1
Note: this classification doesn't make much sense but it is what the
algorithm is recommending
Note 2: The reason there is no weighting variable here as it is built
into the result
"""

"""
fifth index that the algorithm requires
if there is a load rating, then assign maximum weighting
if there is no load rating, then the rating is 0
"""
return weighting
else:
return 0

def bus_route_index(bus_route, weighting=0):
"""
Sixth index that the algorithm needed
if the road is a bus route, then assign maximum weighting
if the road isn't a bus route, then rating is 0
"""
if bus_route:
return weighting
else:
return 0

def recent_rehab_index(rehab_year, weighting=0):
"""
seventh index that the model needs
if a recent rehab occurred in the previous 10 years then assign
maximum weighting
else assign 0
"""
if CURRENT_YEAR - rehab_year <= YEARS_SINCE_LAST_REHAB:
return weighting
else:
return 0

bus_route_index, recent_rehab_index):
"""
redundancy index = 100 - sum of all seven indicies
"""
return 100 - (BCR + BCI + age_index + road_class_index + load_rating_index +
bus_route_index + recent_rehab_index)

def main(input_data):
# Step 1: Calculate bridge Benefit over cost
bridge_route_class = route_class(input_data['route'])
bridge_structure_class = structure_class(bridge_route_class)
bridge_expected_life = expected_life(bridge_structure_class)
bridge_unit_cost = unit_cost(input_data['below bridge'])
bridge_b_over_c = benefit_over_cost(input_data['detour_distance'],
bridge_expected_life,
MILEAGE_COST_PER_KM,
ASSUMED_HOURLY_WAGE_PER_HOUR,
DETOUR_DEFAULT_SPEED,
input_data['area'],
bridge_unit_cost)

# Step 2: calculate if bridge is potentially redundant
bridige_potentially_redundant = potentially_redundant(
input_data['initial criteria'])

# Step 3: Calculate redundancy index
bridge_bcr = BCR_index(bridge_b_over_c,
input_data['max BCR'],
B_OVER_C_WEIGHTING)

bridge_bci = BCI_index(input_data['BCI'],
BCI_WEIGHTING)

bridge_age = age(input_data['year built'])
bridge_age_index = age_index(bridge_expected_life,
bridge_age,
AGE_INDEX_WEIGHTING)

on_bus_route = bus_route(input_data['route'])
bridge_bus_route_index = bus_route_index(on_bus_route,
BUS_ROUTE_INDEX_WEIGHTING)

bridge_recent_rehab_index = recent_rehab_index(input_data['rehab year'],
RECENT_REHAB_INDEX_WEIGHTING)

bridge_redundancy_index = redundancy_index(bridge_bcr,
bridge_bci,
bridge_age_index,
bridge_bus_route_index,
bridge_recent_rehab_index)

return bridige_potentially_redundant, bridge_b_over_c, bridge_redundancy_index
#"bridge potentially redundant: %s, b/c: %s, and redundancy index: %s" % \

# tests

class TestBridgeEliminationAlgorithm(unittest.TestCase):
maxDiff = None

def test_structure_class(self):
self.assertEqual(structure_class('arterial'), 'major')
self.assertEqual(structure_class('colLector'), 'major')
self.assertEqual(structure_class('naTIOnal'), 'major')
self.assertEqual(structure_class('banana'), 'standard')

def test_expected_life(self):
self.assertEqual(expected_life('majOr'), 75)
self.assertEqual(expected_life('StandArd'), 50)

def test_potentially_redundant(self):
self.assertEqual(potentially_redundant('not IN use'), 'redundant')
self.assertEqual(potentially_redundant('cOVered'), 'not redundant')
self.assertEqual(potentially_redundant('HerITage'), 'not redundant')
self.assertEqual(potentially_redundant('LANDmark'), 'not redundant')
self.assertEqual(potentially_redundant('no Detour'), 'not redundant')
self.assertEqual(potentially_redundant('lifeLine'), 'not redundant')
self.assertEqual(
potentially_redundant('banana'), 'potentially redundant')

def test_unit_cost(self):
self.assertEqual(unit_cost('water with pier'), 5500)
self.assertEqual(unit_cost('water without pier'), 4500)

def test_benefit_over_cost(self):
self.assertAlmostEqual(
benefit_over_cost(10, 500, 20, 0.5, 16, 70, 150, 3000),
59.095, 3)

def test_age(self):
self.assertEqual(age(2000), 13)

def test_route_class(self):
self.assertEqual(route_class(1), 'arterial')
self.assertEqual(route_class(50), 'arterial')
self.assertEqual(route_class(99), 'arterial')
self.assertEqual(route_class(100), 'collector')
self.assertEqual(route_class(150), 'collector')
self.assertEqual(route_class(199), 'collector')
self.assertEqual(route_class(200), 'local numbered')
self.assertEqual(route_class(650), 'local numbered')
self.assertEqual(route_class(9052412), 'local numbered')
self.assertEqual(route_class("banana"), 'local named')

def test_bus_routes(self):
self.assertEqual(bus_route(1), True)
self.assertEqual(bus_route(2), True)
self.assertEqual(bus_route(7), True)
self.assertEqual(bus_route(15), True)
self.assertEqual(bus_route(16), True)
self.assertEqual(bus_route(17, 1), False)
self.assertEqual(bus_route(17), False)
self.assertEqual(bus_route(11, 1), True)
self.assertEqual(bus_route(11, 2), True)
self.assertEqual(bus_route(11, 3), True)
self.assertEqual(bus_route(11, 89), False)
self.assertEqual(bus_route(8, 1), True)
self.assertEqual(bus_route(8, 5), False)

def test_BCR_index(self):
self.assertAlmostEqual(BCR_index(50, 85, 30), 0.0196, 4)

def test_BCI_index(self):
self.assertEqual(BCI_index(35.4, 60), 21.24)

def test_age_index(self):
self.assertEqual(age_index(60, 6, 5), 4.5)
self.assertEqual(age_index(30, 15, 0.85), 0.425)
self.assertEqual(age_index(30, 15, -1), 0)
self.assertEqual(age_index(30, 15, 75), 37.5)

def test_bus_route_index(self):
self.assertEqual(bus_route_index('5000', .5), .5)
self.assertEqual(bus_route_index(True, .8), .8)
self.assertEqual(bus_route_index(False), 0)
self.assertEqual(bus_route_index(''), False)

def test_recent_rehab_index(self):
self.assertEqual(recent_rehab_index(2005, 45), 45)
self.assertEqual(recent_rehab_index(1998, 45), 0)
self.assertEqual(recent_rehab_index(2013, 45), 45)
self.assertEqual(recent_rehab_index(2003, 45), 45)

if __name__ == '__main__':

if DEBUG:
TestBridgeEliminationAlgorithm)
unittest.TextTestRunner(verbosity=2).run(suite_bridge_elim)
doctest.testmod()
else:
input_data = {'id': 3514,
'alpha_id': "M444",
'year built': 1935,
'route': 134,
'area': 800,
'initial criteria': '',
'detour_distance': 30,
'BCI': 43,
'max BCR': 30,
'rehab year': 1998
}
print main(input_data)

• if 0 < age_index >= weighting: - shouldn't it be just if age_index >= weighting:? Sep 12, 2013 at 21:09

Although people are often advised to break down their code into functions more, my main comment would be that you have lots of functions not doing very much. This might be worthwhile in some cases if there is a possibility of expanding the code and re-using those functions. As it is it just makes the code more complicated and harder to read and probably debug.

e.g. you find the route class of the inputted route, then check its structure class, but you only use the latter to check the expected life, which is a very simple function of the structure class.

In particular there is not much point in having a function that simply looks something up in a dictionary.

You store a lot of data in constants that I would tend to put together with the logic in the functions, unless you are likely to change the constants without having to change any other part of the functions. You are also passing global constants as arguments to functions, which seems pointless.

You define a weight for road_class_index but never apply it; is this a bug?

I would apply weightings to each of the indices in the main loop rather than within the functions that calculate each index. This is more transparent and you don't have to pass the weightings to the functions, and you can do it easily using something like sum(weight * index for weight, index in zip(weights, indices)).

Your note concerning max and min seems to be confused - these are built-in python functions so you don't need numpy for that.

I don't see much call for OOP in this programme, except that it might be useful to have a simple class Route which takes in the route data and stores some characteristics of the route.

I would tend to remove a lot of the simpler functions and write something like this:

class Route:
""" Classifies a route based on its number """
def __init__(self, route_number):
if type(route_number) is str:
self.type = 'local named'
self.index = 1
elif type(route_number) is int:
if 1 <= route_number < 100:
self.type = 'arterial'
self.index = 5
elif 100 <= route_number < 200:
self.type = 'collector'
self.index = 4
elif route_number >= 200:
self.type = 'local numbered'
self.index = 2
if self.type in ('collector', 'arterial'):
self.life = 75
else:
self.life = 50
self.bus = route_number in BUS_ROUTES

def get_potentially_redundant(bridge_typography):
"""
NOTE: In the algorithm, this is called "initial_criteria
if the bridge is 'NOT IN USE' then it's redundant
If the bridge is covered, heritage, no detour, lifeline, or landmark then
it is NOT REDUNDANT
All Other bridges should be considered for redundancy
"""
if bridge_typography.upper() in BRIDGE_TYPOGRAPHY_ALWAYS_REDUNDANT:
return 'redundant'
elif bridge_typography.upper() in BRIDGE_TYPOGRAPHY_SHOW_BLOCKER:
return 'not redundant'
else:
return "potentially redundant"

def get_b_over_c(detour_distance, AADT, life, deck_area, below_bridge):
"""
Equation developed by the algorithm writers to return the benefit
over cost of a bridge
"""
unit_cost = 3000
elif below_bridge.lower() == 'water with pier':
unit_cost = 5500
elif below_bridge.lower() == 'water without pier':
unit_cost = 4500
return (detour_distance * AADT * 365 * life *
(float(MILEAGE_COST_PER_KM) +
(ASSUMED_HOURLY_WAGE_PER_HOUR / float(DETOUR_DEFAULT_SPEED))) /
(deck_area * float(unit_cost)))

def age_index(route, year_built):
"""
Third index that the algorithm requires
age_index = [()expected life - age) / expected_life] * weighting
max_age_index = weighting and min_age_index = 0
"""
age = CURRENT_YEAR - year_built
index = float(route.life - age) / route.life
return max(min(index, 1), 0)

def main(input_data):
# Step 1: Calculate bridge Benefit over cost
route = Route(input_data['route'])
b_over_c = get_b_over_c(input_data['detour_distance'],
route.life,
input_data['area'],
input_data['below_bridge'])

# Step 2: calculate if bridge is potentially redundant
potentially_redundant = get_potentially_redundant(input_data['initial criteria'])

# Step 3: Calculate redundancy index
indices = {
'bcr': b_over_c / float(input_data['max BCR']),
'bci': float(input_data['BCI']) / 100,
'route': route.index,
'age': age_index(route, float(input_data['year built'])),
'bus': route.bus,
'recent rehab': CURRENT_YEAR - rehab_year <= YEARS_SINCE_LAST_REHAB
}
weights = {
'bcr': 1.0 / 30,
'bci': 25,
'age': 20,
'route': 5,
'bus': 10,
'recent rehab': 5
}
redundancy_index = 100 - sum(indices[k] * weights[k] for k in indices.keys())
return potentially_redundant, b_over_c, redundancy_index


With all those bridge_... variables, how about moving the logic into a class representing a bridge? There is much to be done with this code but the basic restructuring is done.

EDIT: Refactored the code somewhat

"""
This file solves the algorithm to develop a redundancy index
"""

# CONSTANTS
BRIDGE_BELOW_TYPE = {'ROADWAY': 3000, 'WATER WITH PIER': 5500, 'WATER WITHOUT PIER': 4500}
CURRENT_YEAR = 2013
YEARS_SINCE_LAST_REHAB = 10

# TODO change the following to include all CS that buses go on
BUS_ROUTES = (
1., 2., 7., 15., 16., 8.001, 8.002, 11.001, 11.002,
11.003, 11.004, 11.005, 11.006, 11.007, 11.008, 11.009
)

# The following are Constant Assumptions that are made by the initial model:
MILEAGE_COST_PER_KM = 0.5
ASSUMED_HOURLY_WAGE = 25.0
DETOUR_DEFAULT_SPEED = 60.0

# the following are the weighting for each subindex of the redundancy index
B_OVER_C_WEIGHTING = 30.0
BCI_WEIGHTING = 0.25
AGE_INDEX_WEIGHTING = 20
BUS_ROUTE_INDEX_WEIGHTING = 10
RECENT_REHAB_INDEX_WEIGHTING = 5

class Bridge(object):
def __init__(self, **kwargs):
self.id = kwargs['id']
self.route = kwargs['route']
self.detour_distance = kwargs['detour_distance']
self.deck_area = kwargs['area']
self.initial_criteria = kwargs['initial criteria'].upper()
self.max_bcr = kwargs['max BCR']
self.bci = kwargs['BCI']
self.rehab_year = kwargs['rehab year']
self.unit_cost = BRIDGE_BELOW_TYPE[kwargs['below bridge'].upper()]
self.age = CURRENT_YEAR - kwargs['year built']

if isinstance(self.route, basestring):
self.route_class = 'LOCAL NAMED'
self.structure_class = 'standard'
self.expected_life = 50
elif isinstance(self.route, int):
if self.route < 100:
self.route_class = 'ARTERIAL'
self.structure_class = 'major'
self.expected_life = 75
elif self.route < 200:
self.route_class = 'COLLECTOR'
self.structure_class = 'major'
self.expected_life = 75
else:
self.route_class = 'LOCAL NUMBERED'
self.structure_class = 'standard'
self.expected_life = 50

remaining_life = max(0, self.expected_life - self.age)
self.age_index = min(max(0, (remaining_life / self.expected_life)), 1)

if self.initial_criteria in ('NOT IN USE'):
self.potentially_redundant = 'redundant'
elif self.initial_criteria in ('COVERED', 'HERITAGE', 'NO DETOUR', 'LIFELINE', 'LANDMARK'):
self.potentially_redundant = 'not redundant'
else:
self.potentially_redundant = "potentially redundant"

self.recent_rehab = (CURRENT_YEAR - self.rehab_year <= YEARS_SINCE_LAST_REHAB)

"""if route 1, 2, 7, 15, 16, 11 (CS 001->009), 8 (CS 001-002)"""
control_section = 0
self.on_bus_route = (control_section and self.route + (float(control_section) / 1000) in BUS_ROUTES) or (control_section == 0 and self.route in BUS_ROUTES)
self.cost = self.deck_area * self.unit_cost

def benefit(self):
"""Equation developed by the algorithm writers to return the benefit of a bridge"""
return self.detour_distance * self.AADT * 365 * self.expected_life * (MILEAGE_COST_PER_KM + (ASSUMED_HOURLY_WAGE / DETOUR_DEFAULT_SPEED))

def benefit_over_cost(self):
return self.benefit() / self.cost

def bcr_index(self):
"""
BCR = Benefit over cost ratio of bridge
max_BCR = max BCR in the system (highest BCR)
"""
return self.benefit_over_cost() / (self.max_bcr * B_OVER_C_WEIGHTING)

def redundancy_index(self):
"""redundancy index = 100 - weighted sum of all seven indicies"""
return 100 - (
self.bcr_index() +
self.bci * BCI_WEIGHTING +
self.age_index * AGE_INDEX_WEIGHTING +
self.on_bus_route * BUS_ROUTE_INDEX_WEIGHTING +
self.recent_rehab * RECENT_REHAB_INDEX_WEIGHTING
)

def main(input_data):
for bridge in input_data:
bridge = Bridge(**bridge)
print (bridge.potentially_redundant, bridge.benefit_over_cost(), bridge.redundancy_index())

if __name__ == '__main__':
input_data = [{
'id': 3514,
'alpha_id': "M444",
'year built': 1935,
'route': 134,
'area': 800,
'initial criteria': '',
'detour_distance': 30,