65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
from .database import *
|
|
from .s0_base import *
|
|
|
|
def get_pipe_risk_probability_now(name: str, pipe_id: str) -> dict[str, Any]:
|
|
t = try_read(name, f"select * from pipe_risk_probability where pipeid = '{pipe_id}'")
|
|
if t == None:
|
|
return {}
|
|
|
|
d = {}
|
|
d['pipeid'] = str(t['pipeid'])
|
|
d['pipeage'] = t['pipeage']
|
|
d['risk_probability_now'] = t['risk_probability_now']
|
|
|
|
return d
|
|
|
|
def get_pipe_risk_probability(name: str, pipe_id: str) -> dict[str, Any]:
|
|
t = try_read(name, f"select * from pipe_risk_probability where pipeid = '{pipe_id}'")
|
|
if t == None:
|
|
return {}
|
|
|
|
d = {}
|
|
d['pipeid'] = t['pipeid']
|
|
d['x'] = t['x']
|
|
d['y'] = t['y']
|
|
|
|
return d
|
|
|
|
def get_network_pipe_risk_probability_now(name: str) -> list[dict[str, Any]]:
|
|
pipe_risk_probability_list = []
|
|
with conn[name].cursor(row_factory=dict_row) as cur:
|
|
cur.execute(f"select * from pipe_risk_probability")
|
|
for record in cur:
|
|
#pipe_risk_probability_list.append(record)
|
|
t = {}
|
|
t['pipeid'] = record['pipeid']
|
|
t['pipeage'] = record['pipeage']
|
|
t['risk_probability_now'] = record['risk_probability_now']
|
|
pipe_risk_probability_list.append(t)
|
|
|
|
return pipe_risk_probability_list
|
|
|
|
def get_pipes_risk_probability(name: str, pipe_ids: list[str]) -> list[dict[str, Any]]:
|
|
pipe_risk_probability_list = []
|
|
with conn[name].cursor(row_factory=dict_row) as cur:
|
|
cur.execute(f"select * from pipe_risk_probability")
|
|
for record in cur:
|
|
if record['pipeid'] in pipe_ids:
|
|
t = {}
|
|
t['pipeid'] = record['pipeid']
|
|
t['x'] = record['x']
|
|
t['y'] = record['y']
|
|
pipe_risk_probability_list.append(t)
|
|
|
|
return pipe_risk_probability_list
|
|
|
|
# node : {}, link : {}
|
|
def get_pipe_risk_probability_geometries(name: str) -> dict[str, Any]:
|
|
pipe_risk_probability_geometries = {}
|
|
|
|
with conn[name].cursor(row_factory=dict_row) as cur:
|
|
cur.execute(f"select ST_AsGeoJSON(geometry) AS geometry")
|
|
for record in cur:
|
|
print(record['geometry'])
|
|
|
|
return pipe_risk_probability_geometries |