python-thingset/python_thingset/client.py at a637b139518bb290c52f69454ba81553adc49962 · Brill-Power/python-thingset · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
#
# Copyright (c) 2024-2025 Brill Power.
#
# SPDX-License-Identifier: Apache-2.0
#
from abc import ABC, abstractmethod
from typing import Any, List, Union

from .backends import ThingSetBackend
from .response import ThingSetResponse, ThingSetStatus, ThingSetValue
from .log import get_logger


logger = get_logger()


class ThingSetClient(ABC):
    def fetch(
        self,
        parent_id: Union[int, str],
        ids: List[Union[int, str]],
        node_id: Union[int, None] = None,
        get_paths: bool = False,
    ) -> ThingSetResponse:
        values = []

        self._send(self.encode_fetch(parent_id, ids), node_id)

        msg = self._recv()
        tmp = ThingSetResponse(self.backend, msg)

        if tmp.status_code is not None:
            if tmp.status_code <= ThingSetStatus.CONTENT:
                if len(ids) == 0:
                    if self.backend == ThingSetBackend.Serial:
                        values.append(ThingSetValue(None, tmp.data, parent_id))
                    else:
                        values.append(
                            self._create_value(parent_id, node_id, tmp.data, get_paths)
                        )
                else:
                    for idx, id in enumerate(ids):
                        if self.backend == ThingSetBackend.Serial:
                            values.append(ThingSetValue(None, tmp.data[idx], id))
                        else:
                            values.append(
                                self._create_value(
                                    id, node_id, tmp.data[idx], get_paths
                                )
                            )

        return ThingSetResponse(self.backend, msg, values)

    def get(
        self,
        value_id: Union[int, str],
        node_id: Union[int, None] = None,
        get_paths: bool = False,
    ) -> ThingSetResponse:
        values = []

        self._send(self.encode_get(value_id), node_id)

        msg = self._recv()
        tmp = ThingSetResponse(self.backend, msg)

        if tmp.status_code is not None:
            if tmp.status_code <= ThingSetStatus.CONTENT:
                if self.backend == ThingSetBackend.Serial:
                    values.append(ThingSetValue(None, tmp.data, value_id))
                else:
                    values.append(
                        self._create_value(value_id, node_id, tmp.data, get_paths)
                    )

        return ThingSetResponse(self.backend, msg, values)

    def update(
        self,
        value_id: Union[int, str],
        value: Any,
        node_id: Union[int, None] = None,
        parent_id: Union[int, None] = None,
    ) -> ThingSetResponse:
        self._send(self.encode_update(parent_id, value_id, value), node_id)
        return ThingSetResponse(self.backend, self._recv())

    def exec(
        self,
        value_id: Union[int, str],
        args: Union[List[Any], None],
        node_id: Union[int, None] = None,
    ) -> ThingSetResponse:
        self._send(self.encode_exec(value_id, args), node_id)
        return ThingSetResponse(self.backend, self._recv())

    def _create_value(
        self, value_id: int, node_id: int, value: Any, get_paths: bool
    ) -> ThingSetValue:
        path = None

        if get_paths:
            if value_id == ThingSetValue.ID_ROOT:
                path = "Root"
            else:
                self._send(self.encode_get_path(value_id), node_id)
                tmp = ThingSetResponse(self.backend, self._recv())

                if tmp.data is not None:
                    path = tmp.data[0]
                else:
                    logger.warning("Failed to read value path")

        return ThingSetValue(value_id, value, path)

    @abstractmethod
    def disconnect(self) -> None:
        pass

    @abstractmethod
    def _send(self, data: bytes, node_id: Union[int, None]) -> None:
        pass

    @abstractmethod
    def _recv(self) -> bytes:
        pass

    @property
    def is_connected(self) -> bool:
        return self._is_connected

    @is_connected.setter
    def is_connected(self, _is_connected: bool) -> None:
        self._is_connected = _is_connected

    @property
    def backend(self) -> str:
        return self._backend

    @backend.setter
    def backend(self, _backend) -> None:
        self._backend = _backend