__init__.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683
  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 PacketLengthMode, 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. RX = 0x0D
  52. RXFIFO_OVERFLOW = 0x11
  53. TX = 0x13
  54. # TXFIFO_UNDERFLOW = 0x16
  55. class CC1101:
  56. # pylint: disable=too-many-public-methods
  57. # > All transfers on the SPI interface are done
  58. # > most significant bit first.
  59. # > All transactions on the SPI interface start with
  60. # > a header byte containing a R/W bit, a access bit (B),
  61. # > and a 6-bit address (A5 - A0).
  62. # > [...]
  63. # > Table 45: SPI Address Space
  64. _WRITE_SINGLE_BYTE = 0x00
  65. # > Registers with consecutive addresses can be
  66. # > accessed in an efficient way by setting the
  67. # > burst bit (B) in the header byte. The address
  68. # > bits (A5 - A0) set the start address in an
  69. # > internal address counter. This counter is
  70. # > incremented by one each new byte [...]
  71. _WRITE_BURST = 0x40
  72. _READ_SINGLE_BYTE = 0x80
  73. _READ_BURST = 0xC0
  74. # 29.3 Status Register Details
  75. _SUPPORTED_PARTNUM = 0
  76. _SUPPORTED_VERSION = 0x14
  77. _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ = 26e6
  78. # see "21 Frequency Programming"
  79. # > f_carrier = f_XOSC / 2**16 * (FREQ + CHAN * ((256 + CHANSPC_M) * 2**CHANSPC_E-2))
  80. _FREQUENCY_CONTROL_WORD_HERTZ_FACTOR = _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2 ** 16
  81. def __init__(self) -> None:
  82. self._spi = spidev.SpiDev()
  83. @staticmethod
  84. def _log_chip_status_byte(chip_status: int) -> None:
  85. # see "10.1 Chip Status Byte" & "Table 23: Status Byte Summary"
  86. # > The command strobe registers are accessed by transferring
  87. # > a single header byte [...]. That is, only the R/W̄ bit,
  88. # > the burst access bit (set to 0), and the six address bits [...]
  89. # > The R/W̄ bit can be either one or zero and will determine how the
  90. # > FIFO_BYTES_AVAILABLE field in the status byte should be interpreted.
  91. _LOGGER.debug(
  92. "chip status byte: CHIP_RDYn=%d STATE=%s FIFO_BYTES_AVAILBLE=%d",
  93. chip_status >> 7,
  94. bin((chip_status >> 4) & 0b111),
  95. chip_status & 0b1111,
  96. )
  97. def _read_single_byte(
  98. self, register: typing.Union[ConfigurationRegisterAddress, FIFORegisterAddress]
  99. ) -> int:
  100. response = self._spi.xfer([register | self._READ_SINGLE_BYTE, 0])
  101. assert len(response) == 2, response
  102. self._log_chip_status_byte(response[0])
  103. return response[1]
  104. def _read_burst(
  105. self,
  106. start_register: typing.Union[ConfigurationRegisterAddress, FIFORegisterAddress],
  107. length: int,
  108. ) -> typing.List[int]:
  109. response = self._spi.xfer([start_register | self._READ_BURST] + [0] * length)
  110. assert len(response) == length + 1, response
  111. self._log_chip_status_byte(response[0])
  112. return response[1:]
  113. def _read_status_register(self, register: StatusRegisterAddress) -> int:
  114. # > For register addresses in the range 0x30-0x3D,
  115. # > the burst bit is used to select between
  116. # > status registers when burst bit is one, and
  117. # > between command strobes when burst bit is
  118. # > zero. [...]
  119. # > Because of this, burst access is not available
  120. # > for status registers and they must be accessed
  121. # > one at a time. The status registers can only be
  122. # > read.
  123. response = self._spi.xfer([register | self._READ_BURST, 0])
  124. assert len(response) == 2, response
  125. self._log_chip_status_byte(response[0])
  126. return response[1]
  127. def _command_strobe(self, register: StrobeAddress) -> None:
  128. # see "10.4 Command Strobes"
  129. _LOGGER.debug("sending command strobe 0x%02x", register)
  130. response = self._spi.xfer([register | self._WRITE_SINGLE_BYTE])
  131. assert len(response) == 1, response
  132. self._log_chip_status_byte(response[0])
  133. def _write_burst(
  134. self,
  135. start_register: typing.Union[ConfigurationRegisterAddress, FIFORegisterAddress],
  136. values: typing.List[int],
  137. ) -> None:
  138. _LOGGER.debug(
  139. "writing burst: start_register=0x%02x values=%s", start_register, values
  140. )
  141. response = self._spi.xfer([start_register | self._WRITE_BURST] + values)
  142. assert len(response) == len(values) + 1, response
  143. self._log_chip_status_byte(response[0])
  144. assert all(v == response[0] for v in response[1:]), response
  145. def _reset(self) -> None:
  146. self._command_strobe(StrobeAddress.SRES)
  147. @classmethod
  148. def _filter_bandwidth_floating_point_to_real(
  149. cls, mantissa: int, exponent: int
  150. ) -> float:
  151. """
  152. See "13 Receiver Channel Filter Bandwidth"
  153. """
  154. return cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / (
  155. 8 * (4 + mantissa) * (2 ** exponent)
  156. )
  157. def _get_filter_bandwidth_hertz(self) -> float:
  158. """
  159. See "13 Receiver Channel Filter Bandwidth"
  160. MDMCFG4.CHANBW_E & MDMCFG4.CHANBW_M
  161. """
  162. mdmcfg4 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4)
  163. return self._filter_bandwidth_floating_point_to_real(
  164. exponent=mdmcfg4 >> 6, mantissa=(mdmcfg4 >> 4) & 0b11
  165. )
  166. def _get_symbol_rate_exponent(self) -> int:
  167. """
  168. MDMCFG4.DRATE_E
  169. """
  170. return self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4) & 0b00001111
  171. def _set_symbol_rate_exponent(self, exponent: int):
  172. mdmcfg4 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4)
  173. mdmcfg4 &= 0b11110000
  174. mdmcfg4 |= exponent
  175. self._write_burst(
  176. start_register=ConfigurationRegisterAddress.MDMCFG4, values=[mdmcfg4]
  177. )
  178. def _get_symbol_rate_mantissa(self) -> int:
  179. """
  180. MDMCFG3.DRATE_M
  181. """
  182. return self._read_single_byte(ConfigurationRegisterAddress.MDMCFG3)
  183. def _set_symbol_rate_mantissa(self, mantissa: int) -> None:
  184. self._write_burst(
  185. start_register=ConfigurationRegisterAddress.MDMCFG3, values=[mantissa]
  186. )
  187. @classmethod
  188. def _symbol_rate_floating_point_to_real(cls, mantissa: int, exponent: int) -> float:
  189. # see "12 Data Rate Programming"
  190. return (
  191. (256 + mantissa)
  192. * (2 ** exponent)
  193. * cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ
  194. / (2 ** 28)
  195. )
  196. @classmethod
  197. def _symbol_rate_real_to_floating_point(cls, real: float) -> typing.Tuple[int, int]:
  198. # see "12 Data Rate Programming"
  199. assert real > 0, real
  200. exponent = math.floor(
  201. math.log2(real / cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ) + 20
  202. )
  203. mantissa = round(
  204. real * 2 ** 28 / cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2 ** exponent
  205. - 256
  206. )
  207. if mantissa == 256:
  208. exponent += 1
  209. mantissa = 0
  210. assert 0 < exponent <= 2 ** 4, exponent
  211. assert mantissa <= 2 ** 8, mantissa
  212. return mantissa, exponent
  213. def get_symbol_rate_baud(self) -> float:
  214. return self._symbol_rate_floating_point_to_real(
  215. mantissa=self._get_symbol_rate_mantissa(),
  216. exponent=self._get_symbol_rate_exponent(),
  217. )
  218. def set_symbol_rate_baud(self, real: float) -> None:
  219. # > The data rate can be set from 0.6 kBaud to 500 kBaud [...]
  220. mantissa, exponent = self._symbol_rate_real_to_floating_point(real)
  221. self._set_symbol_rate_mantissa(mantissa)
  222. self._set_symbol_rate_exponent(exponent)
  223. def get_modulation_format(self) -> ModulationFormat:
  224. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  225. return ModulationFormat((mdmcfg2 >> 4) & 0b111)
  226. def _set_modulation_format(self, modulation_format: ModulationFormat) -> None:
  227. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  228. mdmcfg2 &= ~(modulation_format << 4)
  229. mdmcfg2 |= modulation_format << 4
  230. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  231. def enable_manchester_code(self) -> None:
  232. """
  233. MDMCFG2.MANCHESTER_EN
  234. Enable manchester encoding & decoding for the entire packet,
  235. including the preamble and synchronization word.
  236. """
  237. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  238. mdmcfg2 |= 0b1000
  239. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  240. def get_sync_mode(self) -> SyncMode:
  241. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  242. return SyncMode(mdmcfg2 & 0b11)
  243. def set_sync_mode(self, mode: SyncMode) -> None:
  244. """
  245. MDMCFG2.SYNC_MODE
  246. see "14.3 Byte Synchronization"
  247. """
  248. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  249. mdmcfg2 &= 0b11111100
  250. mdmcfg2 |= mode
  251. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  252. def get_preamble_length_bytes(self) -> int:
  253. """
  254. MDMCFG1.NUM_PREAMBLE
  255. Minimum number of preamble bytes to be transmitted.
  256. See "15.2 Packet Format"
  257. """
  258. index = (
  259. self._read_single_byte(ConfigurationRegisterAddress.MDMCFG1) >> 4
  260. ) & 0b111
  261. return 2 ** (index >> 1) * (2 + (index & 0b1))
  262. def _set_preamble_length_index(self, index: int) -> None:
  263. assert 0 <= index <= 0b111
  264. mdmcfg1 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG1)
  265. mdmcfg1 &= 0b10001111
  266. mdmcfg1 |= index << 4
  267. self._write_burst(ConfigurationRegisterAddress.MDMCFG1, [mdmcfg1])
  268. def set_preamble_length_bytes(self, length: int) -> None:
  269. """
  270. see .get_preamble_length_bytes()
  271. """
  272. if length < 1:
  273. raise ValueError(
  274. "invalid preamble length {} given".format(length)
  275. + "\ncall .set_sync_mode(cc1101.SyncMode.NO_PREAMBLE_AND_SYNC_WORD)"
  276. + " to disable preamble"
  277. )
  278. if length % 3 == 0:
  279. index = math.log2(length / 3) * 2 + 1
  280. else:
  281. index = math.log2(length / 2) * 2
  282. if not index.is_integer() or index < 0 or index > 0b111:
  283. raise ValueError(
  284. "unsupported preamble length: {} bytes".format(length)
  285. + "\nsee MDMCFG1.NUM_PREAMBLE in cc1101 docs"
  286. )
  287. self._set_preamble_length_index(int(index))
  288. def _set_power_amplifier_setting_index(self, setting_index: int) -> None:
  289. """
  290. FREND0.PA_POWER
  291. > This value is an index to the PATABLE,
  292. > which can be programmed with up to 8 different PA settings.
  293. > In OOK/ASK mode, this selects the PATABLE index to use
  294. > when transmitting a '1'.
  295. > PATABLE index zero is used in OOK/ASK when transmitting a '0'.
  296. > The PATABLE settings from index 0 to the PA_POWER value are
  297. > used for > ASK TX shaping, [...]
  298. see "Figure 32: Shaping of ASK Signal"
  299. > If OOK modulation is used, the logic 0 and logic 1 power levels
  300. > shall be programmed to index 0 and 1 respectively.
  301. """
  302. frend0 = self._read_single_byte(ConfigurationRegisterAddress.FREND0)
  303. frend0 &= 0b000
  304. frend0 |= setting_index
  305. self._write_burst(ConfigurationRegisterAddress.FREND0, [setting_index])
  306. def __enter__(self) -> "CC1101":
  307. # https://docs.python.org/3/reference/datamodel.html#object.__enter__
  308. self._spi.open(0, 0)
  309. self._spi.max_speed_hz = 55700 # empirical
  310. self._reset()
  311. partnum = self._read_status_register(StatusRegisterAddress.PARTNUM)
  312. if partnum != self._SUPPORTED_PARTNUM:
  313. raise ValueError(
  314. "unexpected chip part number {} (expected: {})".format(
  315. partnum, self._SUPPORTED_PARTNUM
  316. )
  317. )
  318. version = self._read_status_register(StatusRegisterAddress.VERSION)
  319. if version != self._SUPPORTED_VERSION:
  320. raise ValueError(
  321. "unexpected chip version number {} (expected: {})".format(
  322. version, self._SUPPORTED_VERSION
  323. )
  324. )
  325. # 6:4 MOD_FORMAT: OOK (default: 2-FSK)
  326. self._set_modulation_format(ModulationFormat.ASK_OOK)
  327. self._set_power_amplifier_setting_index(1)
  328. self._disable_data_whitening()
  329. # 7:6 unused
  330. # 5:4 FS_AUTOCAL: calibrate when going from IDLE to RX or TX
  331. # 3:2 PO_TIMEOUT: default
  332. # 1 PIN_CTRL_EN: default
  333. # 0 XOSC_FORCE_ON: default
  334. self._write_burst(ConfigurationRegisterAddress.MCSM0, [0b010100])
  335. marcstate = self.get_main_radio_control_state_machine_state()
  336. if marcstate != MainRadioControlStateMachineState.IDLE:
  337. raise ValueError("expected marcstate idle (actual: {})".format(marcstate))
  338. return self
  339. def __exit__(self, exc_type, exc_value, traceback): # -> typing.Literal[False]
  340. # https://docs.python.org/3/reference/datamodel.html#object.__exit__
  341. self._spi.close()
  342. return False
  343. def get_main_radio_control_state_machine_state(
  344. self,
  345. ) -> MainRadioControlStateMachineState:
  346. return MainRadioControlStateMachineState(
  347. self._read_status_register(StatusRegisterAddress.MARCSTATE)
  348. )
  349. def get_marc_state(self) -> MainRadioControlStateMachineState:
  350. """
  351. alias for get_main_radio_control_state_machine_state()
  352. """
  353. return self.get_main_radio_control_state_machine_state()
  354. @classmethod
  355. def _frequency_control_word_to_hertz(cls, control_word: typing.List[int]) -> float:
  356. return (
  357. int.from_bytes(control_word, byteorder="big", signed=False)
  358. * cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR
  359. )
  360. @classmethod
  361. def _hertz_to_frequency_control_word(cls, hertz: float) -> typing.List[int]:
  362. return list(
  363. round(hertz / cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR).to_bytes(
  364. length=3, byteorder="big", signed=False
  365. )
  366. )
  367. def _get_base_frequency_control_word(self) -> typing.List[int]:
  368. # > The base or start frequency is set by the 24 bitfrequency
  369. # > word located in the FREQ2, FREQ1, FREQ0 registers.
  370. return self._read_burst(
  371. start_register=ConfigurationRegisterAddress.FREQ2, length=3
  372. )
  373. def _set_base_frequency_control_word(self, control_word: typing.List[int]) -> None:
  374. self._write_burst(
  375. start_register=ConfigurationRegisterAddress.FREQ2, values=control_word
  376. )
  377. def get_base_frequency_hertz(self) -> float:
  378. return self._frequency_control_word_to_hertz(
  379. self._get_base_frequency_control_word()
  380. )
  381. def set_base_frequency_hertz(self, freq: float) -> None:
  382. self._set_base_frequency_control_word(
  383. self._hertz_to_frequency_control_word(freq)
  384. )
  385. def __str__(self) -> str:
  386. sync_mode = self.get_sync_mode()
  387. attrs = (
  388. "marcstate={}".format(
  389. self.get_main_radio_control_state_machine_state().name.lower()
  390. ),
  391. "base_frequency={:.2f}MHz".format(
  392. self.get_base_frequency_hertz() / 10 ** 6
  393. ),
  394. "symbol_rate={:.2f}kBaud".format(self.get_symbol_rate_baud() / 1000),
  395. "modulation_format={}".format(self.get_modulation_format().name),
  396. "sync_mode={}".format(sync_mode.name),
  397. "preamble_length={}B".format(self.get_preamble_length_bytes())
  398. if sync_mode != SyncMode.NO_PREAMBLE_AND_SYNC_WORD
  399. else None,
  400. "sync_word=0x{:02x}{:02x}".format(*self.get_sync_word())
  401. if sync_mode != SyncMode.NO_PREAMBLE_AND_SYNC_WORD
  402. else None,
  403. "packet_length{}{}B".format(
  404. "≤"
  405. if self.get_packet_length_mode() == PacketLengthMode.VARIABLE
  406. else "=",
  407. self.get_packet_length_bytes(),
  408. ),
  409. )
  410. return "CC1101({})".format(", ".join(filter(None, attrs)))
  411. def get_configuration_register_values(
  412. self,
  413. start_register: ConfigurationRegisterAddress = min(
  414. ConfigurationRegisterAddress
  415. ),
  416. end_register: ConfigurationRegisterAddress = max(ConfigurationRegisterAddress),
  417. ) -> typing.Dict[ConfigurationRegisterAddress, int]:
  418. assert start_register <= end_register, (start_register, end_register)
  419. values = self._read_burst(
  420. start_register=start_register, length=end_register - start_register + 1
  421. )
  422. return {
  423. ConfigurationRegisterAddress(start_register + i): v
  424. for i, v in enumerate(values)
  425. }
  426. def get_sync_word(self) -> bytes:
  427. """
  428. SYNC1 & SYNC0
  429. See "15.2 Packet Format"
  430. The first byte's most significant bit is transmitted first.
  431. """
  432. return bytes(
  433. self._read_burst(
  434. start_register=ConfigurationRegisterAddress.SYNC1, length=2
  435. )
  436. )
  437. def set_sync_word(self, sync_word: bytes) -> None:
  438. """
  439. See .set_sync_word()
  440. """
  441. if len(sync_word) != 2:
  442. raise ValueError("expected two bytes, got {!r}".format(sync_word))
  443. self._write_burst(
  444. start_register=ConfigurationRegisterAddress.SYNC1, values=list(sync_word)
  445. )
  446. def get_packet_length_bytes(self) -> int:
  447. """
  448. PKTLEN
  449. Packet length in fixed packet length mode,
  450. maximum packet length in variable packet length mode.
  451. > In variable packet length mode, [...]
  452. > any packet received with a length byte
  453. > with a value greater than PKTLEN will be discarded.
  454. """
  455. return self._read_single_byte(ConfigurationRegisterAddress.PKTLEN)
  456. def set_packet_length_bytes(self, packet_length: int) -> None:
  457. """
  458. see get_packet_length_bytes()
  459. """
  460. assert 1 <= packet_length <= 255, "unsupported packet length {}".format(
  461. packet_length
  462. )
  463. self._write_burst(
  464. start_register=ConfigurationRegisterAddress.PKTLEN, values=[packet_length]
  465. )
  466. def _disable_data_whitening(self):
  467. """
  468. PKTCTRL0.WHITE_DATA
  469. see "15.1 Data Whitening"
  470. > By setting PKTCTRL0.WHITE_DATA=1 [default],
  471. > all data, except the preamble and the sync word
  472. > will be XOR-ed with a 9-bit pseudo-random (PN9)
  473. > sequence before being transmitted.
  474. """
  475. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  476. pktctrl0 &= 0b10111111
  477. self._write_burst(
  478. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  479. )
  480. def disable_checksum(self) -> None:
  481. """
  482. PKTCTRL0.CRC_EN
  483. Disable automatic 2-byte cyclic redundancy check (CRC) sum
  484. appending in TX mode and checking in RX mode.
  485. See "Figure 19: Packet Format".
  486. """
  487. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  488. pktctrl0 &= 0b11111011
  489. self._write_burst(
  490. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  491. )
  492. def _get_transceive_mode(self) -> _TransceiveMode:
  493. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  494. return _TransceiveMode((pktctrl0 >> 4) & 0b11)
  495. def _set_transceive_mode(self, mode: _TransceiveMode) -> None:
  496. _LOGGER.info("changing transceive mode to %s", mode.name)
  497. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  498. pktctrl0 &= ~0b00110000
  499. pktctrl0 |= mode << 4
  500. self._write_burst(
  501. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  502. )
  503. def get_packet_length_mode(self) -> PacketLengthMode:
  504. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  505. return PacketLengthMode(pktctrl0 & 0b11)
  506. def set_packet_length_mode(self, mode: PacketLengthMode) -> None:
  507. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  508. pktctrl0 &= 0b11111100
  509. pktctrl0 |= mode
  510. self._write_burst(
  511. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  512. )
  513. def _flush_tx_fifo_buffer(self) -> None:
  514. # > Only issue SFTX in IDLE or TXFIFO_UNDERFLOW states.
  515. _LOGGER.debug("flushing tx fifo buffer")
  516. self._command_strobe(StrobeAddress.SFTX)
  517. def transmit(self, payload: bytes) -> None:
  518. """
  519. The most significant bit is transmitted first.
  520. In variable packet length mode,
  521. a byte indicating the packet's length will be prepended.
  522. > In variable packet length mode,
  523. > the packet length is configured by the first byte [...].
  524. > The packet length is defined as the payload data,
  525. > excluding the length byte and the optional CRC.
  526. from "15.2 Packet Format"
  527. Call .set_packet_length_mode(cc1101.PacketLengthMode.FIXED)
  528. to switch to fixed packet length mode.
  529. """
  530. # see "15.2 Packet Format"
  531. # > In variable packet length mode, [...]
  532. # > The first byte written to the TXFIFO must be different from 0.
  533. packet_length_mode = self.get_packet_length_mode()
  534. packet_length = self.get_packet_length_bytes()
  535. if packet_length_mode == PacketLengthMode.VARIABLE:
  536. if not payload:
  537. raise ValueError("empty payload {!r}".format(payload))
  538. if len(payload) > packet_length:
  539. raise ValueError(
  540. "payload exceeds maximum payload length of {} bytes".format(
  541. packet_length
  542. )
  543. + "\nsee .get_packet_length_bytes()"
  544. + "\npayload: {!r}".format(payload)
  545. )
  546. payload = int.to_bytes(len(payload), length=1, byteorder="big") + payload
  547. elif (
  548. packet_length_mode == PacketLengthMode.FIXED
  549. and len(payload) != packet_length
  550. ):
  551. raise ValueError(
  552. "expected payload length of {} bytes, got {}".format(
  553. packet_length, len(payload)
  554. )
  555. + "\nsee .set_packet_length_mode() and .get_packet_length_bytes()"
  556. + "\npayload: {!r}".format(payload)
  557. )
  558. marcstate = self.get_main_radio_control_state_machine_state()
  559. if marcstate != MainRadioControlStateMachineState.IDLE:
  560. raise Exception(
  561. "device must be idle before transmission (current marcstate: {})".format(
  562. marcstate.name
  563. )
  564. )
  565. self._flush_tx_fifo_buffer()
  566. self._write_burst(FIFORegisterAddress.TX, list(payload))
  567. _LOGGER.info(
  568. "transmitting 0x%s (%r)",
  569. "".join("{:02x}".format(b) for b in payload),
  570. payload,
  571. )
  572. self._command_strobe(StrobeAddress.STX)
  573. @contextlib.contextmanager
  574. def asynchronous_transmission(self) -> typing.Iterator[Pin]:
  575. """
  576. see "27.1 Asynchronous Serial Operation"
  577. >>> with cc1101.CC1101() as transceiver:
  578. >>> transceiver.set_base_frequency_hertz(433.92e6)
  579. >>> transceiver.set_symbol_rate_baud(600)
  580. >>> print(transceiver)
  581. >>> with transceiver.asynchronous_transmission():
  582. >>> # send digital signal to GDO0 pin
  583. """
  584. self._set_transceive_mode(_TransceiveMode.ASYNCHRONOUS_SERIAL)
  585. self._command_strobe(StrobeAddress.STX)
  586. try:
  587. # > In TX, the GDO0 pin is used for data input (TX data).
  588. yield Pin.GDO0
  589. finally:
  590. self._command_strobe(StrobeAddress.SIDLE)
  591. self._set_transceive_mode(_TransceiveMode.FIFO)