39 lines
1.5 KiB
Python
39 lines
1.5 KiB
Python
from .database import *
|
|
from .s0_base import *
|
|
from .s24_coordinates 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'] = str(si['id'])
|
|
d['type'] = str(si['type'])
|
|
d['x'] = float(si['x'])
|
|
d['y'] = float(si['y'])
|
|
d['api_query_id'] = str(si['api_query_id'])
|
|
d['associated_element_id'] = str(si['associated_element_id'])
|
|
|
|
return d
|
|
|
|
def get_all_scada_info(name: str) -> list[dict[str, Any]]:
|
|
sis = try_read(name, f"select * from scada_info")
|
|
if sis is None:
|
|
return []
|
|
|
|
d = []
|
|
for si in sis:
|
|
d.append({ 'id': str(si['id']), 'type': str(si['type']), 'x': float(si['x']), 'y': float(si['y']), 'api_query_id': str(si['api_query_id']), 'associated_element_id': str(si['associated_element_id']) })
|
|
|
|
return d |