8877213793

Committed 29 Apr 2024 10:39AM UTC coverage: 66.76% (+4.8%) from 61.918%

Build # 8877213793

Build Type

Pull #59

github

Committed by

web-flow

Commit Message

Merge e04c2406b into e2f205274

Pull Request Pull Request #59: Bookworm/Pi5 Compatibility: Upgrade to latest boilerplate, port to gpiod

Run Details

14 of 23 new or added lines in 1 file covered. (60.87%)

1 existing line in 1 file now uncovered.

239 of 358 relevant lines covered (66.76%)

0.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.45

/automationhat/__init__.py

import atexit
import time
import warnings

import ads1015
import gpiod
import gpiodevice
import sn3218
from gpiod.line import Bias, Direction, Value

from .pins import AsyncWorker, ObjectCollection, StoppableThread  # noqa: F401

__version__ = '0.4.1'


RELAY_1 = 13
RELAY_2 = 19
RELAY_3 = 16

INPUT_1 = 26
INPUT_2 = 20
INPUT_3 = 21

OUTPUT_1 = 5
OUTPUT_2 = 12
OUTPUT_3 = 6

UPDATES_PER_SECOND = 30

i2c = None
lights = None

automation_hat = False
automation_phat = True

_led_states = [0] * 18
_lights_need_updating = False
_is_setup = False
_t_update_lights = None
_ads1015 = None
_gpiochip = None


class SNLight(object):
    def __init__(self, index):
        self.index = index
        self._max_brightness = float(128)

    def toggle(self):
        """Toggle the light from on to off or off to on"""
        self.write(1 - self.read())

    def on(self):
        """Turn the light on"""
        self.write(1)

    def off(self):
        """Turn the light off"""
        self.write(0)

    def read(self):
        """Read the current status of the light"""
        if self.index is None:
            return

        return _led_states[self.index] / self._max_brightness

    def write(self, value):
        """Write a specific value to the light

        :param value: Brightness of the light, from 0.0 to 1.0
        """
        global _lights_need_updating

        if self.index is None:
            return

        if not isinstance(value, (int, float)):
            raise TypeError("Value must be int or float")

        if value >= 0 and value <= 1.0:
            _led_states[self.index] = int(self._max_brightness * value)
            if _t_update_lights is None and lights is not None:
                lights.output(_led_states)
            else:
                _lights_need_updating = True

        else:
            raise ValueError("Value must be between 0.0 and 1.0")


class AnalogInput(object):
    type = 'Analog Input'

    def __init__(self, channel, max_voltage, led):
        self._en_auto_lights = True
        self.channel = channel
        self.value = 0
        self.max_voltage = float(max_voltage)
        self.light = SNLight(led)
        self._is_setup = False

    def setup(self):
        if self._is_setup:
            return

        setup()
        self._is_setup = True

    def auto_light(self, value=None):
        if value is not None:
            self._en_auto_lights = value
        return self._en_auto_lights

    def read(self):
        """Return the read voltage of the analog input"""
        if self.name == "four" and is_automation_phat():
            warnings.warn("Analog Four is not supported on Automation pHAT")

        self._update()
        return round(self.value * self.max_voltage, 3)

    def _update(self):
        self.setup()
        self.value = _ads1015.get_voltage("in{}/gnd".format(self.channel)) / 3.3

        if self._en_auto_lights:
            adc = self.value
            self.light.write(max(0.0, min(1.0, adc)))


class Pin(object):
    type = 'Pin'

    def __init__(self, pin):
        self.pin = pin
        self._last_value = None
        self._is_setup = False
        self._gpioline = None

    def __call__(self):
        return filter(lambda x: x[0] != '_', dir(self))

    def read(self):
        self.setup()
        return self._gpioline.get_value(self.pin) == Value.ACTIVE

    def setup(self):
        pass

    def has_changed(self):
        value = self.read()

        if self._last_value is None:
            self._last_value = value

        if value is not self._last_value:
            self._last_value = value
            return True

        return False

    def is_on(self):
        return self.read() is True

    def is_off(self):
        return self.read() is False


