<jupyter_start><jupyter_text>Fuzzy Framework for Speed Estimation Bio-Bio and Maule Region in central Chile This is a frameworked developed for our study which is published in the ISPRS International Journal of Geo-Information.HERE will be the link to the article once it is published.<jupyter_code>%pylab inline
import numpy as np
import skfuzzy as fuzz
from skfuzzy import control as ctrl # for API example
import psycopg2
import matplotlib.pyplot as plt
import matplotlib.ticker as ticker
import time
from mpl_toolkits.mplot3d import Axes3D  # Required for 3D plotting<jupyter_output><empty_output><jupyter_text>Membership functions for TerrainDepending on the input values, the membership function for terrain can either be approximated using road slope or using the support points per kilometer.Either membership function can be used.<jupyter_code>### With Slope Percentage as Slope

terrain = ctrl.Antecedent(np.arange(0, 101, 0.1), 'terrain')

terrain['level'] = fuzz.trapmf(terrain.universe, [-1, -1, 2, 7])
terrain['rolling'] = fuzz.trapmf(terrain.universe, [2, 7, 7, 20])
terrain['mountainous'] = fuzz.trapmf(terrain.universe, [7, 20, 20, 40])
terrain['steep'] = fuzz.trapmf(terrain.universe, [20, 40, 500, 500])
terrain.view()

terrain.defuzzify_method = 'centroid'
### With Support Points as Slope

terrain = ctrl.Antecedent(np.arange(0, 20, 0.05), 'terrain')

terrain['level'] = fuzz.trapmf(terrain.universe, [-1, -1, 1, 4])
terrain['rolling'] = fuzz.trapmf(terrain.universe, [1, 4, 10, 12])
terrain['mountainous'] = fuzz.trapmf(terrain.universe, [10, 12, 13, 15])
terrain['steep'] = fuzz.trapmf(terrain.universe, [12, 15, 20, 20])
terrain.view()
axes = plt.gca()
axes.set_xlim([0,15])

terrain.defuzzify_method = 'centroid'<jupyter_output><empty_output><jupyter_text>Membership function for link length<jupyter_code>length = ctrl.Antecedent(np.arange(0, 40000, 1), 'length')

length['very short'] = fuzz.trapmf(length.universe, [-1, -1, 200, 500])
length['short'] = fuzz.trapmf(length.universe, [200, 500, 500, 1000])
length['medium'] = fuzz.trapmf(length.universe, [500, 1500, 1500, 4000])
length['long'] = fuzz.trapmf(length.universe, [1500, 4000, 4000, 10000])
length['very long'] = fuzz.trapmf(length.universe, [4000, 10000, 100000, 100000])
length.view()
axes = plt.gca()
axes.set_xlim([0,10000])

