__init__.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519
  1. # python-cc1101 - Python Library to Transmit RF Signals via C1101 Transceivers
  2. #
  3. # Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
  4. #
  5. # This program is free software: you can redistribute it and/or modify
  6. # it under the terms of the GNU General Public License as published by
  7. # the Free Software Foundation, either version 3 of the License, or
  8. # any later version.
  9. #
  10. # This program is distributed in the hope that it will be useful,
  11. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. # GNU General Public License for more details.
  14. #
  15. # You should have received a copy of the GNU General Public License
  16. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. import contextlib
  18. import enum
  19. import logging
  20. import math
  21. import typing
  22. import spidev
  23. from cc1101.addresses import (
  24. StrobeAddress,
  25. ConfigurationRegisterAddress,
  26. StatusRegisterAddress,
  27. FIFORegisterAddress,
  28. )
  29. from cc1101.options import SyncMode, ModulationFormat
  30. _LOGGER = logging.getLogger(__name__)
  31. class Pin(enum.Enum):
  32. GDO0 = "GDO0"
  33. class _TransceiveMode(enum.IntEnum):
  34. """
  35. PKTCTRL0.PKT_FORMAT
  36. """
  37. FIFO = 0b00
  38. SYNCHRONOUS_SERIAL = 0b01
  39. RANDOM_TRANSMISSION = 0b10
  40. ASYNCHRONOUS_SERIAL = 0b11
  41. class MainRadioControlStateMachineState(enum.IntEnum):
  42. """
  43. MARCSTATE - Main Radio Control State Machine State
  44. """
  45. # see "Figure 13: Simplified State Diagram"
  46. # and "Figure 25: Complete Radio Control State Diagram"
  47. IDLE = 0x01
  48. STARTCAL = 0x08 # after IDLE
  49. BWBOOST = 0x09 # after STARTCAL
  50. FS_LOCK = 0x0A
  51. TX = 0x13
  52. class CC1101:
  53. # > All transfers on the SPI interface are done
  54. # > most significant bit first.
  55. # > All transactions on the SPI interface start with
  56. # > a header byte containing a R/W bit, a access bit (B),
  57. # > and a 6-bit address (A5 - A0).
  58. # > [...]
  59. # > Table 45: SPI Address Space
  60. _WRITE_SINGLE_BYTE = 0x00
  61. # > Registers with consecutive addresses can be
  62. # > accessed in an efficient way by setting the
  63. # > burst bit (B) in the header byte. The address
  64. # > bits (A5 - A0) set the start address in an
  65. # > internal address counter. This counter is
  66. # > incremented by one each new byte [...]
  67. _WRITE_BURST = 0x40
  68. _READ_SINGLE_BYTE = 0x80
  69. _READ_BURST = 0xC0
  70. # 29.3 Status Register Details
  71. _SUPPORTED_PARTNUM = 0
  72. _SUPPORTED_VERSION = 0x14
  73. _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ = 26e6
  74. # see "21 Frequency Programming"
  75. # > f_carrier = f_XOSC / 2**16 * (FREQ + CHAN * ((256 + CHANSPC_M) * 2**CHANSPC_E-2))
  76. _FREQUENCY_CONTROL_WORD_HERTZ_FACTOR = _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2 ** 16
  77. def __init__(self) -> None:
  78. self._spi = spidev.SpiDev()
  79. @staticmethod
  80. def _log_chip_status_byte(chip_status: int) -> None:
  81. # see "10.1 Chip Status Byte" & "Table 23: Status Byte Summary"
  82. _LOGGER.debug(
  83. "chip status byte: CHIP_RDYn=%d STATE=%s FIFO_BYTES_AVAILBLE=%d",
  84. chip_status >> 7,
  85. bin((chip_status >> 4) & 0b111),
  86. chip_status & 0b1111,
  87. )
  88. def _read_single_byte(
  89. self, register: typing.Union[ConfigurationRegisterAddress, FIFORegisterAddress]
  90. ) -> int:
  91. response = self._spi.xfer([register | self._READ_SINGLE_BYTE, 0])
  92. assert len(response) == 2, response
  93. self._log_chip_status_byte(response[0])
  94. return response[1]
  95. def _read_burst(
  96. self,
  97. start_register: typing.Union[ConfigurationRegisterAddress, FIFORegisterAddress],
  98. length: int,
  99. ) -> 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: StatusRegisterAddress) -> int:
  105. # > For register addresses in the range 0x30-0x3D,
  106. # > the burst bit is used to select between
  107. # > status registers when burst bit is one, and
  108. # > between command strobes when burst bit is
  109. # > zero. [...]
  110. # > Because of this, burst access is not available
  111. # > for status registers and they must be accessed
  112. # > one at a time. The status registers can only be
  113. # > read.
  114. response = self._spi.xfer([register | self._READ_BURST, 0])
  115. assert len(response) == 2, response
  116. self._log_chip_status_byte(response[0])
  117. return response[1]
  118. def _command_strobe(self, register: StrobeAddress) -> None:
  119. # see "10.4 Command Strobes"
  120. _LOGGER.debug("sending command strobe 0x%02x", register)
  121. response = self._spi.xfer([register | self._WRITE_SINGLE_BYTE])
  122. assert len(response) == 1, response
  123. self._log_chip_status_byte(response[0])
  124. def _write_burst(
  125. self,
  126. start_register: typing.Union[ConfigurationRegisterAddress, FIFORegisterAddress],
  127. values: typing.List[int],
  128. ) -> None:
  129. _LOGGER.debug(
  130. "writing burst: start_register=0x%02x values=%s", start_register, values
  131. )
  132. response = self._spi.xfer([start_register | self._WRITE_BURST] + values)
  133. assert len(response) == len(values) + 1, response
  134. self._log_chip_status_byte(response[0])
  135. assert all(v == response[0] for v in response[1:]), response
  136. def _reset(self) -> None:
  137. self._command_strobe(StrobeAddress.SRES)
  138. def _get_symbol_rate_exponent(self) -> int:
  139. """
  140. MDMCFG4.DRATE_E
  141. """
  142. return self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4) & 0b00001111
  143. def _set_symbol_rate_exponent(self, exponent: int):
  144. mdmcfg4 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4)
  145. mdmcfg4 &= 0b11110000
  146. mdmcfg4 |= exponent
  147. self._write_burst(
  148. start_register=ConfigurationRegisterAddress.MDMCFG4, values=[mdmcfg4]
  149. )
  150. def _get_symbol_rate_mantissa(self) -> int:
  151. """
  152. MDMCFG3.DRATE_M
  153. """
  154. return self._read_single_byte(ConfigurationRegisterAddress.MDMCFG3)
  155. def _set_symbol_rate_mantissa(self, mantissa: int) -> None:
  156. self._write_burst(
  157. start_register=ConfigurationRegisterAddress.MDMCFG3, values=[mantissa]
  158. )
  159. @classmethod
  160. def _symbol_rate_floating_point_to_real(cls, mantissa: int, exponent: int) -> float:
  161. # see "12 Data Rate Programming"
  162. return (
  163. (256 + mantissa)
  164. * (2 ** exponent)
  165. * cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ
  166. / (2 ** 28)
  167. )
  168. @classmethod
  169. def _symbol_rate_real_to_floating_point(cls, real: float) -> typing.Tuple[int, int]:
  170. # see "12 Data Rate Programming"
  171. assert real > 0, real
  172. exponent = math.floor(
  173. math.log2(real / cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ) + 20
  174. )
  175. mantissa = round(
  176. real * 2 ** 28 / cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2 ** exponent
  177. - 256
  178. )
  179. if mantissa == 256:
  180. exponent += 1
  181. mantissa = 0
  182. assert 0 < exponent <= 2 ** 4, exponent
  183. assert mantissa <= 2 ** 8, mantissa
  184. return mantissa, exponent
  185. def get_symbol_rate_baud(self) -> float:
  186. return self._symbol_rate_floating_point_to_real(
  187. mantissa=self._get_symbol_rate_mantissa(),
  188. exponent=self._get_symbol_rate_exponent(),
  189. )
  190. def set_symbol_rate_baud(self, real: float) -> None:
  191. # > The data rate can be set from 0.6 kBaud to 500 kBaud [...]
  192. mantissa, exponent = self._symbol_rate_real_to_floating_point(real)
  193. self._set_symbol_rate_mantissa(mantissa)
  194. self._set_symbol_rate_exponent(exponent)
  195. def get_modulation_format(self) -> ModulationFormat:
  196. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  197. return ModulationFormat((mdmcfg2 >> 4) & 0b111)
  198. def _set_modulation_format(self, modulation_format: ModulationFormat) -> None:
  199. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  200. mdmcfg2 &= ~(modulation_format << 4)
  201. mdmcfg2 |= modulation_format << 4
  202. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  203. def enable_manchester_code(self) -> None:
  204. """
  205. MDMCFG2.MANCHESTER_EN
  206. Enable manchester encoding & decoding.
  207. """
  208. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  209. mdmcfg2 |= 0b1000
  210. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  211. def get_sync_mode(self) -> SyncMode:
  212. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  213. return SyncMode(mdmcfg2 & 0b11)
  214. def set_sync_mode(self, mode: SyncMode) -> None:
  215. """
  216. MDMCFG2.SYNC_MODE
  217. see "14.3 Byte Synchronization"
  218. """
  219. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  220. mdmcfg2 &= 0b11111100
  221. mdmcfg2 |= mode
  222. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  223. def _set_power_amplifier_setting_index(self, setting_index: int) -> None:
  224. """
  225. FREND0.PA_POWER
  226. > This value is an index to the PATABLE,
  227. > which can be programmed with up to 8 different PA settings.
  228. > In OOK/ASK mode, this selects the PATABLE index to use
  229. > when transmitting a '1'.
  230. > PATABLE index zero is used in OOK/ASK when transmitting a '0'.
  231. > The PATABLE settings from index 0 to the PA_POWER value are
  232. > used for > ASK TX shaping, [...]
  233. see "Figure 32: Shaping of ASK Signal"
  234. > If OOK modulation is used, the logic 0 and logic 1 power levels
  235. > shall be programmed to index 0 and 1 respectively.
  236. """
  237. frend0 = self._read_single_byte(ConfigurationRegisterAddress.FREND0)
  238. frend0 &= 0b000
  239. frend0 |= setting_index
  240. self._write_burst(ConfigurationRegisterAddress.FREND0, [setting_index])
  241. def __enter__(self) -> "CC1101":
  242. # https://docs.python.org/3/reference/datamodel.html#object.__enter__
  243. self._spi.open(0, 0)
  244. self._spi.max_speed_hz = 55700 # empirical
  245. self._reset()
  246. partnum = self._read_status_register(StatusRegisterAddress.PARTNUM)
  247. if partnum != self._SUPPORTED_PARTNUM:
  248. raise ValueError(
  249. "unexpected chip part number {} (expected: {})".format(
  250. partnum, self._SUPPORTED_PARTNUM
  251. )
  252. )
  253. version = self._read_status_register(StatusRegisterAddress.VERSION)
  254. if version != self._SUPPORTED_VERSION:
  255. raise ValueError(
  256. "unexpected chip version number {} (expected: {})".format(
  257. version, self._SUPPORTED_VERSION
  258. )
  259. )
  260. # 6:4 MOD_FORMAT: OOK (default: 2-FSK)
  261. self._set_modulation_format(ModulationFormat.ASK_OOK)
  262. self._set_power_amplifier_setting_index(1)
  263. self._disable_data_whitening()
  264. # 7:6 unused
  265. # 5:4 FS_AUTOCAL: calibrate when going from IDLE to RX or TX
  266. # 3:2 PO_TIMEOUT: default
  267. # 1 PIN_CTRL_EN: default
  268. # 0 XOSC_FORCE_ON: default
  269. self._write_burst(ConfigurationRegisterAddress.MCSM0, [0b010100])
  270. marcstate = self.get_main_radio_control_state_machine_state()
  271. if marcstate != MainRadioControlStateMachineState.IDLE:
  272. raise ValueError("expected marcstate idle (actual: {})".format(marcstate))
  273. return self
  274. def __exit__(self, exc_type, exc_value, traceback): # -> typing.Literal[False]
  275. # https://docs.python.org/3/reference/datamodel.html#object.__exit__
  276. self._spi.close()
  277. return False
  278. def get_main_radio_control_state_machine_state(
  279. self,
  280. ) -> MainRadioControlStateMachineState:
  281. return MainRadioControlStateMachineState(
  282. self._read_status_register(StatusRegisterAddress.MARCSTATE)
  283. )
  284. def get_marc_state(self) -> MainRadioControlStateMachineState:
  285. """
  286. alias for get_main_radio_control_state_machine_state()
  287. """
  288. return self.get_main_radio_control_state_machine_state()
  289. @classmethod
  290. def _frequency_control_word_to_hertz(cls, control_word: typing.List[int]) -> float:
  291. return (
  292. int.from_bytes(control_word, byteorder="big", signed=False)
  293. * cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR
  294. )
  295. @classmethod
  296. def _hertz_to_frequency_control_word(cls, hertz: float) -> typing.List[int]:
  297. return list(
  298. round(hertz / cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR).to_bytes(
  299. length=3, byteorder="big", signed=False
  300. )
  301. )
  302. def _get_base_frequency_control_word(self) -> typing.List[int]:
  303. # > The base or start frequency is set by the 24 bitfrequency
  304. # > word located in the FREQ2, FREQ1, FREQ0 registers.
  305. return self._read_burst(
  306. start_register=ConfigurationRegisterAddress.FREQ2, length=3
  307. )
  308. def _set_base_frequency_control_word(self, control_word: typing.List[int]) -> None:
  309. self._write_burst(
  310. start_register=ConfigurationRegisterAddress.FREQ2, values=control_word
  311. )
  312. def get_base_frequency_hertz(self) -> float:
  313. return self._frequency_control_word_to_hertz(
  314. self._get_base_frequency_control_word()
  315. )
  316. def set_base_frequency_hertz(self, freq: float) -> None:
  317. self._set_base_frequency_control_word(
  318. self._hertz_to_frequency_control_word(freq)
  319. )
  320. def __str__(self) -> str:
  321. attrs = (
  322. "marcstate={}".format(
  323. self.get_main_radio_control_state_machine_state().name.lower()
  324. ),
  325. "base_frequency={:.2f}MHz".format(
  326. self.get_base_frequency_hertz() / 10 ** 6
  327. ),
  328. "symbol_rate={:.2f}kBaud".format(self.get_symbol_rate_baud() / 1000),
  329. "modulation_format={}".format(self.get_modulation_format().name),
  330. "sync_mode={}".format(self.get_sync_mode().name),
  331. )
  332. return "CC1101({})".format(", ".join(attrs))
  333. def _get_packet_length(self) -> int:
  334. """
  335. packet length in fixed packet length mode,
  336. maximum packet length in variable packet length mode.
  337. """
  338. return self._read_single_byte(ConfigurationRegisterAddress.PKTLEN)
  339. def get_configuration_register_values(
  340. self,
  341. start_register: ConfigurationRegisterAddress = min(
  342. ConfigurationRegisterAddress
  343. ),
  344. end_register: ConfigurationRegisterAddress = max(ConfigurationRegisterAddress),
  345. ) -> typing.Dict[ConfigurationRegisterAddress, int]:
  346. assert start_register <= end_register, (start_register, end_register)
  347. values = self._read_burst(
  348. start_register=start_register, length=end_register - start_register + 1
  349. )
  350. return {
  351. ConfigurationRegisterAddress(start_register + i): v
  352. for i, v in enumerate(values)
  353. }
  354. def _disable_data_whitening(self):
  355. """
  356. PKTCTRL0.WHITE_DATA
  357. see "15.1 Data Whitening"
  358. > By setting PKTCTRL0.WHITE_DATA=1 [default],
  359. > all data, except the preamble and the sync word
  360. > will be XOR-ed with a 9-bit pseudo-random (PN9)
  361. > sequence before being transmitted.
  362. """
  363. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  364. pktctrl0 &= 0b10111111
  365. self._write_burst(
  366. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  367. )
  368. def disable_checksum(self) -> None:
  369. """
  370. PKTCTRL0.CRC_EN
  371. Disable automatic 2-byte cyclic redundancy check (CRC) sum
  372. appending in TX mode and checking in RX mode.
  373. See "Figure 19: Packet Format".
  374. """
  375. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  376. pktctrl0 &= 0b11111011
  377. self._write_burst(
  378. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  379. )
  380. def _get_transceive_mode(self) -> _TransceiveMode:
  381. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  382. return _TransceiveMode((pktctrl0 >> 4) & 0b11)
  383. def _set_transceive_mode(self, mode: _TransceiveMode) -> None:
  384. _LOGGER.info("changing transceive mode to %s", mode.name)
  385. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  386. pktctrl0 &= ~0b00110000
  387. pktctrl0 |= mode << 4
  388. self._write_burst(
  389. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  390. )
  391. def _flush_tx_fifo_buffer(self) -> None:
  392. # > Only issue SFTX in IDLE or TXFIFO_UNDERFLOW states.
  393. _LOGGER.debug("flushing tx fifo buffer")
  394. self._command_strobe(StrobeAddress.SFTX)
  395. def transmit(self, payload: bytes) -> None:
  396. """
  397. The most significant bit is transmitted first.
  398. """
  399. # see "15.2 Packet Format"
  400. # > In variable packet length mode, [...]
  401. # > The first byte written to the TXFIFO must be different from 0.
  402. if payload[0] == 0:
  403. raise ValueError(
  404. "in variable packet length mode the first byte of payload must not be null"
  405. + "\npayload: {!r}".format(payload)
  406. )
  407. marcstate = self.get_main_radio_control_state_machine_state()
  408. if marcstate != MainRadioControlStateMachineState.IDLE:
  409. raise Exception(
  410. "device must be idle before transmission (current marcstate: {})".format(
  411. marcstate.name
  412. )
  413. )
  414. max_packet_length = self._get_packet_length()
  415. if len(payload) > max_packet_length:
  416. raise ValueError(
  417. "payload exceeds maximum payload length of {} bytes".format(
  418. max_packet_length
  419. )
  420. + "\npayload: {!r}".format(payload)
  421. )
  422. self._flush_tx_fifo_buffer()
  423. self._write_burst(FIFORegisterAddress.TX, list(payload))
  424. _LOGGER.info(
  425. "transmitting 0x%s (%r)",
  426. "".join("{:02x}".format(b) for b in payload),
  427. payload,
  428. )
  429. self._command_strobe(StrobeAddress.STX)
  430. @contextlib.contextmanager
  431. def asynchronous_transmission(self) -> typing.Iterator[Pin]:
  432. """
  433. see "27.1 Asynchronous Serial Operation"
  434. >>> with cc1101.CC1101() as transceiver:
  435. >>> transceiver.set_base_frequency_hertz(433.92e6)
  436. >>> transceiver.set_symbol_rate_baud(600)
  437. >>> print(transceiver)
  438. >>> with transceiver.asynchronous_transmission():
  439. >>> # send digital signal to GDO0 pin
  440. """
  441. self._set_transceive_mode(_TransceiveMode.ASYNCHRONOUS_SERIAL)
  442. self._command_strobe(StrobeAddress.STX)
  443. try:
  444. # > In TX, the GDO0 pin is used for data input (TX data).
  445. yield Pin.GDO0
  446. finally:
  447. self._command_strobe(StrobeAddress.SIDLE)
  448. self._set_transceive_mode(_TransceiveMode.FIFO)