Методы клиента

Информация

В этом разделе описаны методы, которые соответствуют версии библиотеки sl5api ver 1.0.2.

узнать текущую установленную версию: pip show sl5api
обновить библиотеку до последней версии: pip install --upgrade sl5api

read_value

Запрашивает текущее значение тега и его тип данных по имени.

async def read_value(self, tag_name: str) -> Any

Параметры:

Имя	Тип	Описание
tag_name	str	Имя тега в системе

Возвращаемое значение Any:

Тип	Описание
bool	Логическое значение
int	Целочисленное значение (32 или 64 бита)
float	Значение с плавающей точкой (double)
str	Строковое значение
datetime	Значение времени (преобразуется в локальный часовой пояс)

Исключения:

Тип	Условие возникновения
grpc.aio.AioRpcError	Сервер недоступен, неверные учетные данные, тег не найден, превышен таймаут

Пример:

Получение текущего значения тега "Температура 1".

import asyncio
from sl5api import Client, Config

cfg = Config("127.0.0.1:23505", "sxe92rd9slu74ol...")

async def main():
    c = Client(cfg)

    value = await c.read_value("Температура 1")

    print(f"Текущее значение: {value}")
    print(f"Тип значения: {type(value).__name__}")

if __name__ == "__main__":
    asyncio.run(main())

read_many

Запрашивает данные нескольких тегов по их именам. Возвращает текущие значения, качество, настройки шкал и статусы аварий для каждого.

async def read_many(self, tag_names: Iterable[str]) -> Iterable[Tag]

Параметры:

Имя	Тип	Описание
tag_names	Iterable[str]	Имена тегов в системе

Возвращаемое значение Iterable[Tag]:

Имя	Тип	Описание
tag_id	int	Уникальный идентификатор тега
name	str	Имя тега
value	Any	Текущее значение тега (тип зависит от тега)
quality	int	Статус качества текущего значения
high_scale	float	Верхняя граница шкалы
low_scale	float	Нижняя граница шкалы
lo_lo_alarm	float	Нижнее пороговое значение аварийного события
lo_alarm	float	Нижнее пороговое значение предупредительного события
hi_alarm	float	Верхнее пороговое значение предупредительного события
hi_hi_alarm	float	Верхнее пороговое значение аварийного события
alarm_status	int	Текущий статус аварии

Исключения:

Тип	Условие возникновения
grpc.aio.AioRpcError	Сервер недоступен, неверные учетные данные, превышен таймаут

Пример:

Получение данных тегов с именем "Температура 1", "Давление 2", "Уровень 3".

import asyncio
from sl5api import Client, Config

cfg = Config("127.0.0.1:23505", "sxe92rd9slu74ol...")

async def main():
    c = Client(cfg)

    tags = await c.read_many(["Температура 1", "Давление 2", "Уровень 3"])

    for tag in tags:
        print(f"{tag.name}: {tag.value} (качество={tag.quality})")
        print(f"  Шкала: {tag.low_scale} - {tag.high_scale}")
        print(f"  Алармы: lo={tag.lo_alarm}, hi={tag.hi_alarm}")

if __name__ == "__main__":
    asyncio.run(main())

write_value

Производит запись нового значения в тег. Для физических тегов выполняется запись через драйвер устройства. Для виртуальных — значение устанавливается непосредственно на стороне сервера.

async def write_value(self, tag_name: str, new_value: Any, new_quality: int) -> None

Параметры:

Имя	Тип	Описание
tag_name	str	Имя тега в системе
new_value	Any	Новое значение тега. Поддерживаются типы: bool, int, float, str, datetime
new_quality	int	Значение качества. Используется только для виртуальных тегов. Для физических тегов игнорируется.

Возвращаемое значение: None

Исключения:

Тип	Условие возникновения
grpc.aio.AioRpcError	Сервер недоступен, неверные учетные данные, тег не найден, превышен таймаут
ValueError	Неподдерживаемый тип значения new_value

Пример:

Запись значений в теги (физические/виртуальные) с разными типами данных. Название тегов: "Уставка температуры", "Состояние", "Время события", "Флаг готовности".

import asyncio
from datetime import datetime
from sl5api import Client, Config

cfg = Config("127.0.0.1:23505", "sxe92rd9slu74olj0il...")

async def main():
    c = Client(cfg)

    # Запись числа в виртуальный тег (качество учитывается)
    await c.write_value("Уставка температуры", 75, 192)

    # Запись строки в физический тег (качество игнорируется)
    await c.write_value("Состояние", False, 0)

    # Запись времени
    await c.write_value("Время события", datetime.now(), 192)

    # Запись логического значения
    await c.write_value("Флаг готовности", True, 192)

if __name__ == "__main__":
    asyncio.run(main())

мoments_get_range

Запрашивает сохраненные значения (моменты) указанного тега за заданный временной интервал. Данные возвращаются в хронологическом порядке.

async def moments_get_range(self, options: MomentGetOptions) -> Iterable[Moment]

Параметры:

Имя	Тип	Описание
options	MomentGetOptions	Объект с параметрами запроса

Структура MomentsGetOptions:

Имя	Тип	Описание
tag_name	str	Идентификатор тега в системе
start	datetime	Начало интервала (должно быть указано в UTC)
end	datetime	Конец интервала (должно быть указано в UTC)

Возвращаемое значение Iterable[Moment]:

Имя	Тип	Описание
value	any	значение (тип зависит от тега)
quality	int	статус качества
time_dt	datetime	Временная метка, преобразованная в локальное время текущего устройства

Исключения:

Тип	Условие возникновения
grpc.aio.AioRpcError	Сервер недоступен, неверные учетные данные, тег не найден, превышен таймаут

Пример:

Получение моментов тега "Температура 1" за сутки.

import asyncio
from datetime import datetime, timedelta, timezone
from sl5api import Client, Config, MomentGetOptions

cfg = Config("127.0.0.1:23505", "sxe92rd9slu74ol...")

async def main():
    c = Client(cfg)

    options = MomentGetOptions(
        tag_name="Температура 1",
        start=datetime.now(timezone.utc) - timedelta(minutes=30),
        end=datetime.now(timezone.utc)
    )

    moments = await c.moments_get_range(options)

    print("Moments:")

    for moment in moments:
        print(f"{moment.time_dt}: {moment.value} (quality={moment.quality})")

if __name__ == "__main__":
    asyncio.run(main())