...
| Code Block | ||
|---|---|---|
| ||
import csv as _csv
from datetime import datetime
import Libraries.ChartFunctions as _cf
import Libraries.SobekWaterFlowFunctions as _wff
import Libraries.StandardFunctions as _st
import Libraries.NetworkFunctions as _nf
import Libraries.MapFunctions as _mf
from Libraries.NetworkFunctions import *
from DelftTools.Hydro.Structures import BridgeFrictionType
def CreateNetworkFromBranchShapeFile(branchShapeFile):
"""Creates a network using the provided shapefile (with line geometries)."""
# create network
network = _nf.HydroNetwork(Name = "Network")
network.CoordinateSystem = _mf.GetShapeFileCoordinateSystem(branchShapeFile)
number = 1
for feature in _mf.GetShapeFileFeatures(branchShapeFile):
firstCoordinate = feature.Geometry.Coordinates[0]
lastCoordinate = feature.Geometry.Coordinates[feature.Geometry.Coordinates.Length -1]
# create nodes
startNode = _nf.HydroNode(Name= feature.Attributes["Source"] ,Geometry = _mf.CreatePointGeometry(firstCoordinate.X, firstCoordinate.Y))
endNode = _nf.HydroNode(Name= feature.Attributes["Target"], Geometry = _mf.CreatePointGeometry(lastCoordinate.X, lastCoordinate.Y))
# create channel
channel = _nf.Channel(startNode, endNode, Name = feature.Attributes["Name"])
channel.Geometry = feature.Geometry
# add channel and nodes to network
network.Nodes.AddRange([startNode, endNode])
network.Branches.Add(channel)
number += 1
_nf.MergeNodesWithSameGeometry(network)
return network
def _CreateCsvLookup(profileCsvFile):
lookup = {}
with open(profileCsvFile) as csvfile:
lines = _csv.reader(csvfile, delimiter=',')
dataRows = None
name = None
for line in lines:
if (len(line) == 0): # empty row
continue
if (len(line) == 1): # new definition
if (name != None and dataRows != None):
lookup[name] = dataRows # add definition to lookup
name = line[0]
dataRows = []
continue
dataRows.append([float(s) for s in line])
# commit last row
lookup[name] = dataRows
return lookup
def AddBranchObjectsFromShapeFile(branchObjectType, shapeFileName, network):
for feature in _mf.GetShapeFileFeatures(shapeFileName):
name = feature.Attributes ["Name"]
branchName = feature.Attributes ["Branch"]
chainage = feature.Attributes ["Chainage"]
longName = feature.Attributes ["LongName"]
branch = _st.GetItemByName(network.Branches, branchName)
branchObject = _nf.CreateBranchObjectOnBranchUsingChainage(branchObjectType, name, branch, chainage)
branchObject.LongName = longName
if branchObjectType == _nf.BranchObjectType.Weir:
branchObject.CrestWidth = feature.Attributes ["CrestWidth"]
branchObject.CrestLevel = feature.Attributes ["CrestLevel"]
def AddCrossSectionsToNetwork(csShapeFile, profileCsvFile, network):
# get crossSection definitions as lookup (name, values)
lookup = _CreateCsvLookup(profileCsvFile)
for feature in _mf.GetShapeFileFeatures(csShapeFile):
# get attributes
name = feature.Attributes ["Name"]
branchName = feature.Attributes ["Branch"]
chainage = feature.Attributes ["Chainage"]
crossSectionType = feature.Attributes ["CrossSectio"]
thalweg = feature.Attributes ["Thalweg"]
definitionName = feature.Attributes ["DefName"]
# search branch
branch = _st.GetItemByName(network.Branches, branchName)
if (crossSectionType == "ZW"):
type = _nf.BranchObjectType.CrossSectionZW
elif (crossSectionType == "YZ"):
type = _nf.BranchObjectType.CrossSectionYZ
else :
continue
crossSection = _nf.CreateBranchObjectOnBranchUsingChainage(type, name, branch, chainage)
_nf.SetCrossSectionProfile(crossSection, lookup[definitionName], thalweg)
def AddBridgesToNetwork(csShapeFile, profileCsvFile, network):
# get crossSection definitions as lookup (name, values)
lookup = _CreateCsvLookup(profileCsvFile)
for feature in _mf.GetShapeFileFeatures(csShapeFile):
# get attributes
name = feature.Attributes ["Name"]
branchName = feature.Attributes ["Branch"]
chainage = feature.Attributes ["Chainage"]
length = feature.Attributes ["Length"]
roughType = feature.Attributes ["RoughType"]
roughnessValue = feature.Attributes ["Roughness"]
inletLoss = feature.Attributes ["InLossCoef"]
outletLoss = feature.Attributes ["OutLossCoef"]
groundLayerRoughness = feature.Attributes ["GLRoughness"]
pillarWidth= feature.Attributes ["PillarWidth"]
shapeFactor= feature.Attributes ["ShapeFactor"]
longName= feature.Attributes ["LongName"]
# search branch
branch = _st.GetItemByName(network.Branches, branchName)
bridge = _nf.CreateBranchObjectOnBranchUsingChainage(_nf.BranchObjectType.Bridge, name, branch, chainage)
bridge.Length = length
bridge.InletLossCoefficient = inletLoss
bridge.OutletLossCoefficient = outletLoss
"""brFrictionTypeDictionary = {"StricklerKs":BridgeFrictionType.StricklerKs,
"Chezy":BridgeFrictionType.Chezy,
"Manning":BridgeFrictionType.Manning,
"StricklerKn":BridgeFrictionType.StricklerKn,
"WhiteColebrook":BridgeFrictionType.WhiteColebrook}"""
bridge.FrictionType = _st.ParseToEnum(roughType,BridgeFrictionType)
bridge.Friction = roughnessValue
bridge.GroundLayerRoughness = groundLayerRoughness
bridge.PillarWidth = pillarWidth
bridge.ShapeFactor = shapeFactor
bridge.LongName = longName
if (lookup.has_key(name)):
bridge.BridgeType = _nf.BridgeType.Tabulated
bridge.TabulatedCrossSectionDefinition.ZWDataTable.Clear()
for item in lookup[name]:
bridge.TabulatedCrossSectionDefinition.ZWDataTable.AddCrossSectionZWRow(item[0], item[1], item[2])
def ImportBoundaryConditions(path, flowModel):
bcDictionary = {"H": _wff.BoundaryConditionType.WaterLevelConstant,
"Q": _wff.BoundaryConditionType.FlowConstant,
"None": _wff.BoundaryConditionType.NoBoundary}
with open(path) as csvfile:
lines = _csv.reader(csvfile,delimiter = ",")
for line in lines:
bcName = line[0]
bcType = bcDictionary[line[1]]
bcValue = float(line[2])
_wff.SetBoundaryCondition(flowModel, bcName, bcType, bcValue)
def ImportLateralData(path, flowModel):
latDictionary = {"FlowConstant": _wff.LateralDataType.FlowConstant}
with open(path) as csvfile:
lines = _csv.reader(csvfile,delimiter = ",")
for line in lines:
latName = line[0]
latType = latDictionary[line[1]]
latValue = float(line[2])
_wff.SetLateralData(flowModel, latName, latType, latValue)
def GetMinYMaxYofCrossSectionProfile(crossSection):
min = crossSection.Definition.YZDataTable[0].Yq
max = crossSection.Definition.YZDataTable[crossSection.Definition.YZDataTable.Count-1].Yq
return min, max
def CreateOutputChart(flowModel, outputName, locationNames, colors):
"""Creates an output chart for the provided locations."""
chart = _cf.CreateChart([])
index = 0
for locationName in locationNames:
location = _wff.GetComputationGridLocationByName(flowModel, locationName)
timeSeries = _wff.GetTimeSeriesFromWaterFlowModel(flowModel, location, outputName)
lineSeries = _cf.CreateLineSeries(timeSeries)
lineSeries.Title = locationName
lineSeries.Color = colors[index]
lineSeries.Width = 3
lineSeries.PointerVisible = False
lineSeries.Transparency = 50
chart.Series.Add(lineSeries)
index += 1
chart.Name = outputName
chart.TitleVisible = True
chart.Title = "Compare " + outputName + " zwolle"
chart.Legend.Visible = True
chart.BottomAxis.Title = "Time"
chart.LeftAxis.Title = outputName
_st.OpenView(chart)
return chart
def OpenModelMapViewWithBackground(model):
Gui.DocumentViewsResolver.OpenViewForData(model)
view = Gui.DocumentViews[Gui.DocumentViews.Count-1]
map = view.MapView.Map
map.CoordinateSystem = _mf.Map.CoordinateSystemFactory.CreateFromEPSG(3857) # webmercator
# Add background (Satellite image) map
layer = BingLayer()
layer.Name = "Satellite Hybrid (Bing Maps)"
map.Layers.Add(layer)
map.ZoomToExtents()
def _GetInterpolatedValue(timeLeft, valueLeft, timeRight, valueRight, timeToSearch):
perc = (timeToSearch - timeLeft).total_seconds() / (timeRight - timeLeft).total_seconds()
return ((valueRight - valueLeft) * perc) + valueLeft
def GetAverageDeviation(ts1, ts2, startAt = 0):
ts1Lookup = sorted(ts1, key=lambda tup: tup[0]) # measurements
ts2Lookup = sorted(ts2, key=lambda tup: tup[0]) # calculated values
ts1Min = ts1Lookup[0][0]
ts1Max = ts1Lookup[-1][0]
ts2Min = ts2Lookup[0][0]
ts2Max = ts2Lookup[-1][0]
if (ts1Min < ts2Min or ts1Max > ts2Max):
raise Exception("Measurements out of range")
totalSum = 0
minValueIndex = 0
for timeValueList in ts1Lookup:
timeMeasurement = timeValueList[0]
valueMeasurement = timeValueList[1]
for i in range(len(ts2Lookup))[minValueIndex:]:
timeCalc = ts2Lookup[i][0]
valueCalc = ts2Lookup[i][1]
if (timeCalc > timeMeasurement):
minValueIndex = i -1
timeleft = ts2Lookup[i-1][0]
valueleft = ts2Lookup[i-1][1]
totalSum += abs(valueMeasurement - _GetInterpolatedValue(timeleft, valueleft, timeCalc, valueCalc, timeMeasurement))
break
return totalSum/ (len(ts1Lookup)) # average deviation
|
...