mirror of
https://github.com/sinseman44/PyCNC.git
synced 2026-07-16 08:37:09 +00:00
reorganise project
This commit is contained in:
Nikolay Khabarov
Executable
+289
@@ -0,0 +1,289 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
from rpgpio_private import *
|
||||
|
||||
import time
|
||||
import logging
|
||||
import sys
|
||||
import struct
|
||||
|
||||
|
||||
class GPIO(object):
|
||||
MODE_OUTPUT = 1
|
||||
MODE_INPUT_NOPULL = 2
|
||||
MODE_INPUT_PULLUP = 3
|
||||
MODE_INPUT_PULLDOWN = 4
|
||||
|
||||
def __init__(self):
|
||||
""" Create object which can control GPIO.
|
||||
This class writes directly to CPU registers and doesn't use any libs
|
||||
or kernel modules.
|
||||
"""
|
||||
self._mem = PhysicalMemory(PERI_BASE + GPIO_REGISTER_BASE)
|
||||
|
||||
def _pullupdn(self, pin, mode):
|
||||
p = self._mem.read_int(GPIO_PULLUPDN_OFFSET)
|
||||
p &= ~3
|
||||
if mode == self.MODE_INPUT_PULLUP:
|
||||
p |= 2
|
||||
elif mode == self.MODE_INPUT_PULLDOWN:
|
||||
p |= 1
|
||||
self._mem.write_int(GPIO_PULLUPDN_OFFSET, p)
|
||||
addr = 4 * int(pin / 32) + GPIO_PULLUPDNCLK_OFFSET
|
||||
self._mem.write_int(addr, 1 << (pin % 32))
|
||||
p = self._mem.read_int(GPIO_PULLUPDN_OFFSET)
|
||||
p &= ~3
|
||||
self._mem.write_int(GPIO_PULLUPDN_OFFSET, p)
|
||||
self._mem.write_int(addr, 0)
|
||||
|
||||
def init(self, pin, mode):
|
||||
""" Initialize or re-initialize GPIO pin.
|
||||
:param pin: pin number.
|
||||
:param mode: one of MODE_* variables in this class.
|
||||
"""
|
||||
addr = 4 * int(pin / 10) + GPIO_FSEL_OFFSET
|
||||
v = self._mem.read_int(addr)
|
||||
v &= ~(7 << ((pin % 10) * 3)) # input value
|
||||
if mode == self.MODE_OUTPUT:
|
||||
v |= (1 << ((pin % 10) * 3)) # output value, base on input
|
||||
self._mem.write_int(addr, v)
|
||||
else:
|
||||
self._mem.write_int(addr, v)
|
||||
self._pullupdn(pin, mode)
|
||||
|
||||
def set(self, pin):
|
||||
""" Set pin to HIGH state.
|
||||
:param pin: pin number.
|
||||
"""
|
||||
addr = 4 * int(pin / 32) + GPIO_SET_OFFSET
|
||||
self._mem.write_int(addr, 1 << (pin % 32))
|
||||
|
||||
def clear(self, pin):
|
||||
""" Set pin to LOW state.
|
||||
:param pin: pin number.
|
||||
"""
|
||||
addr = 4 * int(pin / 32) + GPIO_CLEAR_OFFSET
|
||||
self._mem.write_int(addr, 1 << (pin % 32))
|
||||
|
||||
def read(self, pin):
|
||||
""" Read pin current value.
|
||||
:param pin: pin number.
|
||||
:return: integer value 0 or 1.
|
||||
"""
|
||||
addr = 4 * int(pin / 32) + GPIO_INPUT_OFFSET
|
||||
v = self._mem.read_int(addr)
|
||||
v &= 1 << (pin % 32)
|
||||
if v == 0:
|
||||
return 0
|
||||
return 1
|
||||
|
||||
|
||||
class DMAGPIO(object):
|
||||
_DMA_CONTROL_BLOCK_SIZE = 32
|
||||
_DMA_CHANNEL = 5
|
||||
|
||||
def __init__(self):
|
||||
""" Create object which control GPIO pins via DMA(Direct Memory
|
||||
Access).
|
||||
This object allows to add arbitrary sequence of pulses to any GPIO
|
||||
outputs and run this sequence in background without using CPU since
|
||||
DMA is a separated hardware module.
|
||||
Note: keep this object out of garbage collector until it stops,
|
||||
otherwise memory will be unlocked and it could be overwritten by
|
||||
operating system.
|
||||
"""
|
||||
# allocate buffer for control blocks, always 32 MB
|
||||
self._physmem = CMAPhysicalMemory(32 * 1024 * 1024)
|
||||
self.__current_address = 0
|
||||
|
||||
# prepare dma registers memory map
|
||||
self._dma = PhysicalMemory(PERI_BASE + DMA_BASE)
|
||||
self._pwm = PhysicalMemory(PERI_BASE + PWM_BASE)
|
||||
self._clock = PhysicalMemory(PERI_BASE + CM_BASE)
|
||||
|
||||
# pre calculated variables for control blocks
|
||||
self._delay_info = DMA_TI_NO_WIDE_BURSTS | DMA_SRC_IGNORE \
|
||||
| DMA_TI_PER_MAP(DMA_TI_PER_MAP_PWM) \
|
||||
| DMA_TI_DEST_DREQ
|
||||
self._delay_destination = PHYSICAL_PWM_BUS + 0x18
|
||||
self._delay_stride = 0
|
||||
|
||||
self._pulse_info = DMA_TI_NO_WIDE_BURSTS | DMA_TI_TDMODE \
|
||||
| DMA_TI_WAIT_RESP
|
||||
self._pulse_destination = PHYSICAL_GPIO_BUS + GPIO_SET_OFFSET
|
||||
# YLENGTH is transfers count and XLENGTH size of each transfer
|
||||
self._pulse_length = DMA_TI_TXFR_LEN_YLENGTH(2) \
|
||||
| DMA_TI_TXFR_LEN_XLENGTH(4)
|
||||
self._pulse_stride = DMA_TI_STRIDE_D_STRIDE(12) \
|
||||
| DMA_TI_STRIDE_S_STRIDE(4)
|
||||
|
||||
def add_pulse(self, pins_mask, length_us):
|
||||
""" Add single pulse at the current position.
|
||||
:param pins_mask: bitwise mask of GPIO pins to trigger. Only for first 32 pins.
|
||||
:param length_us: length in us.
|
||||
"""
|
||||
next_cb = self.__current_address + 3 * self._DMA_CONTROL_BLOCK_SIZE
|
||||
if next_cb > self._physmem.get_size():
|
||||
raise MemoryError("Out of allocated memory.")
|
||||
next3 = next_cb + self._physmem.get_bus_address()
|
||||
next2 = next3 - self._DMA_CONTROL_BLOCK_SIZE
|
||||
next1 = next2 - self._DMA_CONTROL_BLOCK_SIZE
|
||||
|
||||
source1 = next1 - 8 # last 8 bytes are padding, use it to store data
|
||||
length2 = 16 * length_us
|
||||
source3 = next3 - 8
|
||||
|
||||
data = (
|
||||
self._pulse_info, source1, self._pulse_destination,
|
||||
self._pulse_length,
|
||||
self._pulse_stride, next1, pins_mask, 0,
|
||||
self._delay_info, 0, self._delay_destination, length2,
|
||||
self._delay_stride, next2, 0, 0,
|
||||
self._pulse_info, source3, self._pulse_destination,
|
||||
self._pulse_length,
|
||||
self._pulse_stride, next3, 0, pins_mask
|
||||
)
|
||||
self._physmem.write(self.__current_address, data)
|
||||
self.__current_address = next_cb
|
||||
|
||||
def add_delay(self, delay_us):
|
||||
""" Add delay at the current position.
|
||||
:param delay_us: delay in us.
|
||||
"""
|
||||
next_cb = self.__current_address + self._DMA_CONTROL_BLOCK_SIZE
|
||||
if next_cb > self._physmem.get_size():
|
||||
raise MemoryError("Out of allocated memory.")
|
||||
next = self._physmem.get_bus_address() + next_cb
|
||||
source = next - 8 # last 8 bytes are padding, use it to store data
|
||||
length = 16 * delay_us
|
||||
data = (
|
||||
self._delay_info, source, self._delay_destination, length,
|
||||
self._delay_stride, next, 0, 0
|
||||
)
|
||||
self._physmem.write(self.__current_address, data)
|
||||
self.__current_address = next_cb
|
||||
|
||||
def finalize_stream(self):
|
||||
""" Mark last added block as the last one.
|
||||
"""
|
||||
self._physmem.write_int(self.__current_address + 20
|
||||
- self._DMA_CONTROL_BLOCK_SIZE, 0)
|
||||
logging.info("DMA took {}MB of memory".
|
||||
format(round(self.__current_address / 1024.0 / 1024.0, 2)))
|
||||
|
||||
def run_stream(self):
|
||||
""" Run DMA module in stream mode, i.e. does'n finalize last block
|
||||
and do not check if there is anything to do.
|
||||
"""
|
||||
# configure PWM hardware module which will clocks DMA
|
||||
self._pwm.write_int(PWM_CTL, 0)
|
||||
self._clock.write_int(CM_CNTL, CM_PASSWORD | CM_SRC_PLLD) # disable
|
||||
while (self._clock.read_int(CM_CNTL) & (1 << 7)) != 0:
|
||||
time.sleep(0.00001) # 10 us, wait until BUSY bit is clear
|
||||
self._clock.write_int(CM_DIV, CM_PASSWORD | CM_DIV_VALUE(50)) # 10MHz
|
||||
self._clock.write_int(CM_CNTL, CM_PASSWORD | CM_SRC_PLLD | CM_ENABLE)
|
||||
self._pwm.write_int(PWM_RNG1, 100)
|
||||
self._pwm.write_int(PWM_DMAC, PWM_DMAC_ENAB
|
||||
| PWM_DMAC_PANIC(15) | PWM_DMAC_DREQ(15))
|
||||
self._pwm.write_int(PWM_CTL, PWM_CTL_CLRF)
|
||||
self._pwm.write_int(PWM_CTL, PWM_CTL_USEF1 | PWM_CTL_PWEN1)
|
||||
# configure DMA
|
||||
addr = 0x100 * self._DMA_CHANNEL
|
||||
cs = self._dma.read_int(addr)
|
||||
cs |= DMA_CS_END
|
||||
self._dma.write_int(addr, cs)
|
||||
self._dma.write_int(addr + 4, self._physmem.get_bus_address())
|
||||
cs = DMA_CS_PRIORITY(7) | DMA_CS_PANIC_PRIORITY(7) | DMA_CS_DISDEBUG
|
||||
self._dma.write_int(addr, cs)
|
||||
cs |= DMA_CS_ACTIVE
|
||||
self._dma.write_int(addr, cs)
|
||||
|
||||
def run(self, loop=False):
|
||||
""" Run DMA module and start sending specified pulses.
|
||||
:param loop: If true, run pulse sequence in infinite loop. Otherwise
|
||||
"""
|
||||
if self.__current_address == 0:
|
||||
raise RuntimeError("Nothing was added.")
|
||||
# fix 'next' field in previous control block
|
||||
if loop:
|
||||
self._physmem.write_int(self.__current_address + 20
|
||||
- self._DMA_CONTROL_BLOCK_SIZE,
|
||||
self._physmem.get_bus_address())
|
||||
else:
|
||||
self.finalize_stream()
|
||||
self.run_stream()
|
||||
|
||||
def stop(self):
|
||||
""" Stop any DMA activities.
|
||||
"""
|
||||
self._pwm.write_int(PWM_CTL, 0)
|
||||
addr = 0x100 * self._DMA_CHANNEL
|
||||
cs = self._dma.read_int(addr)
|
||||
cs |= DMA_CS_ABORT
|
||||
self._dma.write_int(addr, cs)
|
||||
cs &= ~DMA_CS_ACTIVE
|
||||
self._dma.write_int(addr, cs)
|
||||
cs |= DMA_CS_RESET
|
||||
self._dma.write_int(addr, cs)
|
||||
|
||||
def is_active(self):
|
||||
""" Check if DMA is working. Method can check if single sent sequence
|
||||
still active.
|
||||
:return: boolean value
|
||||
"""
|
||||
addr = 0x100 * self._DMA_CHANNEL
|
||||
cs = self._dma.read_int(addr)
|
||||
if cs & DMA_CS_ACTIVE == DMA_CS_ACTIVE:
|
||||
return True
|
||||
return False
|
||||
|
||||
def clear(self):
|
||||
""" Remove any specified pulses.
|
||||
"""
|
||||
self.__current_address = 0
|
||||
|
||||
|
||||
# for testing purpose
|
||||
def main():
|
||||
pin = 21
|
||||
g = GPIO()
|
||||
g.init(pin, GPIO.MODE_INPUT_NOPULL)
|
||||
print("nopull " + str(g.read(pin)))
|
||||
g.init(pin, GPIO.MODE_INPUT_PULLDOWN)
|
||||
print("pulldown " + str(g.read(pin)))
|
||||
g.init(pin, GPIO.MODE_INPUT_PULLUP)
|
||||
print("pullup " + str(g.read(pin)))
|
||||
time.sleep(1)
|
||||
g.init(pin, GPIO.MODE_OUTPUT)
|
||||
g.set(pin)
|
||||
print("set " + str(g.read(pin)))
|
||||
time.sleep(1)
|
||||
g.clear(pin)
|
||||
print("clear " + str(g.read(pin)))
|
||||
time.sleep(1)
|
||||
cma = CMAPhysicalMemory(1*1024*1024)
|
||||
print(str(cma.get_size() / 1024 / 1024) + "MB of memory allocated at " \
|
||||
+ hex(cma.get_phys_address()))
|
||||
a = cma.read_int(0)
|
||||
print("was " + hex(a))
|
||||
cma.write_int(0, 0x12345678)
|
||||
a = cma.read_int(0)
|
||||
assert a == 0x12345678, "Memory isn't written or read correctly"
|
||||
print("now " + hex(a))
|
||||
del cma
|
||||
dg = DMAGPIO()
|
||||
dg.add_pulse(1 << pin, 4000)
|
||||
dg.add_delay(12000)
|
||||
dg.run(True)
|
||||
print("dmagpio is started")
|
||||
try:
|
||||
print("press enter to stop...")
|
||||
sys.stdin.readline()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
dg.stop()
|
||||
g.clear(pin)
|
||||
print("dma stopped")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user