flipperzero-firmware/lib/subghz/subghz_tx_rx_worker.c
SG 274c12fc56
[FL-2274] Inventing streams and moving FFF to them (#981)
* Streams: string stream
* String stream: updated insert/delete api
* Streams: generic stream interface and string stream implementation
* Streams: helpers for insert and delete_and_insert
* FFF: now compatible with streams
* MinUnit: introduced tests with arguments
* FFF: stream access violation
* Streams: copy data between streams
* Streams: file stream
* FFF: documentation
* FFStream: documentation
* FFF: alloc as file
* MinUnit: support for nested tests
* Streams: changed delete_and_insert, now it returns success flag. Added ability dump stream inner parameters and data to cout.
* FFF: simplified file open function
* Streams: unit tests
* FFF: tests
* Streams: declare cache_size constant as define, to allow variable modified arrays
* FFF: lib moved to a separate folder
* iButton: new FFF
* RFID: new FFF
* Animations: new FFF
* IR: new FFF
* NFC: new FFF
* Flipper file format: delete lib
* U2F: new FFF
* Subghz: new FFF and streams
* Streams: read line
* Streams: split
* FuriCore: implement memset with extra asserts
* FuriCore: implement extra heap asserts without inventing memset
* Scene manager: protected access to the scene id stack with a size check
* NFC worker: dirty fix for issue where hal_nfc was busy on app start
* Furi: update allocator to erase memory on allocation. Replace furi_alloc with malloc.
* FuriCore: cleanup memmgr code.
* Furi HAL: furi_hal_init is split into critical and non-critical parts. The critical part is currently clock and console.
* Memmgr: added ability to track allocations and deallocations through console.
* FFStream: some speedup
* Streams, FF: minor fixes
* Tests: restore
* File stream: a slightly more thread-safe version of file_stream_delete_and_insert

Co-authored-by: Aleksandr Kutuzov <alleteam@gmail.com>
2022-02-18 22:53:46 +03:00

260 lines
8.2 KiB
C

#include "subghz_tx_rx_worker.h"
#include <stream_buffer.h>
#include <furi.h>
#define TAG "SubGhzTxRxWorker"
#define SUBGHZ_TXRX_WORKER_BUF_SIZE 2048
//you can not set more than 62 because it will not fit into the FIFO CC1101
#define SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE 60
#define SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF 40
struct SubGhzTxRxWorker {
FuriThread* thread;
StreamBufferHandle_t stream_tx;
StreamBufferHandle_t stream_rx;
volatile bool worker_running;
volatile bool worker_stoping;
SubGhzTxRxWorkerStatus status;
uint32_t frequency;
SubGhzTxRxWorkerCallbackHaveRead callback_have_read;
void* context_have_read;
};
bool subghz_tx_rx_worker_write(SubGhzTxRxWorker* instance, uint8_t* data, size_t size) {
furi_assert(instance);
bool ret = false;
size_t stream_tx_free_byte = xStreamBufferSpacesAvailable(instance->stream_tx);
if(size && (stream_tx_free_byte >= size)) {
if(xStreamBufferSend(
instance->stream_tx, data, size, SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF) ==
size) {
ret = true;
}
}
return ret;
}
size_t subghz_tx_rx_worker_available(SubGhzTxRxWorker* instance) {
furi_assert(instance);
return xStreamBufferBytesAvailable(instance->stream_rx);
}
size_t subghz_tx_rx_worker_read(SubGhzTxRxWorker* instance, uint8_t* data, size_t size) {
furi_assert(instance);
return xStreamBufferReceive(instance->stream_rx, data, size, 0);
}
void subghz_tx_rx_worker_set_callback_have_read(
SubGhzTxRxWorker* instance,
SubGhzTxRxWorkerCallbackHaveRead callback,
void* context) {
furi_assert(instance);
furi_assert(callback);
furi_assert(context);
instance->callback_have_read = callback;
instance->context_have_read = context;
}
bool subghz_tx_rx_worker_rx(SubGhzTxRxWorker* instance, uint8_t* data, uint8_t* size) {
uint8_t timeout = 100;
bool ret = false;
if(instance->status != SubGhzTxRxWorkerStatusRx) {
furi_hal_subghz_rx();
instance->status = SubGhzTxRxWorkerStatusRx;
osDelay(1);
}
//waiting for reception to complete
while(hal_gpio_read(&gpio_cc1101_g0)) {
osDelay(1);
if(!--timeout) {
FURI_LOG_W(TAG, "RX cc1101_g0 timeout");
furi_hal_subghz_flush_rx();
furi_hal_subghz_rx();
break;
}
}
if(furi_hal_subghz_rx_pipe_not_empty()) {
FURI_LOG_I(
TAG, "RSSI: %03.1fdbm LQI: %d", furi_hal_subghz_get_rssi(), furi_hal_subghz_get_lqi());
if(furi_hal_subghz_is_rx_data_crc_valid()) {
furi_hal_subghz_read_packet(data, size);
ret = true;
}
furi_hal_subghz_flush_rx();
furi_hal_subghz_rx();
}
return ret;
}
void subghz_tx_rx_worker_tx(SubGhzTxRxWorker* instance, uint8_t* data, size_t size) {
uint8_t timeout = 200;
if(instance->status != SubGhzTxRxWorkerStatusIDLE) {
furi_hal_subghz_idle();
}
furi_hal_subghz_write_packet(data, size);
furi_hal_subghz_tx(); //start send
instance->status = SubGhzTxRxWorkerStatusTx;
while(!hal_gpio_read(&gpio_cc1101_g0)) { // Wait for GDO0 to be set -> sync transmitted
osDelay(1);
if(!--timeout) {
FURI_LOG_W(TAG, "TX !cc1101_g0 timeout");
break;
}
}
while(hal_gpio_read(&gpio_cc1101_g0)) { // Wait for GDO0 to be cleared -> end of packet
osDelay(1);
if(!--timeout) {
FURI_LOG_W(TAG, "TX cc1101_g0 timeout");
break;
}
}
furi_hal_subghz_idle();
instance->status = SubGhzTxRxWorkerStatusIDLE;
}
/** Worker thread
*
* @param context
* @return exit code
*/
static int32_t subghz_tx_rx_worker_thread(void* context) {
SubGhzTxRxWorker* instance = context;
FURI_LOG_I(TAG, "Worker start");
furi_hal_subghz_reset();
furi_hal_subghz_idle();
furi_hal_subghz_load_preset(FuriHalSubGhzPresetGFSK9_99KbAsync);
//furi_hal_subghz_load_preset(FuriHalSubGhzPresetMSK99_97KbAsync);
hal_gpio_init(&gpio_cc1101_g0, GpioModeInput, GpioPullNo, GpioSpeedLow);
furi_hal_subghz_set_frequency_and_path(instance->frequency);
furi_hal_subghz_flush_rx();
uint8_t data[SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE + 1] = {0};
size_t size_tx = 0;
uint8_t size_rx[1] = {0};
uint8_t timeout_tx = 0;
bool callback_rx = false;
while(instance->worker_running) {
//transmit
size_tx = xStreamBufferBytesAvailable(instance->stream_tx);
if(size_tx > 0 && !timeout_tx) {
timeout_tx = 10; //20ms
if(size_tx > SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE) {
xStreamBufferReceive(
instance->stream_tx,
&data,
SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE,
SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF);
subghz_tx_rx_worker_tx(instance, data, SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE);
} else {
//todo checking that he managed to write all the data to the TX buffer
xStreamBufferReceive(
instance->stream_tx, &data, size_tx, SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF);
subghz_tx_rx_worker_tx(instance, data, size_tx);
}
} else {
//recive
if(subghz_tx_rx_worker_rx(instance, data, size_rx)) {
if(xStreamBufferSpacesAvailable(instance->stream_rx) >= size_rx[0]) {
if(instance->callback_have_read &&
xStreamBufferBytesAvailable(instance->stream_rx) == 0) {
callback_rx = true;
}
//todo checking that he managed to write all the data to the RX buffer
xStreamBufferSend(
instance->stream_rx,
&data,
size_rx[0],
SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF);
if(callback_rx) {
instance->callback_have_read(instance->context_have_read);
callback_rx = false;
}
} else {
//todo RX buffer overflow
}
}
}
if(timeout_tx) timeout_tx--;
osDelay(1);
}
furi_hal_subghz_set_path(FuriHalSubGhzPathIsolate);
furi_hal_subghz_sleep();
FURI_LOG_I(TAG, "Worker stop");
return 0;
}
SubGhzTxRxWorker* subghz_tx_rx_worker_alloc() {
SubGhzTxRxWorker* instance = malloc(sizeof(SubGhzTxRxWorker));
instance->thread = furi_thread_alloc();
furi_thread_set_name(instance->thread, "SubghzTxRxWorker");
furi_thread_set_stack_size(instance->thread, 2048);
furi_thread_set_context(instance->thread, instance);
furi_thread_set_callback(instance->thread, subghz_tx_rx_worker_thread);
instance->stream_tx =
xStreamBufferCreate(sizeof(uint8_t) * SUBGHZ_TXRX_WORKER_BUF_SIZE, sizeof(uint8_t));
instance->stream_rx =
xStreamBufferCreate(sizeof(uint8_t) * SUBGHZ_TXRX_WORKER_BUF_SIZE, sizeof(uint8_t));
instance->status = SubGhzTxRxWorkerStatusIDLE;
instance->worker_stoping = true;
return instance;
}
void subghz_tx_rx_worker_free(SubGhzTxRxWorker* instance) {
furi_assert(instance);
furi_assert(!instance->worker_running);
vStreamBufferDelete(instance->stream_tx);
vStreamBufferDelete(instance->stream_rx);
furi_thread_free(instance->thread);
free(instance);
}
bool subghz_tx_rx_worker_start(SubGhzTxRxWorker* instance, uint32_t frequency) {
furi_assert(instance);
furi_assert(!instance->worker_running);
bool res = false;
xStreamBufferReset(instance->stream_tx);
xStreamBufferReset(instance->stream_rx);
instance->worker_running = true;
furi_thread_start(instance->thread);
if(furi_hal_subghz_is_tx_allowed(frequency)) {
instance->frequency = frequency;
res = true;
}
return res;
}
void subghz_tx_rx_worker_stop(SubGhzTxRxWorker* instance) {
furi_assert(instance);
furi_assert(instance->worker_running);
instance->worker_running = false;
furi_thread_join(instance->thread);
}
bool subghz_tx_rx_worker_is_running(SubGhzTxRxWorker* instance) {
furi_assert(instance);
return instance->worker_running;
}