1. 首页
  2. 文章中心
  3. CANoe

Python 解析Vector DBC信号自动生成代码https://download.csdn.net/download/fz835304205/10819610DBC Format

83 阅读 0 评论
我是靠谱客的博主 现代铅笔，这篇文章主要介绍Python 解析Vector DBC信号自动生成代码https://download.csdn.net/download/fz835304205/10819610DBC Format，现在分享给大家，希望可以做个参考。

http://socialledge.com/sjsu/index.php/Main_Page

https://download.csdn.net/download/fz835304205/10819610

DBC Format

DBC file is a proprietary format that describes the data over a CAN bus.

In this article, you will learn the basic syntax of a DBC file that defines up to 8 bytes of CAN message data. A lot of CAN bus related tools can read the DBC file and display values next to each "signal" that you define in the DBC file.

The second part of this article discusses how the auto-generated code can help you read and write the CAN message data. Essentially, each "message" defined in a DBC becomes a C structure with the signals being the members of the C structure.

Contents

 [hide] 

  • 1DBC Data
    • 1.1Simple DBC Message
    • 1.2Signed Signal
    • 1.3Fractional Signals (float)
    • 1.4Enumeration Types
    • 1.5Multiplexed Message
    • 1.6DBC Example
  • 2Auto-Generated Code
  • 3CAN Communication Handling in C
    • 3.1MIA Handling
    • 3.2CAN RX
    • 3.3CAN TX
    • 3.4CAN TX with callback

DBC Data

 

Simple DBC Message

A simple DBC message contains the Message ID (MID), and at least one signal. Let's demonstrate by showing a message that contains a single 8-bit signal. Spaces and the syntax is really strict, so if you get a single space incorrect, the auto-generation script will likely fail.

复制代码
1
2
BO_ 500 IO_DEBUG: 4 IO
 SG_ IO_DEBUG_test_unsigned : 0|8@1+ (1,0) [0|0] "" DBG

Observations:

  • The message name is IO_DEBUG and MID is 500 (decimal), and the length is 4 bytes (though we only need 1 for 8-bit signal)
  • The sender is IO
  • 0|8: The unsigned signal starts at bit position 0, and the size of this signal is 8
  • (1,0): The scale and offset (discussed later)
  • [0|0]: Min and Max is not defined (discussed later)
  • "": There are no units (it could be, for instance "inches")
  • @1+: Defines that the signal is little-endian, and unsigned: Never change this!

 

Signed Signal

A signed signal can be sent by simply applying a negative offset to a signal. Let's add a signed signal to the previous message.

复制代码
1
2
3
BO_ 500 IO_DEBUG: 4 IO
 SG_ IO_DEBUG_test_unsigned : 0|8@1+ (1,0) [0|0] "" DBG
 SG_ IO_DEBUG_test_signed : 8|8@1- (1,-128) [0|0] "" DBG

 

Fractional Signals (float)

A floating point variable can be sent by deciding the range, and the precision that you require. For example, if we choose 8-bits, with 0.1 as a fraction, we can send the data range of 0.0 -> 25.5. On the other hand, if we want more precision and negative representation, we could use 12-bits with 0.01 as a fraction, and an offset. The second fractional signal also contains an explicit minimum and maximum, which is limited by 12-bit that can represent 4096 different numbers, and by factoring in the offset, and using half of the range for negative representation, it ends up with the limited range of -20.48 -> 20.47

复制代码
1
2
3
4
5
BO_ 500 IO_DEBUG: 4 IO
 SG_ IO_DEBUG_test_unsigned : 0|8@1+ (1,0) [0|0] "" DBG
 SG_ IO_DEBUG_test_signed : 8|8@1- (1,-128) [0|0] "" DBG
 SG_ IO_DEBUG_test_float1 : 16|8@1+ (0.1,0) [0|0] "" DBG
 SG_ IO_DEBUG_test_float2 : 24|12@1+ (0.01,-20.48) [-20.48|20.47] "" DBG

 

Enumeration Types

An enumeration type is used where the user wishes to use or see names, instead of numbers. For example, instead of a state machine showing up as "0, 1, 2", we could see it as "stopped, running, paused". It is accomplished by adding two new lines in the DBC file.

A "BA_" field needs to be added, and for the sake of simplicity, you can follow the example below to list an enumeration as a "FieldType" first. Then, you need a "VAL_" field that actually defines the enumeration values.

复制代码
1
2
3
4
5
6
BO_ 500 IO_DEBUG: 4 IO
 SG_ IO_DEBUG_test_enum : 8|8@1+ (1,0) [0|0] "" DBG

BA_ "FieldType" SG_ 500 IO_DEBUG_test_enum "IO_DEBUG_test_enum";

VAL_ 500 IO_DEBUG_test_enum 2 "IO_DEBUG_test2_enum_two" 1 "IO_DEBUG_test2_enum_one" ;

 

Multiplexed Message

A multiplexed message can be used (indirectly) to send more than 8 bytes using a single message ID. For example, if we use a 2-bit MUX, we can send 62-bits of data with four multiplexers (M0, M1, M2, M3). In other words, your message could state that:

  • If first 2-bits are 0 (M0), then 62-bits of data is for the car's front sensors
  • If first 2-bits are 1 (M1), then 62-bits of data is for the car's rear sensors

Likewise, we could use 8-bit multiplexer, and then we can send 7 bytes of unique data * 256 multiplexers using the same MID. Multiplexed messages are used quite often when:

  • Certain information should be grouped under a single MID.

    If there are 100 sensors that use 32-bit value, it is better to use a multipexer rather than 100 different message IDs.

  • 11-bit MID does not provide any space to send information under a different MID
  • We wish to use the same MID for CAN priority purposes

Observe the following things in the example below:

  • There are two multiplexed messages, m0, and m1.

    m0 has the value of 0b0000 for the MUX, and m1 has the value of 0b0001

    We could have a theoretical m15 with value 0b1111 since there is only a 4 bit MUX.

  • The 4-bit "M" needs to be defined first.
  • In this rather advanced example, we also have a non-mux'd signal called SENSOR_SONARS_err_count that is sent with all multiplexed messages
  • There are four sensor values sent with multiplexor m0
  • There are four "un filtered" sensor values sent with multipexor m1

In conclusion, you can define a multiplexed message that uses a single message ID, however, they are treated, and decoded differently depending on which multipexed value was sent. In order to send a multiplexed message below, you will have to send two separate messages, one for the m0 and one for the m1.

