__init__.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310
  1. import enum
  2. import logging
  3. import typing
  4. import spidev
  5. _LOGGER = logging.getLogger(__name__)
  6. class CC1101:
  7. # > All transfers on the SPI interface are done
  8. # > most significant bit first.
  9. # > All transactions on the SPI interface start with
  10. # > a header byte containing a R/W bit, a access bit (B),
  11. # > and a 6-bit address (A5 - A0).
  12. # > [...]
  13. # > Table 45: SPI Address Space
  14. _WRITE_SINGLE_BYTE = 0x00
  15. # > Registers with consecutive addresses can be
  16. # > accessed in an efficient way by setting the
  17. # > burst bit (B) in the header byte. The address
  18. # > bits (A5 - A0) set the start address in an
  19. # > internal address counter. This counter is
  20. # > incremented by one each new byte [...]
  21. _WRITE_BURST = 0x40
  22. _READ_SINGLE_BYTE = 0x80
  23. _READ_BURST = 0xC0
  24. class _SPIAddress(enum.IntEnum):
  25. # see "Table 45: SPI Address Space"
  26. # > The configuration registers on the CC1101 are
  27. # > located on SPI addresses from 0x00 to 0x2E.
  28. PKTLEN = 0x06
  29. FREQ2 = 0x0D
  30. FREQ1 = 0x0E
  31. FREQ0 = 0x0F
  32. MDMCFG4 = 0x10
  33. MDMCFG3 = 0x11
  34. MDMCFG2 = 0x12
  35. MCSM0 = 0x18
  36. # > For register addresses in the range 0x30-0x3D,
  37. # > the burst bit is used to select between
  38. # > status registers when burst bit is one, and
  39. # > between command strobes when burst bit is
  40. # > zero. [...]
  41. # > Because of this, burst access is not available
  42. # > for status registers and they must be accessed
  43. # > one at a time. The status registers can only be
  44. # > read.
  45. SRES = 0x30
  46. STX = 0x35
  47. SFTX = 0x3B
  48. PARTNUM = 0x30
  49. VERSION = 0x31
  50. MARCSTATE = 0x35
  51. # see "10.5 FIFO Access"
  52. # > When the R/W-bit is zero, the TX FIFO is
  53. # > accessed, and the RX FIFO is accessed when
  54. # > the R/W-bit is one.
  55. TXFIFO = 0x3F
  56. class ModulationFormat(enum.IntEnum):
  57. """
  58. MDMCFG2.MOD_FORMAT
  59. """
  60. FSK2 = 0b000
  61. GFSK = 0b001
  62. ASK_OOK = 0b011
  63. FSK4 = 0b100
  64. MSK = 0b111
  65. class MainRadioControlStateMachineState(enum.IntEnum):
  66. """
  67. MARCSTATE - Main Radio Control State Machine State
  68. """
  69. # see "Figure 13: Simplified State Diagram"
  70. # and "Figure 25: Complete Radio Control State Diagram"
  71. IDLE = 0x01
  72. STARTCAL = 0x08 # after IDLE
  73. BWBOOST = 0x09 # after STARTCAL
  74. FS_LOCK = 0x0A
  75. TX = 0x13
  76. # 29.3 Status Register Details
  77. _SUPPORTED_PARTNUM = 0
  78. _SUPPORTED_VERSION = 0x14
  79. _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ = 26e6
  80. # see "21 Frequency Programming"
  81. # > f_carrier = f_XOSC / 2**16 * (FREQ + CHAN * ((256 + CHANSPC_M) * 2**CHANSPC_E-2))
  82. _FREQUENCY_CONTROL_WORD_HERTZ_FACTOR = _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2 ** 16
  83. def __init__(self) -> None:
  84. self._spi = spidev.SpiDev()
  85. @staticmethod
  86. def _log_chip_status_byte(chip_status: int) -> None:
  87. # see "10.1 Chip Status Byte" & "Table 23: Status Byte Summary"
  88. _LOGGER.debug(
  89. "chip status byte: CHIP_RDYn=%d STATE=%s FIFO_BYTES_AVAILBLE=%d",
  90. chip_status >> 7,
  91. bin((chip_status >> 4) & 0b111),
  92. chip_status & 0b1111,
  93. )
  94. def _read_single_byte(self, register: _SPIAddress) -> int:
  95. response = self._spi.xfer([register | self._READ_SINGLE_BYTE, 0])
  96. assert len(response) == 2, response
  97. self._log_chip_status_byte(response[0])
  98. return response[1]
  99. def _read_burst(self, start_register: _SPIAddress, length: int) -> typing.List[int]:
  100. response = self._spi.xfer([start_register | self._READ_BURST] + [0] * length)
  101. assert len(response) == length + 1, response
  102. self._log_chip_status_byte(response[0])
  103. return response[1:]
  104. def _read_status_register(self, register: _SPIAddress) -> int:
  105. _LOGGER.debug("reading status register 0x%02x", register)
  106. values = self._read_burst(start_register=register, length=1)
  107. assert len(values) == 1, values
  108. return values[0]
  109. def _command_strobe(self, register: _SPIAddress) -> None:
  110. # see "10.4 Command Strobes"
  111. _LOGGER.debug("sending command strobe 0x%02x", register)
  112. response = self._spi.xfer([register | self._WRITE_SINGLE_BYTE])
  113. assert len(response) == 1, response
  114. self._log_chip_status_byte(response[0])
  115. def _write_burst(
  116. self, start_register: _SPIAddress, values: typing.List[int]
  117. ) -> None:
  118. _LOGGER.debug(
  119. "writing burst: start_register=0x%02x values=%s", start_register, values
  120. )
  121. response = self._spi.xfer([start_register | self._WRITE_BURST] + values)
  122. assert len(response) == len(values) + 1, response
  123. self._log_chip_status_byte(response[0])
  124. assert all(v == 0x0F for v in response[1:]), response # TODO why?
  125. def _reset(self) -> None:
  126. self._command_strobe(self._SPIAddress.SRES)
  127. def _get_symbol_rate_exponent(self) -> int:
  128. """
  129. MDMCFG4.DRATE_E
  130. """
  131. return self._read_single_byte(self._SPIAddress.MDMCFG4) & 0b00001111
  132. def _get_symbol_rate_mantissa(self) -> int:
  133. """
  134. MDMCFG3.DRATE_M
  135. """
  136. return self._read_single_byte(self._SPIAddress.MDMCFG3)
  137. def get_symbol_rate_baud(self) -> float:
  138. # see "12 Data Rate Programming"
  139. return (
  140. (256 + self._get_symbol_rate_mantissa())
  141. * (2 ** self._get_symbol_rate_exponent())
  142. * self._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ
  143. / (2 ** 28)
  144. )
  145. def get_modulation_format(self) -> ModulationFormat:
  146. mdmcfg2 = self._read_single_byte(self._SPIAddress.MDMCFG2)
  147. return self.ModulationFormat((mdmcfg2 >> 4) & 0b111)
  148. def _set_modulation_format(self, modulation_format: ModulationFormat) -> None:
  149. mdmcfg2 = self._read_single_byte(self._SPIAddress.MDMCFG2)
  150. mdmcfg2 &= ~(modulation_format << 4)
  151. mdmcfg2 |= modulation_format << 4
  152. self._write_burst(self._SPIAddress.MDMCFG2, [mdmcfg2])
  153. def __enter__(self) -> "CC1101":
  154. # https://docs.python.org/3/reference/datamodel.html#object.__enter__
  155. self._spi.open(0, 0)
  156. self._spi.max_speed_hz = 55700 # empirical
  157. self._reset()
  158. partnum = self._read_status_register(self._SPIAddress.PARTNUM)
  159. if partnum != self._SUPPORTED_PARTNUM:
  160. raise ValueError(
  161. "unexpected chip part number {} (expected: {})".format(
  162. partnum, self._SUPPORTED_PARTNUM
  163. )
  164. )
  165. version = self._read_status_register(self._SPIAddress.VERSION)
  166. if version != self._SUPPORTED_VERSION:
  167. raise ValueError(
  168. "unexpected chip version number {} (expected: {})".format(
  169. version, self._SUPPORTED_VERSION
  170. )
  171. )
  172. # 6:4 MOD_FORMAT: OOK (default: 2-FSK)
  173. self._set_modulation_format(self.ModulationFormat.ASK_OOK)
  174. # 7:6 unused
  175. # 5:4 FS_AUTOCAL: calibrate when going from IDLE to RX or TX
  176. # 3:2 PO_TIMEOUT: default
  177. # 1 PIN_CTRL_EN: default
  178. # 0 XOSC_FORCE_ON: default
  179. self._write_burst(self._SPIAddress.MCSM0, [0b010100])
  180. marcstate = self.get_main_radio_control_state_machine_state()
  181. if marcstate != self.MainRadioControlStateMachineState.IDLE:
  182. raise ValueError("expected marcstate idle (actual: {})".format(marcstate))
  183. return self
  184. def __exit__(self, exc_type, exc_value, traceback) -> bool:
  185. # https://docs.python.org/3/reference/datamodel.html#object.__exit__
  186. self._spi.close()
  187. return False
  188. def get_main_radio_control_state_machine_state(
  189. self
  190. ) -> MainRadioControlStateMachineState:
  191. return self.MainRadioControlStateMachineState(
  192. self._read_status_register(self._SPIAddress.MARCSTATE)
  193. )
  194. def get_marc_state(self) -> MainRadioControlStateMachineState:
  195. """
  196. alias for get_main_radio_control_state_machine_state()
  197. """
  198. return self.get_main_radio_control_state_machine_state()
  199. @classmethod
  200. def _frequency_control_word_to_hertz(cls, control_word: typing.List[int]) -> float:
  201. return (
  202. int.from_bytes(control_word, byteorder="big", signed=False)
  203. * cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR
  204. )
  205. @classmethod
  206. def _hertz_to_frequency_control_word(cls, hertz: float) -> typing.List[int]:
  207. return list(
  208. round(hertz / cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR).to_bytes(
  209. length=3, byteorder="big", signed=False
  210. )
  211. )
  212. def _get_base_frequency_control_word(self) -> typing.List[int]:
  213. # > The base or start frequency is set by the 24 bitfrequency
  214. # > word located in the FREQ2, FREQ1, FREQ0 registers.
  215. return self._read_burst(start_register=self._SPIAddress.FREQ2, length=3)
  216. def _set_base_frequency_control_word(self, control_word: typing.List[int]) -> None:
  217. self._write_burst(start_register=self._SPIAddress.FREQ2, values=control_word)
  218. def get_base_frequency_hertz(self) -> float:
  219. return self._frequency_control_word_to_hertz(
  220. self._get_base_frequency_control_word()
  221. )
  222. def set_base_frequency_hertz(self, freq: float) -> None:
  223. self._set_base_frequency_control_word(
  224. self._hertz_to_frequency_control_word(freq)
  225. )
  226. def __str__(self) -> str:
  227. return "CC1101(marcstate={}, base_frequency={:.2f}MHz, symbol_rate={:.2f}kBaud, modulation_format={})".format(
  228. self.get_main_radio_control_state_machine_state().name.lower(),
  229. self.get_base_frequency_hertz() / 10 ** 6,
  230. self.get_symbol_rate_baud() / 1000,
  231. self.get_modulation_format().name,
  232. )
  233. def _get_packet_length(self) -> int:
  234. """
  235. packet length in fixed packet length mode,
  236. maximum packet length in variable packet length mode.
  237. """
  238. return self._read_burst(start_register=self._SPIAddress.PKTLEN, length=1)[0]
  239. def _flush_tx_fifo_buffer(self) -> str:
  240. # > Only issue SFTX in IDLE or TXFIFO_UNDERFLOW states.
  241. _LOGGER.debug("flushing tx fifo buffer")
  242. self._command_strobe(self._SPIAddress.SFTX)
  243. def transmit(self, payload: typing.List[int]) -> None:
  244. # see "15.2 Packet Format"
  245. # > In variable packet length mode, [...]
  246. # > The first byte written to the TXFIFO must be different from 0.
  247. if payload[0] == 0:
  248. raise ValueError(
  249. "in variable packet length mode the first byte of payload must not be null"
  250. + "\npayload: {}".format(payload)
  251. )
  252. marcstate = self.get_main_radio_control_state_machine_state()
  253. if marcstate != self.MainRadioControlStateMachineState.IDLE:
  254. raise Exception(
  255. "device must be idle before transmission (current marcstate: {})".format(
  256. marcstate.name
  257. )
  258. )
  259. max_packet_length = self._get_packet_length()
  260. if len(payload) > max_packet_length:
  261. raise ValueError(
  262. "payload exceeds maximum payload length of {} bytes".format(
  263. max_packet_length
  264. )
  265. + "\npayload: {}".format(payload)
  266. )
  267. self._flush_tx_fifo_buffer()
  268. self._write_burst(self._SPIAddress.TXFIFO, payload)
  269. _LOGGER.info("transmitting %s", payload)
  270. self._command_strobe(self._SPIAddress.STX)