Repository navigation
Expand file tree
/
Copy pathraw_433_backend.py
More file actions
222 lines (190 loc) · 8.14 KB
/
Copy pathraw_433_backend.py
File metadata and controls
222 lines (190 loc) · 8.14 KB
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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
#!/usr/bin/python3
import time
class Raw433Config:
DEFAULT_TXGPIO = 4
def __init__(self, tx_gpio=DEFAULT_TXGPIO):
if tx_gpio is None:
tx_gpio = self.DEFAULT_TXGPIO
self.tx_gpio = int(tx_gpio)
if not (0 <= self.tx_gpio <= 27):
raise ValueError(f"TXGPIO {self.tx_gpio} is invalid; valid Pi GPIO numbers are 0–27")
@classmethod
def from_app_config(cls, config):
if hasattr(config, "ReadValue"):
return cls(
tx_gpio=config.ReadValue(
"TXGPIO",
return_type=int,
default=cls.DEFAULT_TXGPIO,
section="General",
)
)
return cls(tx_gpio=getattr(config, "TXGPIO", cls.DEFAULT_TXGPIO))
class Raw433Transmitter:
def __init__(
self,
config,
is_pi5=False,
pigpio_module=None,
lgpio_module=None,
lgpio_chip=4,
):
self.config = config
self.is_pi5 = is_pi5
self.pigpio = pigpio_module
self.lgpio = lgpio_module
self.lgpio_chip = lgpio_chip
self._lgpio_handle = None
def transmit(self, frame, repetition):
if self.is_pi5:
self._send_lgpio(frame, repetition)
else:
self._send_pigpio(frame, repetition)
def set_idle_low(self):
"""Drive the TX GPIO low (before CC1101 starts the carrier, and at startup)."""
if self.is_pi5:
self._set_idle_low_lgpio()
else:
self._set_idle_low_pigpio()
def close(self):
if self._lgpio_handle is None:
return
lgpio = self._load_lgpio()
handle, self._lgpio_handle = self._lgpio_handle, None
try:
lgpio.gpio_free(handle, self.config.tx_gpio)
finally:
lgpio.gpiochip_close(handle)
def _load_pigpio(self):
if self.pigpio is not None:
return self.pigpio
try:
import pigpio
except ImportError:
raise ImportError("pigpio not installed — run: sudo apt-get install python3-pigpio") from None
return pigpio
def _load_lgpio(self):
if self.lgpio is not None:
return self.lgpio
try:
import lgpio
except ImportError:
raise ImportError("lgpio not installed — run: sudo apt-get install python3-lgpio") from None
return lgpio
def _send_pigpio(self, frame, repetition):
pigpio = self._load_pigpio()
pi = pigpio.pi()
if not pi.connected:
pi.stop()
raise RuntimeError("pigpio connection failed — ensure pigpiod is running (sudo pigpiod)")
try:
tx_gpio = self.config.tx_gpio
pi.wave_add_new()
pi.set_mode(tx_gpio, pigpio.OUTPUT)
pi.write(tx_gpio, 0)
wf = []
wf.append(pigpio.pulse(1 << tx_gpio, 0, 9415)) # wake up pulse
wf.append(pigpio.pulse(0, 1 << tx_gpio, 89565)) # silence
for i in range(2): # hardware synchronization
wf.append(pigpio.pulse(1 << tx_gpio, 0, 2560))
wf.append(pigpio.pulse(0, 1 << tx_gpio, 2560))
wf.append(pigpio.pulse(1 << tx_gpio, 0, 4550)) # software synchronization
wf.append(pigpio.pulse(0, 1 << tx_gpio, 640))
for i in range(0, 56): # manchester encoding of payload data
if ((frame[int(i / 8)] >> (7 - (i % 8))) & 1):
wf.append(pigpio.pulse(0, 1 << tx_gpio, 640))
wf.append(pigpio.pulse(1 << tx_gpio, 0, 640))
else:
wf.append(pigpio.pulse(1 << tx_gpio, 0, 640))
wf.append(pigpio.pulse(0, 1 << tx_gpio, 640))
wf.append(pigpio.pulse(0, 1 << tx_gpio, 30415)) # interframe gap
for j in range(1, repetition): # repeating frames
for i in range(7): # hardware synchronization
wf.append(pigpio.pulse(1 << tx_gpio, 0, 2560))
wf.append(pigpio.pulse(0, 1 << tx_gpio, 2560))
wf.append(pigpio.pulse(1 << tx_gpio, 0, 4550)) # software synchronization
wf.append(pigpio.pulse(0, 1 << tx_gpio, 640))
for i in range(0, 56): # manchester encoding of payload data
if ((frame[int(i / 8)] >> (7 - (i % 8))) & 1):
wf.append(pigpio.pulse(0, 1 << tx_gpio, 640))
wf.append(pigpio.pulse(1 << tx_gpio, 0, 640))
else:
wf.append(pigpio.pulse(1 << tx_gpio, 0, 640))
wf.append(pigpio.pulse(0, 1 << tx_gpio, 640))
wf.append(pigpio.pulse(0, 1 << tx_gpio, 30415)) # interframe gap
pi.wave_add_generic(wf)
wid = pi.wave_create()
pi.wave_send_once(wid)
while pi.wave_tx_busy():
pass
pi.wave_delete(wid)
finally:
pi.stop()
def _set_idle_low_pigpio(self):
pigpio = self._load_pigpio()
pi = pigpio.pi()
if not pi.connected:
pi.stop()
raise RuntimeError("pigpio connection failed — ensure pigpiod is running (sudo pigpiod)")
try:
tx_gpio = self.config.tx_gpio
pi.set_mode(tx_gpio, pigpio.OUTPUT)
pi.write(tx_gpio, 0)
finally:
pi.stop()
def _send_lgpio(self, frame, repetition):
lgpio = self._load_lgpio()
tx_gpio = self.config.tx_gpio
h = self._claim_lgpio_output()
pulses = []
pulses.append(lgpio.pulse(1, 1, 9415)) # wake up pulse
pulses.append(lgpio.pulse(0, 1, 89565)) # silence
for i in range(2): # hardware synchronization
pulses.append(lgpio.pulse(1, 1, 2560))
pulses.append(lgpio.pulse(0, 1, 2560))
pulses.append(lgpio.pulse(1, 1, 4550)) # software synchronization
pulses.append(lgpio.pulse(0, 1, 640))
for i in range(0, 56): # manchester encoding of payload data
if ((frame[int(i / 8)] >> (7 - (i % 8))) & 1):
pulses.append(lgpio.pulse(0, 1, 640))
pulses.append(lgpio.pulse(1, 1, 640))
else:
pulses.append(lgpio.pulse(1, 1, 640))
pulses.append(lgpio.pulse(0, 1, 640))
pulses.append(lgpio.pulse(0, 1, 30415)) # interframe gap
for j in range(1, repetition): # repeating frames
for i in range(7): # hardware synchronization
pulses.append(lgpio.pulse(1, 1, 2560))
pulses.append(lgpio.pulse(0, 1, 2560))
pulses.append(lgpio.pulse(1, 1, 4550)) # software synchronization
pulses.append(lgpio.pulse(0, 1, 640))
for i in range(0, 56): # manchester encoding of payload data
if ((frame[int(i / 8)] >> (7 - (i % 8))) & 1):
pulses.append(lgpio.pulse(0, 1, 640))
pulses.append(lgpio.pulse(1, 1, 640))
else:
pulses.append(lgpio.pulse(1, 1, 640))
pulses.append(lgpio.pulse(0, 1, 640))
pulses.append(lgpio.pulse(0, 1, 30415)) # interframe gap
lgpio.tx_wave(h, tx_gpio, pulses)
while lgpio.tx_busy(h, tx_gpio, lgpio.TX_WAVE):
time.sleep(0.001)
lgpio.gpio_write(h, tx_gpio, 0)
def _claim_lgpio_output(self):
"""Claim the TX GPIO as a low output once and keep the claim.
Releasing it would hand the line back to the kernel as a floating
input, which makes the 433 MHz module transmit continuously (#154).
"""
if self._lgpio_handle is None:
lgpio = self._load_lgpio()
handle = lgpio.gpiochip_open(self.lgpio_chip)
try:
lgpio.gpio_claim_output(handle, self.config.tx_gpio, 0)
except Exception:
lgpio.gpiochip_close(handle)
raise
self._lgpio_handle = handle
return self._lgpio_handle
def _set_idle_low_lgpio(self):
lgpio = self._load_lgpio()
lgpio.gpio_write(self._claim_lgpio_output(), self.config.tx_gpio, 0)