+++ /dev/null
-"""Pin definitions for Raspberry Pi Compute Modules."""
-
-from adafruit_blinka.microcontroller.bcm2711 import pin
-
-D2 = pin.D2
-SDA = pin.SDA
-D3 = pin.D3
-SCL = pin.SCL
-
-D4 = pin.D4
-D5 = pin.D5
-D6 = pin.D6
-
-D7 = pin.D7
-CE1 = pin.D7
-D8 = pin.D8
-CE0 = pin.D8
-D9 = pin.D9
-MISO = pin.D9
-D10 = pin.D10
-MOSI = pin.D10
-D11 = pin.D11
-SCLK = pin.D11
-SCK = pin.D11
-
-D12 = pin.D12
-D13 = pin.D13
-
-D14 = pin.D14
-TXD = pin.D14
-D15 = pin.D15
-RXD = pin.D15
-# create alias for most of the examples
-TX = pin.D14
-RX = pin.D15
-
-D16 = pin.D16
-D17 = pin.D17
-D18 = pin.D18
-D19 = pin.D19
-MISO_1 = pin.D19
-D20 = pin.D20
-MOSI_1 = pin.D20
-D21 = pin.D21
-SCLK_1 = pin.D21
-SCK_1 = pin.D21
-D22 = pin.D22
-D23 = pin.D23
-D24 = pin.D24
-D25 = pin.D25
-D26 = pin.D26
-D27 = pin.D27
-D28 = pin.D28
-D29 = pin.D29
-D30 = pin.D30
-D31 = pin.D31
-D32 = pin.D32
-D33 = pin.D33
-D34 = pin.D34
-D35 = pin.D35
-D36 = pin.D36
-D37 = pin.D37
-D38 = pin.D38
-D39 = pin.D39
-D40 = pin.D40
-MISO_2 = pin.D40
-D41 = pin.D41
-MOSI_2 = pin.D41
-D42 = pin.D42
-SCLK_2 = pin.D42
-SCK_2 = pin.D43
-D43 = pin.D43
-D44 = pin.D44
-D45 = pin.D45
+++ /dev/null
-"""BCM2711 NeoPixel Driver Class"""
-import time
-import atexit
-import _rpi_ws281x as ws
-
-# LED configuration.
-# pylint: disable=redefined-outer-name,too-many-branches,too-many-statements
-# pylint: disable=global-statement,protected-access
-LED_CHANNEL = 0
-LED_FREQ_HZ = 800000 # Frequency of the LED signal. We only support 800KHz
-LED_DMA_NUM = 10 # DMA channel to use, can be 0-14.
-LED_BRIGHTNESS = 255 # We manage the brightness in the neopixel library
-LED_INVERT = 0 # We don't support inverted logic
-LED_STRIP = None # We manage the color order within the neopixel library
-
-# a 'static' object that we will use to manage our PWM DMA channel
-# we only support one LED strip per raspi
-_led_strip = None
-_buf = None
-
-
-def neopixel_write(gpio, buf):
- """NeoPixel Writing Function"""
- global _led_strip # we'll have one strip we init if its not at first
- global _buf # we save a reference to the buf, and if it changes we will cleanup and re-init.
-
- if _led_strip is None or buf is not _buf:
- # This is safe to call since it doesn't do anything if _led_strip is None
- neopixel_cleanup()
-
- # Create a ws2811_t structure from the LED configuration.
- # Note that this structure will be created on the heap so you
- # need to be careful that you delete its memory by calling
- # delete_ws2811_t when it's not needed.
- _led_strip = ws.new_ws2811_t()
- _buf = buf
-
- # Initialize all channels to off
- for channum in range(2):
- channel = ws.ws2811_channel_get(_led_strip, channum)
- ws.ws2811_channel_t_count_set(channel, 0)
- ws.ws2811_channel_t_gpionum_set(channel, 0)
- ws.ws2811_channel_t_invert_set(channel, 0)
- ws.ws2811_channel_t_brightness_set(channel, 0)
-
- channel = ws.ws2811_channel_get(_led_strip, LED_CHANNEL)
-
- # Initialize the channel in use
- count = 0
- if len(buf) % 3 == 0:
- # most common, divisible by 3 is likely RGB
- LED_STRIP = ws.WS2811_STRIP_RGB
- count = len(buf) // 3
- elif len(buf) % 4 == 0:
- LED_STRIP = ws.SK6812_STRIP_RGBW
- count = len(buf) // 4
- else:
- raise RuntimeError("We only support 3 or 4 bytes-per-pixel")
-
- ws.ws2811_channel_t_count_set(
- channel, count
- ) # we manage 4 vs 3 bytes in the library
- ws.ws2811_channel_t_gpionum_set(channel, gpio._pin.id)
- ws.ws2811_channel_t_invert_set(channel, LED_INVERT)
- ws.ws2811_channel_t_brightness_set(channel, LED_BRIGHTNESS)
- ws.ws2811_channel_t_strip_type_set(channel, LED_STRIP)
-
- # Initialize the controller
- ws.ws2811_t_freq_set(_led_strip, LED_FREQ_HZ)
- ws.ws2811_t_dmanum_set(_led_strip, LED_DMA_NUM)
-
- resp = ws.ws2811_init(_led_strip)
- if resp != ws.WS2811_SUCCESS:
- if resp == -5:
- raise RuntimeError(
- "NeoPixel support requires running with sudo, please try again!"
- )
- message = ws.ws2811_get_return_t_str(resp)
- raise RuntimeError(
- "ws2811_init failed with code {0} ({1})".format(resp, message)
- )
- atexit.register(neopixel_cleanup)
-
- channel = ws.ws2811_channel_get(_led_strip, LED_CHANNEL)
- if gpio._pin.id != ws.ws2811_channel_t_gpionum_get(channel):
- raise RuntimeError("Raspberry Pi neopixel support is for one strip only!")
-
- if ws.ws2811_channel_t_strip_type_get(channel) == ws.WS2811_STRIP_RGB:
- bpp = 3
- else:
- bpp = 4
- # assign all colors!
- for i in range(len(buf) // bpp):
- r = buf[bpp * i]
- g = buf[bpp * i + 1]
- b = buf[bpp * i + 2]
- if bpp == 3:
- pixel = (r << 16) | (g << 8) | b
- else:
- w = buf[bpp * i + 3]
- pixel = (w << 24) | (r << 16) | (g << 8) | b
- ws.ws2811_led_set(channel, i, pixel)
-
- resp = ws.ws2811_render(_led_strip)
- if resp != ws.WS2811_SUCCESS:
- message = ws.ws2811_get_return_t_str(resp)
- raise RuntimeError(
- "ws2811_render failed with code {0} ({1})".format(resp, message)
- )
- time.sleep(0.001 * ((len(buf) // 100) + 1)) # about 1ms per 100 bytes
-
-
-def neopixel_cleanup():
- """Cleanup when we're done"""
- global _led_strip
-
- if _led_strip is not None:
- # Ensure ws2811_fini is called before the program quits.
- ws.ws2811_fini(_led_strip)
- # Example of calling delete function to clean up structure memory. Isn't
- # strictly necessary at the end of the program execution here, but is good practice.
- ws.delete_ws2811_t(_led_strip)
- _led_strip = None
+++ /dev/null
-"""Custom PWMOut Wrapper for Rpi.GPIO PWM Class"""
-import RPi.GPIO as GPIO
-
-GPIO.setmode(GPIO.BCM) # Use BCM pins D4 = GPIO #4
-GPIO.setwarnings(False) # shh!
-
-
-# pylint: disable=unnecessary-pass
-class PWMError(IOError):
- """Base class for PWM errors."""
-
- pass
-
-
-# pylint: enable=unnecessary-pass
-
-
-class PWMOut:
- """Pulse Width Modulation Output Class"""
-
- def __init__(self, pin, *, frequency=500, duty_cycle=0, variable_frequency=False):
- self._pwmpin = None
- self._period = 0
- self._open(pin, duty_cycle, frequency, variable_frequency)
-
- def __del__(self):
- self.deinit()
-
- def __enter__(self):
- return self
-
- def __exit__(self, t, value, traceback):
- self.deinit()
-
- def _open(self, pin, duty=0, freq=500, variable_frequency=False):
- self._pin = pin
- GPIO.setup(pin.id, GPIO.OUT)
- self._pwmpin = GPIO.PWM(pin.id, freq)
-
- if variable_frequency:
- print("Variable Frequency is not supported, continuing without it...")
-
- # set frequency
- self.frequency = freq
- # set duty
- self.duty_cycle = duty
-
- self.enabled = True
-
- def deinit(self):
- """Deinit the PWM."""
- if self._pwmpin is not None:
- self._pwmpin.stop()
- GPIO.cleanup(self._pin.id)
- self._pwmpin = None
-
- def _is_deinited(self):
- if self._pwmpin is None:
- raise ValueError(
- "Object has been deinitialize and can no longer "
- "be used. Create a new object."
- )
-
- @property
- def period(self):
- """Get or set the PWM's output period in seconds.
-
- Raises:
- PWMError: if an I/O or OS error occurs.
- TypeError: if value type is not int or float.
-
- :type: int, float
- """
- return 1.0 / self.frequency
-
- @period.setter
- def period(self, period):
- if not isinstance(period, (int, float)):
- raise TypeError("Invalid period type, should be int or float.")
-
- self.frequency = 1.0 / period
-
- @property
- def duty_cycle(self):
- """Get or set the PWM's output duty cycle which is the fraction of
- each pulse which is high. 16-bit
-
- Raises:
- PWMError: if an I/O or OS error occurs.
- TypeError: if value type is not int or float.
- ValueError: if value is out of bounds of 0.0 to 1.0.
-
- :type: int, float
- """
- return int(self._duty_cycle * 65535)
-
- @duty_cycle.setter
- def duty_cycle(self, duty_cycle):
- if not isinstance(duty_cycle, (int, float)):
- raise TypeError("Invalid duty cycle type, should be int or float.")
-
- if not 0 <= duty_cycle <= 65535:
- raise ValueError("Invalid duty cycle value, should be between 0 and 65535")
-
- # convert from 16-bit
- duty_cycle /= 65535.0
-
- self._duty_cycle = duty_cycle
- self._pwmpin.ChangeDutyCycle(round(self._duty_cycle * 100))
-
- @property
- def frequency(self):
- """Get or set the PWM's output frequency in Hertz.
-
- Raises:
- PWMError: if an I/O or OS error occurs.
- TypeError: if value type is not int or float.
-
- :type: int, float
- """
-
- return self._frequency
-
- @frequency.setter
- def frequency(self, frequency):
- if not isinstance(frequency, (int, float)):
- raise TypeError("Invalid frequency type, should be int or float.")
-
- self._pwmpin.ChangeFrequency(round(frequency))
- self._frequency = frequency
-
- @property
- def enabled(self):
- """Get or set the PWM's output enabled state.
-
- Raises:
- PWMError: if an I/O or OS error occurs.
- TypeError: if value type is not bool.
-
- :type: bool
- """
- return self._enabled
-
- @enabled.setter
- def enabled(self, value):
- if not isinstance(value, bool):
- raise TypeError("Invalid enabled type, should be string.")
-
- if value:
- self._pwmpin.start(round(self._duty_cycle * 100))
- else:
- self._pwmpin.stop()
-
- self._enabled = value
-
- # String representation
- def __str__(self):
- return "pin %s (freq=%f Hz, duty_cycle=%f%%)" % (
- self._pin,
- self.frequency,
- self.duty_cycle,
- )
+++ /dev/null
-"""Custom PulseIn Class to read PWM signals"""
-import time
-import subprocess
-import os
-import atexit
-import random
-import struct
-import sysv_ipc
-
-DEBUG = False
-queues = []
-procs = []
-
-# The message queues live outside of python space, and must be formally cleaned!
-def final():
- """In case the program is cancelled or quit, we need to clean up the PulseIn
- helper process and also the message queue, this is called at exit to do so"""
- if DEBUG:
- print("Cleaning up message queues", queues)
- print("Cleaning up processes", procs)
- for q in queues:
- q.remove()
- for proc in procs:
- proc.terminate()
-
-
-atexit.register(final)
-
-# pylint: disable=c-extension-no-member
-class PulseIn:
- """PulseIn Class to read PWM signals"""
-
- def __init__(self, pin, maxlen=2, idle_state=False):
- """Create a PulseIn object associated with the given pin.
- The object acts as a read-only sequence of pulse lengths with
- a given max length. When it is active, new pulse lengths are
- added to the end of the list. When there is no more room
- (len() == maxlen) the oldest pulse length is removed to make room."""
- self._pin = pin
- self._maxlen = maxlen
- self._idle_state = idle_state
- self._queue_key = random.randint(1, 9999)
- try:
- self._mq = sysv_ipc.MessageQueue(None, flags=sysv_ipc.IPC_CREX)
- if DEBUG:
- print("Message Queue Key: ", self._mq.key)
- queues.append(self._mq)
- except sysv_ipc.ExistentialError:
- raise RuntimeError(
- "Message queue creation failed"
- ) from sysv_ipc.ExistentialError
-
- # Check if OS is 64-bit
- if struct.calcsize("P") * 8 == 64:
- libgpiod_filename = "libgpiod_pulsein64"
- else:
- libgpiod_filename = "libgpiod_pulsein"
-
- dir_path = os.path.dirname(os.path.realpath(__file__))
- cmd = [
- dir_path + "/" + libgpiod_filename,
- "--pulses",
- str(maxlen),
- "--queue",
- str(self._mq.key),
- ]
- if idle_state:
- cmd.append("-i")
- cmd.append("gpiochip0")
- cmd.append(str(pin))
- if DEBUG:
- print(cmd)
-
- self._process = subprocess.Popen(cmd)
- procs.append(self._process)
-
- # wait for it to start up
- if DEBUG:
- print("Waiting for startup success message from subprocess")
- message = self._wait_receive_msg(timeout=0.25)
- if message[0] != b"!":
- raise RuntimeError("Could not establish message queue with subprocess")
- self._paused = False
-
- # pylint: disable=redefined-builtin
- def _wait_receive_msg(self, timeout=0, type=2):
- """Internal helper that will wait for new messages of a given type,
- and throw an exception on timeout"""
- if timeout > 0:
- stamp = time.monotonic()
- while (time.monotonic() - stamp) < timeout:
- try:
- message = self._mq.receive(block=False, type=type)
- return message
- except sysv_ipc.BusyError:
- time.sleep(0.001) # wait a bit then retry!
- # uh-oh timed out
- raise RuntimeError(
- "Timed out waiting for PulseIn message. Make sure libgpiod is installed."
- )
- message = self._mq.receive(block=True, type=type)
- return message
-
- # pylint: enable=redefined-builtin
-
- def deinit(self):
- """Deinitialises the PulseIn and releases any hardware and software
- resources for reuse."""
- # Clean up after ourselves
- self._process.terminate()
- procs.remove(self._process)
- self._mq.remove()
- queues.remove(self._mq)
-
- def __enter__(self):
- """No-op used by Context Managers."""
- return self
-
- def __exit__(self, exc_type, exc_value, tb):
- """Automatically deinitializes the hardware when exiting a context."""
- self.deinit()
-
- def resume(self, trigger_duration=0):
- """Resumes pulse capture after an optional trigger pulse."""
- if trigger_duration != 0:
- self._mq.send("t%d" % trigger_duration, True, type=1)
- else:
- self._mq.send("r", True, type=1)
- self._paused = False
-
- def pause(self):
- """Pause pulse capture"""
- self._mq.send("p", True, type=1)
- self._paused = True
-
- @property
- def paused(self):
- """True when pulse capture is paused as a result of pause() or
- an error during capture such as a signal that is too fast."""
- return self._paused
-
- @property
- def maxlen(self):
- """The maximum length of the PulseIn. When len() is equal to maxlen,
- it is unclear which pulses are active and which are idle."""
- return self._maxlen
-
- def clear(self):
- """Clears all captured pulses"""
- self._mq.send("c", True, type=1)
-
- def popleft(self):
- """Removes and returns the oldest read pulse."""
- self._mq.send("^", True, type=1)
- message = self._wait_receive_msg()
- reply = int(message[0].decode("utf-8"))
- # print(reply)
- if reply == -1:
- raise IndexError("pop from empty list")
- return reply
-
- def __len__(self):
- """Returns the current pulse length"""
- self._mq.send("l", True, type=1)
- message = self._wait_receive_msg()
- return int(message[0].decode("utf-8"))
-
- # pylint: disable=redefined-builtin
- def __getitem__(self, index, type=None):
- """Returns the value at the given index or values in slice."""
- self._mq.send("i%d" % index, True, type=1)
- message = self._wait_receive_msg()
- ret = int(message[0].decode("utf-8"))
- if ret == -1:
- raise IndexError("list index out of range")
- return ret
-
- # pylint: enable=redefined-builtin
elif board_id == ap_board.RASPBERRY_PI_PICO:
from adafruit_blinka.board.raspberrypi.pico import *
-elif detector.board.RASPBERRY_PI_4B or detector.board.RASPBERRY_PI_400:
+elif (
+ detector.board.RASPBERRY_PI_4B
+ or detector.board.RASPBERRY_PI_CM4
+ or detector.board.RASPBERRY_PI_400
+):
from adafruit_blinka.board.raspberrypi.raspi_4b import *
-elif detector.board.RASPBERRY_PI_CM4:
- from adafruit_blinka.board.raspberrypi.raspi_cm4 import *
-
elif detector.board.any_raspberry_pi_40_pin:
from adafruit_blinka.board.raspberrypi.raspi_40pin import *
projects / Adafruit_Blinka-hackapet.git / commitdiff