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
|
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018-2021 Maxim Devaev <[email protected]> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
import re
import functools
import errno
import time
from typing import Tuple
from typing import Dict
from typing import Optional
import serial
import socket
import binascii
from ...logging import get_logger
from ... import aiotools
from ... import aiomulti
from ... import aioproc
from ...yamlconf import Option
from ...validators.basic import valid_number
from ...validators.basic import valid_float_f01
from ...validators.os import valid_abs_path
from ...validators.hw import valid_tty_speed
from ...validators.net import valid_ip_or_host
from ...validators.net import valid_port
from . import BaseUserGpioDriver
# =====
class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attributes
def __init__(
self,
instance_name: str,
notifier: aiotools.AioNotifier,
tesmart_host: str,
tesmart_port: int,
max_ports: int,
) -> None:
super().__init__(instance_name, notifier)
self.__tesmart_host = tesmart_host
self.__tesmart_port = tesmart_port
self.__max_ports = max_ports
self.__switch_state: Dict[int, bool] = {}
self.__tes_socket: socket
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"tesmart_host": Option("192.168.1.10", type=valid_ip_or_host),
"tesmart_port": Option(5000, type=valid_port),
"max_ports": Option(8, type=functools.partial(valid_number, min=4, max=16)),
}
def register_input(self, pin: int, debounce: float) -> None:
_ = pin
_ = debounce
def register_output(self, port: int, initial: Optional[bool]) -> None:
if port <= self.__max_ports:
self.__switch_state[port] = initial
def prepare(self) -> None:
self.__tes_socket = socket.create_connection((self.__tesmart_host,self.__tesmart_port))
self.__update_state()
def __update_state(self) -> None:
for port in self.__switch_state:
self.__switch_state[port] = False
selport = self.__get_selected_port()
if selport in self.__switch_state:
self.__switch_state[selport] = True
def __get_selected_port(self) -> int:
retint = self.__send_tesmart_command("1000")
return retint+1
def __send_tesmart_command(self,tes_cmd: str) -> int:
full_cmd="AABB03"+tes_cmd+"EE"
binstr = binascii.unhexlify(full_cmd)
self.__tes_socket.sendall(binstr)
retstr=self.__tes_socket.recv(6)
return int(bytearray(retstr)[4])
async def run(self) -> None:
pass
def cleanup(self) -> None:
pass
async def read(self, pin: int) -> bool:
if pin in self.__switch_state:
return self.__switch_state[pin]
return False
async def write(self, pin: int, state: bool) -> None:
if state == False:
return
part_cmd="01"+format(pin,"#04x")[2:4]
writeret = self.__send_tesmart_command(part_cmd)
self.__update_state()
# =====
def __str__(self) -> str:
return f"tesmart({self._instance_name})"
__repr__ = __str__
|