length.defuzzify_method = 'centroid'<jupyter_output><empty_output><jupyter_text>Defining the membership functions for the different road classes The membership functions have to be designed using expert knowledge about the study region. Values have to be inserted where 'INSERTVALUE' is written.For reference see the two application examples for the Biobio and Maule regions and for north New South Wales.<jupyter_code>### Membership Funktions for velocity -- different classes
## Motorway
velocity[0]['slow'] = fuzz.trapmf(velocity[0].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[0]['medium'] = fuzz.trapmf(velocity[0].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[0]['fast'] = fuzz.trapmf(velocity[0].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Trunk
velocity[1]['slow'] = fuzz.trapmf(velocity[1].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[1]['medium'] = fuzz.trapmf(velocity[1].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[1]['fast'] = fuzz.trapmf(velocity[1].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Primary paved
velocity[2]['slow'] = fuzz.trapmf(velocity[2].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[2]['medium'] = fuzz.trapmf(velocity[2].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[2]['fast'] = fuzz.trapmf(velocity[2].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Primary unpaved
velocity[3]['slow'] = fuzz.trapmf(velocity[3].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[3]['medium'] = fuzz.trapmf(velocity[3].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[3]['fast'] = fuzz.trapmf(velocity[3].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Secondary paved
velocity[4]['slow'] = fuzz.trapmf(velocity[4].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[4]['medium'] = fuzz.trapmf(velocity[4].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[4]['fast'] = fuzz.trapmf(velocity[4].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Secondary unpaved
velocity[5]['slow'] = fuzz.trapmf(velocity[5].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[5]['medium'] = fuzz.trapmf(velocity[5].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[5]['fast'] = fuzz.trapmf(velocity[5].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Tertiary paved
velocity[6]['slow'] = fuzz.trapmf(velocity[6].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[6]['medium'] = fuzz.trapmf(velocity[6].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[6]['fast'] = fuzz.trapmf(velocity[6].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Tertiary unpaved
velocity[7]['slow'] = fuzz.trapmf(velocity[7].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[7]['medium'] = fuzz.trapmf(velocity[7].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[7]['fast'] = fuzz.trapmf(velocity[7].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Unclassified paved
velocity[8]['slow'] = fuzz.trapmf(velocity[8].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[8]['medium'] = fuzz.trapmf(velocity[8].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[8]['fast'] = fuzz.trapmf(velocity[8].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])

## Unclassified unpaved
velocity[9]['slow'] = fuzz.trapmf(velocity[9].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[9]['medium'] = fuzz.trapmf(velocity[9].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])
velocity[9]['fast'] = fuzz.trapmf(velocity[9].universe, ['INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE', 'INSERTVALUE'])


#plt.subplot(6, 1, 1)
velocity[0].view()
velocity[1].view()
velocity[2].view()
velocity[3].view()
velocity[4].view()
velocity[5].view()
velocity[6].view()
velocity[7].view()
velocity[8].view()
velocity[9].view()<jupyter_output><empty_output><jupyter_text>Definition of the rule base for the Fuzzy Control System<jupyter_code>## Fuzzy Ruleset
ruleset = [[],[],[],[],[],[],[],[],[],[]]
for i in range(10):
    # very long
    ruleset[i].append(ctrl.Rule(terrain['level'] & length['very long'], velocity[i]['fast']))
    ruleset[i].append(ctrl.Rule(terrain['rolling'] & length['very long'], consequent = (velocity[i]['fast'], velocity[i]['medium'])))
    ruleset[i].append(ctrl.Rule(terrain['mountainous'] & length['very long'], velocity[i]['medium']))
    ruleset[i].append(ctrl.Rule(terrain['steep'] & length['very long'], consequent = (velocity[i]['medium'], velocity[i]['slow'])))
    
    # long
    ruleset[i].append(ctrl.Rule(terrain['level'] & length['long'], consequent = (velocity[i]['fast']%0.6, velocity[i]['medium']%0.4)))
    ruleset[i].append(ctrl.Rule(terrain['rolling'] & length['long'], consequent = (velocity[i]['fast']%0.2, velocity[i]['medium']%0.8)))
    ruleset[i].append(ctrl.Rule(terrain['mountainous'] & length['long'], consequent = (velocity[i]['medium']%0.9, velocity[i]['slow']%0.1)))
    ruleset[i].append(ctrl.Rule(terrain['steep'] & length['long'], consequent = (velocity[i]['medium']%0.4, velocity[i]['slow']%0.6)))

    # medium
    ruleset[i].append(ctrl.Rule(terrain['level'] & length['medium'], consequent = (velocity[i]['fast']%0.4, velocity[i]['medium']%0.6)))
    ruleset[i].append(ctrl.Rule(terrain['rolling'] & length['medium'], consequent = (velocity[i]['fast']%0.1, velocity[i]['medium']%0.9)))
    ruleset[i].append(ctrl.Rule(terrain['mountainous'] & length['medium'], consequent = (velocity[i]['medium']%0.6, velocity[i]['slow']%0.4)))
    ruleset[i].append(ctrl.Rule(terrain['steep'] & length['medium'], consequent = (velocity[i]['medium']%0.1, velocity[i]['slow']%0.9)))
    
    # short
    ruleset[i].append(ctrl.Rule(terrain['level'] & length['short'], consequent = (velocity[i]['fast']%0.2, velocity[i]['medium']%0.8)))
    ruleset[i].append(ctrl.Rule(terrain['rolling'] & length['short'], consequent = (velocity[i]['medium']%0.7, velocity[i]['slow']%0.3)))
    ruleset[i].append(ctrl.Rule(terrain['mountainous'] & length['short'], consequent = (velocity[i]['medium']%0.3, velocity[i]['slow']%0.8)))
    ruleset[i].append(ctrl.Rule(terrain['steep'] & length['short'], velocity[i]['slow']))

    # very short
    ruleset[i].append(ctrl.Rule(terrain['level'] & length['very short'], velocity[i]['medium']))
    ruleset[i].append(ctrl.Rule(terrain['rolling'] & length['very short'], consequent = (velocity[i]['medium'], velocity[i]['slow'])))
    ruleset[i].append(ctrl.Rule(terrain['mountainous'] & length['very short'], velocity[i]['slow']))
    ruleset[i].append(ctrl.Rule(terrain['steep'] & length['very short'], velocity[i]['slow']))
velocity_calc = []
for i in range(10):
    velocity_calc.append(ctrl.ControlSystemSimulation(ctrl.ControlSystem(ruleset[i])))<jupyter_output><empty_output><jupyter_text>Database connectionThe roads are administrated in a pgsql database. The Fuzzy-FSE is applied to estimate speed values, then the values are inserted in the database.The FCS need the input parameters road class, road surface, link length and road slope.To input the data from a csv file the code in the next two lines must be altered to import from the csv file.<jupyter_code>## Database parameters
host = 'xxx'
port = 'xxx'
database = 'xxx'
user = 'xxx'
pw = '<PASSWORD>'
region = '_bm'
ways_table = 'ways' + region
## Fetching the ways table
conn = psycopg2.connect("host='" + host + "' port='" + port + "' dbname='" + database + "' user='" + user + "' password='" + pw + "'")
cur = conn.cursor()
cur.execute('SELECT ' + ways_table + '.gid, ' + ## = Unique identifier of the object
            ' ' + ways_table + '.class_id, ' + ## = Road class id
            ' ' + ways_table + '.surf_val, ' + ## = Surface Value
            ' ' + ways_table + '.length_m, ' + ## = Length of the link
            ' ' + ways_table + '.nnsw_slope ' + ## = Road slope
            ' FROM ' + ways_table) 
ways = cur.fetchall()
cur.close()
conn.close()

## Defining the class ids and surface values
id_motorway = 101
id_motorwaylink = 102
id_trunk = 103
id_trunklink = 104
id_primary = 105
id_primarylink = 106
id_secondary = 107
id_secondarylink = 108
id_tertiary = 109
id_tertiarylink = 110
id_unclassified = 111

paved = "paved"
unpaved = "unpaved"
nosurf = "NULL"
## Rulebased fetching of the FCS
tic = time.time()
results = []

for item in ways:
    # Motorways
    if item[1] == id_motorway or item[1] == id_motorwaylink:
        velocity_calc[0].input['terrain'] = item[4]
        velocity_calc[0].input['length'] = item[3]
        velocity_calc[0].compute()
        results.append(item + (velocity_calc[0].output['velocity_mw'],))
    # Trunk
    elif item[1] == id_trunk or item[1] == id_trunklink:
        velocity_calc[1].input['terrain'] = item[4]
        velocity_calc[1].input['length'] = item[3]
        velocity_calc[1].compute()
        results.append(item + (velocity_calc[1].output['velocity_tr'],))
    # Primary paved
    elif (item[1] == id_primary or item[1] == id_primarylink) and (item[2] == paved or item[2]== nosurf): # Default: if no surface value: assumed paved
        velocity_calc[2].input['terrain'] = item[4]
        velocity_calc[2].input['length'] = item[3]
        velocity_calc[2].compute()
        results.append(item + (velocity_calc[2].output['velocity_pr-pv'],))
    # Primary unpaved
    elif (item[1] == id_primary or item[1] == id_primarylink) and item[2] == unpaved: # Default: if no surface value: assumed paved
        velocity_calc[3].input['terrain'] = item[4]
        velocity_calc[3].input['length'] = item[3]
        velocity_calc[3].compute()
        results.append(item + (velocity_calc[3].output['velocity_pr-up'],))
    # Secondary paved
    elif (item[1] == id_secondary or item[1] == id_secondarylink) and (item[2] == paved or item[2]== nosurf): # Default: if no surface value: assumed paved
        velocity_calc[4].input['terrain'] = item[4]
        velocity_calc[4].input['length'] = item[3]
        velocity_calc[4].compute()
        results.append(item + (velocity_calc[4].output['velocity_sc-pv'],))
    # Secondary unpaved
    elif (item[1] == id_secondary or item[1] == id_secondarylink) and item[2] == unpaved: # Default: if no surface value: assumed paved
        velocity_calc[5].input['terrain'] = item[4]
        velocity_calc[5].input['length'] = item[3]
        velocity_calc[5].compute()
        results.append(item + (velocity_calc[5].output['velocity_sc-up'],))
    # Tertiary paved
    elif (item[1] == id_tertiary or item[1] == id_tertiarylink) and (item[2] == paved): # Default: if no surface value: assumed unpaved
        velocity_calc[6].input['terrain'] = item[4]
        velocity_calc[6].input['length'] = item[3]
        velocity_calc[6].compute()
        results.append(item + (velocity_calc[6].output['velocity_ty-pv'],))
    # Tertiary unpaved
    elif (item[1] == id_tertiary or item[1] == id_tertiarylink) and (item[2] == unpaved or item[2] == nosurf): # Default: if no surface value: assumed unpaved
        velocity_calc[7].input['terrain'] = item[4]
        velocity_calc[7].input['length'] = item[3]
        velocity_calc[7].compute()
        results.append(item + (velocity_calc[7].output['velocity_ty-up'],))
    # Unclassified paved
    elif (item[1] == id_unclassified) and (item[2] == paved): # Default: if no surface value: assumed unpaved
        velocity_calc[8].input['terrain'] = item[4]
        velocity_calc[8].input['length'] = item[3]
        velocity_calc[8].compute()
        results.append(item + (velocity_calc[8].output['velocity_uc-pv'],))
    # Unclassified unpaved
    elif (item[1] == id_unclassified) and (item[2] == unpaved or item[2] == nosurf): # Default: if no surface value: assumed unpaved
        velocity_calc[9].input['terrain'] = item[4]
        velocity_calc[9].input['length'] = item[3]
        velocity_calc[9].compute()
        results.append(item + (velocity_calc[9].output['velocity_uc-up'],))

toc = time.time() 
print('Fetching the FCS took ' + str(toc-tic) + ' seconds ('+ str((toc-tic)/60) + ' minutes)')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18544773624021935, 'fraction_numerical': 0.027084865913194676, 'mean_word_length': 5.784936479128857, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '167619', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 5365, 'n_tokens_neox': 4940, 'n_words': 1043, 'path': 'Fuzzy-FSE.ipynb', 'repo_name': 'jojonny17/Fuzzy-Framework-for-Speed-Estimation', 'size': '14946'}