3pp-mirror/ahoy
1
0
Fork 0
mirror of https://github.com/lumapu/ahoy.git synced 2025-04-28 17:56:21 +02:00
Code Issues Projects Releases Packages Wiki Activity

restructured communication queue and queue element

Browse source
This commit is contained in:
lumapu 2024-09-29 21:56:53 +02:00
parent 6f04379cb1
commit fe265d78a7
2 changed files with 78 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

114
src/hm/CommQueue.h
View file

@ -11,12 +11,13 @@
#include "hmInverter.h" #include "hmInverter.h"
#include "../utils/dbg.h" #include "../utils/dbg.h"
#define DEFAULT_ATTEMPS 5
#define MORE_ATTEMPS_ALARMDATA 3 // 8
#define MORE_ATTEMPS_GRIDONPROFILEPARA 0 // 5
template <uint8_t N=100> template <uint8_t N=100>
class CommQueue { class CommQueue {
protected: /* types */
static constexpr uint8_t DefaultAttempts = 5;
static constexpr uint8_t MoreAttemptsAlarmData = 3;
static constexpr uint8_t MoreAttemptsGridProfile = 0;
public: public:
CommQueue() CommQueue()
: wrPtr {0} : wrPtr {0}
@ -31,7 +32,7 @@ class CommQueue {
} }
void addImportant(Inverter<> *iv, uint8_t cmd) { void addImportant(Inverter<> *iv, uint8_t cmd) {
queue_s q(iv, cmd, true); QueueElement q(iv, cmd, true);
xSemaphoreTake(this->mutex, portMAX_DELAY); xSemaphoreTake(this->mutex, portMAX_DELAY);
if(!isIncluded(&q)) { if(!isIncluded(&q)) {
dec(&this->rdPtr); dec(&this->rdPtr);
@ -42,7 +43,7 @@ class CommQueue {
void add(Inverter<> *iv, uint8_t cmd) { void add(Inverter<> *iv, uint8_t cmd) {
xSemaphoreTake(this->mutex, portMAX_DELAY); xSemaphoreTake(this->mutex, portMAX_DELAY);
queue_s q(iv, cmd, false); QueueElement q(iv, cmd, false);
if(!isIncluded(&q)) { if(!isIncluded(&q)) {
mQueue[this->wrPtr] = q; mQueue[this->wrPtr] = q;
inc(&this->wrPtr); inc(&this->wrPtr);
@ -50,12 +51,6 @@ class CommQueue {
xSemaphoreGive(this->mutex); xSemaphoreGive(this->mutex);
} }
void chgCmd(Inverter<> *iv, uint8_t cmd) {
xSemaphoreTake(this->mutex, portMAX_DELAY);
mQueue[this->wrPtr] = queue_s(iv, cmd, false);
xSemaphoreGive(this->mutex);
}
uint8_t getFillState(void) const { uint8_t getFillState(void) const {
//DPRINTLN(DBG_INFO, "wr: " + String(this->wrPtr) + ", rd: " + String(this->rdPtr)); //DPRINTLN(DBG_INFO, "wr: " + String(this->wrPtr) + ", rd: " + String(this->rdPtr));
return abs(this->rdPtr - this->wrPtr); return abs(this->rdPtr - this->wrPtr);
@ -66,7 +61,7 @@ class CommQueue {
} }
protected: protected:
struct queue_s { struct QueueElement {
Inverter<> *iv; Inverter<> *iv;
uint8_t cmd; uint8_t cmd;
uint8_t attempts; uint8_t attempts;
@ -74,18 +69,25 @@ class CommQueue {
uint32_t ts; uint32_t ts;
bool isDevControl; bool isDevControl;
queue_s() {} QueueElement()
: iv {nullptr}
queue_s(Inverter<> *i, uint8_t c, bool dev) , cmd {0}
: iv {i} , attempts {0}
, cmd {c} , attemptsMax {0}
, attempts {DEFAULT_ATTEMPS}
, attemptsMax {DEFAULT_ATTEMPS}
, ts {0} , ts {0}
, isDevControl {dev} , isDevControl {false}
{} {}
queue_s(const queue_s &other) // copy constructor QueueElement(Inverter<> *iv, uint8_t cmd, bool devCtrl)
: iv {iv}
, cmd {cmd}
, attempts {DefaultAttempts}
, attemptsMax {DefaultAttempts}
, ts {0}
, isDevControl {devCtrl}
{}
QueueElement(const QueueElement &other) // copy constructor
: iv {other.iv} : iv {other.iv}
, cmd {other.cmd} , cmd {other.cmd}
, attempts {other.attempts} , attempts {other.attempts}
@ -93,35 +95,48 @@ class CommQueue {
, ts {other.ts} , ts {other.ts}
, isDevControl {other.isDevControl} , isDevControl {other.isDevControl}
{} {}
void changeCmd(uint8_t cmd) {
this->cmd = cmd;
this->isDevControl = false;
}
void setTs(const uint32_t ts) {
this->ts = ts;
}
void setAttempt() {
if(this->attempts)
this->attempts--;
}
void incrAttempt(uint8_t attempts = 1) {
this->attempts += attempts;
if (this->attempts > this->attemptsMax)
this->attemptsMax = this->attempts;
}
}; };
protected: protected:
void add(queue_s q) { void add(QueueElement q) {
xSemaphoreTake(this->mutex, portMAX_DELAY); xSemaphoreTake(this->mutex, portMAX_DELAY);
mQueue[this->wrPtr] = q; mQueue[this->wrPtr] = q;
inc(&this->wrPtr); inc(&this->wrPtr);
xSemaphoreGive(this->mutex); xSemaphoreGive(this->mutex);
} }
void add(queue_s *q, bool rstAttempts = false) { void add(QueueElement *q, bool rstAttempts = false) {
xSemaphoreTake(this->mutex, portMAX_DELAY); xSemaphoreTake(this->mutex, portMAX_DELAY);
mQueue[this->wrPtr] = *q; mQueue[this->wrPtr] = *q;
if(rstAttempts) { if(rstAttempts) {
mQueue[this->wrPtr].attempts = DEFAULT_ATTEMPS; mQueue[this->wrPtr].attempts = DefaultAttempts;
mQueue[this->wrPtr].attemptsMax = DEFAULT_ATTEMPS; mQueue[this->wrPtr].attemptsMax = DefaultAttempts;
} }
inc(&this->wrPtr); inc(&this->wrPtr);
xSemaphoreGive(this->mutex); xSemaphoreGive(this->mutex);
} }
void chgCmd(queue_s *q, uint8_t cmd) { void get(std::function<void(bool valid, QueueElement *q)> cb) {
xSemaphoreTake(this->mutex, portMAX_DELAY);
q->cmd = cmd;
q->isDevControl = false;
xSemaphoreGive(this->mutex);
}
void get(std::function<void(bool valid, queue_s *q)> cb) {
if(this->rdPtr == this->wrPtr) if(this->rdPtr == this->wrPtr)
cb(false, nullptr); // empty cb(false, nullptr); // empty
else { else {
@ -132,11 +147,11 @@ class CommQueue {
} }
} }
void cmdDone(queue_s *q, bool keep = false) { void cmdDone(QueueElement *q, bool keep = false) {
xSemaphoreTake(this->mutex, portMAX_DELAY); xSemaphoreTake(this->mutex, portMAX_DELAY);
if(keep) { if(keep) {
q->attempts = DEFAULT_ATTEMPS; q->attempts = DefaultAttempts;
q->attemptsMax = DEFAULT_ATTEMPS; q->attemptsMax = DefaultAttempts;
xSemaphoreGive(this->mutex); xSemaphoreGive(this->mutex);
add(q); // add to the end again add(q); // add to the end again
xSemaphoreTake(this->mutex, portMAX_DELAY); xSemaphoreTake(this->mutex, portMAX_DELAY);
@ -145,27 +160,6 @@ class CommQueue {
xSemaphoreGive(this->mutex); xSemaphoreGive(this->mutex);
} }
void setTs(queue_s *q, const uint32_t *ts) {
xSemaphoreTake(this->mutex, portMAX_DELAY);
q->ts = *ts;
xSemaphoreGive(this->mutex);
}
void setAttempt(queue_s *q) {
xSemaphoreTake(this->mutex, portMAX_DELAY);
if(q->attempts)
q->attempts--;
xSemaphoreGive(this->mutex);
}
void incrAttempt(queue_s *q, uint8_t attempts = 1) {
xSemaphoreTake(this->mutex, portMAX_DELAY);
q->attempts += attempts;
if (q->attempts > q->attemptsMax)
q->attemptsMax = q->attempts;
xSemaphoreGive(this->mutex);
}
private: private:
void inc(uint8_t *ptr) { void inc(uint8_t *ptr) {
if(++(*ptr) >= N) if(++(*ptr) >= N)
@ -178,7 +172,7 @@ class CommQueue {
--(*ptr); --(*ptr);
} }
bool isIncluded(const queue_s *q) { bool isIncluded(const QueueElement *q) {
uint8_t ptr = this->rdPtr; uint8_t ptr = this->rdPtr;
while (ptr != this->wrPtr) { while (ptr != this->wrPtr) {
if(mQueue[ptr].cmd == q->cmd) { if(mQueue[ptr].cmd == q->cmd) {
@ -193,7 +187,7 @@ class CommQueue {
} }
protected: protected:
std::array<queue_s, N> mQueue; std::array<QueueElement, N> mQueue;
private: private:
uint8_t wrPtr; uint8_t wrPtr;

48
src/hm/Communication.h
View file

@ -52,7 +52,7 @@ class Communication : public CommQueue<> {
} }
void loop() { void loop() {
get([this](bool valid, queue_s *q) { get([this](bool valid, QueueElement *q) {
if(!valid) { if(!valid) {
if(mPrintSequenceDuration) { if(mPrintSequenceDuration) {
mPrintSequenceDuration = false; mPrintSequenceDuration = false;
@ -72,7 +72,7 @@ class Communication : public CommQueue<> {
} }
private: private:
inline void innerLoop(queue_s *q) { inline void innerLoop(QueueElement *q) {
switch(mState) { switch(mState) {
case States::RESET: case States::RESET:
if (!mWaitTime.isTimeout()) if (!mWaitTime.isTimeout())
@ -102,13 +102,13 @@ class Communication : public CommQueue<> {
mFramesExpected = getFramesExpected(q); // function to get expected frame count. mFramesExpected = getFramesExpected(q); // function to get expected frame count.
mTimeout = DURATION_TXFRAME + mFramesExpected*DURATION_ONEFRAME + duration_reserve[q->iv->ivRadioType]; mTimeout = DURATION_TXFRAME + mFramesExpected*DURATION_ONEFRAME + duration_reserve[q->iv->ivRadioType];
if((q->iv->ivGen == IV_MI) && ((q->cmd == MI_REQ_CH1) || (q->cmd == MI_REQ_4CH))) if((q->iv->ivGen == IV_MI) && ((q->cmd == MI_REQ_CH1) || (q->cmd == MI_REQ_4CH)))
incrAttempt(q, q->iv->channels); // 2 more attempts for 2ch, 4 more for 4ch q->incrAttempt(q->iv->channels); // 2 more attempts for 2ch, 4 more for 4ch
mState = States::START; mState = States::START;
break; break;
case States::START: case States::START:
setTs(q, mTimestamp); q->setTs(*mTimestamp);
if(INV_RADIO_TYPE_CMT == q->iv->ivRadioType) { if(INV_RADIO_TYPE_CMT == q->iv->ivRadioType) {
// frequency was changed during runtime // frequency was changed during runtime
if(q->iv->curCmtFreq != q->iv->config->frequency) { if(q->iv->curCmtFreq != q->iv->config->frequency) {
@ -127,10 +127,10 @@ class Communication : public CommQueue<> {
//q->iv->radioStatistics.txCnt++; //q->iv->radioStatistics.txCnt++;
q->iv->radio->mRadioWaitTime.startTimeMonitor(mTimeout); q->iv->radio->mRadioWaitTime.startTimeMonitor(mTimeout);
if((!mIsRetransmit && (q->cmd == AlarmData)) || (q->cmd == GridOnProFilePara)) if((!mIsRetransmit && (q->cmd == AlarmData)) || (q->cmd == GridOnProFilePara))
incrAttempt(q, (q->cmd == AlarmData)? MORE_ATTEMPS_ALARMDATA : MORE_ATTEMPS_GRIDONPROFILEPARA); q->incrAttempt((q->cmd == AlarmData)? CommQueue::MoreAttemptsAlarmData : CommQueue::MoreAttemptsGridProfile);
mIsRetransmit = false; mIsRetransmit = false;
setAttempt(q); q->setAttempt();
mState = States::WAIT; mState = States::WAIT;
break; break;
@ -292,7 +292,7 @@ class Communication : public CommQueue<> {
} }
} }
setAttempt(q); q->setAttempt();
if(*mSerialDebug) { if(*mSerialDebug) {
DPRINT_IVID(DBG_WARN, q->iv->id); DPRINT_IVID(DBG_WARN, q->iv->id);
@ -321,7 +321,7 @@ class Communication : public CommQueue<> {
} }
} }
inline void printRxInfo(queue_s *q, packet_t *p) { inline void printRxInfo(QueueElement *q, packet_t *p) {
DPRINT_IVID(DBG_INFO, q->iv->id); DPRINT_IVID(DBG_INFO, q->iv->id);
DBGPRINT(F("RX ")); DBGPRINT(F("RX "));
if(p->millis < 100) if(p->millis < 100)
@ -355,7 +355,7 @@ class Communication : public CommQueue<> {
} }
inline uint8_t getFramesExpected(queue_s *q) { inline uint8_t getFramesExpected(QueueElement *q) {
if(q->isDevControl) if(q->isDevControl)
return 1; return 1;
@ -419,7 +419,7 @@ class Communication : public CommQueue<> {
return (ah::crc8(buf, len - 1) == buf[len-1]); return (ah::crc8(buf, len - 1) == buf[len-1]);
} }
inline bool parseFrame(queue_s *q, packet_t *p) { inline bool parseFrame(QueueElement *q, packet_t *p) {
uint8_t *frameId = &p->packet[9]; uint8_t *frameId = &p->packet[9];
if(0x00 == *frameId) { if(0x00 == *frameId) {
DPRINTLN(DBG_WARN, F("invalid frameId 0x00")); DPRINTLN(DBG_WARN, F("invalid frameId 0x00"));
@ -438,7 +438,7 @@ class Communication : public CommQueue<> {
if((*frameId & ALL_FRAMES) == ALL_FRAMES) { if((*frameId & ALL_FRAMES) == ALL_FRAMES) {
mMaxFrameId = (*frameId & 0x7f); mMaxFrameId = (*frameId & 0x7f);
if(mMaxFrameId > 8) // large payloads, e.g. AlarmData if(mMaxFrameId > 8) // large payloads, e.g. AlarmData
incrAttempt(q, mMaxFrameId - 6); q->incrAttempt(mMaxFrameId - 6);
} }
frame_t *f = &mLocalBuf[(*frameId & 0x7f) - 1]; frame_t *f = &mLocalBuf[(*frameId & 0x7f) - 1];
@ -449,7 +449,7 @@ class Communication : public CommQueue<> {
return true; return true;
} }
inline void parseMiFrame(packet_t *p, queue_s *q) { inline void parseMiFrame(packet_t *p, QueueElement *q) {
if((!mIsRetransmit && p->packet[9] == 0x00) && (p->millis < LIMIT_FAST_IV_MI)) //first frame is fast? if((!mIsRetransmit && p->packet[9] == 0x00) && (p->millis < LIMIT_FAST_IV_MI)) //first frame is fast?
mHeu.setIvRetriesGood(q->iv,p->millis < LIMIT_VERYFAST_IV_MI); mHeu.setIvRetriesGood(q->iv,p->millis < LIMIT_VERYFAST_IV_MI);
if ((p->packet[0] == MI_REQ_CH1 + ALL_FRAMES) if ((p->packet[0] == MI_REQ_CH1 + ALL_FRAMES)
@ -473,7 +473,7 @@ class Communication : public CommQueue<> {
} }
} }
inline bool parseDevCtrl(const packet_t *p, queue_s *q) { inline bool parseDevCtrl(const packet_t *p, QueueElement *q) {
switch(p->packet[12]) { switch(p->packet[12]) {
case ActivePowerContr: case ActivePowerContr:
if(p->packet[13] != 0x00) if(p->packet[13] != 0x00)
@ -511,7 +511,7 @@ class Communication : public CommQueue<> {
return accepted; return accepted;
} }
inline bool compilePayload(queue_s *q) { inline bool compilePayload(QueueElement *q) {
uint16_t crc = 0xffff, crcRcv = 0x0000; uint16_t crc = 0xffff, crcRcv = 0x0000;
for(uint8_t i = 0; i < mMaxFrameId; i++) { for(uint8_t i = 0; i < mMaxFrameId; i++) {
if(i == (mMaxFrameId - 1)) { if(i == (mMaxFrameId - 1)) {
@ -611,7 +611,7 @@ class Communication : public CommQueue<> {
return true; return true;
} }
void sendRetransmit(queue_s *q, uint8_t i) { void sendRetransmit(QueueElement *q, uint8_t i) {
mFramesExpected = 1; mFramesExpected = 1;
q->iv->radio->setExpectedFrames(mFramesExpected); q->iv->radio->setExpectedFrames(mFramesExpected);
q->iv->radio->sendCmdPacket(q->iv, TX_REQ_INFO, (SINGLE_FRAME + i), true); q->iv->radio->sendCmdPacket(q->iv, TX_REQ_INFO, (SINGLE_FRAME + i), true);
@ -622,7 +622,7 @@ class Communication : public CommQueue<> {
} }
private: private:
void closeRequest(queue_s *q, bool crcPass) { void closeRequest(QueueElement *q, bool crcPass) {
mHeu.evalTxChQuality(q->iv, crcPass, (q->attemptsMax - 1 - q->attempts), q->iv->curFrmCnt); mHeu.evalTxChQuality(q->iv, crcPass, (q->attemptsMax - 1 - q->attempts), q->iv->curFrmCnt);
if(crcPass) if(crcPass)
q->iv->radioStatistics.rxSuccess++; q->iv->radioStatistics.rxSuccess++;
@ -646,7 +646,7 @@ class Communication : public CommQueue<> {
DBGPRINTLN(F("-----")); DBGPRINTLN(F("-----"));
} }
inline void miHwDecode(packet_t *p, queue_s *q) { inline void miHwDecode(packet_t *p, QueueElement *q) {
record_t<> *rec = q->iv->getRecordStruct(InverterDevInform_All); // choose the record structure record_t<> *rec = q->iv->getRecordStruct(InverterDevInform_All); // choose the record structure
rec->ts = q->ts; rec->ts = q->ts;
/* /*
@ -760,7 +760,7 @@ class Communication : public CommQueue<> {
} }
} }
inline void miGPFDecode(packet_t *p, queue_s *q) { inline void miGPFDecode(packet_t *p, QueueElement *q) {
record_t<> *rec = q->iv->getRecordStruct(InverterDevInform_Simple); // choose the record structure record_t<> *rec = q->iv->getRecordStruct(InverterDevInform_Simple); // choose the record structure
rec->ts = q->ts; rec->ts = q->ts;
rec->mqttSentStatus = MqttSentStatus::NEW_DATA; rec->mqttSentStatus = MqttSentStatus::NEW_DATA;
@ -786,7 +786,7 @@ class Communication : public CommQueue<> {
q->iv->miMultiParts = 7; // indicate we are ready q->iv->miMultiParts = 7; // indicate we are ready
} }
inline void miDataDecode(packet_t *p, queue_s *q) { inline void miDataDecode(packet_t *p, QueueElement *q) {
record_t<> *rec = q->iv->getRecordStruct(RealTimeRunData_Debug); // choose the parser record_t<> *rec = q->iv->getRecordStruct(RealTimeRunData_Debug); // choose the parser
rec->ts = q->ts; rec->ts = q->ts;
//mState = States::RESET; //mState = States::RESET;
@ -839,7 +839,7 @@ class Communication : public CommQueue<> {
q->iv->miMultiParts += 6; // indicate we are ready q->iv->miMultiParts += 6; // indicate we are ready
} }
void miNextRequest(uint8_t cmd, queue_s *q) { void miNextRequest(uint8_t cmd, QueueElement *q) {
mHeu.evalTxChQuality(q->iv, true, (q->attemptsMax - 1 - q->attempts), q->iv->curFrmCnt); mHeu.evalTxChQuality(q->iv, true, (q->attemptsMax - 1 - q->attempts), q->iv->curFrmCnt);
mHeu.getTxCh(q->iv); mHeu.getTxCh(q->iv);
q->iv->radioStatistics.ivSent++; q->iv->radioStatistics.ivSent++;
@ -859,12 +859,12 @@ class Communication : public CommQueue<> {
DBGHEXLN(cmd); DBGHEXLN(cmd);
} }
mIsRetransmit = true; mIsRetransmit = true;
chgCmd(q, cmd); q->changeCmd(cmd);
//mState = States::WAIT; //mState = States::WAIT;
} }
void miRepeatRequest(queue_s *q) { void miRepeatRequest(QueueElement *q) {
setAttempt(q); // if function is called, we got something, and we necessarily need more transmissions for MI types... q->setAttempt(); // if function is called, we got something, and we necessarily need more transmissions for MI types...
q->iv->radio->sendCmdPacket(q->iv, q->cmd, 0x00, true); q->iv->radio->sendCmdPacket(q->iv, q->cmd, 0x00, true);
q->iv->radioStatistics.retransmits++; q->iv->radioStatistics.retransmits++;
q->iv->radio->mRadioWaitTime.startTimeMonitor(DURATION_TXFRAME + DURATION_ONEFRAME + duration_reserve[q->iv->ivRadioType]); q->iv->radio->mRadioWaitTime.startTimeMonitor(DURATION_TXFRAME + DURATION_ONEFRAME + duration_reserve[q->iv->ivRadioType]);
@ -878,7 +878,7 @@ class Communication : public CommQueue<> {
//mIsRetransmit = false; //mIsRetransmit = false;
} }
void miStsConsolidate(queue_s *q, uint8_t stschan, record_t<> *rec, uint8_t uState, uint8_t uEnum, uint8_t lState = 0, uint8_t lEnum = 0) { void miStsConsolidate(QueueElement *q, uint8_t stschan, record_t<> *rec, uint8_t uState, uint8_t uEnum, uint8_t lState = 0, uint8_t lEnum = 0) {
//uint8_t status = (p->packet[11] << 8) + p->packet[12]; //uint8_t status = (p->packet[11] << 8) + p->packet[12];
uint16_t statusMi = 3; // regular status for MI, change to 1 later? uint16_t statusMi = 3; // regular status for MI, change to 1 later?
if ( uState == 2 ) { if ( uState == 2 ) {