-
Notifications
You must be signed in to change notification settings - Fork 15
/
base.py
119 lines (93 loc) · 3 KB
/
base.py
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
import spidev
import gpiod
from typing import Iterable
class LoRaSpi():
SPI_SPEED = 8000000
def __init__(self, bus: int, cs: int, speed: int = SPI_SPEED):
self.bus = bus
self.cs = cs
self.speed = speed
def transfer(self, buf: Iterable) -> tuple:
spi = spidev.SpiDev()
spi.open(self.bus, self.cs)
spi.lsbfirst = False
spi.mode = 0
spi.max_speed_hz = self.speed
ret = spi.xfer2(buf)
spi.close()
return ret
class LoRaGpio:
LOW = 0
HIGH = 1
def __init__(self, chip: int, offset: int):
self.chip = "gpiochip" + str(chip)
self.offset = offset
def output(self, value: int):
chip = gpiod.Chip(self.chip)
line = chip.get_line(self.offset)
try:
line.request(consumer="LoRaGpio", type=gpiod.LINE_REQ_DIR_OUT)
line.set_value(value)
except: return
finally:
line.release()
chip.close()
def input(self) -> int:
chip = gpiod.Chip(self.chip)
line = chip.get_line(self.offset)
try:
line.request(consumer="LoRaGpio", type=gpiod.LINE_REQ_DIR_IN)
value = line.get_value()
except: return -1
finally:
line.release()
chip.close()
return value
def monitor(self, callback, timeout: float):
seconds = int(timeout)
chip = gpiod.Chip(self.chip)
line = chip.get_line(self.offset)
try:
line.request(consumer="LoRaGpio", type=gpiod.LINE_REQ_EV_RISING_EDGE)
if line.event_wait(seconds, int((timeout - seconds) * 1000000000)):
callback()
except: return
finally:
line.release()
chip.close()
def monitor_continuous(self, callback, timeout: float):
seconds = int(timeout)
while True:
chip = gpiod.Chip(self.chip)
line = chip.get_line(self.offset)
try:
line.request(consumer="LoRaGpio", type=gpiod.LINE_REQ_EV_RISING_EDGE)
if line.event_wait(seconds, int((timeout - seconds) * 1000000000)):
callback()
except: continue
finally:
line.release()
chip.close()
class BaseLoRa :
def begin(self):
raise NotImplementedError
def end(self):
raise NotImplementedError
def reset(self):
raise NotImplementedError
def beginPacket(self):
raise NotImplementedError
def endPacket(self, timeout: int)-> bool:
raise NotImplementedError
def write(self, data, length: int):
raise NotImplementedError
def request(self, timeout: int)-> bool:
raise NotImplementedError
def available(self):
raise NotImplementedError
def read(self, length: int):
raise NotImplementedError
def wait(self, timeout: int)-> bool:
raise NotImplementedError
def status(self):
raise NotImplementedError