peterlu14
196 lines
4.5 KiB
Python
196 lines
4.5 KiB
Python
import abc
|
|
from typing import Optional, List, Tuple, Union
|
|
|
|
import spidev
|
|
|
|
from . import LowLevelHardwareInterface
|
|
|
|
_RUNTIME_COMPILE = False
|
|
|
|
|
|
class AbstractSpiInterface(LowLevelHardwareInterface, metaclass=abc.ABCMeta):
|
|
__slots__ = ()
|
|
|
|
@abc.abstractmethod
|
|
def send_byte(self, data: bytes) -> Optional[List[int]]:
|
|
pass
|
|
|
|
|
|
class HardwareImplSpiInterface(LowLevelHardwareInterface):
|
|
"""
|
|
**Linux driver**
|
|
|
|
The default Linux driver is spi-bcm2708.
|
|
|
|
The following information was valid 2014-07-05.
|
|
|
|
**SPEED**
|
|
|
|
The driver supports the following speeds:
|
|
|
|
======= ============ =================================
|
|
cdiv speed note
|
|
======= ============ =================================
|
|
2 125.0 MHz
|
|
4 62.5 MHz
|
|
8 31.2 MHz
|
|
16 15.6 MHz
|
|
32 7.8 MHz cc2650 master mode max (12MHz)
|
|
64 3.9 MHz cc2650 slave mode max (4MHz)
|
|
128 1953 kHz
|
|
256 976 kHz
|
|
512 488 kHz
|
|
1024 244 kHz
|
|
2048 122 kHz
|
|
4096 61 kHz
|
|
8192 30.5 kHz
|
|
16384 15.2 kHz
|
|
32768 7629 Hz
|
|
======= ============ =================================
|
|
|
|
**required pins**
|
|
|
|
*MISO*, *SCLK*, *GE0_N*
|
|
"""
|
|
|
|
__slots__ = ('_device', '_spi', '_spi_mode', 'pin_request', '_spi_speed')
|
|
|
|
def __init__(self,
|
|
device: Tuple[int, int] = None,
|
|
spi_mode=0x00,
|
|
spi_speed=12_000_000):
|
|
# SPI parameter
|
|
if device is None:
|
|
self._device = (0, 0)
|
|
else:
|
|
self._device = device
|
|
|
|
# SPI instance
|
|
self._spi = spidev.SpiDev()
|
|
self._spi_mode = spi_mode
|
|
self._spi_speed = spi_speed
|
|
|
|
def reset(self):
|
|
self._spi.open(*self._device)
|
|
|
|
self._spi.mode = self._spi_mode
|
|
self._spi.max_speed_hz = self._spi_speed
|
|
|
|
def close(self):
|
|
self._spi.close()
|
|
|
|
def flush(self):
|
|
pass
|
|
|
|
def send_byte(self, data: Union[bytes, List[int]]) -> List[int]:
|
|
if isinstance(data, bytes):
|
|
return self._spi.xfer([v for v in data])
|
|
else:
|
|
return self._spi.xfer(data)
|
|
|
|
def recv_byte(self, size: int) -> Optional[bytes]:
|
|
raise RuntimeError()
|
|
|
|
|
|
class SoftwareImplSpiInterface(AbstractSpiInterface):
|
|
BPS_SPI = None
|
|
|
|
__slots__ = ('_spi', '_select')
|
|
|
|
def __init__(self,
|
|
clk: int,
|
|
mosi: int,
|
|
miso: Optional[int] = None,
|
|
cs: Union[None, int, Tuple[int]] = None,
|
|
cpol=False,
|
|
cpha=True):
|
|
|
|
self._check_import()
|
|
|
|
self._spi = self.BPS_SPI.Bpsspi(clk, mosi, miso, cs, cpol, cpha)
|
|
|
|
self._select = 0
|
|
|
|
@classmethod
|
|
def _check_import(cls):
|
|
if cls.BPS_SPI is None:
|
|
import biopro.ext.bpsspi as bpsspi
|
|
|
|
cls.BPS_SPI = bpsspi
|
|
|
|
@property
|
|
def select(self) -> int:
|
|
return self._select
|
|
|
|
@select.setter
|
|
def select(self, value: int):
|
|
self._select = value
|
|
|
|
def reset(self):
|
|
self._spi.reset()
|
|
|
|
super().reset()
|
|
|
|
def close(self):
|
|
self._spi = None
|
|
|
|
def flush(self):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
if self._spi is None:
|
|
raise RuntimeError('spi has closed')
|
|
|
|
self._spi.set_cs(self._select, False)
|
|
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self._spi is not None:
|
|
self._spi.set_cs(self._select, True)
|
|
|
|
def send_byte(self, data: bytes) -> Optional[List[int]]:
|
|
if self._spi is not None:
|
|
with self:
|
|
if self._spi.miso >= 0:
|
|
return self.send_byte_exchange(data)
|
|
else:
|
|
return self.send_byte_transmit(data)
|
|
|
|
return None
|
|
|
|
def send_byte_transmit(self, data: bytes) -> Optional[List[int]]:
|
|
if self._spi is not None:
|
|
self._spi.send_byte_transmit(data)
|
|
|
|
return None
|
|
|
|
def send_byte_exchange(self, data: bytes) -> Optional[List[int]]:
|
|
if self._spi is not None:
|
|
return self._spi.send_byte_exchange(data)
|
|
|
|
return None
|
|
|
|
def recv_byte(self, size: int) -> Optional[bytes]:
|
|
return None
|
|
|
|
|
|
def _main():
|
|
from time import sleep
|
|
from biopro.util.console import hex_line
|
|
|
|
spi = HardwareImplSpiInterface()
|
|
|
|
spi.reset()
|
|
|
|
data = list(range(100))
|
|
|
|
while True:
|
|
hex_line(spi.send_byte(data))
|
|
|
|
sleep(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
_main()
|