class Input(Pin):
    type = 'Digital Input'

    def __init__(self, pin, led):
        self._en_auto_lights = True
        Pin.__init__(self, pin)
        self.light = SNLight(led)

    def setup(self):
        if self._is_setup:
            return False

        setup()

        self.pin = _gpiochip.line_offset_from_id(self.pin)

        self._gpioline = _gpiochip.request_lines(consumer="AH", config={
            self.pin: gpiod.LineSettings(direction=Direction.INPUT, bias=Bias.DISABLED)
        })

        self._is_setup = True

    def auto_light(self, value=None):
        if value is not None:
            self._en_auto_lights = value
        return self._en_auto_lights

    def read(self):
        self.setup()
        value = self._gpioline.get_value(self.pin) == Value.ACTIVE
        if self._en_auto_lights:
            self.light.write(value)
        return value


class Output(Pin):
    type = 'Digital Output'

    def __init__(self, pin, led):
        self._en_auto_lights = True
        Pin.__init__(self, pin)
        self.light = SNLight(led)

    def setup(self):
        if self._is_setup:
            return False

        setup()

        self.pin = _gpiochip.line_offset_from_id(self.pin)

        self._gpioline = _gpiochip.request_lines(consumer="AH", config={
            self.pin: gpiod.LineSettings(direction=Direction.OUTPUT, bias=Bias.DISABLED, output_value=Value.INACTIVE)
        })

        self._is_setup = True
        return True

    def auto_light(self, value=None):
        if value is not None:
            self._en_auto_lights = value
        return self._en_auto_lights

    def write(self, value):
        """Write a value to the output.

        :param value: Value to write, either 1 for HIGH or 0 for LOW
        """
        self.setup()
        self._gpioline.set_value(self.pin, Value.ACTIVE if value else Value.INACTIVE)
        if self._en_auto_lights:
            self.light.write(1 if value else 0)

    def on(self):
        """Turn the output on/HIGH"""
        self.write(1)

    def off(self):
        """Turn the output off/LOW"""
        self.write(0)

    def toggle(self):
        """Toggle the output."""
        self.write(not self.read())


class Relay(Output):
    type = 'Relay'

    def __init__(self, pin, led_no, led_nc):
        Pin.__init__(self, pin)
        self.light_no = SNLight(led_no)
        self.light_nc = SNLight(led_nc)
        self._en_auto_lights = True

    def auto_light(self, value=None):
        if value is not None:
            self._en_auto_lights = value
        return self._en_auto_lights

    def setup(self):
        if self._is_setup:
            return False

        setup()

        self.pin = _gpiochip.line_offset_from_id(self.pin)

        if is_automation_phat() and self.name == "one":
            self.pin = RELAY_3

        self._gpioline = _gpiochip.request_lines(consumer="AH", config={
            self.pin: gpiod.LineSettings(direction=Direction.OUTPUT, bias=Bias.DISABLED, output_value=Value.INACTIVE)
        })

        self._is_setup = True
        return True

    def write(self, value):
        """Write a value to the relay.

        :param value: Value to write, either 0 for LOW or 1 for HIGH
        """
        self.setup()

        if is_automation_phat() and self.name in ["two", "three"]:
            warnings.warn("Relay '{}' is not supported on Automation pHAT".format(self.name))

        self._gpioline.set_value(self.pin, Value.ACTIVE if value else Value.INACTIVE)

        if self._en_auto_lights:
            if value:
                self.light_no.write(1)
                self.light_nc.write(0)
            else:
                self.light_no.write(0)
                self.light_nc.write(1)


def _update_lights():
    global _lights_need_updating

    analog.read()
    input.read()

    if _lights_need_updating:
        lights.output(_led_states)
        _lights_need_updating = False

    time.sleep(1.0 / UPDATES_PER_SECOND)


def is_automation_hat():
    setup()
    return lights is not None


def is_automation_phat():
    setup()
    return lights is None


