IEC 60870-5-101/104 Test Client 1.4.3

from System.Collections.Generic import List

#Connect to host 192.168.178.76 on port 2404

# Create and set APCI parameters and ApplicationLayer parameters

apci = CS104TestMaster.APCIParameters()
apci.K = 12
apci.W = 8
apci.T0 = 10
apci.T1 = 15
apci.T2 = 10
apci.T3 = 20
print("APCI Parameters Set")

alp = CS104TestMaster.ApplicationLayerParameters()
alp.OA = 10
alp.SizeOfCOT = 2
alp.SizeOfIOA = 3
alp.SizeOfCA = 2
print("Application Layer Parameters Set")

CS104TestMaster.SetApplicationLayerParameters(alp)
CS104TestMaster.SetAPCIParameters(apci)

#Connect to host 192.168.178.76 on port 2404

CS104TestMaster.Connect("192.168.178.76",2404)

#Exit code that can be checked after run script from command line

exitCode = ExitCode.UnknownError
CS104TestMaster.SetExitCode(exitCode)

#Create single point information object with timestamp

quality = CS101.QualityDescriptor()
quality.Blocked = True;
quality.Invalid = True;
quality.NonTopical = True;
quality.Substituted = True;
quality.Overflow = False;

time = CP24Time2a()
time.Millisecond = 2
time.Second = 3
time.Minute = 5
time.Invalid = True
time.Substitued = False

#List of information elements

InfoElementsList = List[InformationElement]()

valueSIQ = ElementValue()
valueSIQ.boolValue = True
valueSIQ.qualitySIQValue = quality

#InformationElement(string type, ElementValue value, int size, byte[] byteValue)
element1 = InformationElement("SIQ" ,valueSIQ, 1, None)

valueTime = ElementValue()
valueTime .cp24Time2aValue= time

#InformationElement(string type, ElementValue value, int size, byte[] byteValue)
element2 = InformationElement("CP24Time2a" ,valueTime , 3, None)

InfoElementsList .Add(element1)
InfoElementsList .Add(element2)

typeId = CS101.TypeID.M_SP_TA_1

#UserDefinedType((TypeID)typeDescription.TypeID, ioa, InformationElementsList)
io = UserDefinedType(typeId, 101, InfoElementsList)

#Create ASDU
alp = CS104TestMaster.GetApplicationLayerParameters()

asdu = CS101.ASDU(alp, CS101.CauseOfTransmission.ACTIVATION, False, False, alp.OA, 1, False)

#Add IO do ASDU
asdu.AddInformationObject(io)

#Send ASDU
CS104TestMaster.SendASDU(asdu)

#The following code wait 500ms and print messages received as well as ASDU

list = List[str]()
list.Add("I")

messageList = CS104TestMaster.WaitForMessages(5000,list)

print("Received message:")

for message in messageList :
  print(message.RawMessage.Timestamp)
  if(message.Apci!= None):
     print(message.Apci)
  if(message.Asdu != None):
      print(message.Asdu)

asduList=CS104TestMaster.WaitForAsdus(5000)

for i in asduList:
  print(i)

CS104TestMaster.Disconnect()

from System.Collections.Generic import List

#Set LPCI, Serial and Application layer parameters as well as the protocol mode

lpci = LPCIParameters()
lpci.PollInterval = 500
lpci.SetDirBit = True
lpci.DefaultSlaveAddress = 1
lpci.OwnAddress = 3
lpci.AutomaticPoll = False
lpci.LlAddressSize = 2

serial = SerialParameters()
serial.BaudRate = 9600
serial.Parity = "E"
serial.DeviceName = "COM9"

alp = CS101.ApplicationLayerParameters()
alp.OA = 10
alp.SizeOfCOT = 2
alp.SizeOfIOA = 3
alp.SizeOfCA = 2

mode = ProtocolMode.CS101_BALANCED

master=CS101TestMaster
master.SetLPCIParameters(lpci)
master.SetSerialParameters(serial)
master.SetMode(mode)
master.SetApplicationLayerParameters(alp)

#Connect after set the parameters
master.Connect()

#Wait and print the information of messages received in a interval of 5000ms
messageList = master.WaitForMessages(5000)

print("Received message:")

for message in messageList :
  print(message.RawMessage.Timestamp)
  if(message.Apci!= None):
     print(message.Apci);
  if(message.Asdu != None):
      print(message.Asdu);

#Wait and print the information of ASDUs received in a interval of 5000ms
asduList=master.WaitForAsdus(5000)

for i in asduList:
  print(i)

#Create single information object with timestamp, add to ASDU and send it

quality = CS101.QualityDescriptor()
quality.Blocked = True;
quality.Invalid = True;
quality.NonTopical = True;
quality.Substituted = True;
quality.Overflow = False;

time = CP24Time2a()
time.Millisecond = 2
time.Second = 3
time.Minute = 5
time.Invalid = True
time.Substitued = False

InfoElementsList = List[InformationElement]()

#using IOFactory.CreateElement (type, value, value) to return InformationElement and add to InfoElementsList
InfoElementsList .Add(IOFactory.CreateElement("SIQ", True, quality))
InfoElementsList .Add(IOFactory.CreateElement("CP24Time2a",time) )

#Using IOFactory.CreateIO to create de information object
io = IOFactory.CreateIO("M_SP_TA_1", 101, InfoElementsList)

alp = CS104TestMaster.GetApplicationLayerParameters()

asdu = CS101.ASDU(alp, CS101.CauseOfTransmission.ACTIVATION, False, False, alp.OA, 1, False)

asdu.AddInformationObject(io)

master.SendASDU(asdu)

master.Disconnect()

size = CS104TestMaster.MessageBufferSize()
print("Size " + str(size))

#Get message in the position 3
rawMessage = CS104TestMaster.MessageBufferGetRawMessage(3)

print(rawMessage)

parsedMessage= CS104TestMaster.MessageBufferGetParsedMessage(3)

print(parsedMessage)

#Print rawMessage TimeStamp
print (rawMessage .Timestamp)

#Print parsedMessage Apci
if parsedMessage.Apci is not None:
  print(parsedMessage.Apci)

#Get ASDU
if parsedMessage.Asdu is not None:
  asdu = parsedMessage.Asdu
  print(asdu)
  print("Number of Elements: " + str(asdu.NumberOfElements))

  #Get all IO information from the ASDU
  for x in range(0, asdu.NumberOfElements) :
    InformationObject = asdu.GetElement(x)
    print(InformationObject)
    print("IOA(" + str(x) + "): " + str(InformationObject .ObjectAddress))

    #Get IO information details
    for info in CS104TestMaster.GetIODetails(asdu,x):
      print(info)

.\IEC104TestTool.exe script.py

.\IEC104TestTool.exe -l ALL ScriptExamples\CS104\ScripExampleCS104.py

.\IEC104TestTool.exe -l ALL ScriptExamples\CS104\ScripExampleCS104.py -s C:\Users\micha\Desktop\savedLog.scp

.\IEC104TestTool.exe -l ALL ScriptExamples\CS104\ScripExampleCS104.py -o C:\Users\micha\Desktop\savedLog.scp