复制代码
1
2
3
4
5
6
7
8
9
10
11
BO_ 200 SENSOR_SONARS: 8 SENSOR
 SG_ SENSOR_SONARS_mux M : 0|4@1+ (1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_err_count : 4|12@1+ (1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_left m0 : 16|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_middle m0 : 28|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_right m0 : 40|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_rear m0 : 52|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_no_filt_left m1 : 16|12@1+ (0.1,0) [0|0] "" DBG
 SG_ SENSOR_SONARS_no_filt_middle m1 : 28|12@1+ (0.1,0) [0|0] "" DBG
 SG_ SENSOR_SONARS_no_filt_right m1 : 40|12@1+ (0.1,0) [0|0] "" DBG
 SG_ SENSOR_SONARS_no_filt_rear m1 : 52|12@1+ (0.1,0) [0|0] "" DBG


 

DBC Example

复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
VERSION ""

NS_ :
	BA_
	BA_DEF_
	BA_DEF_DEF_
	BA_DEF_DEF_REL_
	BA_DEF_REL_
	BA_DEF_SGTYPE_
	BA_REL_
	BA_SGTYPE_
	BO_TX_BU_
	BU_BO_REL_
	BU_EV_REL_
	BU_SG_REL_
	CAT_
	CAT_DEF_
	CM_
	ENVVAR_DATA_
	EV_DATA_
	FILTER
	NS_DESC_
	SGTYPE_
	SGTYPE_VAL_
	SG_MUL_VAL_
	SIGTYPE_VALTYPE_
	SIG_GROUP_
	SIG_TYPE_REF_
	SIG_VALTYPE_
	VAL_
	VAL_TABLE_

BS_:

BU_: DBG DRIVER IO MOTOR SENSOR


BO_ 100 DRIVER_HEARTBEAT: 1 DRIVER
 SG_ DRIVER_HEARTBEAT_cmd : 0|8@1+ (1,0) [0|0] "" SENSOR,MOTOR

BO_ 500 IO_DEBUG: 4 IO
 SG_ IO_DEBUG_test_unsigned : 0|8@1+ (1,0) [0|0] "" DBG
 SG_ IO_DEBUG_test_enum : 8|8@1+ (1,0) [0|0] "" DBG
 SG_ IO_DEBUG_test_signed : 16|8@1- (1,0) [0|0] "" DBG
 SG_ IO_DEBUG_test_float : 24|8@1+ (0.5,0) [0|0] "" DBG

BO_ 101 MOTOR_CMD: 1 DRIVER
 SG_ MOTOR_CMD_steer : 0|4@1- (1,-5) [-5|5] "" MOTOR
 SG_ MOTOR_CMD_drive : 4|4@1+ (1,0) [0|9] "" MOTOR

BO_ 400 MOTOR_STATUS: 3 MOTOR
 SG_ MOTOR_STATUS_wheel_error : 0|1@1+ (1,0) [0|0] "" DRIVER,IO
 SG_ MOTOR_STATUS_speed_kph : 8|16@1+ (0.001,0) [0|0] "kph" DRIVER,IO

BO_ 200 SENSOR_SONARS: 8 SENSOR
 SG_ SENSOR_SONARS_mux M : 0|4@1+ (1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_err_count : 4|12@1+ (1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_left m0 : 16|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_middle m0 : 28|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_right m0 : 40|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_rear m0 : 52|12@1+ (0.1,0) [0|0] "" DRIVER,IO
 SG_ SENSOR_SONARS_no_filt_left m1 : 16|12@1+ (0.1,0) [0|0] "" DBG
 SG_ SENSOR_SONARS_no_filt_middle m1 : 28|12@1+ (0.1,0) [0|0] "" DBG
 SG_ SENSOR_SONARS_no_filt_right m1 : 40|12@1+ (0.1,0) [0|0] "" DBG
 SG_ SENSOR_SONARS_no_filt_rear m1 : 52|12@1+ (0.1,0) [0|0] "" DBG




CM_ BU_ DRIVER "The driver controller driving the car";
CM_ BU_ MOTOR "The motor controller of the car";
CM_ BU_ SENSOR "The sensor controller of the car";
CM_ BO_ 100 "Sync message used to synchronize the controllers";

BA_DEF_ "BusType" STRING ;
BA_DEF_ BO_ "GenMsgCycleTime" INT 0 0;
BA_DEF_ SG_ "FieldType" STRING ;

BA_DEF_DEF_ "BusType" "CAN";
BA_DEF_DEF_ "FieldType" "";
BA_DEF_DEF_ "GenMsgCycleTime" 0;

BA_ "GenMsgCycleTime" BO_ 100 1000;
BA_ "GenMsgCycleTime" BO_ 500 100;
BA_ "GenMsgCycleTime" BO_ 101 100;
BA_ "GenMsgCycleTime" BO_ 400 100;
BA_ "GenMsgCycleTime" BO_ 200 100;
BA_ "FieldType" SG_ 100 DRIVER_HEARTBEAT_cmd "DRIVER_HEARTBEAT_cmd";
BA_ "FieldType" SG_ 500 IO_DEBUG_test_enum "IO_DEBUG_test_enum";


VAL_ 100 DRIVER_HEARTBEAT_cmd 2 "DRIVER_HEARTBEAT_cmd_REBOOT" 1 "DRIVER_HEARTBEAT_cmd_SYNC" 0 "DRIVER_HEARTBEAT_cmd_NOOP" ;
VAL_ 500 IO_DEBUG_test_enum 2 "IO_DEBUG_test2_enum_two" 1 "IO_DEBUG_test2_enum_one" ;

 

Auto-Generated Code

In a nutshell, the AGC (Auto-Generated-Code) allows a C/C++ application to deal with structures, and avoid the work of packing and unpacking data from a CAN message. Here is an example of a message defined in the DBC, and its auto-generated code artifact:

复制代码
1
2
3
BO_ 101 MOTOR_CMD: 1 DRIVER
 SG_ MOTOR_CMD_steer : 0|4@1- (1,-5) [-5|5] "" MOTOR
 SG_ MOTOR_CMD_drive : 4|4@1+ (1,0) [0|9] "" MOTOR
-->
复制代码
1
2
3
4
typedef struct {
    int8_t MOTOR_CMD_steer;
    uint8_t MOTOR_CMD_drive;
} MOTOR_CMD_t;

 

DBC File

 

CAN Communication Handling in C

I created auto-generation tool that operates on the DBC file that produces the AGC. AGC contains useful artifacts such as:

  • C structures that store data variables as described in the DBC message
    • Each signal that your controller is a recipient of will occupy a member variable of the struct
    • If your controller is the sender of the message, then all signals of the DBC message will appear in the structs
  • Receive handling to convert a CAN message to the C structure
  • Transmission handling to convert the C structure into the CAN message's data bytes

The AGC essentially removes the burden of encoding and decoding the CAN message data. For example, you don't have to parse fields individually by doing something such as "bit 5 of byte 6 means foo", and instead, you will simply have a C structure where the CAN message data can be parsed and stored upon.

MIA Handling

MIA means "Missing in Action", and the objective is that if a periodic message is not received as expected, then the contents of the parsed message can be defaulted to the data of your choice. For example, if a temperature sensor reading is not received, then we can tell the AGC to default the sensor reading value to 68 degrees Fahrenheit. This reduces code clutter because each time you use the temperature sensor's value, you don't have to check if the corresponding CAN data was received in the last few seconds.

CAN RX

To handling the messages that are to be received, we must have "glue code" that pieces together the data of a received messages to the target structure. Unfortunately the AGC cannot generate the glue code because it is decoupled from the CAN driver and furthermore, it leaves it up the application to provide the destination structure to convert the CAN data onto.

复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
#include "_can_dbc/generated_can.h"
#include "can.h"

// The MIA functions require that you define the:
//   - Time when the handle_mia() functions will replace the data with the MIA
//   - The MIA data itself (ie: MOTOR_STATUS__MIA_MSG)
const uint32_t            SENSOR_SONARS_m0__MIA_MS = 3000;
const SENSOR_SONARS_m0_t  SENSOR_SONARS_m0__MIA_MSG = { 0 };
const uint32_t            SENSOR_SONARS_m1__MIA_MS = 3000;
const SENSOR_SONARS_m1_t  SENSOR_SONARS_m1__MIA_MSG = { 0 };
const uint32_t            MOTOR_STATUS__MIA_MS = 3000;
const MOTOR_STATUS_t      MOTOR_STATUS__MIA_MSG = { 0 };

// For the sake of example, we use global data storage for messages that we receive
SENSOR_SONARS_t sensor_can_msg = { 0 };
MOTOR_STATUS_t motor_status_msg = { 0 };

// Sample periodic task that receives and stores messages
void period_100Hz(void)
{
    can_msg_t can_msg;

    // Empty all of the queued, and received messages within the last 10ms (100Hz callback frequency)
    while (CAN_rx(can1, &can_msg, 0))
    {
        // Form the message header from the metadata of the arriving message
        dbc_msg_hdr_t can_msg_hdr;
        can_msg_hdr.dlc = can_msg.frame_fields.data_len;
        can_msg_hdr.mid = can_msg.msg_id;

        // Attempt to decode the message (brute force, but should use switch/case with MID)
        dbc_decode_SENSOR_SONARS(&sensor_can_msg, can_msg.data.bytes, &can_msg_hdr);
        dbc_decode_MOTOR_STATUS(&motor_status_msg, can_msg.data.bytes, &can_msg_hdr);
    }

    // Service the MIA counters of the MUX'd message
    // successful decoding resets the MIA counter, otherwise it will increment to
    // its MIA value and upon the MIA trigger, it will get replaced by your MIA struct
    dbc_handle_mia_SENSOR_SONARS_m0(&sensor_can_msg.m0, 10);  // 10ms due to 100Hz
    dbc_handle_mia_SENSOR_SONARS_m1(&sensor_can_msg.m1, 10);

    // Service the MIA counter of a regular (non MUX'd) message
    dbc_handle_mia_MOTOR_STATUS(&motor_status_msg, 10);
}

 

CAN TX

The transmission side is simpler than receive because you simply provide an 8-byte data storage for the CAN message, and the encode_ functions convert the C structure onto the 8-bytes that you can send out as part of the CAN message. The encode_ functions also return the message ID and the length of the CAN message and this was done in an effort to decouple the CAN driver from the AGC.

复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
void period_10Hz(void)
{
    // Send out Motor command at 10Hz
    MOTOR_CMD_t motor_cmd = { 0 };
    motor_cmd.MOTOR_CMD_drive = 3;
    motor_cmd.MOTOR_CMD_steer = 0;

    can_msg_t can_msg = { 0 };

    // Encode the CAN message's data bytes, get its header and set the CAN message's DLC and length
    dbc_msg_hdr_t msg_hdr = dbc_encode_MOTOR_CMD(can_msg.data.bytes, &motor_cmd);
    can_msg.msg_id = msg_hdr.mid;
    can_msg.frame_fields.data_len = msg_hdr.dlc;

    // Queue the CAN message to be sent out
    CAN_tx(can1, &can_msg, 0);
}

 

CAN TX with callback

There is an alternate method of sending CAN messages that may make your life easier by defining a callback function to send a CAN message. This way, your application doesn't have to deal with a CAN message at all!

复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// This method needs to be defined once, and AGC will call it for all dbc_encode_and_send_FOO() functions
bool dbc_app_send_can_msg(uint32_t mid, uint8_t dlc, uint8_t bytes[8])
{
    can_msg_t can_msg = { 0 };
    can_msg.msg_id                = mid;
    can_msg.frame_fields.data_len = dlc;
    memcpy(can_msg.data.bytes, bytes, dlc);

    return CAN_tx(can1, &can_msg, 0);
}

void period_10Hz(void)
{
    // Send out Motor command at 10Hz
    MOTOR_CMD_t motor_cmd = { 0 };
    motor_cmd.MOTOR_CMD_drive = 3;
    motor_cmd.MOTOR_CMD_steer = 0;

    // This function will encode the CAN data bytes, and send the CAN msg using dbc_app_send_can_msg()
    dbc_encode_and_send_MOTOR_CMD(&motor_cmd);
}
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
#!/usr/bin/python

import sys, getopt
import re
from collections import OrderedDict

"""
@Author: Preet
This parses the Vector DBC file to generate code to marshal and unmarshal DBC defined messages

Use Python (I used Python 3.5)
python dbc_parse.py -i 243.dbc -s GPS
Generate all code: dbc_parse.py -i 243.dbc -s Driver -a all > generated.h
"""

LINE_BEG = '%'


def is_empty(s):
    if s:
        return False
    else:
        return True


def MIN(x, y):
    if (x < y):
        return x
    else:
        return y


class Signal(object):
    def __init__(self, name, bit_start, bit_size, endian_and_sign, scale, offset, min_val, max_val, recipients, mux, signal_min, signal_max):
        self.has_field_type = False
        self.name = name
        self.bit_start = int(bit_start)
        self.bit_size = int(bit_size)
        self.endian_and_sign = endian_and_sign

        self.offset = float(offset)
        self.offset_str = offset
        self.scale = float(scale)
        self.scale_str = scale
        self.min_val = float(min_val)
        self.min_val_str = min_val
        self.max_val = float(max_val)
        self.max_val_str = max_val
        self.signal_min = signal_min
        self.signal_max = signal_max

        self.recipients = recipients
        self.enum_info = {}
        self.mux = mux
        if self.mux == '':
            self.mux = '__NO__MUX__'

    # Returns true if the signal uses an enumeration type
    def is_enum_type(self):
        return not is_empty(self.enum_info)

    # Returns true if the signal is part of MUX'd data
    def is_muxed(self):
        return '__NO__MUX__' != self.mux

    # Returns true if the signal should be an unsigned type
    def is_unsigned_var(self):
        t = self.get_code_var_type()
        return t.find("uint") == 0

    # Returns true if the signal is defined in the DBC as a signed type
    def is_real_signed(self):
        return '-' == self.endian_and_sign[1]

    # Returns the variable type (float, int, or enum) based ont he signal data range
    def get_code_var_type(self):
       if self.scale_str.count(".00000")>=1:
           return "double"
       elif '.' in self.scale_str:
           return "float"
       else:
           if not is_empty(self.enum_info):
               return self.name + "_E"

           _max = (2 ** self.bit_size) * self.scale
           if self.is_real_signed():
               _max *= 2

           t = "uint32_t"
           if _max <= 256:
               t = "uint8_t"
           elif _max <= 65536:
               t = "uint16_t"

           # If the signal is signed, or the offset is negative, remove "u" to use "int" type.
           if self.is_real_signed() or self.offset < 0:
               t = t[1:]

           return t

    # Get the signal declaration with the variable type and bit size
    def get_signal_code(self):
        code = ""
        code += "    " + self.get_code_var_type() + " " + self.name
        if self.bit_size <= 4:
            code += " : " + str(self.bit_size) + ";"
        else:
            code += ";"

        # Align the start of the comments
        for i in range(len(code), 45):
            code += " "

        # Comment with Min/Max
        code += " ///< B" + str(self.bit_start + self.bit_size - 1) + ":" + str(self.bit_start)
        if self.min_val != 0 or self.max_val != 0:
            code += "  Min: " + self.min_val_str + " Max: " + self.max_val_str

        # Comment with destination nodes:
        code += "   Destination: "
        for r in self.recipients:
            if r == self.recipients[0]:
                code += r
            else:
                code += "," + r

        return code + "n"

    # Get the encode code of the signal
    def get_encode_code(self, raw_sig_name, var_name):
        code = ''

        # Min/Max check
        if self.min_val != 0 or self.max_val != 0:
            # If signal is unsigned, and min value is zero, then do not check for '< 0'
            if not (self.is_unsigned_var() and self.min_val == 0):
                code += ("    if(" + var_name + " < " + self.min_val_str + ") { " + var_name + " = " + self.min_val_str + "; } // Min value: " + self.min_val_str + "n")
            else:
                code += "    // Not doing min value check since the signal is unsigned alreadyn"
            code += ("    if(" + var_name + " > " + self.max_val_str + ") { " + var_name + " = " + self.max_val_str + "; } // Max value: " + self.max_val_str + "n")

        # Compute binary value
        # Encode should subtract offset then divide
        # TODO: Might have to add -0.5 for a negative signal
        raw_sig_code = "    " + raw_sig_name + " = "
        raw_sig_code += "((uint32_t)(((" + var_name + " - (" + self.offset_str + ")) / " + str(self.scale) + ") + 0.5))"
        if self.is_real_signed():
            s = "    // Stuff a real signed number into the DBC " + str(self.bit_size) + "-bit signaln"
            s += raw_sig_code + (" & 0x" + format(2 ** self.bit_size - 1, '02x') + ";n")
        else:
            s = raw_sig_code + (" & 0x" + format(2 ** self.bit_size - 1, '02x') + ";n")

        # Optimize
        s = s.replace(" - (0)", "")
        s = s.replace(" / 1.0)", ")")
        if self.scale == 1:
            s = s.replace(" + 0.5", "")

        # Add the code
        code += s

        # Stuff the raw data into individual bytes
        bit_pos = self.bit_start
        remaining = self.bit_size
        byte_num = int(self.bit_start / 8)
        while remaining > 0:
            bits_in_this_byte = MIN(8 - (bit_pos % 8), remaining)

            s = ""
            s += ("    bytes[" + str(byte_num) + "] |= (((uint8_t)(" + raw_sig_name + " >> " + str(
                bit_pos - self.bit_start) + ")")
            s += (" & 0x" + format(2 ** bits_in_this_byte - 1, '02x') + ") << " + str(bit_pos % 8) + ")")
            s += ("; ///< " + str(bits_in_this_byte) + " bit(s) starting from B" + str(bit_pos) + "n")

            # Optimize
            s = s.replace(" >> 0", "")
            s = s.replace(" << 0", "")
            # Cannot optimize by removing 0xff just for code safety
            #s = s.replace(" & 0xff", "")

            code += s
            byte_num += 1

            bit_pos += bits_in_this_byte
            remaining -= bits_in_this_byte
        return code

    # Get the decode code of the signal
    def get_decode_code(self, raw_sig_name, prefix=''):
        # Little and Big Endian:
        bit_pos = self.bit_start
        remaining = self.bit_size
        byte_num = int(self.bit_start / 8)
        bit_count = 0
        code = ''

        while remaining > 0:
            bits_in_this_byte = MIN(8 - (bit_pos % 8), remaining)

            s = ""
            s += (
            LINE_BEG + raw_sig_name + " |= ((uint32_t)((bytes[" + str(byte_num) + "] >> " + str(bit_pos % 8) + ")")
            s += (" & 0x" + format(2 ** bits_in_this_byte - 1, '02x') + ")) << " + str(bit_count) + ";")
            s += (" ///< " + str(bits_in_this_byte) + " bit(s) from B" + str(bit_pos) + "n")

            # Optimize
            s = s.replace(" >> 0", "")
            s = s.replace(" << 0", "")
            s = s.replace(" & 0xff", "")

            code += s
            if bit_count == 0:
                code = code.replace("|=", " =")

            byte_num += 1
            bit_pos += bits_in_this_byte
            remaining -= bits_in_this_byte
            bit_count += bits_in_this_byte

        # Decode/get should multiply then add the offset
        enum_cast = ''
        if self.is_enum_type():
            enum_cast = "(" + self.get_code_var_type() + ")"

        # If the signal is not defined as a signed, then we will use this code
        unsigned_code = (prefix + self.name + " = " + enum_cast + "((" + raw_sig_name + " * " + str(self.scale) + ") + (" + self.offset_str + "));n")

        if self.is_real_signed():
            mask = "(1 << " + str(self.bit_size - 1) + ")"
            s = LINE_BEG + "if (" + raw_sig_name + " & " + mask + ") { // Check signed bitn"
            s += LINE_BEG + prefix + self.name + " = " + enum_cast
            s += "((((0xFFFFFFFF << " + str(self.bit_size - 1) + ") | " + raw_sig_name + ") * " + str(self.scale) + ") + (" + self.offset_str + "));n"
            s += LINE_BEG + "} " + "else {n"
            s += LINE_BEG + unsigned_code
            s += LINE_BEG + "}n"
        else:
            s = unsigned_code

        # Optimize
        s = s.replace(" + (0)", "")
        s = s.replace(" * 1.0)", ")")
        code += s

        return code


class Message(object):
    """
    Message Object that contains the list of signals inside
    """

    def __init__(self, mid, name, dlc, sender):
        self.mid = mid
        self.name = name
        self.dlc = dlc
        self.sender = sender
        self.signals = OrderedDict()

    # Adds the signal to the dictionary of signals of this message
    def add_signal(self, s):
        self.signals[s.name] = s

    # Returns the struct name derived from the message name
    def get_struct_name(self):
        return "%s_t" % (self.name)
        # return "%s_TX_%s_t" % (self.sender, self.name)

    # Returns true if the node is a recipient of at least one signal contained in the message
    def is_recipient_of_at_least_one_sig(self, node):
        for key in self.signals:
            if node in self.signals[key].recipients:
                return True
        return False

    # Returns true if at least one message signal is a MUX'd type
    def contains_muxed_signals(self):
        for key in self.signals:
            if self.signals[key].is_muxed():
                return True
        return False

    # Returne true if one or more of the message signal is an enumeration type
    def contains_enums(self):
        for key in self.signals:
            if not is_empty(self.signals[key].enum_info):
                return True
        return False

    def get_muxes(self):
        muxes = []
        for key in self.signals:
            if self.signals[key].is_muxed() and self.signals[key].mux not in muxes:
                muxes.append(self.signals[key].mux)
        return muxes

    # Returns the message signal that defines the MUX value
    def get_mux_index_signal(self):
        for key in self.signals:
            if self.signals[key].is_muxed() and self.signals[key].mux == "M":
                return self.signals[key]
        return ""

    # TODO: Do not generate this struct if we are not the recipient of any of the signals of this MUX
    def get_struct_for_mux(self, mux, non_muxed_signals, gen_mia_struct):
        code = 'n'
        code += ("/// Struct for MUX: " + mux + " (used for transmitting)n")
        code += ("typedef struct {n")
        code += non_muxed_signals

        for key in self.signals:
            if self.signals[key].mux == mux:
                code += (self.signals[key].get_signal_code())
        if gen_mia_struct:
            code += ("n    dbc_mia_info_t mia_info;")
        else:
            code += ("n    // No dbc_mia_info_t for a message that we will send")
        code += ("n} " + self.get_struct_name()[:-2] + "_" + str(mux) + "_t;n")
        return code

    def gen_converted_struct(self, self_node, gen_all):
        code = ''
        if self.contains_muxed_signals():
            # Non Muxed signals in this struct, exclude the MUXED index
            non_muxed_signals = ''
            for key in self.signals:
                if not self.signals[key].is_muxed() and not self.signals[key].mux == "M":
                    non_muxed_signals += (self.signals[key].get_signal_code())

            # MUX'd data structures
            code = ("/// @{ MUX'd message: " + self.name + "n")
            muxes = self.get_muxes()
            gen_mia_struct = gen_all or self_node != self.sender
            for m in muxes[1:]:
                code += self.get_struct_for_mux(m, non_muxed_signals, gen_mia_struct)

            # Parent data structure
            code += "n/// Struct with all the child MUX'd signals (Used for receiving)n"
            code += "typedef struct {n"

            # Child struct instances of the Mux'd signals
            for m in muxes[1:]:
                code += ("    " + self.get_struct_name()[:-2] + "_" + str(m) + "_t " + str(m) + "; ///< MUX'd structuren")
            code += ("} " + self.get_struct_name() + ";n")

            code += ("/// @} MUX'd messagen")
        else:
            code += (
            "n/// Message: " + self.name + " from '" + self.sender + "', DLC: " + self.dlc + " byte(s), MID: " + self.mid + "n")
            code += ("typedef struct {n")
            for key in self.signals:
                if gen_all or self_node in self.signals[key].recipients or self.sender == self_node:
                    code += (self.signals[key].get_signal_code())

            if gen_all or self_node != self.sender:
                code += ("n    dbc_mia_info_t mia_info;")
            else:
                code += ("n    // No dbc_mia_info_t for a message that we will send")
            code += ("n} " + self.get_struct_name() + ";n")

        return code

    def get_encode_and_send(self, name):
        code = ''
        code += ("n/// Encode and send for dbc_encode_" + name + "() messagen")
        code += ("static inline bool dbc_encode_and_send_" + name + "(" + name + "_t *from)n")
        code += "{n"
        code += ("    uint8_t bytes[8];n")
        code += ("    const dbc_msg_hdr_t hdr = dbc_encode_" + name + "(bytes, from);n")
        code += ("    return dbc_app_send_can_msg(hdr.mid, hdr.dlc, bytes);n")
        code += "}n"
        code += "n"
        return code

    def get_encode_code(self):
        code = ''
        if self.contains_muxed_signals():
            muxes = self.get_muxes()
            for mux in muxes:
                if "M" == mux:
                    continue

                name = self.get_struct_name()
                name_with_mux = name[:-2] + "_" + str(mux)
                code += ("n/// Encode " + self.sender + "'s '" + self.name + "' MUX(" + str(mux) + ") messagen")
                code += ("/// @returns the message header of this messagen")
                code += ("static inline dbc_msg_hdr_t dbc_encode_" + name_with_mux)
                code += ("(uint8_t bytes[8], " + name_with_mux + "_t *from)n")
                code += ("{n")
                code += ("    uint32_t raw;n")
                code += ("    bytes[0]=bytes[1]=bytes[2]=bytes[3]=bytes[4]=bytes[5]=bytes[6]=bytes[7]=0;nn")
                code += ("    // Set the MUX index valuen")
                muxed_idx = self.get_mux_index_signal()
                code += muxed_idx.get_encode_code("raw", str(mux)[1:])
                code += ("n")

                # Non Muxed signals in this struct, exclude the MUXED index
                code += "    // Set non MUX'd signals that need to go out with this MUX'd messagen"
                for key in self.signals:
                    if not self.signals[key].is_muxed() and not self.signals[key].mux == "M":
                        code += self.signals[key].get_encode_code("raw", "from->" + key)

                # Rest of the signals that are part of this MUX
                code += ("n")
                code += ("    // Set the rest of the signals within this MUX (" + mux + ")n")
                for key in self.signals:
                    if mux == self.signals[key].mux:
                        code += self.signals[key].get_encode_code("raw", "from->" + key)

                code += ("n")
                code += ("    return " + name[:-2] + "_HDR;n")
                code += ("}n")

                # Encode and send function
                code += self.get_encode_and_send(name_with_mux)

        else:
            name = self.get_struct_name()
            code += ("n/// Encode " + self.sender + "'s '" + self.name + "' messagen")
            code += ("/// @returns the message header of this messagen")
            code += ("static inline dbc_msg_hdr_t dbc_encode_" + name[:-2] + "(uint8_t bytes[8], " + name + " *from)n")
            code += ("{n")
            code += ("    uint32_t raw;n")
            code += ("    bytes[0]=bytes[1]=bytes[2]=bytes[3]=bytes[4]=bytes[5]=bytes[6]=bytes[7]=0;n")
            code += ("n")

            for key in self.signals:
                code += self.signals[key].get_encode_code("raw", "from->" + key) + "n"

            code += ("    return " + self.get_struct_name()[:-2] + "_HDR;n")
            code += ("}n")

            # Encode and send function
            code += self.get_encode_and_send(name[:-2])

        return code

    def get_non_mux_signal_decode_code(self, raw_sig_name, prefix=''):
        code = ''
        for key in self.signals:
            if not self.signals[key].is_muxed():
                code += self.signals[key].get_decode_code(raw_sig_name, prefix)
        return code

    def get_signal_decode_code_for_mux(self, mux, raw_sig_name, prefix=''):
        code = ''
        for key in self.signals:
            if self.signals[key].mux == mux:
                code += self.signals[key].get_decode_code(raw_sig_name, prefix)
        return code

    def get_decode_code(self):
        raw_sig_name = "raw"
        code = ''
        code += ("n/// Decode " + self.sender + "'s '" + self.name + "' messagen")
        code += (
        "/// @param hdr  The header of the message to validate its DLC and MID; this can be NULL to skip this checkn")
        code += ("static inline bool dbc_decode_" + self.get_struct_name()[
                                                    :-2] + "(" + self.get_struct_name() + " *to, const uint8_t bytes[8], const dbc_msg_hdr_t *hdr)n")
        code += ("{n")
        code += ("    const bool success = true;n")
        code += ("    // If msg header is provided, check if the DLC and the MID matchn")
        code += ("    if (NULL != hdr && (hdr->dlc != " + self.get_struct_name()[:-2] + "_HDR.dlc || hdr->mid != " + self.get_struct_name()[:-2] + "_HDR.mid)) {n")
        code += ("        return !success;n")
        code += ("    }nn")
        code += ("    uint32_t " + raw_sig_name + ";n")

        if self.contains_muxed_signals():
            # Decode the Mux and store it into it own variable type
            muxed_sig = self.get_mux_index_signal()
            code += ("    // Decode the MUXn")
            code += (muxed_sig.get_decode_code(raw_sig_name).replace(LINE_BEG, "    ")).replace(muxed_sig.name,
                                                                                                "    const " + muxed_sig.get_code_var_type() + " MUX")
            code += ("n")

            # Decode the Mux'd signal(s)
            muxes = self.get_muxes()
            for mux in muxes[1:]:
                prefix = "%to->" + mux + "."

                # Each MUX'd message may also have non muxed signals:
                non_mux_code = self.get_non_mux_signal_decode_code(raw_sig_name, prefix)
                mux_code = self.get_signal_decode_code_for_mux(mux, raw_sig_name, prefix)

                if mux == muxes[1]:
                    code += ("    if (" + str(mux)[1:] + " == MUX) {n")
                else:
                    code += ("    else if (" + str(mux)[1:] + " == MUX) {n")

                if non_mux_code != '':
                    code += "        // Non Muxed signals (part of all MUX'd structures)n"
                    code += non_mux_code.replace(LINE_BEG, "        ")
                    code += "n"
                code += mux_code.replace(LINE_BEG, "        ")

                code += ("n        to->" + str(mux) + ".mia_info.mia_counter_ms = 0; ///< Reset the MIA countern")
                code += ("    }n")
            code += "    else {n        return !success;n    }n"
        else:
            code += self.get_non_mux_signal_decode_code(raw_sig_name, "    to->").replace(LINE_BEG, "    ")
            code += ("n")
            code += ("    to->mia_info.mia_counter_ms = 0; ///< Reset the MIA countern")

        code += ("n    return success;n")
        code += ("}n")
        return code


class DBC(object):
    def __init__(self, name, self_node, gen_all):
        self.name = name
        self.self_node = self_node
        self.gen_all = gen_all

        # Dictionary of messages with the MSG-ID as the key
        self.messages = OrderedDict()
        self.nodes = []

    def gen_file_header(self):
        code = ''
        code += ("/// DBC file: %s    Self node: '%s'  (ALL = %u)n" % (self.name, self.self_node, self.gen_all))
        code += ("/// This file can be included by a source file, for example: #include "generated.h"n")
        code += ("#ifndef __GENEARTED_DBC_PARSERn")
        code += ("#define __GENERATED_DBC_PARSERn")
        code += ("#include <stdbool.h>n")
        code += ("#include <stdint.h>n")
        code += ("#include <stdlib.h>n")
        return code

    def gen_msg_hdr_struct(self):
        code = ("/// CAN message header structuren")
        code += ("typedef struct { n")
        code += ("    uint32_t mid; ///< Message ID of the messagen")
        code += ("    uint8_t  dlc; ///< Data length of the messagen")
        code += ("} dbc_msg_hdr_t; n")
        return code

    def gen_enum_types(self):
        code = ''
        for mkey in self.messages:
            m = self.messages[mkey]
            if not m.contains_enums():
                continue
            if self.gen_all or m.is_recipient_of_at_least_one_sig(self.self_node) or self.self_node == m.sender:
                code += ("/// Enumeration(s) for Message: '" + m.name + "' from '" + m.sender + "'n")
                for key in m.signals:
                    if m.signals[key].is_enum_type():
                        code += "typedef enum {n"
                        for enum_key in m.signals[key].enum_info:
                            code += "    " + enum_key + " = " + m.signals[key].enum_info[enum_key] + ",n"
                        code += "} " + m.signals[key].name + "_E ;nn"
        code += "n"
        return code

    def gen_msg_hdr_instances(self):
        code = ''
        for mkey in self.messages:
            m = self.messages[mkey]
            if not self.gen_all and not m.is_recipient_of_at_least_one_sig(
                    self.self_node) and self.self_node != m.sender:
                code += "// "
            code += ("static const dbc_msg_hdr_t " + (m.get_struct_name()[:-2] + "_HDR = ").ljust(32 + 7))
            code += ("{ " + str(m.mid).rjust(4) + ", " + m.dlc + " };n")
        return code

    def gen_mia_struct(self):
        code = ("/// Missing in Action structuren")
        code += ("typedef struct {n")
        code += ("    uint32_t is_mia : 1;          ///< Missing in action flagn")
        code += ("    uint32_t mia_counter_ms : 31; ///< Missing in action countern")
        code += ("} dbc_mia_info_t;n")
        return code

    def _gen_mia_func_header(self, sender, msg_name):
        code = ''
        code += ("n/// Handle the MIA for " + sender + "'s " + msg_name + " messagen")
        code += ("/// @param   time_incr_ms  The time to increment the MIA counter withn")
        code += ("/// @returns true if the MIA just occurredn")
        code += ("/// @post    If the MIA counter reaches the MIA threshold, MIA struct will be copied to *msgn")
        return code

    def _get_mia_func_body(self, msg_name):
        code = ''
        code += ("{n")
        code += ("    bool mia_occurred = false;n")
        code += ("    const dbc_mia_info_t old_mia = msg->mia_info;n")
        code += ("    msg->mia_info.is_mia = (msg->mia_info.mia_counter_ms >= " + msg_name + "__MIA_MS);n")
        code += ("n")
        code += ("    if (!msg->mia_info.is_mia) { // Not MIA yet, so keep incrementing the MIA countern")
        code += ("        msg->mia_info.mia_counter_ms += time_incr_ms;n")
        code += ("    }n")
        code += ("    else if(!old_mia.is_mia)   { // Previously not MIA, but it is MIA nown")
        code += ("        // Copy MIA struct, then re-write the MIA counter and is_mia that is overwritenn")
        code += ("        *msg = " + msg_name + "__MIA_MSG;n")
        code += ("        msg->mia_info.mia_counter_ms = 0;n")
        code += ("        msg->mia_info.is_mia = true;n")
        code += ("        mia_occurred = true;n")
        code += ("    }n")
        code += ("n    return mia_occurred;n")
        code += ("}n")
        return code

    def gen_mia_funcs(self):
        code = ''

        # Generate MIA handler for the dbc.messages we are a recipient of
        for mkey in self.messages:
            m = self.messages[mkey]
            if not self.gen_all and not m.is_recipient_of_at_least_one_sig(self.self_node):
                continue
            if m.contains_muxed_signals():
                muxes = m.get_muxes()
                for mux in muxes[1:]:
                    code += self._gen_mia_func_header(m.sender, m.name + " for MUX "" + mux + '"')
                    code += ("static inline bool dbc_handle_mia_" + m.get_struct_name()[:-2] + "_" + mux + "(")
                    code += (m.get_struct_name()[:-2] + "_" + mux + "_t *msg, uint32_t time_incr_ms)n")
                    code += self._get_mia_func_body(m.name + "_" + mux)
            else:
                code += self._gen_mia_func_header(m.sender, m.name)
                code += ("static inline bool dbc_handle_mia_" + m.get_struct_name()[
                                                               :-2] + "(" + m.get_struct_name() + " *msg, uint32_t time_incr_ms)n")
                code += self._get_mia_func_body(m.name)

        return code


def main(argv):
    dbcfile = '243.dbc'  # Default value unless overriden
    self_node = 'DRIVER'  # Default value unless overriden
    gen_all = False
    muxed_signal = False
    mux_bit_width = 0
    msg_ids_used = []
    try:
        opts, args = getopt.getopt(argv, "i:s:a", ["ifile=", "self=", "all"])
    except getopt.GetoptError:
        print('dbc_parse.py -i <dbcfile> -s <self_node> <-a>')
        sys.exit(2)
    for opt, arg in opts:
        if opt == '-h':
            print('dbc_parse.py -i <dbcfile> -s <self_node> <-a> <-b>')
            sys.exit()
        elif opt in ("-i", "--ifile"):
            dbcfile = arg
        elif opt in ("-s", "--self"):
            self_node = arg
        elif opt in ("-a", "--all"):
            gen_all = True

    # Parse the DBC file
    dbc = DBC(dbcfile, self_node, gen_all)
    f = open(dbcfile, "r")
    last_mid = -1
    validFile = True
    while 1:
        line = f.readline()
        if not line:
            break

        # Nodes in the DBC file
        if line.startswith("BU_:"):
            nodes = line.strip("n").split(' ')
            dbc.nodes = (nodes[1:])
            if self_node not in dbc.nodes:
                print('/// ERROR /')
                print('#error "Self node: ' + self_node + ' not found in _BU nodes in the DBC file"')
                print('/// ERROR /')
                print('')
                raise ValueError('#error "Self node: ' + self_node + ' not found in _BU nodes in the DBC file"')

        # Start of a message
        # BO_ 100 DRIVER_HEARTBEAT: 1 DRIVER
        if line.startswith("BO_ "):
            muxed_signal = False
            mux_bit_width = 0
            tokens = line.split(' ')
            msg_id = tokens[1]
            msg_name = tokens[2].strip(":")
            dbc.messages[msg_id] = Message(msg_id, msg_name, tokens[3], tokens[4].strip("n"))
            msg_length = tokens[3]
            last_mid = msg_id
            fixed_mux_signal = False
            fixed_signal_end = 0
            prev_signal_end = 0
            prev_mux_index = 0

            if (int(msg_id) < 0) or (int(msg_id) > 2047):
                print('/// ERROR /')
                print('#error msg id '+ msg_id + ' is out of bounds')
                print('/// ERROR /')
                print('')
                raise ValueError('#error msg id '+ msg_id + ' is out of bounds')

            if msg_id not in msg_ids_used:
                msg_id = msg_ids_used.append(msg_id)
            else:
                print('/// ERROR /')
                print('#error '+ msg_id + ' has already been used')
                print('/// ERROR /')
                print('')
                raise ValueError('#error msg id '+ msg_id + ' has already been used')

            if int(msg_length) > 8 or int(msg_length) < 1:
                print('/// ERROR /')
                print('#error '+ msg_id + ' has an incorrect number of bytes. It must be between 1 and 8 bytes.')
                print('/// ERROR /')
                print('')
                raise ValueError('#error msg id '+ msg_id + ' has an incorrect number of bytes. It must be between 1 and 8 bytes.')

        # Signals: SG_ IO_DEBUG_test_signed : 16|8@1+ (1,-128) [0|0] "" DBG
        if line.startswith(" SG_ "):
            t = line[1:].split(' ')

            # If this is a MUX'd symbol
            mux = ''
            if t[3] == ":":
                mux = t[2]
                line = line.replace(mux + " ", '')
                t = line[1:].split(' ')

            # Split the bit start and the bit size
            s = re.split('[|@]', t[3])
            bit_start = s[0]
            bit_size = s[1]

            if mux == 'M':
                muxed_signal = True
                mux_bit_width = int(bit_size)

            if not muxed_signal:
                if (int(bit_start) < prev_signal_end):
                    print('/// ERROR /')
                    print('#error ' + t[1] + ' start bit overwrites previous signal') 
                    print('/// ERROR /')
                    print('')
                    raise ValueError('#error ' + t[1] + ' start bit overwrites previous signal')
                prev_signal_end = int(bit_start) + int(bit_size)
            # Ensure a mux index 
            if muxed_signal:
                if mux == '':
                    fixed_mux_signal = True
                    fixed_signal_end = mux_bit_width + int(bit_size)
                elif mux[0] == 'm':
                    fixed_mux_signal = False
                    if int(mux[1:]) != prev_mux_index:
                        prev_signal_end = fixed_signal_end

                if fixed_mux_signal:
                    if int(bit_start) < mux_bit_width:
                        print('/// ERROR /')
                        print('#error ' + t[1] + ' start bit overwrites mux index') 
                        print('/// ERROR /')
                        print('')
                        raise ValueError('#error ' + t[1] + ' start bit overwrites mux index')
                else:
                    if mux != 'M':
                        # Do not allow the signal to use the indexing bits
                        if int(bit_start) < fixed_signal_end:
                            print('/// ERROR /')
                            print('#error ' + t[1] + ' start bit overwrites mux index') 
                            print('/// ERROR /')
                            print('')
                            raise ValueError('#error ' + t[1] + ' start bit overwrites previous fixed signal')
                        if mux[0] == 'm':
                        # Check for mux index out of bounds
                            if (int(mux[1:]) >= pow(2,mux_bit_width)) or (int(mux[1:]) < 0):
                                print('/// ERROR /')
                                print('#error ' + t[1] + ' mux index out of bounds.') 
                                print('/// ERROR /')
                                print('')
                                raise ValueError('#error ' + t[1] + ' mux index out of bounds.')

                            if int(bit_start) < prev_signal_end:
                                print('/// ERROR /')
                                print('#error ' + t[1] + ' start bit overwrites previous signal') 
                                print('/// ERROR /')
                                print('')
                                raise ValueError('#error ' + t[1] + ' start bit overwrites previous signal')
                            prev_signal_end = int(bit_start) + int(bit_size)
                        prev_mux_index = int(mux[1:])

            # If we have an invalid message length then invalidate the DBC and print the offending signal
            # Signal bit width is <= 0
            if (int(bit_size) <= 0):
                print('/// ERROR /')
                print('#error ' + t[1] + ' has invalid size. Signal bit width is: ' + str(int(bit_size))) 
                print('/// ERROR /')
                print('')
                raise ValueError('#error ' + t[1] + ' has invalid size. Signal bit width is: ' + str(int(bit_size)))
            
            # Signal is too wide for message
            if (int(bit_start) + int(bit_size)) > (int(msg_length) * 8):
                print('/// ERROR /')
                print('#error ' + t[1] + ' too large. Message needs ' + str(int(bit_start) + int(bit_size)) + ' bits.') 
                print('/// ERROR /')
                print('')
                raise ValueError('#error ' + t[1] + ' too large. Message needs ' + str(int(bit_start) + int(bit_size)) + ' bits.')

            endian_and_sign = s[2]
            # Split (0.1,1) to two tokens by removing the ( and the )
            s = t[4][1:-1].split(',')
            scale = s[0]
            offset = s[1]

            # Split the [0|0] to min and max
            s = t[5][1:-1].split('|')
            min_val = s[0]
            max_val = s[1]

            signal_min = 0
            signal_max = (float(scale) * pow(2,int(bit_size)))
            if '-' in t[3]:
                signal_min = -(float(scale) * pow(2,int(bit_size))) / 2
                signal_max = (float(scale) * pow(2,int(bit_size)) / 2)
            # If our min / max values are incorrect then clamping will not work correctly. 
            # Invalidate the DBC and print out the offending signal.
            signal_min = signal_min + float(offset)
            signal_max = signal_max + float(offset) - float(scale)

            # Min for signal is too low.
            if (float(min_val) != 0) and (float(min_val) < float(signal_min)):
                print('/// ERROR /')
                print('#error ' + t[1] + ' min value too low. Min value is: ' + str(signal_min))
                print('/// ERROR /')
                print('')
                raise ValueError('#error ' + t[1] + ' min value too low. Min value is: ' + str(signal_min))

            # Max for signal is too high
            if (float(max_val) != 0) and (float(max_val)) > (float(signal_max)):
                print('/// ERROR /')
                print('#error ' + t[1] + ' max value too high. Max value is: ' + str(signal_max))
                print('/// ERROR /')
                print('')
                raise ValueError('#error ' + t[1] + ' max value too high. Max value is: ' + str(signal_max))

            recipients = t[7].strip('n').split(',')

            # Add the signal the last message object
            sig = Signal(t[1], bit_start, bit_size, endian_and_sign, scale, offset, min_val, max_val, recipients, mux, signal_min, signal_max)
            dbc.messages[last_mid].add_signal(sig)

        # Parse the "FieldType" which is the trigger to use enumeration type for certain signals
        if line.startswith('BA_ "FieldType"'):
            t = line[1:].split(' ')  # BA_ "FieldType" SG_ 123 Some_sig "Some_sig";
            sig_mid = t[3]
            sig_name = t[4]

            # Locate the message and the signal whom this "FieldType" type belongs to
            if sig_mid in dbc.messages:
                if sig_name in dbc.messages[sig_mid].signals:
                    dbc.messages[sig_mid].signals[sig_name].has_field_type = True

        # Enumeration types
        # VAL_ 100 DRIVER_HEARTBEAT_cmd 2 "DRIVER_HEARTBEAT_cmd_REBOOT" 1 "DRIVER_HEARTBEAT_cmd_SYNC" ;
        if line.startswith("VAL_ "):
            t = line[1:].split(' ')
            sig_mid = t[1]
            enum_name = t[2]
            pairs = {}
            t = t[3:]
            for i in range(0, int(len(t) / 2)):
                pairs[t[i * 2 + 1].replace('"', '').replace(';n', '')] = t[i * 2]

            # Locate the message and the signal whom this enumeration type belongs to
            if sig_mid in dbc.messages:
                if enum_name in dbc.messages[sig_mid].signals:
                    if dbc.messages[sig_mid].signals[enum_name].has_field_type:
                        dbc.messages[sig_mid].signals[enum_name].enum_info = pairs
    
    # If there were errors in parsing the DBC file then do not continue with generation.
    if not validFile:
        sys.exit(-1)

    print(dbc.gen_file_header())
    print("n")

    # Generate the application send extern function
    print("/// Extern function needed for dbc_encode_and_send()")
    print("extern bool dbc_app_send_can_msg(uint32_t mid, uint8_t dlc, uint8_t bytes[8]);")
    print("")

    # Generate header structs and MIA struct
    print(dbc.gen_mia_struct())
    print(dbc.gen_msg_hdr_struct())
    print(dbc.gen_msg_hdr_instances())
    print(dbc.gen_enum_types())

    # Generate converted struct types for each message
    for mid in dbc.messages:
        m = dbc.messages[mid]
        if not gen_all and not m.is_recipient_of_at_least_one_sig(self_node) and m.sender != self_node:
            code = ("n// Not generating '" + m.get_struct_name() + "' since we are not the sender or a recipient of any of its signals")
        else:
            print(m.gen_converted_struct(self_node, gen_all))

    # Generate MIA handler "externs"
    print("n/// @{ These 'externs' need to be defined in a source file of your project")
    for mid in dbc.messages:
        m = dbc.messages[mid]
        if gen_all or m.is_recipient_of_at_least_one_sig(self_node):
            if m.contains_muxed_signals():
                muxes = m.get_muxes()
                for mux in muxes[1:]:
                    print(str("extern const uint32_t ").ljust(50) + (m.name + "_" + mux + "__MIA_MS;"))
                    print(str("extern const " + m.get_struct_name()[:-2] + "_" + mux + "_t").ljust(49) + " " + (
                    m.name + "_" + mux + "__MIA_MSG;"))
            else:
                print(str("extern const uint32_t ").ljust(50) + (m.name + "__MIA_MS;"))
                print(str("extern const " + m.get_struct_name()).ljust(49) + " " + (m.name + "__MIA_MSG;"))
    print("/// @}n")

    # Generate encode methods
    for mid in dbc.messages:
        m = dbc.messages[mid]
        if not gen_all and m.sender != self_node:
            print ("n/// Not generating code for dbc_encode_" + m.get_struct_name()[:-2] + "() since the sender is " + m.sender + " and we are " + self_node)
        else:
            print(m.get_encode_code())

    # Generate decode methods
    for mid in dbc.messages:
        m = dbc.messages[mid]
        if not gen_all and not m.is_recipient_of_at_least_one_sig(self_node):
            print ("n/// Not generating code for dbc_decode_" + m.get_struct_name()[:-2] + "() since '" + self_node + "' is not the recipient of any of the signals")
        else:
            print(m.get_decode_code())

    print(dbc.gen_mia_funcs())
    print("#endif")


if __name__ == "__main__":
    main(sys.argv[1:])
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
#include "../include/can_encode_decode_inl.h"

#include <iostream>
#include <stdlib.h>

#include <stack>
#include <ctime>
#include <bitset>
#include <cassert>

std::stack<clock_t> tictoc_stack;
unsigned int num_tests = 0;

void tic()
{
  tictoc_stack.push(clock());
}

void toc()
{
  std::cout << "All " << num_tests << " tests passed within "
            << ((double)(clock() - tictoc_stack.top())) / CLOCKS_PER_SEC << " seconds!" << std::endl;
  tictoc_stack.pop();
}

inline void TEST(float k, float v, float eps = 0.01)
{
  num_tests++;
  bool near = (k >= v - eps && k <= v + eps);
  if (!near)
  {
    std::cerr << "Expected " << v << " but got " << k << std::endl;
    exit(1);
  }
}

inline void PRINT(uint64_t value, int64_t signed_value)
{
  std::cout << "Values signed vs unsigned:" << std::endl;
  std::cout << std::bitset<64>(value) << ": " << value << std::endl;
  std::cout << std::bitset<64>(signed_value) << ": " << signed_value << " (signed)" << std::endl;
}

inline void PRINT(uint8_t* array)
{
  std::cout << ">> Array:" << std::endl;
  for (unsigned int i = 0; i < 8; ++i)
    std::cout << "t" << std::bitset<8>(array[i]) << std::endl;
}

inline void TEST_STORE_EXTRACT(int64_t value, unsigned int startbit, unsigned int length, bool is_big_endian,
                               bool is_signed)
{
  // std::cout << "--n TEST_STORE_EXTRACT(" << value << ", " << startbit << ", " << length << ", " << is_big_endian <<
  // ", " << is_signed << ")" << std::endl;
  uint8_t src_array[8] = { 0, 0, 0, 0, 0, 0, 0, 0 };

  storeSignal(src_array, value, startbit, length, is_big_endian, is_signed);
  TEST((int32_t)extractSignal(src_array, startbit, length, is_big_endian, is_signed), value);
}

inline void TEST_ENCODE_DECODE(float value, unsigned int startbit, unsigned int length, bool is_big_endian,
                               bool is_signed, float factor, float offset)
{
  // std::cout << "--n TEST_ENCODE_DECODE(" << value << ", " << startbit << ", " << length << ", " << is_big_endian <<
  // ", " << is_signed << ", " << factor << ", " << offset << ")" << std::endl;
  uint8_t src_array[8] = { 0, 0, 0, 0, 0, 0, 0, 0 };

  // Only negative allowed when signed
  assert(is_signed || (int64_t)fromPhysicalValue(value, factor, offset) >= 0);

  encode(src_array, value, startbit, length, is_big_endian, is_signed, factor, offset);

  TEST(decode(src_array, startbit, length, is_big_endian, is_signed, factor, offset), value);
}

inline void TEST_IQ_STORE_EXTRACT(float value, unsigned int startbit, unsigned int length, unsigned int float_length,
                                  bool is_big_endian, bool is_signed)
{
  uint8_t src_array[8] = { 0, 0, 0, 0, 0, 0, 0, 0 };
  storeIQ(src_array, value, startbit, length, float_length, is_big_endian, is_signed);
  TEST(extractIQ(src_array, startbit, length, float_length, is_big_endian, is_signed), value);
}

int main()
{
  tic();

  // Decode test 1 (big endian unsigned)
  {
    uint64_t src = 0x182000110008d;
    uint8_t src_array[8] = { 141, 0, 16, 1, 0, 130, 1, 0 };

    TEST(extractSignal(src_array, 0, 2, true, false), 1);
    TEST(decode(src_array, 0, 2, true, false, 1.000000, 0), 1.000000);

    TEST(extractSignal(src_array, 2, 6, true, false), 35);
    TEST(decode(src_array, 2, 6, true, false, 1.000000, 0), 35.000000);

    TEST(extractSignal(src_array, 21, 11, true, false), 0);
    TEST(decode(src_array, 21, 11, true, false, 0.100000, 0), 0.000000);

    TEST(extractSignal(src_array, 25, 12, true, false), 2048);
    TEST(decode(src_array, 25, 12, true, false, 0.062500, -128), 0.000000);

    TEST(extractSignal(src_array, 32, 9, true, false), 256);
    TEST(decode(src_array, 32, 9, true, false, 0.062500, -16), 0.000000);

    TEST(extractSignal(src_array, 48, 3, true, false), 1);
    TEST(decode(src_array, 48, 3, true, false, 1.000000, 0), 1.000000);

    TEST(extractSignal(src_array, 51, 3, true, false), 0);
    TEST(decode(src_array, 51, 3, true, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 54, 10, true, false), 520);
    TEST(decode(src_array, 54, 10, true, false, 0.100000, -52), 0.000000);

    TEST(extractSignal(src_array, 56, 3, true, false), 0);
    TEST(decode(src_array, 56, 3, true, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 59, 3, true, false), 0);
    TEST(decode(src_array, 59, 3, true, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 62, 2, true, false), 0);
    TEST(decode(src_array, 62, 2, true, false, 1.000000, 0), 0.000000);
  }

  // Decode test 2 (little endian, signed and unsigned)
  {
    // src, startbit, bitlength, is_big_endian, is_signed, factor, offset, name, value
    uint8_t src_array[8] = { 12, 0, 5, 112, 3, 205, 31, 131 };

    TEST(extractSignal(src_array, 60, 2, false, false), 0);
    TEST(decode(src_array, 60, 2, false, true, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 55, 1, false, false), 0);
    TEST(decode(src_array, 55, 1, false, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 20, 4, false, false), 0);
    TEST(decode(src_array, 20, 4, false, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 62, 2, false, false), 2);
    TEST(decode(src_array, 62, 2, false, false, 1.000000, 0), 2.000000);

    TEST(extractSignal(src_array, 34, 3, false, false), 0);
    TEST(decode(src_array, 34, 3, false, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 37, 3, false, false), 0);
    TEST(decode(src_array, 37, 3, false, false, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 59, 1, false, true), 0);
    TEST(decode(src_array, 59, 1, false, true, 1.000000, 0), 0.000000);

    TEST(extractSignal(src_array, 56, 3, false, false), 3);
    TEST(decode(src_array, 56, 3, false, false, 1.000000, 0), 3.000000);

    TEST(extractSignal(src_array, 52, 3, false, false), 1);
    TEST(decode(src_array, 52, 3, false, false, 1.000000, 0), 1.000000);

    TEST(extractSignal(src_array, 8, 12, false, false), 1280);
    TEST(decode(src_array, 8, 12, false, false, 0.062500, 0), 80.000000);

    TEST(decode(src_array, 40, 12, false, true, 0.062500, 0), -3.187500);

    TEST(extractSignal(src_array, 24, 10, false, true), (uint64_t)-144);
    TEST(decode(src_array, 24, 10, false, true, 0.062500, 0), -9.000000);

    TEST(extractSignal(src_array, 0, 8, false, false), 12);
    TEST(decode(src_array, 0, 8, false, false, 1.000000, 0), 12.000000);
  }

  // Store extract
  {
    TEST_STORE_EXTRACT(512, 8, 12, false, true);

    // Limits
    TEST_STORE_EXTRACT(512, 0, 32, false, false);
    TEST_STORE_EXTRACT(512, 56, 32, true, false);
    TEST_STORE_EXTRACT(512, 32, 32, false, false);
    TEST_STORE_EXTRACT(512, 32, 32, true, false);

    // Limits
    TEST_STORE_EXTRACT(-512, 0, 32, false, true);
    TEST_STORE_EXTRACT(-512, 56, 32, true, true);
    TEST_STORE_EXTRACT(-512, 32, 32, false, true);
    TEST_STORE_EXTRACT(-512, 32, 32, true, true);

    // Some other
    TEST_STORE_EXTRACT(26, 4, 8, false, false);
    TEST_STORE_EXTRACT(26, 62, 8, true, false);
    TEST_STORE_EXTRACT(-12, 4, 8, false, true);
    TEST_STORE_EXTRACT(-13, 62, 8, true, true);
    TEST_STORE_EXTRACT(-190, 62, 24, true, true);
  }

  // Encoding decoding
  {
    TEST_ENCODE_DECODE(-1.0, 0, 7, false, true, 1.0, 0.0);
    TEST_ENCODE_DECODE(-100.0, 0, 7, false, true, 100, 0.0);
    TEST_ENCODE_DECODE(3.0, 56, 3, false, false, 1.000000, 0.0);
    TEST_ENCODE_DECODE(35, 2, 6, true, false, 1.0, 0);
    TEST_ENCODE_DECODE(1.0, 0, 2, true, false, 1.0, 0);

    TEST_ENCODE_DECODE(1.0, 62, 24, true, true, 0.1, 20);
  }

  // IQ notations store extract (little endian only)
  {
    TEST_IQ_STORE_EXTRACT(-1.0, 0, 7, 3, false, true);
    TEST_IQ_STORE_EXTRACT(-8.25, 0, 16, 8, false, true);
    TEST_IQ_STORE_EXTRACT(3.0, 56, 3, 0, false, false);
    TEST_IQ_STORE_EXTRACT(8.32, 2, 16, 8, false, false);
    TEST_IQ_STORE_EXTRACT(1.0, 0, 2, 0, false, false);
    TEST_IQ_STORE_EXTRACT(1.6, 0, 24, 8, false, true);
  }

  toc();
}

 

最后

以上就是现代铅笔最近收集整理的关于Python 解析Vector DBC信号自动生成代码https://download.csdn.net/download/fz835304205/10819610DBC Format的全部内容，更多相关Python内容请搜索靠谱客的其他文章。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(55)

相关文章

Simulink框图整理心得Simulink 框图整理心得
Simulink框图整理心得Simulink 框图整理心得
simulink快速上手
simulink快速上手
simulink中gnd无法连线问题
simulink中gnd无法连线问题
Matlab/Simulink-----Vector 联合仿真
Matlab/Simulink-----Vector 联合仿真
Python 解析Vector DBC信号自动生成代码https://download.csdn.net/download/fz835304205/10819610DBC Format
Python 解析Vector DBC信号自动生成代码https://download.csdn.net/download/fz835304205/10819610DBC Format
MATLAB点与点之间连线
MATLAB点与点之间连线
生成自动判别类型的Vector
生成自动判别类型的Vector
如何拿vector生成二维数组三维数组
如何拿vector生成二维数组三维数组

评论列表共有 0 条评论

立即
投稿
返回
顶部