def enable_auto_lights(state):
    global _t_update_lights

    setup()

    if lights is None:
        return

    input.auto_light(state)
    output.auto_light(state)
    relay.auto_light(state)
    analog.auto_light(state)

    if state and _t_update_lights is None:
        _t_update_lights = AsyncWorker(_update_lights)
        _t_update_lights.start()

    if not state and _t_update_lights is not None:
        _t_update_lights.stop()
        _t_update_lights.join()
        _t_update_lights = None


def setup():
    global automation_hat, automation_phat, lights, _ads1015, _is_setup, _t_update_lights, _gpiochip

    if _is_setup:
        return True

    _is_setup = True

    _gpiochip = gpiodevice.find_chip_by_platform()

    _ads1015 = ads1015.ADS1015()
    try:
        chip_type = _ads1015.detect_chip_type()
    except IOError:
        raise RuntimeError("No ADC detected, check your connections")

    if chip_type == 'ADS1015':
        _ads1015.set_sample_rate(1600)
    else:
        _ads1015.set_sample_rate(860)

    _ads1015.set_programmable_gain(4.096)


    try:
        lights = sn3218.SN3218()
    except (IOError, OSError):
        pass

    if lights is not None:
        lights.enable()
        lights.enable_leds(0b111111111111111111)
        automation_hat = True
        automation_phat = False
        _t_update_lights = AsyncWorker(_update_lights)
        _t_update_lights.start()

    atexit.register(_exit)


def _exit():
    if _t_update_lights:
        _t_update_lights.stop()
        _t_update_lights.join()

    if lights is not None:
        lights.output([0] * 18)


analog = ObjectCollection()
analog._add(one=AnalogInput(0, 25.85, 0))
analog._add(two=AnalogInput(1, 25.85, 1))
analog._add(three=AnalogInput(2, 25.85, 2))
analog._add(four=AnalogInput(3, 3.3, None))

input = ObjectCollection()
input._add(one=Input(INPUT_1, 14))
input._add(two=Input(INPUT_2, 13))
input._add(three=Input(INPUT_3, 12))

output = ObjectCollection()
output._add(one=Output(OUTPUT_1, 3))
output._add(two=Output(OUTPUT_2, 4))
output._add(three=Output(OUTPUT_3, 5))

relay = ObjectCollection()

relay._add(one=Relay(RELAY_1, 6, 7))
relay._add(two=Relay(RELAY_2, 8, 9))
relay._add(three=Relay(RELAY_3, 10, 11))

light = ObjectCollection()
light._add(power=SNLight(17))
light._add(comms=SNLight(16))
light._add(warn=SNLight(15))


