######################################################################################### # # RTips Technologies, Bangalore (www.rtipsonline.com) # # This sample demonstrates using DDC's AceXtreme SDK in a Python script. The sample # creates three messages (BC->RT, BC<-RT and RT->RT) and schedules them in one minor # frame. It reads message the results and prints them to the console. The sample # demonstrates converting between Phython-type variables to c-types in order to pass to # the SDK APIs. Similarly, return values of APIs are converted back to Python types. # # The console prints have been designed to match that of the bcdemo.c sample in the SDK. # This source code may freely be distributed and used even for commercial purposes # provided this header block is retained. # # Author: Ganesh Okade (ganeshokade@rtipsonline.com) # Last updated: August 05, 2025 ######################################################################################### import ctypes # importing ctypes for using the DDC library which orginates in C, using python script import sys from time import sleep from dataclasses import dataclass # Specify path to the AceXtreme SDK API dll file. ddc_lib = ctypes.CDLL ("c://DDC//aceXtremeSDKv4.9.5//libraries//emacepl//bin//x64//Release//emacepl.dll") # Constants used by this program. This is equivalent to symbol/macro creation in 'C' using #define DBLK1 = 1 DBLK2 = 2 DBLK3 = 3 MSG1 = 0 MSG2 = 1 MSG3 = 2 OP1 = 0 OP2 = 1 OP3 = 2 OP_CAL = 3 MNR1 = 1 MJR = 2 NO_OF_MSGS_IN_MIN_FRAME = 3 # Constants required to be passed as arguements to SDK APIs. In the 'C' SDK, these values are generally available in the header files ACE_ACCESS_CARD = 0 ACE_MODE_BC = 1 ACE_OPCODE_XEQ = 0x0001 ACE_OPCODE_CAL = 0x0003 ACE_CNDTST_ALWAYS = 0x000F ACE_FRAME_MINOR = 0x0002 ACE_FRAME_MAJOR = 0x0000 ACE_BCCTRL_EOM_IRQ = 0x0010 ACE_IMR1_BC_MSG_EOM = 0x00000010 TRUE = 1 FALSE = 0 # Bit masks use for extracting bit information from word values ACE_BCCTRL_RT_TO_RT = 0x0001 ACE_BC_BSW_CHNL = 0x2000 ### Define callback type CALLBACK_TYPE = ctypes.WINFUNCTYPE(None, ctypes.c_short, ctypes.c_uint) ### Create global buffer of 32 unsigned 16-bit integers (U16BIT) to handle data in ISR dblkBuffer = (ctypes.c_ushort * 32)() ### Define ISR, InterruptHandler @CALLBACK_TYPE def InterruptHandler(devNum, status): nResult = 0 nResult = ddc_lib.aceBCDataBlkRead(devNum, DBLK1, dblkBuffer, 32, 0) # Increment first 10 values for i in range(10): dblkBuffer[i] += 1 nResult = ddc_lib.aceBCDataBlkWrite(devNum, DBLK1, dblkBuffer, 32, 0) #### Print header print('\n---------------------------------------------------------------------') print('Demo - Using aceXtreme SDK APIs in Python') print('Copyright (c) 2014 RTips Technologies') print('\nThis sample demonstrates using the AceXtreme SDK in a Python script.') print('The sample creates three messages (BC->RT, BC<-RT and RT->RT) and') print('schedules them in one minor frame. It reads message the ') print('results and prints them to the console.') print('---------------------------------------------------------------------') #### Read Logical Device Number of 1553 channel to run BC on devNum = input("\nSelect BC Logical Device Number (0-31):\n> ") try: val = int(devNum) except ValueError: print("Invalid Logical Device Number entered. Aborting...") sys.exit() devNum = int(devNum) if devNum<0 or 32 ") try: val = int(chnlIn) except ValueError: print("Invalid Bus ID entered. Aborting...") sys.exit() chnlIn = int(chnlIn) if chnlIn == 1: chnlNum = 0x0080 # selecting channel A elif chnlIn == 2: chnlNum = 0x0000 # selecting channel B else: print("Invalid Bus ID entered. Aborting...") sys.exit() #### Initialize card as BC nresult = 0 nresult = ddc_lib.aceInitialize(devNum, ACE_ACCESS_CARD, ACE_MODE_BC, 0, 0, 0) if nresult != 0: print('aceIntialize error') sys.exit() #### Define buffer array for creation of data block buffer_array = [0x1111, 0x2222, 0x3333, 0x4444, 0x1111, 0x2222, 0x3333, 0x4444, \ 0x1111, 0x2222, 0x3333, 0x4444, 0x1111, 0x2222, 0x3333, 0x4444, \ 0x1111, 0x2222, 0x3333, 0x4444, 0x1111, 0x2222, 0x3333, 0x4444, \ 0x1111, 0x2222, 0x3333, 0x4444, 0x1111, 0x2222, 0x3333, 0x4444] #### Set interrupt at end of message interrupt_handler_ref = InterruptHandler nresult = ddc_lib.aceSetIrqConditions(devNum, TRUE, ACE_IMR1_BC_MSG_EOM, interrupt_handler_ref) if nresult != 0: print('aceSetIrqConditions error, Error code:', nresult) sys.exit() # Since the buffer array created is Python type it cannot be passed into a 'C' function. # Therfore we convert it into a compatible type for passing it to function aceBCDataBlkCreate() wBuffer = (ctypes.c_short * len(buffer_array))(*buffer_array) #### Create data block for MSG1 nresult = ddc_lib.aceBCDataBlkCreate(devNum, DBLK1, 32, wBuffer, 32) if nresult != 0: print('aceBCDataBlkCreate error, Error code:', nresult) sys.exit() #### Create data block for MSG2 nresult = ddc_lib.aceBCDataBlkCreate(devNum, DBLK2, 32, wBuffer, 32) if nresult != 0: print('aceBCDataBlkCreate error, Error code:', nresult) sys.exit() #### Create data block for MSG3 nresult = ddc_lib.aceBCDataBlkCreate(devNum, DBLK3, 32, wBuffer, 32) if nresult != 0: print('aceBCDataBlkCreate error, Error code:', nresult) sys.exit() #### Create MSG1 as 01-R-02-10 nresult = ddc_lib.aceBCMsgCreateBCtoRT(devNum, MSG1, DBLK1, 1, 1, 10, 0, chnlNum | ACE_BCCTRL_EOM_IRQ) if nresult != 0: print('aceBCMsgCreateBCtoRT error, Error code:', nresult) sys.exit() #### Create MSG2 as 02-T-05-32 nresult = ddc_lib.aceBCMsgCreateRTtoBC(devNum, MSG2, DBLK2, 2, 5, 0, 0, chnlNum) if nresult != 0: print('aceBCMsgCreateRTtoBC error, Error code:', nresult) sys.exit() #### Create MSG3 as 12-R-07-32 nresult = ddc_lib.aceBCMsgCreateRTtoRT(devNum, MSG3, DBLK3, 2, 7, 32, 3, 3, 0, chnlNum) if nresult != 0: print('aceBCMsgCreateRTtoRT error, Error code:', nresult) sys.exit() #### Create XEQ opcodes for the three messages ddc_lib.aceBCOpCodeCreate(devNum, OP1, ACE_OPCODE_XEQ, ACE_CNDTST_ALWAYS, MSG1, 0, 0) # XEQ opcode for MSG1 if nresult != 0: print('aceBCOpCodeCreate error (XEQ), Error code:', nresult) sys.exit() ddc_lib.aceBCOpCodeCreate(devNum, OP2, ACE_OPCODE_XEQ, ACE_CNDTST_ALWAYS, MSG2, 0, 0) # XEQ opcode for MSG2 if nresult != 0: print('aceBCOpCodeCreate error (XEQ), Error code:', nresult) sys.exit() ddc_lib.aceBCOpCodeCreate(devNum, OP3, ACE_OPCODE_XEQ, ACE_CNDTST_ALWAYS, MSG3, 0, 0) # XEQ opcode for MSG3 if nresult != 0: print('aceBCOpCodeCreate error (XEQ), Error code:', nresult) sys.exit() #### Create CAL opcode that will call minor frame from major frame ddc_lib.aceBCOpCodeCreate(devNum, OP_CAL, ACE_OPCODE_CAL, ACE_CNDTST_ALWAYS, 1, 0, 0) if nresult != 0: print('aceBCOpCodeCreate error (CAL), Error code:', nresult) sys.exit() #### Create an array of XEQ opcodes to send as argument to aceBCFrameCreate() aOpCode_array = [ 0x0000, 0x0000, 0x0000 ] # Since the above array is of a Python type, it cannot be passed directly to a 'C' function of the AceXtreme SDK. # So, convert aOpcode_array to ctypes that is suitable to pass into a 'C' function aOpCodes = (ctypes.c_uint16 * len(aOpCode_array))(*aOpCode_array) aOpCodes[0] = OP1 aOpCodes[1] = OP2 aOpCodes[2] = OP3 #### Create minor frame nresult = ddc_lib.aceBCFrameCreate(devNum, MNR1, ACE_FRAME_MINOR, aOpCodes, len(aOpCodes), 0, 0) if nresult != 0: print('aceBCFrameCreate error (Minor Frame), Error code:', nresult) sys.exit() #### Create major frame with just one minor frame aOpCodes[0] = OP_CAL nresult = ddc_lib.aceBCFrameCreate(devNum, MJR,ACE_FRAME_MAJOR, aOpCodes, 1, 1000, 0) if nresult != 0: print('aceBCFrameCreate error (Major Frame), Error code:', nresult) sys.exit() #### Install Host Buffer to read transaction results nresult = ddc_lib.aceBCInstallHBuf(devNum, 16*1024) if nresult != 0: print('aceBCInstallHBuf error, Error code:', nresult) sys.exit() #### Ask user whether to display messages as Raw or Decoded print("\nWould you like messages displayed as:") print("1. Raw") print("2. Decode") displaySelection = input("> ") try: val = int(displaySelection) except ValueError: print("Invalid display selection method entered. Aborting...") sys.exit() displaySelection = int(displaySelection) if displaySelection<1 or 2 to continue\n") # Start the BC nresult = ddc_lib.aceBCStart(devNum, 2, wReapeatCount) if nresult != 0: print('aceBCStart error, Error code:', nresult) sys.exit() # Define pointer types to pass as arguments into aceBCGetHBufMsgsRaw() c_uint32_p = ctypes.POINTER(ctypes.c_uint32) c_uint16_p = ctypes.POINTER(ctypes.c_uint16) # Read raw messages by calling aceBCGetHBufMsgsRaw() until all scheduled messages have been read msgcount = 0 while dwCurCount < (wReapeatCount * NO_OF_MSGS_IN_MIN_FRAME): # Mention the "type" of each arguement being passed toaceBCGetHBufMsgsRaw ddc_lib.aceBCGetHBufMsgsRaw.argtypes = [ctypes.c_short, c_uint16_p, ctypes.c_uint16, c_uint32_p, c_uint32_p] # Read one raw message from HBuf nresult = ddc_lib.aceBCGetHBufMsgsRaw(devNum, xBuffer, 42, dwMsgCount, dwMsgLost) if nresult != 0: print('aceBCGetHBufMsgsRaw error, Error code:', nresult) sys.exit() # Extract values returned by aceBCGetHBufMsgsRaw in pointer-type variables msgcount = c_uint32_p(dwMsgCount).contents.value msglost = c_uint16_p(dwMsgLost).contents.value # Update various counters if msgcount != 0 : dwLostCount += msglost dwCurCount += msgcount print("BC: Total msgs captured:" + str(dwCurCount + dwLostCount)) elif displaySelection == 2: # On user selection of decoded messages display # Setup 'C' compatible parameters to pass to various API functions dwHBufLost = ctypes.c_ulong() # defining parameter type dwMsgCount = ctypes.c_ulong() wRT = ctypes.c_ulong() wTR1 = ctypes.c_ulong() wTR2 = ctypes.c_ulong() wSA = ctypes.c_ulong() wWC = ctypes.c_ulong() dwCurCount = 0x00000000 wReapeatCount = 3 # No of times BC will run the major frame c_uint16 = ctypes.c_uint16 c_Array = ctypes.c_uint16 * 32 #defining array type and size for passing it to structure init_array = c_Array() # aceBCGetHBufMsgDecoded() uses a 'C' structure named "MSGSTRUCT" which is being defined here class MSGSTRUCT(ctypes.Structure): _fields_ = [('wType', c_uint16), ('wBlkSts', c_uint16), ('wTimeTag', c_uint16), ('wCmdWrd1', c_uint16), ('wCmdWrd2', c_uint16), ('wCmdWrd1Flg', c_uint16), ('wCmdWrd2Flg', c_uint16), ('wStsWrd1', c_uint16), ('wStsWrd2', c_uint16), ('wStsWrd1Flg', c_uint16), ('wStsWrd2Flg', c_uint16), ('wWordCount', c_uint16), ('aDataWrds', c_Array), ('wBCCtrlWrd', c_uint16), ('wBCGapTime', c_uint16), ('wBCLoopBack1', c_uint16), ('wTimeTag2', c_uint16), ('wBCLoopBack1Flg', c_uint16), ('wTimeTag3', c_uint16)] sMsg_p = ctypes.POINTER(MSGSTRUCT) # defining pointer type for structure MSGSTRUCT # Initialize the MSGSTRUCT variable "sMSg" before passing into aceBCGetHBufMsgDecoded() by zeroing out all values sMsg = MSGSTRUCT(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, init_array, 0, 0, 0, 0, 0, 0) c_uint32_p = ctypes.POINTER(ctypes.c_uint32) c_uint16_p = ctypes.POINTER(ctypes.c_uint16) print("\n** The BC will now send 20 messages\n\n** Messages will also be output to the screen\n") enter = input("Press to continue\n") # Start the BC nresult = ddc_lib.aceBCStart(devNum, 2, wReapeatCount) if nresult != 0: print('aceBCStart error, Error code:', nresult) sys.exit() # Read decoded messages by calling aceBCGetHBufMsgDecoded() until all scheduled messages have been read while dwCurCount < (wReapeatCount * NO_OF_MSGS_IN_MIN_FRAME): # Mention the "type" of each arguement being passed aceBCGetHBufMsgDecoded ddc_lib.aceBCGetHBufMsgDecoded.argtypes = [ctypes.c_short, sMsg_p, c_uint32_p, c_uint32_p, c_uint16] # Read one decoded message from HBuf nresult = ddc_lib.aceBCGetHBufMsgDecoded(devNum, sMsg, dwMsgCount, dwHBufLost, 0); if nresult != 0: print('aceBCGetHBufMsgDecoded error, Error code:', nresult) sys.exit() # Extract values returned by aceBCGetHBufMsgDecoded in pointer-type variables msgcount = c_uint32_p(dwMsgCount).contents.value hbuflost = c_uint16_p(dwHBufLost).contents.value if msgcount != 0 : # message found by aceBCGetHBufMsgDecoded dwCurCount += 1 # Update counters # Call aceGetMsgTypeString() to get message type string (e.g. "BC to RT") ddc_lib.aceGetMsgTypeString.restype = ctypes.c_char_p msgtype_string = ddc_lib.aceGetMsgTypeString(sMsg.wType) # Probe BSW to know which bus (A or B) this message was transmitted on if sMsg.wBlkSts & ACE_BC_BSW_CHNL == 0: bus = 'A' elif sMsg.wBlkSts & ACE_BC_BSW_CHNL == 8192: bus = 'B' # Print message number, time tag, Bus and message type print('MSG #{0:04d} TIME = {1}us BUS {2} TYPE{3}: {4}'.format(dwCurCount, '{:08d}'.format(sMsg.wTimeTag), bus, sMsg.wType, msgtype_string.decode("utf-8"))) # Parse the command word and extract individual information elements ddc_lib.aceCmdWordParse.argtypes = [ctypes.c_uint16, c_uint32_p, c_uint32_p, c_uint32_p, c_uint32_p] ddc_lib.aceCmdWordParse(sMsg.wCmdWrd1, wRT, wTR1, wSA, wWC) # Direction bit TR1 = 0 if c_uint32_p(wTR1).contents.value == 1: TR1 = 'T' elif c_uint32_p(wTR1).contents.value == 0: TR1 = 'R' # Print command word, direction bit, Sub Address and Word Count print(' CMD1 {0:04X} --> {1}-{2}-{3}-{4} BSW {5:04X}'.format( sMsg.wCmdWrd1, c_uint32_p(wRT).contents.value, TR1, c_uint32_p(wSA).contents.value, c_uint32_p(wWC).contents.value, sMsg.wBlkSts, sMsg.wBCCtrlWrd)) # Repeat for Command Word 2, if present if sMsg.wCmdWrd2Flg != 0: ddc_lib.aceCmdWordParse(sMsg.wCmdWrd2, wRT, wTR2, wSA, wWC) TR2 = 0 if c_uint32_p(wTR2).contents.value == 1: TR2 = 'T' elif c_uint32_p(wTR2).contents.value == 0: TR2 = 'R' print(' CMD2 {0:04X} --> {1}-{2}-{3}-{4}'.format( sMsg.wCmdWrd2, c_uint32_p(wRT).contents.value, TR2, c_uint32_p(wSA).contents.value, c_uint32_p(wWC).contents.value)) # Display transmit status words if c_uint32_p(wTR1).contents.value == 1: if sMsg.wStsWrd1Flg != 0: print(' STA1 {0:04X}'.format(sMsg.wStsWrd1)) # Print Data Words i = 0 while i < sMsg.wWordCount: if i == 0: print(' DATA', end =" ") print('{0:04X} '.format(sMsg.aDataWrds[i]), end = "") if (i%8) == 7: print('\n ', end = " ") i = i+1 # Display receive status words if c_uint32_p(wTR1).contents.value == 0: if sMsg.wStsWrd1Flg != 0: print('\n STA1 {0:04X}'.format(sMsg.wStsWrd1)) if sMsg.wStsWrd2Flg != 0: print('\n STA2 {0:04X}'.format(sMsg.wStsWrd2)) # Display Error information if sMsg.wBlkSts & 0x1000 != 0: ddc_lib.aceGetBSWErrString.restype = ctypes.c_char_p errortype_string = ddc_lib.aceGetBSWErrString(ACE_MODE_BC, sMsg.wBlkSts) print('\nERROR:', errortype_string.decode("utf-8")) print('\n') enter = input("Press to exit\n") sys.exit()