42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
from .database import *
|
|
|
|
|
|
def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]:
|
|
return { 'id' : {'type': 'str' , 'optional': False , 'readonly': True },
|
|
'type' : {'type': 'str' , 'optional': False , 'readonly': True },
|
|
'x' : {'type': 'float' , 'optional': False , 'readonly': False},
|
|
'y' : {'type': 'float' , 'optional': False , 'readonly': False},
|
|
'query_api_id' : {'type': 'str' , 'optional': False , 'readonly': False},
|
|
'associated_element_id' : {'type': 'str' , 'optional': False , 'readonly': True } }
|
|
|
|
|
|
def get_scada_info(name: str, id: str) -> dict[str, Any]:
|
|
si = try_read(name, f"select * from scada_info where id = '{id}'")
|
|
if si is None:
|
|
return {}
|
|
|
|
d = {}
|
|
d['id'] = si['id']
|
|
d['type'] = si['type']
|
|
d['x'] = float(si['x_coor'])
|
|
d['y'] = float(si['y_coor'])
|
|
d['api_query_id'] = si['api_query_id']
|
|
d['associated_element_id'] = si['associated_element_id']
|
|
|
|
return d
|
|
|
|
def get_all_scada_info(name: str) -> list[dict[str, Any]]:
|
|
sis = read_all(name, f"select * from scada_info")
|
|
if sis is None:
|
|
return []
|
|
|
|
d = []
|
|
for si in sis:
|
|
d.append({ 'id': si['id'],
|
|
'type': si['type'],
|
|
'x': float(si['x_coor']),
|
|
'y': float(si['y_coor']),
|
|
'api_query_id': si['api_query_id'],
|
|
'associated_element_id': si['associated_element_id'] })
|
|
|
|
return d |