1
0

__init__.py 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999
  1. # python-cc1101 - Python Library to Transmit RF Signals via CC1101 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. from __future__ import annotations
  18. import collections.abc
  19. import contextlib
  20. import datetime
  21. import enum
  22. import fcntl
  23. import logging
  24. import math
  25. import typing
  26. import warnings
  27. import spidev
  28. import cc1101._gpio
  29. from cc1101.addresses import (
  30. ConfigurationRegisterAddress,
  31. FIFORegisterAddress,
  32. PatableAddress,
  33. StatusRegisterAddress,
  34. StrobeAddress,
  35. )
  36. from cc1101.options import (
  37. GDOSignalSelection,
  38. ModulationFormat,
  39. PacketLengthMode,
  40. SyncMode,
  41. _TransceiveMode,
  42. )
  43. _LOGGER = logging.getLogger(__name__)
  44. class Pin(enum.Enum):
  45. GDO0 = "GDO0"
  46. class MainRadioControlStateMachineState(enum.IntEnum):
  47. """
  48. MARCSTATE - Main Radio Control State Machine State
  49. """
  50. # see "Figure 13: Simplified State Diagram"
  51. # and "Figure 25: Complete Radio Control State Diagram"
  52. SLEEP = 0x00
  53. IDLE = 0x01
  54. XOFF = 0x02
  55. VCOON_MC = 0x03
  56. REGON_MC = 0x04
  57. MANCAL = 0x05
  58. VCOON = 0x06
  59. REGON = 0x07
  60. STARTCAL = 0x08 # after IDLE
  61. BWBOOST = 0x09 # after STARTCAL
  62. FS_LOCK = 0x0A
  63. IFADCON = 0x0B
  64. ENDCAL = 0x0C
  65. RX = 0x0D
  66. RX_END = 0x0E
  67. RX_RST = 0x0F
  68. TXRX_SWITCH = 0x10
  69. RXFIFO_OVERFLOW = 0x11
  70. FSTXON = 0x12
  71. TX = 0x13
  72. TX_END = 0x14
  73. RXTX_SWITCH = 0x15
  74. TXFIFO_UNDERFLOW = 0x16
  75. class _ReceivedPacket: # unstable
  76. # "Table 31: Typical RSSI_offset Values"
  77. _RSSI_OFFSET_dB = 74
  78. def __init__(
  79. self,
  80. *,
  81. payload: bytes,
  82. rssi_index: int, # byte
  83. checksum_valid: bool,
  84. link_quality_indicator: int, # 7bit
  85. ):
  86. self.payload = payload
  87. self._rssi_index = rssi_index
  88. assert 0 <= rssi_index < (1 << 8), rssi_index
  89. self.checksum_valid = checksum_valid
  90. self.link_quality_indicator = link_quality_indicator
  91. assert 0 <= link_quality_indicator < (1 << 7), link_quality_indicator
  92. @property
  93. def rssi_dbm(self) -> float:
  94. """
  95. Estimated Received Signal Strength Indicator (RSSI) in dBm
  96. see section "17.3 RSSI"
  97. """
  98. if self._rssi_index >= 128:
  99. return (self._rssi_index - 256) / 2 - self._RSSI_OFFSET_dB
  100. return self._rssi_index / 2 - self._RSSI_OFFSET_dB
  101. def __str__(self) -> str:
  102. return f"{type(self).__name__}(RSSI {self.rssi_dbm:.0f}dBm, 0x{self.payload.hex()})"
  103. def _format_patable(
  104. settings: collections.abc.Iterable[int], insert_spaces: bool
  105. ) -> str:
  106. # "Table 39: Optimum PATABLE Settings" uses hexadecimal digits
  107. # "0" for brevity
  108. settings_hex = tuple(map(lambda s: "0" if s == 0 else f"0x{s:x}", settings))
  109. if len(settings_hex) == 1:
  110. return f"({settings_hex[0]},)"
  111. delimiter = ", " if insert_spaces else ","
  112. return f"({delimiter.join(settings_hex)})"
  113. class CC1101:
  114. # pylint: disable=too-many-public-methods
  115. # > All transfers on the SPI interface are done
  116. # > most significant bit first.
  117. # > All transactions on the SPI interface start with
  118. # > a header byte containing a R/W bit, a access bit (B),
  119. # > and a 6-bit address (A5 - A0).
  120. # > [...]
  121. # > Table 45: SPI Address Space
  122. _WRITE_SINGLE_BYTE = 0x00
  123. # > Registers with consecutive addresses can be
  124. # > accessed in an efficient way by setting the
  125. # > burst bit (B) in the header byte. The address
  126. # > bits (A5 - A0) set the start address in an
  127. # > internal address counter. This counter is
  128. # > incremented by one each new byte [...]
  129. _WRITE_BURST = 0x40
  130. _READ_SINGLE_BYTE = 0x80
  131. _READ_BURST = 0xC0
  132. # 29.3 Status Register Details
  133. _SUPPORTED_PARTNUM = 0
  134. # > The two versions of the chip will behave the same.
  135. # https://e2e.ti.com/support/wireless-connectivity/sub-1-ghz/f/156/p/428028/1529544#1529544
  136. _SUPPORTED_VERSIONS = [
  137. 0x04, # https://github.com/fphammerle/python-cc1101/issues/15
  138. 0x14,
  139. ]
  140. _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ = 26e6
  141. # see "21 Frequency Programming"
  142. # > f_carrier = f_XOSC / 2**16 * (FREQ + CHAN * ((256 + CHANSPC_M) * 2**CHANSPC_E-2))
  143. _FREQUENCY_CONTROL_WORD_HERTZ_FACTOR = _CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2**16
  144. # roughly estimated / tested with SDR receiver, docs specify:
  145. # > can [...] be programmed for operation at other frequencies
  146. # > in the 300-348 MHz, 387-464 MHz and 779-928 MHz bands.
  147. _TRANSMIT_MIN_FREQUENCY_HERTZ = 281.7e6
  148. # > The PATABLE is an 8-byte table that defines the PA control settings [...]
  149. _PATABLE_LENGTH_BYTES = 8
  150. def __init__(
  151. self,
  152. spi_bus: int = 0,
  153. spi_chip_select: int = 0,
  154. lock_spi_device: bool = False,
  155. spi_max_speed_hz: int = 55700,
  156. ) -> None:
  157. """
  158. lock_spi_device:
  159. When True, an advisory, exclusive lock will be set on the SPI device file
  160. non-blockingly via flock upon entering the context.
  161. If the SPI device file is already locked (e.g., by a different process),
  162. a BlockingIOError will be raised.
  163. The lock will be removed automatically, when leaving the context.
  164. The lock can optionally be released earlier by calling .unlock_spi_device().
  165. >>> transceiver = cc1101.CC1101(lock_spi_device=True)
  166. >>> # not locked
  167. >>> with transceiver:
  168. >>> # locked
  169. >>> # lock removed
  170. >>> with transceiver:
  171. >>> # locked
  172. >>> transceiver.unlock_spi_device()
  173. >>> # lock removed
  174. """
  175. self._spi = spidev.SpiDev()
  176. self._spi_max_speed_hz = spi_max_speed_hz
  177. self._spi_bus = int(spi_bus)
  178. # > The BCM2835 core common to all Raspberry Pi devices has 3 SPI Controllers:
  179. # > SPI0, with two hardware chip selects, [...]
  180. # > SPI1, with three hardware chip selects, [...]
  181. # > SPI2, also with three hardware chip selects, is only usable on a Compute Module [...]
  182. # https://github.com/raspberrypi/documentation/blob/d41d69f8efa3667b1a8b01a669238b8bd113edc1/hardware/raspberrypi/spi/README.md#hardware
  183. # https://www.raspberrypi.org/documentation/hardware/raspberrypi/spi/README.md
  184. self._spi_chip_select = int(spi_chip_select)
  185. self._lock_spi_device = lock_spi_device
  186. @property
  187. def _spi_device_path(self) -> str:
  188. # https://github.com/doceme/py-spidev/blob/v3.4/spidev_module.c#L1286
  189. return f"/dev/spidev{self._spi_bus}.{self._spi_chip_select}"
  190. @staticmethod
  191. def _log_chip_status_byte(chip_status: int) -> None:
  192. # see "10.1 Chip Status Byte" & "Table 23: Status Byte Summary"
  193. # > The command strobe registers are accessed by transferring
  194. # > a single header byte [...]. That is, only the R/W̄ bit,
  195. # > the burst access bit (set to 0), and the six address bits [...]
  196. # > The R/W̄ bit can be either one or zero and will determine how the
  197. # > FIFO_BYTES_AVAILABLE field in the status byte should be interpreted.
  198. _LOGGER.debug(
  199. "chip status byte: CHIP_RDYn=%d STATE=%s FIFO_BYTES_AVAILBLE=%d",
  200. chip_status >> 7,
  201. bin((chip_status >> 4) & 0b111),
  202. chip_status & 0b1111,
  203. )
  204. def _read_single_byte(
  205. self, register: ConfigurationRegisterAddress | FIFORegisterAddress
  206. ) -> int:
  207. response = self._spi.xfer([register | self._READ_SINGLE_BYTE, 0])
  208. assert len(response) == 2, response
  209. self._log_chip_status_byte(response[0])
  210. return response[1]
  211. def _read_burst(
  212. self,
  213. start_register: (
  214. ConfigurationRegisterAddress | PatableAddress | FIFORegisterAddress
  215. ),
  216. length: int,
  217. ) -> list[int]:
  218. response = self._spi.xfer([start_register | self._READ_BURST] + [0] * length)
  219. assert len(response) == length + 1, response
  220. self._log_chip_status_byte(response[0])
  221. return response[1:]
  222. def _read_status_register(self, register: StatusRegisterAddress) -> int:
  223. # > For register addresses in the range 0x30-0x3D,
  224. # > the burst bit is used to select between
  225. # > status registers when burst bit is one, and
  226. # > between command strobes when burst bit is
  227. # > zero. [...]
  228. # > Because of this, burst access is not available
  229. # > for status registers and they must be accessed
  230. # > one at a time. The status registers can only be
  231. # > read.
  232. response = self._spi.xfer([register | self._READ_BURST, 0])
  233. assert len(response) == 2, response
  234. self._log_chip_status_byte(response[0])
  235. return response[1]
  236. def _command_strobe(self, register: StrobeAddress) -> None:
  237. # see "10.4 Command Strobes"
  238. _LOGGER.debug("sending command strobe 0x%02x", register)
  239. response = self._spi.xfer([register | self._WRITE_SINGLE_BYTE])
  240. assert len(response) == 1, response
  241. self._log_chip_status_byte(response[0])
  242. def _write_burst(
  243. self,
  244. start_register: (
  245. ConfigurationRegisterAddress | PatableAddress | FIFORegisterAddress
  246. ),
  247. values: list[int],
  248. ) -> None:
  249. _LOGGER.debug(
  250. "writing burst: start_register=0x%02x values=%s", start_register, values
  251. )
  252. response = self._spi.xfer([start_register | self._WRITE_BURST] + values)
  253. assert len(response) == len(values) + 1, response
  254. self._log_chip_status_byte(response[0])
  255. assert all(v == response[0] for v in response[1:]), response
  256. def _reset(self) -> None:
  257. self._command_strobe(StrobeAddress.SRES)
  258. @classmethod
  259. def _filter_bandwidth_floating_point_to_real(
  260. cls, *, mantissa: int, exponent: int
  261. ) -> float:
  262. """
  263. See "13 Receiver Channel Filter Bandwidth"
  264. """
  265. return cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / (
  266. 8 * (4 + mantissa) * (2**exponent)
  267. )
  268. def _get_filter_bandwidth_hertz(self) -> float:
  269. """
  270. MDMCFG4.CHANBW_E & MDMCFG4.CHANBW_M
  271. > [...] decimation ratio for the delta-sigma ADC input stream
  272. > and thus the channel bandwidth.
  273. See "13 Receiver Channel Filter Bandwidth"
  274. """
  275. mdmcfg4 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4)
  276. return self._filter_bandwidth_floating_point_to_real(
  277. exponent=mdmcfg4 >> 6, mantissa=(mdmcfg4 >> 4) & 0b11
  278. )
  279. def _set_filter_bandwidth(self, *, mantissa: int, exponent: int) -> None:
  280. """
  281. MDMCFG4.CHANBW_E & MDMCFG4.CHANBW_M
  282. """
  283. mdmcfg4 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4)
  284. mdmcfg4 &= 0b00001111
  285. assert 0 <= exponent <= 0b11, exponent
  286. mdmcfg4 |= exponent << 6
  287. assert 0 <= mantissa <= 0b11, mantissa
  288. mdmcfg4 |= mantissa << 4
  289. self._write_burst(
  290. start_register=ConfigurationRegisterAddress.MDMCFG4, values=[mdmcfg4]
  291. )
  292. def _get_symbol_rate_exponent(self) -> int:
  293. """
  294. MDMCFG4.DRATE_E
  295. """
  296. return self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4) & 0b00001111
  297. def _set_symbol_rate_exponent(self, exponent: int):
  298. mdmcfg4 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG4)
  299. mdmcfg4 &= 0b11110000
  300. mdmcfg4 |= exponent
  301. self._write_burst(
  302. start_register=ConfigurationRegisterAddress.MDMCFG4, values=[mdmcfg4]
  303. )
  304. def _get_symbol_rate_mantissa(self) -> int:
  305. """
  306. MDMCFG3.DRATE_M
  307. """
  308. return self._read_single_byte(ConfigurationRegisterAddress.MDMCFG3)
  309. def _set_symbol_rate_mantissa(self, mantissa: int) -> None:
  310. self._write_burst(
  311. start_register=ConfigurationRegisterAddress.MDMCFG3, values=[mantissa]
  312. )
  313. @classmethod
  314. def _symbol_rate_floating_point_to_real(
  315. cls, *, mantissa: int, exponent: int
  316. ) -> float:
  317. # see "12 Data Rate Programming"
  318. return (
  319. (256 + mantissa)
  320. * (2**exponent)
  321. * cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ
  322. / (2**28)
  323. )
  324. @classmethod
  325. def _symbol_rate_real_to_floating_point(cls, real: float) -> tuple[int, int]:
  326. # see "12 Data Rate Programming"
  327. assert real > 0, real
  328. exponent = math.floor(
  329. math.log2(real / cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ) + 20
  330. )
  331. mantissa = round(
  332. real * 2**28 / cls._CRYSTAL_OSCILLATOR_FREQUENCY_HERTZ / 2**exponent - 256
  333. )
  334. if mantissa == 256:
  335. exponent += 1
  336. mantissa = 0
  337. assert 0 < exponent <= 2**4, exponent
  338. assert mantissa <= 2**8, mantissa
  339. return mantissa, exponent
  340. def get_symbol_rate_baud(self) -> float:
  341. return self._symbol_rate_floating_point_to_real(
  342. mantissa=self._get_symbol_rate_mantissa(),
  343. exponent=self._get_symbol_rate_exponent(),
  344. )
  345. def set_symbol_rate_baud(self, real: float) -> None:
  346. # > The data rate can be set from 0.6 kBaud to 500 kBaud [...]
  347. mantissa, exponent = self._symbol_rate_real_to_floating_point(real)
  348. self._set_symbol_rate_mantissa(mantissa)
  349. self._set_symbol_rate_exponent(exponent)
  350. def get_modulation_format(self) -> ModulationFormat:
  351. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  352. return ModulationFormat((mdmcfg2 >> 4) & 0b111)
  353. def _set_modulation_format(self, modulation_format: ModulationFormat) -> None:
  354. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  355. mdmcfg2 &= 0b10001111
  356. mdmcfg2 |= modulation_format << 4
  357. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  358. def enable_manchester_code(self) -> None:
  359. """
  360. MDMCFG2.MANCHESTER_EN
  361. Enable manchester encoding & decoding for the entire packet,
  362. including the preamble and synchronization word.
  363. """
  364. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  365. mdmcfg2 |= 0b1000
  366. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  367. def get_sync_mode(self) -> SyncMode:
  368. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  369. return SyncMode(mdmcfg2 & 0b11)
  370. def set_sync_mode(
  371. self,
  372. mode: SyncMode,
  373. *,
  374. _carrier_sense_threshold_enabled: bool | None = None, # unstable
  375. ) -> None:
  376. """
  377. MDMCFG2.SYNC_MODE
  378. see "14.3 Byte Synchronization"
  379. Carrier Sense (CS) Threshold (when receiving packets, API unstable):
  380. > Carrier sense can be used as a sync word qualifier
  381. > that requires the signal level to be higher than the threshold
  382. > for a sync word > search to be performed [...]
  383. > CS can be used to avoid interference from other RF sources [...]
  384. True: enable, False: disable, None: keep current setting
  385. See "17.4 Carrier Sense (CS)"
  386. """
  387. mdmcfg2 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG2)
  388. mdmcfg2 &= 0b11111100
  389. mdmcfg2 |= mode
  390. if _carrier_sense_threshold_enabled is not None:
  391. if _carrier_sense_threshold_enabled:
  392. mdmcfg2 |= 0b00000100
  393. else:
  394. mdmcfg2 &= 0b11111011
  395. self._write_burst(ConfigurationRegisterAddress.MDMCFG2, [mdmcfg2])
  396. def get_preamble_length_bytes(self) -> int:
  397. """
  398. MDMCFG1.NUM_PREAMBLE
  399. Minimum number of preamble bytes to be transmitted.
  400. See "15.2 Packet Format"
  401. """
  402. index = (
  403. self._read_single_byte(ConfigurationRegisterAddress.MDMCFG1) >> 4
  404. ) & 0b111
  405. return 2 ** (index >> 1) * (2 + (index & 0b1))
  406. def _set_preamble_length_index(self, index: int) -> None:
  407. assert 0 <= index <= 0b111
  408. mdmcfg1 = self._read_single_byte(ConfigurationRegisterAddress.MDMCFG1)
  409. mdmcfg1 &= 0b10001111
  410. mdmcfg1 |= index << 4
  411. self._write_burst(ConfigurationRegisterAddress.MDMCFG1, [mdmcfg1])
  412. def set_preamble_length_bytes(self, length: int) -> None:
  413. """
  414. see .get_preamble_length_bytes()
  415. """
  416. if length < 1:
  417. raise ValueError(
  418. f"invalid preamble length {length} given"
  419. "\ncall .set_sync_mode(cc1101.SyncMode.NO_PREAMBLE_AND_SYNC_WORD)"
  420. " to disable preamble"
  421. )
  422. if length % 3 == 0: # pylint: disable=consider-ternary-expression
  423. index = math.log2(length / 3) * 2 + 1
  424. else:
  425. index = math.log2(length / 2) * 2
  426. if not index.is_integer() or index < 0 or index > 0b111:
  427. raise ValueError(
  428. f"unsupported preamble length: {length} bytes"
  429. "\nsee MDMCFG1.NUM_PREAMBLE in cc1101 docs"
  430. )
  431. self._set_preamble_length_index(int(index))
  432. def _get_power_amplifier_setting_index(self) -> int:
  433. """
  434. see ._set_power_amplifier_setting_index
  435. """
  436. return self._read_single_byte(ConfigurationRegisterAddress.FREND0) & 0b111
  437. def _set_power_amplifier_setting_index(self, setting_index: int) -> None:
  438. """
  439. FREND0.PA_POWER
  440. > This value is an index to the PATABLE,
  441. > which can be programmed with up to 8 different PA settings.
  442. > In OOK/ASK mode, this selects the PATABLE index to use
  443. > when transmitting a '1'.
  444. > PATABLE index zero is used in OOK/ASK when transmitting a '0'.
  445. > The PATABLE settings from index 0 to the PA_POWER value are
  446. > used for > ASK TX shaping, [...]
  447. see "Figure 32: Shaping of ASK Signal"
  448. > If OOK modulation is used, the logic 0 and logic 1 power levels
  449. > shall be programmed to index 0 and 1 respectively.
  450. """
  451. assert 0 <= setting_index <= 0b111, setting_index
  452. frend0 = self._read_single_byte(ConfigurationRegisterAddress.FREND0)
  453. frend0 &= 0b11111000
  454. frend0 |= setting_index
  455. self._write_burst(ConfigurationRegisterAddress.FREND0, [frend0])
  456. def _verify_chip(self) -> None:
  457. partnum = self._read_status_register(StatusRegisterAddress.PARTNUM)
  458. if partnum != self._SUPPORTED_PARTNUM:
  459. raise ValueError(
  460. f"unexpected chip part number {partnum} (expected: {self._SUPPORTED_PARTNUM})"
  461. )
  462. version = self._read_status_register(StatusRegisterAddress.VERSION)
  463. if version not in self._SUPPORTED_VERSIONS:
  464. msg = f"Unsupported chip version 0x{version:02x}"
  465. supported_versions = ", ".join(
  466. f"0x{v:02x}" for v in self._SUPPORTED_VERSIONS
  467. )
  468. msg += f" (expected one of [{supported_versions}])"
  469. if version == 0:
  470. msg += (
  471. "\n\nPlease verify that all required pins are connected"
  472. " (see https://github.com/fphammerle/python-cc1101#wiring-raspberry-pi)"
  473. " and that you selected the correct SPI bus and chip/slave select line."
  474. )
  475. raise ValueError(msg)
  476. def _configure_defaults(self) -> None:
  477. # next major/breaking release will probably stick closer to CC1101's defaults
  478. # 6:4 MOD_FORMAT: OOK (default: 2-FSK)
  479. self._set_modulation_format(ModulationFormat.ASK_OOK)
  480. self._set_power_amplifier_setting_index(1)
  481. self._disable_data_whitening()
  482. # 7:6 unused
  483. # 5:4 FS_AUTOCAL: calibrate when going from IDLE to RX or TX
  484. # 3:2 PO_TIMEOUT: default
  485. # 1 PIN_CTRL_EN: default
  486. # 0 XOSC_FORCE_ON: default
  487. self._write_burst(ConfigurationRegisterAddress.MCSM0, [0b010100])
  488. # > Default is CLK_XOSC/192 (See Table 41 on page 62).
  489. # > It is recommended to disable the clock output in initialization,
  490. # > in order to optimize RF performance.
  491. self._write_burst(
  492. ConfigurationRegisterAddress.IOCFG0,
  493. # required for _wait_for_packet()
  494. [GDOSignalSelection.RX_FIFO_AT_OR_ABOVE_THRESHOLD_OR_PACKET_END_REACHED],
  495. )
  496. def __enter__(self) -> CC1101:
  497. # https://docs.python.org/3/reference/datamodel.html#object.__enter__
  498. try:
  499. self._spi.open(self._spi_bus, self._spi_chip_select)
  500. except PermissionError as exc:
  501. raise PermissionError(
  502. f"Could not access {self._spi_device_path}"
  503. "\nVerify that the current user has both read and write access."
  504. "\nOn some systems, like Raspberry Pi OS / Raspbian,"
  505. "\n\tsudo usermod -a -G spi $USER"
  506. "\nfollowed by a re-login grants sufficient permissions."
  507. ) from exc
  508. if self._lock_spi_device:
  509. # advisory, exclusive, non-blocking
  510. # lock removed in __exit__ by SpiDev.close()
  511. fcntl.flock(self._spi.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
  512. try:
  513. self._spi.max_speed_hz = self._spi_max_speed_hz
  514. self._reset()
  515. self._verify_chip()
  516. self._configure_defaults()
  517. marcstate = self.get_main_radio_control_state_machine_state()
  518. if marcstate != MainRadioControlStateMachineState.IDLE:
  519. raise ValueError(f"expected marcstate idle (actual: {marcstate.name})")
  520. except:
  521. self._spi.close()
  522. raise
  523. return self
  524. def __exit__(self, exc_type, exc_value, traceback) -> typing.Literal[False]:
  525. # https://docs.python.org/3/reference/datamodel.html#object.__exit__
  526. self._spi.close()
  527. return False
  528. def unlock_spi_device(self) -> None:
  529. """
  530. Manually release the lock set on the SPI device file.
  531. Alternatively, the lock will be released automatically,
  532. when leaving the context.
  533. Method fails silently, if the SPI device file is not locked.
  534. >>> transceiver = cc1101.CC1101(lock_spi_device=True)
  535. >>> # not locked
  536. >>> with transceiver:
  537. >>> # locked
  538. >>> # lock removed
  539. >>> with transceiver:
  540. >>> # locked
  541. >>> transceiver.unlock_spi_device()
  542. >>> # lock removed
  543. """
  544. fileno = self._spi.fileno()
  545. if fileno != -1:
  546. fcntl.flock(fileno, fcntl.LOCK_UN)
  547. def get_main_radio_control_state_machine_state(
  548. self,
  549. ) -> MainRadioControlStateMachineState:
  550. return MainRadioControlStateMachineState(
  551. self._read_status_register(StatusRegisterAddress.MARCSTATE)
  552. )
  553. def get_marc_state(self) -> MainRadioControlStateMachineState:
  554. """
  555. alias for get_main_radio_control_state_machine_state()
  556. """
  557. return self.get_main_radio_control_state_machine_state()
  558. @classmethod
  559. def _frequency_control_word_to_hertz(cls, control_word: list[int]) -> float:
  560. return (
  561. int.from_bytes(control_word, byteorder="big", signed=False)
  562. * cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR
  563. )
  564. @classmethod
  565. def _hertz_to_frequency_control_word(cls, hertz: float) -> list[int]:
  566. return list(
  567. round(hertz / cls._FREQUENCY_CONTROL_WORD_HERTZ_FACTOR).to_bytes(
  568. length=3, byteorder="big", signed=False
  569. )
  570. )
  571. def _get_base_frequency_control_word(self) -> list[int]:
  572. # > The base or start frequency is set by the 24 bitfrequency
  573. # > word located in the FREQ2, FREQ1, FREQ0 registers.
  574. return self._read_burst(
  575. start_register=ConfigurationRegisterAddress.FREQ2, length=3
  576. )
  577. def _set_base_frequency_control_word(self, control_word: list[int]) -> None:
  578. self._write_burst(
  579. start_register=ConfigurationRegisterAddress.FREQ2, values=control_word
  580. )
  581. def get_base_frequency_hertz(self) -> float:
  582. return self._frequency_control_word_to_hertz(
  583. self._get_base_frequency_control_word()
  584. )
  585. def set_base_frequency_hertz(self, freq: float) -> None:
  586. if freq < (self._TRANSMIT_MIN_FREQUENCY_HERTZ - 50e3):
  587. # > [use] warnings.warn() in library code if the issue is avoidable
  588. # > and the client application should be modified to eliminate the warning[.]
  589. # > [use] logging.warning() if there is nothing the client application
  590. # > can do about the situation, but the event should still be noted.
  591. # https://docs.python.org/3/howto/logging.html#when-to-use-logging
  592. warnings.warn(
  593. "CC1101 is unable to transmit at frequencies"
  594. f" below {(self._TRANSMIT_MIN_FREQUENCY_HERTZ / 1e6):.1f} MHz"
  595. )
  596. self._set_base_frequency_control_word(
  597. self._hertz_to_frequency_control_word(freq)
  598. )
  599. def __str__(self) -> str:
  600. sync_mode = self.get_sync_mode()
  601. attrs = (
  602. f"marcstate={self.get_main_radio_control_state_machine_state().name.lower()}",
  603. f"base_frequency={(self.get_base_frequency_hertz() / 1e6):.2f}MHz",
  604. f"symbol_rate={(self.get_symbol_rate_baud() / 1000):.2f}kBaud",
  605. f"modulation_format={self.get_modulation_format().name}",
  606. f"sync_mode={sync_mode.name}",
  607. (
  608. f"preamble_length={self.get_preamble_length_bytes()}B"
  609. if sync_mode != SyncMode.NO_PREAMBLE_AND_SYNC_WORD
  610. else None
  611. ),
  612. (
  613. f"sync_word=0x{self.get_sync_word().hex()}"
  614. if sync_mode != SyncMode.NO_PREAMBLE_AND_SYNC_WORD
  615. else None
  616. ),
  617. "packet_length{}{}B".format( # pylint: disable=consider-using-f-string
  618. (
  619. "≤"
  620. if self.get_packet_length_mode() == PacketLengthMode.VARIABLE
  621. else "="
  622. ),
  623. self.get_packet_length_bytes(),
  624. ),
  625. "output_power="
  626. + _format_patable(self.get_output_power(), insert_spaces=False),
  627. )
  628. # pylint: disable=consider-using-f-string
  629. return "CC1101({})".format(", ".join(filter(None, attrs)))
  630. def get_configuration_register_values(
  631. self,
  632. start_register: ConfigurationRegisterAddress = min(
  633. ConfigurationRegisterAddress
  634. ),
  635. end_register: ConfigurationRegisterAddress = max(ConfigurationRegisterAddress),
  636. ) -> dict[ConfigurationRegisterAddress, int]:
  637. assert start_register <= end_register, (start_register, end_register)
  638. values = self._read_burst(
  639. start_register=start_register, length=end_register - start_register + 1
  640. )
  641. return {
  642. ConfigurationRegisterAddress(start_register + i): v
  643. for i, v in enumerate(values)
  644. }
  645. def get_sync_word(self) -> bytes:
  646. """
  647. SYNC1 & SYNC0
  648. See "15.2 Packet Format"
  649. The first byte's most significant bit is transmitted first.
  650. """
  651. return bytes(
  652. self._read_burst(
  653. start_register=ConfigurationRegisterAddress.SYNC1, length=2
  654. )
  655. )
  656. def set_sync_word(self, sync_word: bytes) -> None:
  657. """
  658. See .set_sync_word()
  659. """
  660. if len(sync_word) != 2:
  661. raise ValueError(f"expected two bytes, got {sync_word!r}")
  662. self._write_burst(
  663. start_register=ConfigurationRegisterAddress.SYNC1, values=list(sync_word)
  664. )
  665. def get_packet_length_bytes(self) -> int:
  666. """
  667. PKTLEN
  668. Packet length in fixed packet length mode,
  669. maximum packet length in variable packet length mode.
  670. > In variable packet length mode, [...]
  671. > any packet received with a length byte
  672. > with a value greater than PKTLEN will be discarded.
  673. """
  674. return self._read_single_byte(ConfigurationRegisterAddress.PKTLEN)
  675. def set_packet_length_bytes(self, packet_length: int) -> None:
  676. """
  677. see get_packet_length_bytes()
  678. """
  679. assert 1 <= packet_length <= 255, f"unsupported packet length {packet_length}"
  680. self._write_burst(
  681. start_register=ConfigurationRegisterAddress.PKTLEN, values=[packet_length]
  682. )
  683. def _disable_data_whitening(self):
  684. """
  685. PKTCTRL0.WHITE_DATA
  686. see "15.1 Data Whitening"
  687. > By setting PKTCTRL0.WHITE_DATA=1 [default],
  688. > all data, except the preamble and the sync word
  689. > will be XOR-ed with a 9-bit pseudo-random (PN9)
  690. > sequence before being transmitted.
  691. """
  692. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  693. pktctrl0 &= 0b10111111
  694. self._write_burst(
  695. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  696. )
  697. def disable_checksum(self) -> None:
  698. """
  699. PKTCTRL0.CRC_EN
  700. Disable automatic 2-byte cyclic redundancy check (CRC) sum
  701. appending in TX mode and checking in RX mode.
  702. See "Figure 19: Packet Format".
  703. """
  704. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  705. pktctrl0 &= 0b11111011
  706. self._write_burst(
  707. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  708. )
  709. def _get_transceive_mode(self) -> _TransceiveMode:
  710. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  711. return _TransceiveMode((pktctrl0 >> 4) & 0b11)
  712. def _set_transceive_mode(self, mode: _TransceiveMode) -> None:
  713. _LOGGER.info("changing transceive mode to %s", mode.name)
  714. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  715. pktctrl0 &= ~0b00110000
  716. pktctrl0 |= mode << 4
  717. self._write_burst(
  718. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  719. )
  720. def get_packet_length_mode(self) -> PacketLengthMode:
  721. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  722. return PacketLengthMode(pktctrl0 & 0b11)
  723. def set_packet_length_mode(self, mode: PacketLengthMode) -> None:
  724. pktctrl0 = self._read_single_byte(ConfigurationRegisterAddress.PKTCTRL0)
  725. pktctrl0 &= 0b11111100
  726. pktctrl0 |= mode
  727. self._write_burst(
  728. start_register=ConfigurationRegisterAddress.PKTCTRL0, values=[pktctrl0]
  729. )
  730. def _get_patable(self) -> tuple[int, ...]:
  731. """
  732. see "10.6 PATABLE Access" and "24 Output Power Programming"
  733. default: (0xC6, 0, 0, 0, 0, 0, 0, 0)
  734. """
  735. return tuple(
  736. self._read_burst(
  737. start_register=PatableAddress.PATABLE, length=self._PATABLE_LENGTH_BYTES
  738. )
  739. )
  740. def _set_patable(self, settings: collections.abc.Iterable[int]):
  741. settings = list(settings)
  742. assert all(0 <= l <= 0xFF for l in settings), settings
  743. assert 0 < len(settings) <= self._PATABLE_LENGTH_BYTES, settings
  744. self._write_burst(start_register=PatableAddress.PATABLE, values=settings)
  745. def get_output_power(self) -> tuple[int, ...]:
  746. """
  747. Returns the enabled output power settings
  748. (up to 8 bytes of the PATABLE register).
  749. see .set_output_power()
  750. """
  751. return self._get_patable()[: self._get_power_amplifier_setting_index() + 1]
  752. def set_output_power(self, power_settings: collections.abc.Iterable[int]) -> None:
  753. """
  754. Configures output power levels by setting PATABLE and FREND0.PA_POWER.
  755. Up to 8 bytes may be provided.
  756. > [PATABLE] provides flexible PA power ramp up and ramp down
  757. > at the start and end of transmission when using 2-FSK, GFSK,
  758. > 4-FSK, and MSK modulation as well as ASK modulation shaping.
  759. For OOK modulation, exactly 2 bytes must be provided:
  760. 0 to turn off the transmission for logical 0,
  761. and a level > 0 to turn on the transmission for logical 1.
  762. >>> transceiver.set_output_power((0, 0xC6))
  763. See "Table 39: Optimum PATABLE Settings for Various Output Power Levels [...]"
  764. and section "24 Output Power Programming".
  765. """
  766. power_settings = list(power_settings)
  767. # checks in sub-methods
  768. self._set_power_amplifier_setting_index(len(power_settings) - 1)
  769. self._set_patable(power_settings)
  770. def _flush_tx_fifo_buffer(self) -> None:
  771. # > Only issue SFTX in IDLE or TXFIFO_UNDERFLOW states.
  772. _LOGGER.debug("flushing tx fifo buffer")
  773. self._command_strobe(StrobeAddress.SFTX)
  774. def transmit(self, payload: bytes) -> None:
  775. """
  776. The most significant bit is transmitted first.
  777. In variable packet length mode,
  778. a byte indicating the packet's length will be prepended.
  779. > In variable packet length mode,
  780. > the packet length is configured by the first byte [...].
  781. > The packet length is defined as the payload data,
  782. > excluding the length byte and the optional CRC.
  783. from "15.2 Packet Format"
  784. Call .set_packet_length_mode(cc1101.PacketLengthMode.FIXED)
  785. to switch to fixed packet length mode.
  786. """
  787. # see "15.2 Packet Format"
  788. # > In variable packet length mode, [...]
  789. # > The first byte written to the TXFIFO must be different from 0.
  790. packet_length_mode = self.get_packet_length_mode()
  791. packet_length = self.get_packet_length_bytes()
  792. if packet_length_mode == PacketLengthMode.VARIABLE:
  793. if not payload:
  794. raise ValueError(f"empty payload {payload!r}")
  795. if len(payload) > packet_length:
  796. raise ValueError(
  797. f"payload exceeds maximum payload length of {packet_length} bytes"
  798. "\nsee .get_packet_length_bytes()"
  799. f"\npayload: {payload!r}"
  800. )
  801. payload = int.to_bytes(len(payload), length=1, byteorder="big") + payload
  802. elif (
  803. packet_length_mode == PacketLengthMode.FIXED
  804. and len(payload) != packet_length
  805. ):
  806. raise ValueError(
  807. f"expected payload length of {packet_length} bytes, got {len(payload)}"
  808. + "\nsee .set_packet_length_mode() and .get_packet_length_bytes()"
  809. + f"\npayload: {payload!r}"
  810. )
  811. marcstate = self.get_main_radio_control_state_machine_state()
  812. if marcstate != MainRadioControlStateMachineState.IDLE:
  813. raise RuntimeError(
  814. f"device must be idle before transmission (current marcstate: {marcstate.name})"
  815. )
  816. self._flush_tx_fifo_buffer()
  817. self._write_burst(FIFORegisterAddress.TX, list(payload))
  818. _LOGGER.info("transmitting 0x%s (%r)", payload.hex(), payload)
  819. self._command_strobe(StrobeAddress.STX)
  820. @contextlib.contextmanager
  821. def asynchronous_transmission(self) -> collections.abc.Iterator[Pin]:
  822. """
  823. > [...] the GDO0 pin is used for data input [...]
  824. > The CC1101 modulator samples the level of the asynchronous input
  825. > 8 times faster than the programmed data rate.
  826. see "27.1 Asynchronous Serial Operation"
  827. >>> with cc1101.CC1101() as transceiver:
  828. >>> transceiver.set_base_frequency_hertz(433.92e6)
  829. >>> transceiver.set_symbol_rate_baud(600)
  830. >>> print(transceiver)
  831. >>> with transceiver.asynchronous_transmission():
  832. >>> # send digital signal to GDO0 pin
  833. """
  834. self._set_transceive_mode(_TransceiveMode.ASYNCHRONOUS_SERIAL)
  835. self._command_strobe(StrobeAddress.STX)
  836. try:
  837. # > In TX, the GDO0 pin is used for data input (TX data).
  838. yield Pin.GDO0
  839. finally:
  840. self._command_strobe(StrobeAddress.SIDLE)
  841. self._set_transceive_mode(_TransceiveMode.FIFO)
  842. def _enable_receive_mode(self) -> None:
  843. self._command_strobe(StrobeAddress.SRX)
  844. def _get_received_packet(self) -> _ReceivedPacket | None: # unstable
  845. """
  846. see section "20 Data FIFO"
  847. """
  848. rxbytes = self._read_status_register(StatusRegisterAddress.RXBYTES)
  849. # PKTCTRL1.APPEND_STATUS is enabled by default
  850. if rxbytes < 2:
  851. return None
  852. buffer = self._read_burst(start_register=FIFORegisterAddress.RX, length=rxbytes)
  853. return _ReceivedPacket(
  854. payload=bytes(buffer[:-2]),
  855. rssi_index=buffer[-2],
  856. checksum_valid=bool(buffer[-1] >> 7),
  857. link_quality_indicator=buffer[-1] & 0b0111111,
  858. )
  859. def _wait_for_packet( # unstable
  860. self,
  861. timeout: datetime.timedelta,
  862. gdo0_gpio_line_name: bytes = b"GPIO24", # recommended in README.md
  863. ) -> _ReceivedPacket | None:
  864. """
  865. depends on IOCFG0 == 0b00000001 (see _configure_defaults)
  866. """
  867. # pylint: disable=protected-access
  868. gdo0 = cc1101._gpio.GPIOLine.find(name=gdo0_gpio_line_name)
  869. self._enable_receive_mode()
  870. if not gdo0.wait_for_rising_edge(consumer=b"CC1101:GDO0", timeout=timeout):
  871. self._command_strobe(StrobeAddress.SIDLE)
  872. _LOGGER.debug(
  873. "reached timeout of %.02f seconds while waiting for packet",
  874. timeout.total_seconds(),
  875. )
  876. return None # timeout
  877. return self._get_received_packet()