59 lines
2.1 KiB
Python
59 lines
2.1 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},
|
|
"transmission_mode": {"type": "str", "optional": True, "readonly": True},
|
|
"transmission_frequency": {"type": "float", "optional": True, "readonly": True},
|
|
"reliability": {"type": "float", "optional": True, "readonly": True},
|
|
"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["transmission_mode"] = si.get("transmission_mode")
|
|
d["transmission_frequency"] = si.get("transmission_frequency")
|
|
d["reliability"] = si.get("reliability")
|
|
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"],
|
|
"transmission_mode": si.get("transmission_mode"),
|
|
"transmission_frequency": si.get("transmission_frequency"),
|
|
"reliability": si.get("reliability"),
|
|
"associated_element_id": si["associated_element_id"],
|
|
}
|
|
)
|
|
|
|
return d
|