digital_twins

Этот модуль содержит следующие функции и классы ошибок для работы с Digital Twins:

1class NoDeviceFoundError(Exception):
2    pass
3
4class NoAgentFoundError(Exception):
5    pass
6
7class BadParamsError(Exception):
8    pass
9
10class UnknownError(Exception):
11    pass
12
13
14def change_state(tag: Tag, new_value: Any) -> Update:
15  """
16  Изменение состояния
17  """
18
19def get_agent_by_id(agent_id: int) -> Agent:
20  """
21  Получить агента по его ID
22  """
23
24
25def get_device_by_id(device_id: int) -> Device:
26  """
27  Получить устройство по его ID
28  """

Классы Agent и Device содержат много информации о соответствующем агенте/устройстве:

1class Driver(NamedTuple):
2    id: int
3    name: str
4
5class DeviceAgentType(NamedTuple):
6    id: int
7    name: str
8    label: str
9    drivers: List[Driver]
10
11class DeviceAgent(NamedTuple):
12    id: int
13    name: str
14    label: str
15    type: DeviceAgentType
16
17class DeviceType(NamedTuple):
18    id: int
19    name: str
20    label: str
21    driver: Driver
22
23class Device:
24    id: int
25    name: str
26    label: str
27    type: DeviceType
28    tag: TagT
29    first_seen_at: int
30    last_seen_at: int
31    agent: Optional[DeviceAgent]
32
33class AgentType:
34    id: int
35    name: str
36    label: str
37    drivers: List[Driver]
38
39class AgentDevice:
40    id: int
41    name: str
42    label: str
43    type: DeviceType
44
45class Agent:
46    id: int
47    name: str
48    label: str
49    first_seen_at: int
50    last_seen_at: int
51    type: AgentType
52    tag: TagT
53    devices: List[AgentDevice]

Интерфейс класса Update содержит только методы change_state и execute:

1class State(TypedDict):
2    received_at: int
3    state_changed: bool
4    tag_id: int
5    timestamp: int
6    value: Any
7
8class ChangedStates(TypedDict):
9    states: List[State]
10
11class Update:
12
13    def change_state(self, tag: Tag, new_value: Any) -> 'Update':
14		  pass
15
16    def execute(self) -> ChangedStates:
17		  pass

Пример изменения состояния тега и получения информации о его состоянии:

1from coiiot_sdk import digital_twins as dt, context, user_logs
2
3logger = user_logs.get_logger()
4ctx = context.current()
5
6# полагаем, тэг имеет целочисленный тип
7dt.change_state(ctx.tag, 7777).execute()
8logger.info(ctx.tag.state.value) # 7777

В этом примере используется свойство state тега. Это не статичное свойство. Если поменять состояние тега tag на новое, то при следующем обращении оно уже будет доступно в tag.state.