1	import atexit	1✔
2	import time	1✔
3	import warnings	1✔
4
5	import ads1015	1✔
6	import gpiod	1✔
7	import gpiodevice	1✔
8	import sn3218	1✔
9	from gpiod.line import Bias, Direction, Value	1✔
10
11	from .pins import AsyncWorker, ObjectCollection, StoppableThread # noqa: F401	1✔
12
13	__version__ = '0.4.1'	1✔
14
15
16	RELAY_1 = 13	1✔
17	RELAY_2 = 19	1✔
18	RELAY_3 = 16	1✔
19
20	INPUT_1 = 26	1✔
21	INPUT_2 = 20	1✔
22	INPUT_3 = 21	1✔
23
24	OUTPUT_1 = 5	1✔
25	OUTPUT_2 = 12	1✔
26	OUTPUT_3 = 6	1✔
27
28	UPDATES_PER_SECOND = 30	1✔
29
30	i2c = None	1✔
31	lights = None	1✔
32
33	automation_hat = False	1✔
34	automation_phat = True	1✔
35
36	_led_states = [0] * 18	1✔
37	_lights_need_updating = False	1✔
38	_is_setup = False	1✔
39	_t_update_lights = None	1✔
40	_ads1015 = None	1✔
41	_gpiochip = None	1✔
42
43
44	class SNLight(object):	1✔
45	def __init__(self, index):	1✔
46	self.index = index	1✔
47	self._max_brightness = float(128)	1✔
48
49	def toggle(self):	1✔
50	"""Toggle the light from on to off or off to on"""
51	self.write(1 - self.read())	×
52
53	def on(self):	1✔
54	"""Turn the light on"""
55	self.write(1)	×
56
57	def off(self):	1✔
58	"""Turn the light off"""
59	self.write(0)	×
60
61	def read(self):	1✔
62	"""Read the current status of the light"""
63	if self.index is None:	×
64	return	×
65
66	return _led_states[self.index] / self._max_brightness	×
67
68	def write(self, value):	1✔
69	"""Write a specific value to the light
70
71	:param value: Brightness of the light, from 0.0 to 1.0
72	"""
73	global _lights_need_updating
74
75	if self.index is None:	1✔
76	return	1✔
77
78	if not isinstance(value, (int, float)):	1✔
79	raise TypeError("Value must be int or float")	×
80
81	if value >= 0 and value <= 1.0:	1✔
82	_led_states[self.index] = int(self._max_brightness * value)	1✔
83	if _t_update_lights is None and lights is not None:	1✔
84	lights.output(_led_states)	×
85	else:
86	_lights_need_updating = True	1✔
87
88	else:
89	raise ValueError("Value must be between 0.0 and 1.0")	×
90
91
92	class AnalogInput(object):	1✔
93	type = 'Analog Input'	1✔
94
95	def __init__(self, channel, max_voltage, led):	1✔
96	self._en_auto_lights = True	1✔
97	self.channel = channel	1✔
98	self.value = 0	1✔
99	self.max_voltage = float(max_voltage)	1✔
100	self.light = SNLight(led)	1✔
101	self._is_setup = False	1✔
102
103	def setup(self):	1✔
104	if self._is_setup:	1✔
105	return	1✔
106
107	setup()	1✔
108	self._is_setup = True	1✔
109
110	def auto_light(self, value=None):	1✔
111	if value is not None:	×
112	self._en_auto_lights = value	×
113	return self._en_auto_lights	×
114
115	def read(self):	1✔
116	"""Return the read voltage of the analog input"""
117	if self.name == "four" and is_automation_phat():	1✔
118	warnings.warn("Analog Four is not supported on Automation pHAT")	×
119
120	self._update()	1✔
121	return round(self.value * self.max_voltage, 3)	1✔
122
123	def _update(self):	1✔
124	self.setup()	1✔
125	self.value = _ads1015.get_voltage("in{}/gnd".format(self.channel)) / 3.3	1✔
126
127	if self._en_auto_lights:	1✔
128	adc = self.value	1✔
129	self.light.write(max(0.0, min(1.0, adc)))	1✔
130
131
132	class Pin(object):	1✔
133	type = 'Pin'	1✔
134
135	def __init__(self, pin):	1✔
136	self.pin = pin	1✔
137	self._last_value = None	1✔
138	self._is_setup = False	1✔
139	self._gpioline = None	1✔
140
141	def __call__(self):	1✔
142	return filter(lambda x: x[0] != '_', dir(self))	×
143
144	def read(self):	1✔
145	self.setup()	×
NEW 146	return self._gpioline.get_value(self.pin) == Value.ACTIVE	×
147
148	def setup(self):	1✔
149	pass	×
150
151	def has_changed(self):	1✔
152	value = self.read()	×
153
154	if self._last_value is None:	×
155	self._last_value = value	×
156
157	if value is not self._last_value:	×
158	self._last_value = value	×
159	return True	×
160
161	return False	×
162
163	def is_on(self):	1✔
NEW 164	return self.read() is True	×
165
166	def is_off(self):	1✔
NEW 167	return self.read() is False	×
168
169
170	class Input(Pin):	1✔
171	type = 'Digital Input'	1✔
172
173	def __init__(self, pin, led):	1✔
174	self._en_auto_lights = True	1✔
175	Pin.__init__(self, pin)	1✔
176	self.light = SNLight(led)	1✔
177
178	def setup(self):	1✔
179	if self._is_setup:	1✔
180	return False	×
181
182	setup()	1✔
183
184	self.pin = _gpiochip.line_offset_from_id(self.pin)	1✔
185
186	self._gpioline = _gpiochip.request_lines(consumer="AH", config={	1✔
187	self.pin: gpiod.LineSettings(direction=Direction.INPUT, bias=Bias.DISABLED)
188	})
189
190	self._is_setup = True	1✔
191
192	def auto_light(self, value=None):	1✔
193	if value is not None:	×
194	self._en_auto_lights = value	×
195	return self._en_auto_lights	×
196
197	def read(self):	1✔
198	self.setup()	1✔
199	value = self._gpioline.get_value(self.pin) == Value.ACTIVE	1✔
200	if self._en_auto_lights:	1✔
201	self.light.write(value)	1✔
202	return value	1✔
203
204
205	class Output(Pin):	1✔
206	type = 'Digital Output'	1✔
207
208	def __init__(self, pin, led):	1✔
209	self._en_auto_lights = True	1✔
210	Pin.__init__(self, pin)	1✔
211	self.light = SNLight(led)	1✔
212
213	def setup(self):	1✔
214	if self._is_setup:	×
215	return False	×
216
217	setup()	×
218
NEW 219	self.pin = _gpiochip.line_offset_from_id(self.pin)	×
220
NEW 221	self._gpioline = _gpiochip.request_lines(consumer="AH", config={	×
222	self.pin: gpiod.LineSettings(direction=Direction.OUTPUT, bias=Bias.DISABLED, output_value=Value.INACTIVE)
223	})
224
225	self._is_setup = True	×
226	return True	×
227
228	def auto_light(self, value=None):	1✔
229	if value is not None:	×
230	self._en_auto_lights = value	×
231	return self._en_auto_lights	×
232
233	def write(self, value):	1✔
234	"""Write a value to the output.
235
236	:param value: Value to write, either 1 for HIGH or 0 for LOW
237	"""
238	self.setup()	×
NEW 239	self._gpioline.set_value(self.pin, Value.ACTIVE if value else Value.INACTIVE)	×
240	if self._en_auto_lights:	×
241	self.light.write(1 if value else 0)	×
242
243	def on(self):	1✔
244	"""Turn the output on/HIGH"""
245	self.write(1)	×
246
247	def off(self):	1✔
248	"""Turn the output off/LOW"""
249	self.write(0)	×
250
251	def toggle(self):	1✔
252	"""Toggle the output."""
253	self.write(not self.read())	×
254
255
256	class Relay(Output):	1✔
257	type = 'Relay'	1✔
258
259	def __init__(self, pin, led_no, led_nc):	1✔
260	Pin.__init__(self, pin)	1✔
261	self.light_no = SNLight(led_no)	1✔
262	self.light_nc = SNLight(led_nc)	1✔
263	self._en_auto_lights = True	1✔
264
265	def auto_light(self, value=None):	1✔
266	if value is not None:	×
267	self._en_auto_lights = value	×
268	return self._en_auto_lights	×
269
270	def setup(self):	1✔
271	if self._is_setup:	×
272	return False	×
273
274	setup()	×
275
NEW 276	self.pin = _gpiochip.line_offset_from_id(self.pin)	×
277
278	if is_automation_phat() and self.name == "one":	×
279	self.pin = RELAY_3	×
280
NEW 281	self._gpioline = _gpiochip.request_lines(consumer="AH", config={	×
282	self.pin: gpiod.LineSettings(direction=Direction.OUTPUT, bias=Bias.DISABLED, output_value=Value.INACTIVE)
283	})
284
285	self._is_setup = True	×
286	return True	×
287
288	def write(self, value):	1✔
289	"""Write a value to the relay.
290
291	:param value: Value to write, either 0 for LOW or 1 for HIGH
292	"""
293	self.setup()	×
294
295	if is_automation_phat() and self.name in ["two", "three"]:	×
296	warnings.warn("Relay '{}' is not supported on Automation pHAT".format(self.name))	×
297
NEW 298	self._gpioline.set_value(self.pin, Value.ACTIVE if value else Value.INACTIVE)	×
299
300	if self._en_auto_lights:	×
301	if value:	×
302	self.light_no.write(1)	×
303	self.light_nc.write(0)	×
304	else:
305	self.light_no.write(0)	×
306	self.light_nc.write(1)	×
307
308
309	def _update_lights():	1✔
310	global _lights_need_updating
311
312	analog.read()	1✔
313	input.read()	1✔
314
315	if _lights_need_updating:	1✔
316	lights.output(_led_states)	1✔
317	_lights_need_updating = False	1✔
318
319	time.sleep(1.0 / UPDATES_PER_SECOND)	1✔
320
321
322	def is_automation_hat():	1✔
323	setup()	×
324	return lights is not None	×
325
326
327	def is_automation_phat():	1✔
328	setup()	1✔
329	return lights is None	1✔
330
331
332	def enable_auto_lights(state):	1✔
333	global _t_update_lights
334
335	setup()	×
336
337	if lights is None:	×
338	return	×
339
340	input.auto_light(state)	×
341	output.auto_light(state)	×
342	relay.auto_light(state)	×
343	analog.auto_light(state)	×
344
345	if state and _t_update_lights is None:	×
346	_t_update_lights = AsyncWorker(_update_lights)	×
347	_t_update_lights.start()	×
348
349	if not state and _t_update_lights is not None:	×
350	_t_update_lights.stop()	×
351	_t_update_lights.join()	×
352	_t_update_lights = None	×
353
354
355	def setup():	1✔
356	global automation_hat, automation_phat, lights, _ads1015, _is_setup, _t_update_lights, _gpiochip
357
358	if _is_setup:	1✔
359	return True	1✔
360
361	_is_setup = True	1✔
362
363	_gpiochip = gpiodevice.find_chip_by_platform()	1✔
364
365	_ads1015 = ads1015.ADS1015()	1✔
366	try:	1✔
367	chip_type = _ads1015.detect_chip_type()	1✔
368	except IOError:	×
369	raise RuntimeError("No ADC detected, check your connections")	×
370
371	if chip_type == 'ADS1015':	1✔
372	_ads1015.set_sample_rate(1600)	1✔
373	else:
UNCOV 374	_ads1015.set_sample_rate(860)	×
375
376	_ads1015.set_programmable_gain(4.096)	1✔
377
378
379	try:	1✔
380	lights = sn3218.SN3218()	1✔
381	except (IOError, OSError):	×
382	pass	×
383
384	if lights is not None:	1✔
385	lights.enable()	1✔
386	lights.enable_leds(0b111111111111111111)	1✔
387	automation_hat = True	1✔
388	automation_phat = False	1✔
389	_t_update_lights = AsyncWorker(_update_lights)	1✔
390	_t_update_lights.start()	1✔
391
392	atexit.register(_exit)	1✔
393
394
395	def _exit():	1✔
396	if _t_update_lights:	×
397	_t_update_lights.stop()	×
398	_t_update_lights.join()	×
399
400	if lights is not None:	×
401	lights.output([0] * 18)	×
402
403
404	analog = ObjectCollection()	1✔
405	analog._add(one=AnalogInput(0, 25.85, 0))	1✔
406	analog._add(two=AnalogInput(1, 25.85, 1))	1✔
407	analog._add(three=AnalogInput(2, 25.85, 2))	1✔
408	analog._add(four=AnalogInput(3, 3.3, None))	1✔
409
410	input = ObjectCollection()	1✔
411	input._add(one=Input(INPUT_1, 14))	1✔
412	input._add(two=Input(INPUT_2, 13))	1✔
413	input._add(three=Input(INPUT_3, 12))	1✔
414
415	output = ObjectCollection()	1✔
416	output._add(one=Output(OUTPUT_1, 3))	1✔
417	output._add(two=Output(OUTPUT_2, 4))	1✔
418	output._add(three=Output(OUTPUT_3, 5))	1✔
419
420	relay = ObjectCollection()	1✔
421
422	relay._add(one=Relay(RELAY_1, 6, 7))	1✔
423	relay._add(two=Relay(RELAY_2, 8, 9))	1✔
424	relay._add(three=Relay(RELAY_3, 10, 11))	1✔
425
426	light = ObjectCollection()	1✔
427	light._add(power=SNLight(17))	1✔
428	light._add(comms=SNLight(16))	1✔
429	light._add(warn=SNLight(15))	1✔
430

pimoroni / automation-hat / 8877213793

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous