From 92927c8a37c4a813a720eb8d67f45ce724bcde86 Mon Sep 17 00:00:00 2001 From: bebopagogo Date: Fri, 11 Nov 2022 20:17:30 -0500 Subject: [PATCH] updated pynorm Python binding to use Python 3 contributions from Honglei --- examples/normSocket.cpp | 2 - examples/normStreamer.cpp | 15 +- examples/python/normCast.py | 423 ++++++++++++++++++++++++++++++++++++ examples/python/normMsgr.py | 42 ++-- src/pynorm/__init__.py | 11 +- src/pynorm/constants.py | 146 ++++++++----- src/pynorm/core.py | 12 + src/pynorm/event.py | 79 +------ src/pynorm/instance.py | 61 +++--- src/pynorm/node.py | 34 +-- src/pynorm/object.py | 89 ++++---- src/pynorm/session.py | 240 ++++++++++++-------- 12 files changed, 832 insertions(+), 322 deletions(-) create mode 100644 examples/python/normCast.py diff --git a/examples/normSocket.cpp b/examples/normSocket.cpp index 829ed08..27b7247 100755 --- a/examples/normSocket.cpp +++ b/examples/normSocket.cpp @@ -519,8 +519,6 @@ bool NormSocket::Listen(UINT16 serverPort, const char* groupAddr, const char* se NormSetSilentReceiver(norm_session, true); // So that the listener can construct (unsent) ACKs without failure - // (I.e., these ACKs are never sent by the listener but this call is - // needed as a "work around" existing NORM "connectionless" behavior) NormSetDefaultUnicastNack(norm_session, true); // Note we use a _small_ buffer size here since a "listening" socket isn't diff --git a/examples/normStreamer.cpp b/examples/normStreamer.cpp index 371e21d..a2b3f9b 100755 --- a/examples/normStreamer.cpp +++ b/examples/normStreamer.cpp @@ -70,7 +70,10 @@ class NormStreamer { loopback = state; if (NORM_SESSION_INVALID != norm_session) + { NormSetMulticastLoopback(norm_session, state); + NormSetLoopback(norm_session, state); // XXX test code + } } void SetFtiInfo(bool state) { @@ -567,8 +570,11 @@ bool NormStreamer::OpenNormSession(NormInstanceHandle instance, const char* addr { NormSetRxPortReuse(norm_session, true); if (loopback) + { NormSetMulticastLoopback(norm_session, true); + } } + NormSetLoopback(norm_session, loopback); // Set some default parameters (maybe we should put parameter setting in Start()) NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM); @@ -889,11 +895,11 @@ unsigned int NormStreamer::WriteToStream(const char* buffer, unsigned int numByt if (ack_ex) { const char* req = "Hello, acker"; - NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1); + NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1);//, true); } else { - NormSetWatermark(norm_session, tx_stream); + NormSetWatermark(norm_session, tx_stream);//, true); } tx_watermark_pending = true; @@ -1178,7 +1184,6 @@ void NormStreamer::HandleNormEvent(const NormEvent& event) break; case NORM_TX_WATERMARK_COMPLETED: - TRACE("NORM_TX_WATERMARK_COMPLETED ...\n"); if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) { //fprintf(stderr, "WATERMARK COMPLETED\n"); @@ -2051,11 +2056,7 @@ int main(int argc, char* argv[]) } // TBD - should provide more error checking of NORM API calls - TRACE("creating instance ...\n"); NormInstanceHandle normInstance = NormCreateInstance(boostPriority); - TRACE("restarting instance ...\n"); - //NormRestartInstance(normInstance); // xxx - TRACE("instance restarted.\n"); NormSetDebugLevel(debugLevel); if ((NULL != logFile) && !NormOpenDebugLog(normInstance, logFile)) diff --git a/examples/python/normCast.py b/examples/python/normCast.py new file mode 100644 index 0000000..f20f9bf --- /dev/null +++ b/examples/python/normCast.py @@ -0,0 +1,423 @@ +''' + +LD_LIBRARY_PATH=/home/honglei/norm_test/ + +''' +# sender/Win10: +#--id 34252 --send "C:\Users\Admin\AppData\Roaming\Wing Pro 9" --repeat 1 --addr 224.1.2.4/6003 --txaddr 10.65.39.191/8002 --cc rate 5000 --ack auto --grttprobing active --debug 4 +# receiver/Debian10 +#--id 172042016 --recv "~/recvFiles" --addr 224.1.2.4/6003 --interface ens33 --unicast_nack --debug 4 +import sys +import os +if os.name =='nt': + os.add_dll_directory(os.path.dirname(__file__)) +import argparse +import traceback +import logging +from typing import Optional + +import enum + +class CCEnum(enum.Enum): + CC = "cc" + CCE = "cce" + CCL = "ccl" + Fixed = "rate" + +class FlushEnum(enum.Enum): + none='none' + passive='passive' + active='active' + +class GrttEnum(enum.Enum): + none='none' + passive='passive' + active='active' + + +class EnumAction(argparse.Action): + """ + Argparse action for handling Enums + from https://stackoverflow.com/questions/43968006/support-for-enum-arguments-in-argparse + """ + def __init__(self, **kwargs): + # Pop off the type value + enum_type = kwargs.pop("type", None) + + # Ensure an Enum subclass is provided + if enum_type is None: + raise ValueError("type must be assigned an Enum when using EnumAction") + if not issubclass(enum_type, enum.Enum): + raise TypeError("type must be an Enum when using EnumAction") + + # Generate choices from the Enum + #kwargs.setdefault("choices", tuple(e.value for e in enum_type)) #if e!=CCEnum.rate else "rate " + + super(EnumAction, self).__init__(**kwargs) + + self._enum = enum_type + + def __call__(self, parser, namespace, values:list|str, option_string=None): + # Convert value back into an Enum + if isinstance(values,list): + if len(values) ==2: + value = int(values[1]) + elif len(values) ==1: + value = self._enum( values[0].lower()) + else: + value = self._enum(values.lower() ) + setattr(namespace, self.dest, value) + +''' + norm.cpp: + + fprintf(stderr, "Usage: normCast {send &| recv } [silent {on|off}]\n" + " [repeat [updatesOnly]] [id ]\n" + " [addr [/]][txaddr [/]][txport ]\n" + " [interface ][reuse][loopback]\n" + " [ack auto|[,,...]] [segment ]\n" + " [block ] [parity ] [auto ]\n" + " [cc|cce|ccl|rate ] [rxloss ]\n" + " [txloss ] [flush {none|passive|active}]\n" + " [grttprobing {none|passive|active}] [grtt ]\n" + " [ptos ] [processor ] [saveaborts]\n" + " [sentprocessor ]\n" + " [purgeprocessor ] [buffer ]\n" + " [txsockbuffer ] [rxsockbuffer ]\n" + " [debug ] [trace] [log ]\n"); + +''' +# Instantiate the parser + +def get_arg_options(): + parser = argparse.ArgumentParser(description='Optional app description') + + ## Required positional argument + parser.add_argument('--id', type=int, #required=True, + help='id ') + + parser.add_argument('--send', '-S',nargs='+', + help='send ') + + parser.add_argument('--recv', '-R', + help='recv ') + + parser.add_argument('--repeat','-r', type=int, default=60, + help='[repeat [updatesOnly]]') + + parser.add_argument('--interface', '-i', + help='addr [/]') + + parser.add_argument('--addr', '-a', + help='addr [/]') + + parser.add_argument('--txaddr', '-t', + help='txaddr [/]') + parser.add_argument('--txport', type=int, + help='txport ') + + parser.add_argument('--ack', nargs='+', #nargs='?', + help='ack auto|[,,...') + + + #To create an option that needs no value, set the action [docs] of it to 'store_const', 'store_true' or 'store_false' + parser.add_argument('--loopback', action='store_true',help='') + parser.add_argument('--segment', type=int, + help='segment ') + parser.add_argument('--block', type=int, + help='[block ] ') + parser.add_argument('--parity', type=int, + help='[parity ]') + parser.add_argument('--auto', type=int, + help='[auto ]') + + + parser.add_argument('--grttprobing', type=GrttEnum, action=EnumAction, + help='[grttprobing {none|passive|active}]') + parser.add_argument('--grtt', type=int, help='[grtt ]') + + parser.add_argument('--ptos', type=int, + help='[ptos ]') + + parser.add_argument('--flush', type=FlushEnum, action=EnumAction, + help='[flush {none|passive|active}]') + + parser.add_argument('--silent', action='store_true',help='') + parser.add_argument('--unicast_nack', action='store_true', help='') + + parser.add_argument('--txloss', type=float, help='[txloss ]') + parser.add_argument('--rxloss', type=float, help='[rxloss ]') + + parser.add_argument('--buffer', type=int, help='[buffer ]') + parser.add_argument('--txsockbuffer', type=int, help=' [txsockbuffer ]') + parser.add_argument('--rxsockbuffer', type=int, help='[rxsockbuffer ]') + parser.add_argument('--debug', type=int, help='[debug ]') + parser.add_argument('--cc',type=CCEnum, nargs='+', action=EnumAction, + help='[cc|cce|ccl|rate ] ' + ) + + args = parser.parse_args() + return args + +import pynorm + +def create_session(instance:pynorm.Instance, opts:argparse.Namespace): + ''' + use opts to create a NORM session + ''' + sessionAddr ='224.1.2.3' + sessionPort = 6003 + + if opts.addr: + args = opts.addr.split('/') + if len(args) >0: + sessionAddr = args[0] + if len(args) ==2: + sessionPort = int(args[1]) + session = instance.createSession(sessionAddr, sessionPort, localId=opts.id) + + + sessionTxAddr:str = None + sessionTxPort = 8002 + if opts.txaddr: + args = opts.txaddr.split('/') + if len(args) >0: + sessionTxAddr = args[0] + if len(args) ==2: + sessionTxPort = int(args[1]) + + if sessionTxAddr: + session.setTxPort(txPort=sessionTxPort, txBindAddr=sessionTxAddr) + + if opts.interface: + session.setMulticastInterface(opts.interface) + if opts.loopback: + session.setLoopback(loopbackEnable=True) + + if opts.silent: + session.setSilentReceiver(True) + + if opts.unicast_nack: + session.setDefaultUnicastNack(enable=True) + + autoAck:bool = False + ackingNodeList:list[int] =[] + if opts.ack: + if 'auto' == opts.ack[0]: + autoAck = True + session.setAutoAckingNodes(pynorm.TrackingStatus.RECEIVERS) + else: + ackingNodeList = [ int(i) for i in opts.ack] + + for ackingNondeID in ackingNodeList: + session.addAckingNode(ackingNondeID) + + if opts.txloss and opts.txloss>0: + session.setTxLoss(opts.txloss) + + if opts.rxloss and opts.rxloss>0: + session.setRxLoss(opts.rxloss) + + # Congestion Control + if opts.cc: + if isinstance(opts.cc,int): + session.setTxRate(opts.cc) + session.setEcnSupport(ecnEnable=False) + else: + if opts.cc == CCEnum.CC: # default TCP-friendly congestion control + session.setEcnSupport(ecnEnable=False) + elif opts.cc == CCEnum.CCE: #"wireless-ready" ECN-only congestion control + session.setEcnSupport(ecnEnable=True, ignoreLoss=True) + elif opts.cc == CCEnum.CCL: # "loss tolerant", non-ECN congestion control + session.setEcnSupport(ecnEnable=False, ignoreLoss=False, tolerateLoss=True) + + defaultBufferSpace = 64*1024*1024 + defaultTxSocketBufferSize = 4*1024*1024 + defaultRxSocketBufferSize = 6*1024*1024 + + bufferSpace = opts.buffer if opts.buffer else defaultBufferSpace + if opts.send: + if opts.ack: + session.setFlowControl(flowControlFactor=0) #// ack-based flow control enabled on command-line, so disable timer-based flow control + session.setBackoffFactor(0) + #FEC + session.startSender(sessionId=opts.id, + bufferSpace=bufferSpace, + segmentSize=opts.segment if opts.segment else 1400, + blockSize=opts.block if opts.block else 64, + numParity=opts.parity if opts.parity else 0, + fecId=0) + + if opts.auto: + session.setAutoParity(opts.auto) + session.setTxSocketBuffer(opts.txsockbuffer if opts.txsockbuffer else defaultTxSocketBufferSize ) + + if opts.recv: + session.startReceiver(bufferSpace=bufferSpace) + session.setRxSocketBuffer(opts.rxsockbuffer if opts.rxsockbuffer else defaultRxSocketBufferSize ) + + if opts.grttprobing: + session.setGrttProbingMode(pynorm.ProbingMode[ opts.grttprobing.value.upper() ] ) + + return session + +from pynorm import EventType + +from typing import Iterable +def listFiles( dirFileList:list[str]) -> Iterable[str]: + for dirFile in dirFileList: + if os.path.isfile(dirFile): + yield dirFile + elif os.path.isdir(dirFile): + for filename in os.scandir(dirFile): + if filename.is_file(): + yield filename.path + else: + raise FileExistsError(f"{dirFile} is not a valid file/dir") + +import logging +import ipaddress + +class NormCaster(): + def __init__(self,instance:pynorm.Instance, opts:argparse.Namespace ): + self.opts:argparse.Namespace = opts + self.recvDir:Optional[str] = os.path.expanduser(opts.recv) if opts.recv else None + + self.instance:pynorm.Instance = instance + + self.session:pynorm.Session = create_session(instance, opts) + self.fileIterator:Iterable[str] = listFiles(opts.send) + self.is_running:bool = True + self.pendingSendFilePath:Optional[str] = None # + + def addOneFile(self,session): + ''' + + ''' + if self.pendingSendFilePath is None: + try: + self.pendingSendFilePath:str = self.fileIterator.__next__() + except StopIteration: + self.is_running = False + return + + obj:Optional[pynorm.Object] = session.fileEnqueue(self.pendingSendFilePath, info= self.pendingSendFilePath.replace("\\","/").encode() ) + if obj: + if self.opts.ack: + session.setWatermark(obj,True) + logging.info(f"add file:{self.pendingSendFilePath} {obj._object}") + self.pendingSendFilePath = None # succeed enqued + else: + logging.warning(f"fileEnqueue: {sendFilePath} failure!") + + def handle_norm_event(self, event:pynorm.Event): + session:pynorm.Session = event.session + + evtType:pynorm.EventType = event.type + logging.info(evtType ) + if evtType in (EventType.TX_QUEUE_EMPTY, EventType.TX_QUEUE_VACANCY): + self.addOneFile(session) + elif evtType == EventType.GRTT_UPDATED: + pass + elif evtType == EventType.TX_WATERMARK_COMPLETED: + if pynorm.AckingStatus.SUCCESS == session.getAckingStatus(): + logging.warning( "normCast: NORM_TX_WATERMARK_COMPLETED, NORM_ACK_SUCCESS"); + self.addOneFile(session) + else: + logging.warning( "normCast: NORM_TX_WATERMARK_COMPLETED, _NOT_ NORM_ACK_SUCCESS"); + obj = event.object + if obj is None: + session.resetWatermark() + elif evtType == EventType.TX_FLUSH_COMPLETED: + pass + elif evtType == EventType.TX_OBJECT_PURGED: + obj = event.object + if obj and obj.type== pynorm.ObjectType.FILE: + logging.info(f"normCast: send file purged: {obj.info.decode()}") + elif evtType == EventType.TX_OBJECT_SENT: + obj = event.object + if obj and obj.type== pynorm.ObjectType.FILE: + logging.info(f"initial send complete: {obj.info.decode()}") + self.addOneFile(session) + elif evtType == EventType.ACKING_NODE_NEW: + + sender = event.sender + logging.info(f"normCast: new acking node: {sender.id} IP address{sender.address}") + elif evtType == EventType.REMOTE_SENDER_INACTIVE: + pass + elif evtType == EventType.RX_OBJECT_ABORTED: + obj = event.object + if obj and obj.type is not pynorm.ObjectType.FILE: + logging.error("normCast: received invalid object type?!") + return + filePath = event.object.filename + event.object.cancel() + #remove temparary file if recv aborted. + os.remove(filePath) + logging.info("") + + elif evtType == EventType.RX_OBJECT_INFO: + pass + elif evtType == EventType.RX_OBJECT_COMPLETED: + obj = event.object + if obj and obj.type is not pynorm.ObjectType.FILE: + logging.error("normCast: received invalid object type?!") + return + file_path = event.object.info.decode() + fileName = os.path.split(file_path)[-1] + path = os.path.join( self.recvDir, fileName ) + oldPath = event.object.filename + + logging.debug (f"{oldPath=}") + try: + if os.path.isfile(path): + os.remove(path) + os.rename(src=oldPath, dst=path) + except Exception as e: + logging.error ( traceback.format_exc() ) + + + def getAckNodesStatus(self,session): + isSuccess, nodeID, nodeAckStatus = session.getNextAckingNode() + while isSuccess: + ip=ipaddress.IPv4Address(nodeID) + if pynorm.AckingStatus.SUCCESS == nodeAckStatus: + logging.info( f"normCast: node {nodeID} (IP address: {str(ip)}) acnkowledged.") + else: + logging.info( f"normCast: node {nodeID} (IP address: {str(ip)}) failed to acnkowledge.") + isSuccess, nodeID, nodeAckStatus = session.getNextAckingNode() + + def run(self): + while self.is_running: + event: Optional[pynorm.Event,bool,None] = self.instance.getNextEvent(timeout=self.opts.repeat) + while event: + self.handle_norm_event(event) + event = self.instance.getNextEvent(timeout=self.opts.repeat) + if event is None and self.opts.send: + self.addOneFile(self.session) # means timeout + logging.warning("normCast exits!") + + +def main(argv): + opts:argparse.Namespace = get_arg_options() + instance = pynorm.Instance() + if opts.debug: + instance.setDebugLevel(level=pynorm.DebugLevel(opts.debug) ) + if opts.recv: + recvPath:str = os.path.expanduser( opts.recv ) + if not os.path.isdir(recvPath): #create is not exists! + os.makedirs( recvPath ) + instance.setCacheDirectory( recvPath ) + + normCast:NormCaster = NormCaster(instance, opts) + if opts.send: + normCast.addOneFile(normCast.session) + + normCast.run() + + + + +if __name__ == '__main__': + logging.getLogger().setLevel(logging.INFO) + sys.exit(main(sys.argv)) diff --git a/examples/python/normMsgr.py b/examples/python/normMsgr.py index 6661280..b31501c 100755 --- a/examples/python/normMsgr.py +++ b/examples/python/normMsgr.py @@ -14,20 +14,19 @@ class InputThread(Thread): def __init__(self, parent, *args, **kwargs): super(InputThread, self).__init__(*args, **kwargs) - #self.setDaemon(True) ;# this is "child" daemon thread (setDaemon is deprecated) self.daemon = True self.msgr = parent def run(self): while True: try: - msgHdr = bytearray(sys.stdin.read(MSG_HDR_SIZE)) + msgHdr = sys.stdin.buffer.read(MSG_HDR_SIZE) except: sys.stderr.write("normMsgr: input thread end-of-file 1 ...\n") return try: msgSize = 256*int(msgHdr[0]) + int(msgHdr[1]) - msgBuffer = sys.stdin.read(msgSize - 2) + msgBuffer = sys.stdin.buffer.read(msgSize - 2) except: sys.stderr.write("normMsgr: input thread end-of-file 2 ...\n") return @@ -38,7 +37,7 @@ class OutputThread(Thread): def __init__(self, parent, *args, **kwargs): super(OutputThread, self).__init__(*args, **kwargs) - self.setDaemon(True) ;# this is "child" daemon thread + self.daemon = True self.msgr = parent def run(self): @@ -48,11 +47,15 @@ class OutputThread(Thread): msgHeader = bytearray(MSG_HDR_SIZE) msgHeader[0] = (msgLen >> 8) & 0x00ff msgHeader[1] = msgLen & 0x00ff - sys.stdout.write(msgHeader) - sys.stdout.write(msg) + try: + sys.stdout.buffer.write(msgHeader) + sys.stdout.buffer.write(msg) + sys.stdout.flush() + except: + sys.stderr.write("normMsgr: output thread exiting ...\n") + return del msg - class NormMsgr: """This class keeps state for NORM tx/rx operations""" @@ -79,10 +82,12 @@ class NormMsgr: # Create a NormSession and set some default parameters self.normSession = self.normInstance.createSession(addr, port, nodeId) self.normSession.setRxCacheLimit(2*self.norm_tx_queue_max) ;# we let the receiver track some extra objects - self.normSession.setDefaultSyncPolicy(pynorm.NORM_SYNC_ALL); + self.normSession.setDefaultSyncPolicy(pynorm.SyncPolicy.ALL); self.normSession.setDefaultUnicastNack(True); self.normSession.setTxCacheBounds(10*1024*1024, self.norm_tx_queue_max, self.norm_tx_queue_max); self.normSession.setCongestionControl(True, True); + self.normSession.setRxPortReuse(True) + self.normSession.setMulticastLoopback(True) return self.normSession def addAckingNode(self, nodeId): @@ -204,7 +209,7 @@ class NormMsgr: def onNormRxObjectCompleted(self, obj): with self.normRxLock: - if pynorm.NORM_OBJECT_DATA == obj.getType(): + if pynorm.ObjectType.DATA == obj.getType(): if 0 != len(self.output_msg_queue): wasEmpty = False else: @@ -250,12 +255,12 @@ class NormEventHandler(Thread): return if event is None: break - if pynorm.NORM_EVENT_INVALID == event.type: + if pynorm.EventType.EVENT_INVALID == event.type: continue - elif pynorm.NORM_TX_QUEUE_EMPTY == event.type or pynorm.NORM_TX_QUEUE_VACANCY == event.type: + elif pynorm.EventType.TX_QUEUE_EMPTY == event.type or pynorm.EventType.TX_QUEUE_VACANCY == event.type: msgr.onNormTxQueueVacancy() - elif pynorm.NORM_TX_WATERMARK_COMPLETED == event.type: - if pynorm.NORM_ACK_SUCCESS == event.session.getAckingStatus(): + elif pynorm.EventType.TX_WATERMARK_COMPLETED == event.type: + if pynorm.EventType.ACK_SUCCESS == event.session.getAckingStatus(): # All receivers acknowledged msgr.onNormTxWatermarkCompleted() else: @@ -263,12 +268,12 @@ class NormEventHandler(Thread): # from our acking list. For now, we are infinitely # persistent by resetting watermark ack request event.session.resetWatermark() - elif pynorm.NORM_TX_OBJECT_PURGED == event.type: + elif pynorm.EventType.TX_OBJECT_PURGED == event.type: msgr.onNormTxObjectPurged(event.object) - elif pynorm.NORM_RX_OBJECT_COMPLETED == event.type: + elif pynorm.EventType.RX_OBJECT_COMPLETED == event.type: msgr.onNormRxObjectCompleted(event.object) #else: - # sys.stderr.write("normMsgr: NormEventHandler warning: unhandled event: %s\n" % str(event)) + #sys.stderr.write("normMsgr: NormEventHandler warning: unhandled event: %s\n" % pynorm.EventType(event.type)) sys.stderr.write("normMsgr: NormEventHandler thread exiting ...\n"); self.lock.release() @@ -395,4 +400,9 @@ except KeyboardInterrupt: #sys.stderr.write("exception while waiting on input thread ..\n"); pass +# TBD - shut the NormEventHandler thread down in a graceful way +if send: + inputThread.join() +if recv: + outputThread.join() sys.stderr.write("normMsgr: Done.\n") diff --git a/src/pynorm/__init__.py b/src/pynorm/__init__.py index 046e585..625af23 100755 --- a/src/pynorm/__init__.py +++ b/src/pynorm/__init__.py @@ -6,9 +6,12 @@ By: Tom Wambold from pynorm.instance import Instance from pynorm.core import libnorm, NormError from pynorm.constants import * +from pynorm.event import Event +from pynorm.session import Session +from pynorm.object import Object -def setDebugLevel(level): - libnorm.NormSetDebugLevel(level) +def setDebugLevel(level: DebugLevel): + libnorm.NormSetDebugLevel(level.value) -def getDebugLevel(): - return libnorm.NormGetDebugLevel() +def getDebugLevel() -> DebugLevel: + return DebugLevel( libnorm.NormGetDebugLevel() ) diff --git a/src/pynorm/constants.py b/src/pynorm/constants.py index f4d06d3..c7e4d0a 100755 --- a/src/pynorm/constants.py +++ b/src/pynorm/constants.py @@ -7,75 +7,117 @@ from __future__ import absolute_import import ctypes +import enum from pynorm.core import libnorm +#enum ProtoDebugLevel {PL_FATAL, PL_ERROR, PL_WARN, PL_INFO, PL_DEBUG, PL_TRACE, PL_DETAIL, PL_MAX, PL_ALWAYS}; +class DebugLevel(enum.IntEnum): + FATAL = 0 + ERROR = 1 + WARNNING = 2 + INFO =3 + DEBUG =4 + TRACE =5 + DETAIL=6 + MAX =7 + ALWAYS=8 # Constants # enum NormObjectType -NORM_OBJECT_NONE = 0 -NORM_OBJECT_DATA = 1 -NORM_OBJECT_FILE = 2 -NORM_OBJECT_STREAM = 3 +class ObjectType(enum.Enum): + NONE = 0 + DATA = 1 + FILE = 2 + STREAM = 3 # enum NormFlushMode -NORM_FLUSH_NONE = 0 -NORM_FLUSH_PASSIVE = 1 -NORM_FLUSH_ACTIVE = 2 +class FlushMode(enum.Enum): + ''' + NormStreamFlush + NormStreamSetAutoFlush + ''' + NONE = 0 + PASSIVE = 1 + ACTIVE = 2 # enum NormNackingMode -NORM_NACK_NONE = 0 -NORM_NACK_INFO_ONLY = 1 -NORM_NACK_NORMAL = 2 +class NackingMode(enum.Enum): + ''' + only used for NormSetDefaultNackingMode + ''' + NONE = 0 + INFO_ONLY = 1 + NORMAL = 2 # enum NormAckingStatus -NORM_ACK_INVALID = 0 -NORM_ACK_FAILURE = 1 -NORM_ACK_PENDING = 2 -NORM_ACK_SUCCESS = 3 +class AckingStatus(enum.Enum): + ''' + the return value of NormGetAckingStatus + ''' + INVALID = 0 + FAILURE = 1 + PENDING = 2 + SUCCESS = 3 # enum NormProbingMode -NORM_PROBE_NONE = 0 -NORM_PROBE_PASSIVE = 1 -NORM_PROBE_ACTIVE = 2 +class ProbingMode(enum.Enum): + ''' + used in NormSetGrttProbingMode + ''' + NONE = 0 + PASSIVE = 1 + ACTIVE = 2 # enum NormSyncPolicy -NORM_SYNC_CURRENT = 0 -NORM_SYNC_STREAM = 1 -NORM_SYNC_ALL = 2 -# enum NormRepairBoundary -NORM_BOUNDARY_BLOCK = 0 -NORM_BOUNDARY_OBJECT = 1 +class SyncPolicy(enum.Enum): + ''' + used in NormSetDefaultSyncPolicy + ''' + CURRENT = 0 + STREAM = 1 + ALL = 2 + +class TrackingStatus(enum.Enum): #enum NormTrackingStatus + NONE = 0 + RECEIVERS = 1 + SENDERS = 2 + ALL = 3 + +class RepairBoundary(enum.Enum): + BLOCK = 0 + OBJECT = 1 # enum NormEventType -NORM_EVENT_INVALID = 0 -NORM_TX_QUEUE_VACANCY = 1 -NORM_TX_QUEUE_EMPTY = 2 -NORM_TX_FLUSH_COMPLETED = 3 -NORM_TX_WATERMARK_COMPLETED = 4 -NORM_TX_CMD_SENT = 5 -NORM_TX_OBJECT_SENT = 6 -NORM_TX_OBJECT_PURGED = 7 -NORM_TX_RATE_CHANGED = 8 -NORM_LOCAL_SENDER_CLOSED = 9 -NORM_REMOTE_SENDER_NEW = 10 -NORM_REMOTE_SENDER_RESET = 11 -NORM_REMOTE_SENDER_ADDRESS = 12 -NORM_REMOTE_SENDER_ACTIVE = 13 -NORM_REMOTE_SENDER_INACTIVE = 14 -NORM_REMOTE_SENDER_PURGED = 15 -NORM_RX_CMD_NEW = 16 -NORM_RX_OBJECT_NEW = 17 -NORM_RX_OBJECT_INFO = 18 -NORM_RX_OBJECT_UPDATED = 19 -NORM_RX_OBJECT_COMPLETED = 20 -NORM_RX_OBJECT_ABORTED = 21 -NORM_RX_ACK_REQUEST = 22 -NORM_GRTT_UPDATED = 23 -NORM_CC_ACTIVE = 24 -NORM_CC_INACTIVE = 25 -NORM_ACKING_NODE_NEW = 26 -NORM_SEND_ERROR = 27 -NORM_USER_TIMEOUT = 28 +class EventType(enum.Enum): + EVENT_INVALID = 0 + TX_QUEUE_VACANCY = 1 + TX_QUEUE_EMPTY = 2 + TX_FLUSH_COMPLETED = 3 + TX_WATERMARK_COMPLETED = 4 + TX_CMD_SENT = 5 + TX_OBJECT_SENT = 6 + TX_OBJECT_PURGED = 7 + TX_RATE_CHANGED = 8 + LOCAL_SENDER_CLOSED = 9 + REMOTE_SENDER_NEW = 10 + REMOTE_SENDER_RESET = 11 + REMOTE_SENDER_ADDRESS = 12 + REMOTE_SENDER_ACTIVE = 13 + REMOTE_SENDER_INACTIVE = 14 + REMOTE_SENDER_PURGED = 15 + RX_CMD_NEW = 16 + RX_OBJECT_NEW = 17 + RX_OBJECT_INFO = 18 + RX_OBJECT_UPDATED = 19 + RX_OBJECT_COMPLETED = 20 + RX_OBJECT_ABORTED = 21 + RX_ACK_REQUEST = 22 + GRTT_UPDATED = 23 + CC_ACTIVE = 24 + CC_INACTIVE = 25 + ACKING_NODE_NEW = 26 + SEND_ERROR = 27 + USER_TIMEOUT = 28 NORM_INSTANCE_INVALID = ctypes.c_void_p.in_dll(libnorm, "NORM_INSTANCE_INVALID").value NORM_SESSION_INVALID = ctypes.c_void_p.in_dll(libnorm, "NORM_SESSION_INVALID").value diff --git a/src/pynorm/core.py b/src/pynorm/core.py index 933d89f..7467a52 100755 --- a/src/pynorm/core.py +++ b/src/pynorm/core.py @@ -169,6 +169,10 @@ def get_libnorm(): libnorm.NormSetTOS.argtypes = [ctypes.c_void_p, ctypes.c_uint8] libnorm.NormSetTOS.errcheck = errcheck_bool + libnorm.NormSetMulticastLoopback.restype = ctypes.c_bool + libnorm.NormSetMulticastLoopback.argtypes = [ctypes.c_void_p, ctypes.c_byte] + libnorm.NormSetMulticastLoopback.errcheck = errcheck_bool + libnorm.NormSetLoopback.restype = ctypes.c_bool libnorm.NormSetLoopback.argtypes = [ctypes.c_void_p, ctypes.c_byte] libnorm.NormSetLoopback.errcheck = errcheck_bool @@ -321,6 +325,7 @@ def get_libnorm(): libnorm.NormStreamMarkEom.restype = None libnorm.NormStreamMarkEom.argtypes = [ctypes.c_void_p] + libnorm.NormSetWatermark.restype = ctypes.c_bool libnorm.NormSetWatermark.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_bool] libnorm.NormSetWatermark.errcheck = errcheck_bool @@ -336,9 +341,16 @@ def get_libnorm(): libnorm.NormRemoveAckingNode.restype = None libnorm.NormRemoveAckingNode.argtypes = [ctypes.c_void_p, ctypes.c_uint32] + + libnorm.NormSetAutoAckingNodes.restype = None + libnorm.NormSetAutoAckingNodes.argtypes = [ctypes.c_void_p, ctypes.c_int] + libnorm.NormGetAckingStatus.restype = ctypes.c_int libnorm.NormGetAckingStatus.argtypes = [ctypes.c_void_p, ctypes.c_uint32] + + libnorm.NormGetNextAckingNode.restype = ctypes.c_bool + libnorm.NormGetNextAckingNode.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p] libnorm.NormSendCommand.restype = ctypes.c_bool libnorm.NormSendCommand.argtypes = [ctypes.c_void_p, ctypes.c_char_p, diff --git a/src/pynorm/event.py b/src/pynorm/event.py index 3db3f7b..aced43a 100755 --- a/src/pynorm/event.py +++ b/src/pynorm/event.py @@ -13,17 +13,17 @@ from pynorm.node import Node from pynorm.object import Object class Event(object): - def __init__(self, eventType, session, sender, normObject): - self._type = eventType - self._session = session - self._sender = sender - self._object = normObject + def __init__(self, eventType:c.EventType, session:Session, sender, normObject): + self._type:c.EventType = c.EventType(eventType) + self._session:Session = session + self._sender:Node = sender + self._object:Object = normObject # Properties - type = property(lambda self: self._type) - session = property(lambda self: self._session) - sender = property(lambda self: self._sender) - object = property(lambda self: self._object) + type:c.EventType = property(lambda self: self._type) + session:Session = property(lambda self: self._session) + sender:Node = property(lambda self: self._sender) + object:Object = property(lambda self: self._object) ## Private functions @property @@ -41,63 +41,4 @@ class Event(object): return cmp(self.type, type) def __str__(self): - if self.type == c.NORM_EVENT_INVALID: - return "NORM_EVENT_INVALID" - elif self.type == c.NORM_TX_QUEUE_VACANCY: - return "NORM_TX_QUEUE_VACANCY" - elif self.type == c.NORM_TX_QUEUE_EMPTY: - return "NORM_TX_QUEUE_EMPTY" - elif self.type == c.NORM_TX_FLUSH_COMPLETED: - return "NORM_TX_FLUSH_COMPLETED" - elif self.type == c.NORM_TX_WATERMARK_COMPLETED: - return "NORM_TX_WATERMARK_COMPLETED" - elif self.type == c.NORM_TX_CMD_SENT: - return "NORM_TX_CMD_SENT" - elif self.type == c.NORM_TX_OBJECT_SENT: - return "NORM_TX_OBJECT_SENT" - elif self.type == c.NORM_TX_OBJECT_PURGED: - return "NORM_TX_OBJECT_PURGED" - elif self.type == c.NORM_TX_RATE_CHANGED: - return "NORM_TX_RATE_CHANGED" - elif self.type == c.NORM_LOCAL_SENDER_CLOSED: - return "NORM_LOCAL_SENDER_CLOSED" - elif self.type == c.NORM_REMOTE_SENDER_NEW: - return "NORM_REMOTE_SENDER_NEW" - elif self.type == c.NORM_REMOTE_SENDER_RESET: - return "NORM_REMOTE_SENDER_RESET" - elif self.type == c.NORM_REMOTE_SENDER_ADDRESS: - return "NORM_REMOTE_SENDER_ADDRESS" - elif self.type == c.NORM_REMOTE_SENDER_ACTIVE: - return "NORM_REMOTE_SENDER_ACTIVE" - elif self.type == c.NORM_REMOTE_SENDER_INACTIVE: - return "NORM_REMOTE_SENDER_INACTIVE" - elif self.type == c.NORM_REMOTE_SENDER_PURGED: - return "NORM_REMOTE_SENDER_PURGED" - elif self.type == c.NORM_RX_CMD_NEW: - return "NORM_RX_CMD_NEW" - elif self.type == c.NORM_RX_OBJECT_NEW: - return "NORM_RX_OBJECT_NEW" - elif self.type == c.NORM_RX_OBJECT_INFO: - return "NORM_RX_OBJECT_INFO" - elif self.type == c.NORM_RX_OBJECT_UPDATED: - return "NORM_RX_OBJECT_UPDATED" - elif self.type == c.NORM_RX_OBJECT_COMPLETED: - return "NORM_RX_OBJECT_COMPLETED" - elif self.type == c.NORM_RX_OBJECT_ABORTED: - return "NORM_RX_OBJECT_ABORTED" - elif self.type == c.NORM_RX_ACK_REQUEST: - return "NORM_RX_ACK_REQUEST" - elif self.type == c.NORM_GRTT_UPDATED: - return "NORM_GRTT_UPDATED" - elif self.type == c.NORM_CC_ACTIVE: - return "NORM_CC_ACTIVE" - elif self.type == c.NORM_CC_INACTIVE: - return "NORM_CC_INACTIVE" - elif self.type == c.NORM_ACKING_NODE_NEW: - return "NORM_ACKING_NODE_NEW" - elif self.type == c.NORM_SEND_ERROR: - return "NORM_SEND_ERROR" - elif self.type == c.NORM_USER_TIMEOUT: - return "NORM_USER_TIMEOUT" - else: - return "Unknown event type" + return self._type.name diff --git a/src/pynorm/instance.py b/src/pynorm/instance.py index f6ca6a5..342a1ca 100755 --- a/src/pynorm/instance.py +++ b/src/pynorm/instance.py @@ -47,60 +47,63 @@ class Instance(object): def restart(self): libnorm.NormRestartInstance(self) - def setCacheDirectory(self, path): - libnorm.NormSetCacheDirectory(self, path.encode('utf-8')) - - def setDebugLevel(self, level): - libnorm.NormSetDebugLevel(level) + def setCacheDirectory(self, path:str): + libnorm.NormSetCacheDirectory(self, path.encode()) - def openDebugLog(self, path): - libnorm.NormOpenDebugLog(self, path.encode('utf-8')) + def setDebugLevel(self, level:c.DebugLevel): + libnorm.NormSetDebugLevel(int(level)) + + def openDebugLog(self, path:str): + libnorm.NormOpenDebugLog(self, path.encode()) def closeDebugLog(self): libnorm.NormCloseDebugLog(self) - def openDebugPipe(self, pipeName): - libnorm.NormOpenDebugPipe(self, pipeName.encode('utf-8')) + def openDebugPipe(self, pipeName:str): + libnorm.NormOpenDebugPipe(self, pipeName.encode()) def getNextEvent(self, timeout=None): # Use python's select because letting the C NormGetNextEvent block # seems to stop signals (CTRL+C) from killing the process if not self._select(timeout): return None - result = libnorm.NormGetNextEvent(self, ctypes.byref(self._estruct), True) - if not result: sys.stderr.write("NormInstance.getNextEvent() warning: no more NORM events\n") return False # Note a NORM_EVENT_INVALID can be OK (with NORM_SESSION_INVALID) - if self._estruct.type == c.NORM_EVENT_INVALID: - return Event(c.NORM_EVENT_INVALID, None, None, None) - + if self._estruct.type == c.EventType.EVENT_INVALID: + return Event(c.EventType.EVENT_INVALID, None, None, None) + if self._estruct.session == c.NORM_SESSION_INVALID: raise NormError("No new event") - - try: - sender = self._senders[self._estruct.sender] - except KeyError: - sender = self._senders[self._estruct.sender] = Node(self._estruct.sender) - try: - obj = self._objects[self._estruct.object] - except KeyError: - obj = self._objects[self._estruct.object] = Object(self._estruct.object) + if self._estruct.sender: + try: + sender = self._senders[self._estruct.sender] + except KeyError: + sender = self._senders[self._estruct.sender] = Node(self._estruct.sender) + else: + sender = None + if self._estruct.object: + try: + obj = self._objects[self._estruct.object] + except KeyError: + obj = self._objects[self._estruct.object] = Object(self._estruct.object) + else: + obj = None return Event(self._estruct.type, self._sessions[self._estruct.session], sender, obj) - def getDescriptor(self): + def getDescriptor(self) -> int: return libnorm.NormGetDescriptor(self) - def fileno(self): + def fileno(self) -> int: return libnorm.NormGetDescriptor(self) - def createSession(self, address, port, localId=None): + def createSession(self, address:str, port:int, localId:int=None): if localId == None: localId = c.NORM_NODE_ANY - session = Session(self, address, port, localId) + session:Session = Session(self, address, port, localId) self._sessions[session._session] = session return session @@ -118,7 +121,7 @@ class Instance(object): else: # Windows wants milliseconds... timeout *= 1000 - rv = win32event.WaitForSingleObject(self.getDescriptor(), timeout) + rv = win32event.WaitForSingleObject(self.getDescriptor(), int(timeout) ) return True if rv == win32event.WAIT_OBJECT_0 else False def _select_everythingelse(self, timeout): @@ -143,6 +146,8 @@ class Instance(object): return __next__() def __cmp__(self, other): + def cmp(a, b): + return (a > b) - (a < b) return cmp(self._as_parameter_, other._as_parameter) def __hash__(self): diff --git a/src/pynorm/node.py b/src/pynorm/node.py index d5f165e..4ceb9ac 100755 --- a/src/pynorm/node.py +++ b/src/pynorm/node.py @@ -7,6 +7,8 @@ from __future__ import absolute_import import ctypes +from typing import Tuple +import ipaddress import pynorm.constants as c from pynorm.core import libnorm, NormError @@ -18,13 +20,13 @@ class Node(object): libnorm.NormNodeRetain(node) self._node = node - def getId(self): + def getId(self) -> int: id = libnorm.NormNodeGetId(self) if id == c.NORM_NODE_ANY: return None return id - def getAddress(self): + def getAddress(self) -> (str,int): try: return self._address except AttributeError: @@ -34,31 +36,35 @@ class Node(object): if not libnorm.NormNodeGetAddress(self, buf, ctypes.byref(size), ctypes.byref(port)): raise NormError("Node getAddress failed") - self._address = (buf.value, port.value) + if len(buf.value)==4: + ip = ipaddress.IPv4Address(buf.value) + elif len(buf.value) == 16: #ipv6 is 128bit + ip = ipaddress.IPv6Address(buf.value) + self._address = (str(ip), port.value) return self._address - def getCommand(self, buf): + def getCommand(self, buf:bytes) -> bool: return libnorm.NormNodeGetCommand(self, buf, len(buf)) - def getGrtt(self): + def getGrtt(self) -> float: grtt = libnorm.NormNodeGetGrtt(self) if grtt == -1.0: raise NormError("getGrtt failed") return grtt - def setUnicastNack(self, mode): + def setUnicastNack(self, mode:bool): libnorm.NormNodeSetUnicastNack(self, mode) - def setNackingMode(self, mode): - libnorm.NormNodeSetNackingMode(self, mode) + def setNackingMode(self, mode:c.NackingMode): + libnorm.NormNodeSetNackingMode(self, mode.value) - def setRepairBoundary(self, boundary): - libnorm.NormNodeSetRepairBoundary(self, boundary) + def setRepairBoundary(self, boundary:c.RepairBoundary): + libnorm.NormNodeSetRepairBoundary(self, boundary.value) ## Properties - id = property(getId) - address = property(getAddress) - grtt = property(getGrtt) + id:int = property(getId) + address:Tuple[str,int] = property(getAddress) + grtt:float = property(getGrtt) ## Private Functions def __del__(self): @@ -72,6 +78,8 @@ class Node(object): return "Node - Id=%i" % self.id def __cmp__(self, other): + def cmp(a, b): + return (a > b) - (a < b) return cmp(self._as_parameter_, other._as_parameter_) def __hash__(self): diff --git a/src/pynorm/object.py b/src/pynorm/object.py index b3673a9..aa8a359 100755 --- a/src/pynorm/object.py +++ b/src/pynorm/object.py @@ -4,23 +4,24 @@ By: Tom Wambold """ from __future__ import absolute_import - import ctypes - +import locale import pynorm.constants as c from pynorm.core import libnorm, NormError from pynorm.node import Node class Object(object): """Represents a NORM object instance""" + locale_encoding:str = locale.getpreferredencoding() ## Public functions - def __init__(self, object): - libnorm.NormObjectRetain(object) - self._object = object + def __init__(self, object_id:int): + libnorm.NormObjectRetain(object_id) + self._object:int = object_id # type NormObjectHandle - def getType(self): - return libnorm.NormObjectGetType(self) + def getType(self) -> c.ObjectType: + value = libnorm.NormObjectGetType(self) + return c.ObjectType(value) def hasObjectInfo(self): return libnorm.NormObjectHasInfo(self) @@ -37,23 +38,25 @@ class Object(object): raise NormError("No object info received yet.") return buf.value - def getSize(self): + def getSize(self) -> int: return libnorm.NormObjectGetSize(self) - def getBytesPending(self): + def getBytesPending(self) -> int: return libnorm.NormObjectGetBytesPending(self) - def cancel(self): + def cancel(self) -> None: libnorm.NormObjectCancel(self) - def getFileName(self): - # TBD - should we do something to get file name size first? + def getFileName(self) -> str: buf = ctypes.create_string_buffer(256) - libnorm.NormFileGetName(self, buf, ctypes.sizeof(buf)) - return buf.value + success = libnorm.NormFileGetName(self, buf, ctypes.sizeof(buf)) + if success: + return buf.value.decode(Object.locale_encoding) + else: + return None - def renameFile(self, name): - libnorm.NormFileRename(self, name.encode('utf-8')) + def renameFile(self, name:str) -> bool: + return libnorm.NormFileRename(self, name.encode(Object.locale_encoding)) # Because 'ctypes.string_at()' makes a _copy_ of the data, we don't # support the usual NORM accessData / detachData options. We use @@ -71,23 +74,23 @@ class Object(object): def getSender(self): return Node(libnorm.NormObjectGetSender(self)) - def setNackingMode(self, mode): - libnorm.NormObjectSetNackingMode(self, mode) + def setNackingMode(self, mode:c.NackingMode): + libnorm.NormObjectSetNackingMode(self, mode.value) ## Stream sending functions def streamClose(self, graceful=False): libnorm.NormStreamClose(self, graceful) - def streamWrite(self, msg): - return libnorm.NormStreamWrite(self, msg.encode('utf-8'), len(msg)) + def streamWrite(self, msg:bytes): + return libnorm.NormStreamWrite(self, msg, len(msg)) - def streamFlush(self, eom=False, flushmode=c.NORM_FLUSH_PASSIVE): - libnorm.NormStreamFlush(self, eom, flushmode) + def streamFlush(self, eom=False, flushmode:c.FlushMode = c.FlushMode.PASSIVE): + libnorm.NormStreamFlush(self, eom, flushmode.value) - def streamSetAutoFlush(self, mode): - libnorm.NormStreamSetAutoFlush(self, mode) + def streamSetAutoFlush(self, flushMode:c.FlushMode): + libnorm.NormStreamSetAutoFlush(self, flushMode.value) - def streamPushEnable(self, push): + def streamPushEnable(self, push:bool): libnorm.NormStreamSetPushEnable(self, push) def streamHasVacancy(self): @@ -97,11 +100,11 @@ class Object(object): libnorm.NormStreamMarkEom(self) ## Stream receiving functions - def streamRead(self, size): + def streamRead(self, size) -> (int,bytes): buf = ctypes.create_string_buffer(size) numBytes = ctypes.c_uint(ctypes.sizeof(buf)) libnorm.NormStreamRead(self, buf, ctypes.byref(numBytes)) - return (numBytes, buf.value) + return (numBytes.value, buf) def streamSeekMsgStart(self): return libnorm.NormStreamSeekMsgStart(self) @@ -110,32 +113,34 @@ class Object(object): return libnorm.NormStreamGetReadOffset(self) ## Properties - type = property(getType) - info = property(getInfo) - size = property(getSize) - bytesPending = property(getBytesPending) - filename = property(getFileName, renameFile) - sender = property(getSender) + type:c.ObjectType = property(getType) + info:bytes = property(getInfo) + size:int = property(getSize) + bytesPending:int = property(getBytesPending) + filename:bytes = property(getFileName, renameFile) + sender:Node = property(getSender) + handle:int = property( lambda self:self._object ) ## Private functions + def __copy__(self): + return Object(self._object) + + def __deepcopy__(self, memo): + return Object(self._object) + def __del__(self): - libnorm.NormObjectRelease(self._object) + libnorm.NormObjectRelease(self) @property def _as_parameter_(self): return self._object def __str__(self): - if self.type == c.NORM_OBJECT_DATA: - return "NORM_OBJECT_DATA" - elif self.type == c.NORM_OBJECT_FILE: - return "NORM_OBJECT_FILE" - elif self.type == c.NORM_OBJECT_STREAM: - return "NORM_OBJECT_STREAM" - elif self.type == c.NORM_OBJECT_NONE: - return "NORM_OBJECT_NONE" + return self.type.name def __cmp__(self, other): + def cmp(a, b): + return (a > b) - (a < b) return cmp(self._as_parameter_, other._as_parameter_) def __hash__(self): diff --git a/src/pynorm/session.py b/src/pynorm/session.py index c2f32d8..1b3463b 100755 --- a/src/pynorm/session.py +++ b/src/pynorm/session.py @@ -6,16 +6,18 @@ By: Tom Wambold from __future__ import absolute_import import ctypes - +import locale +from typing import Optional import pynorm.constants as c from pynorm.core import libnorm, NormError from pynorm.object import Object class Session(object): """This represents a session tied to a particular NORM instance""" + locale_encoding:str = locale.getpreferredencoding() ## Public functions - def __init__(self, instance, address, port, localId=c.NORM_NODE_ANY): + def __init__(self, instance, address:str, port:int, localId=c.NORM_NODE_ANY): """ instance - An instance of NormInstance address - String of multicast address to join @@ -23,7 +25,7 @@ class Session(object): localId - NormNodeId """ self._instance = instance - self._session = libnorm.NormCreateSession(instance, address.encode('utf-8'), port, localId) + self._session:int = libnorm.NormCreateSession(instance, address.encode('utf-8'), port, localId) self.sendGracefulStop = False self.gracePeriod = 0 @@ -31,195 +33,253 @@ class Session(object): libnorm.NormDestroySession(self) del self._instance._sessions[self] - def setUserData(self, data): + def setUserData(self, data:str): """data should be a string""" libnorm.NormSetUserData(self, data.encode('utf-8')) - def getUserData(self): + def getUserData(self)->str: data = libnorm.NormGetUserData(self) return data.decode('utf-8') if data else None def getNodeId(self): return libnorm.NormGetLocalNodeId(self) - def setTxPort(self, txPort, enableReuse=False, txBindAddr=None): - libnorm.NormSetTxPort(self, txPort, enableReuse, txBindAddr) + def setTxPort(self, txPort:int, enableReuse:bool=False, txBindAddr:Optional[str]=None): + libnorm.NormSetTxPort(self, txPort, enableReuse, + txBindAddr.encode('utf-8') if txBindAddr else None ) - def setRxPortReuse(self, enable, rxBindAddr=None, senderAddr=None, senderPort=0): - libnorm.NormSetRxPortReuse(self, enable, rxBindAddr.encode('utf-8'), senderAddr.encode('utf-8'), senderPort) + def setTxOnly(self, txOnly:bool=False, connectToSessionAddress:bool=False): + libnorm.NormSetTxOnly(self, txOnly, connectToSessionAddress) - def setMulticastInterface(self, iface): + + def setRxPortReuse(self, enable:bool, rxBindAddr:Optional[str]=None, senderAddr:Optional[str]=None, senderPort:int=0): + ''' + This function allows the user to control the port reuse and binding behavior for the receive socket used for the given NORM sessionHandle. + When the enablReuse parameter is set to true, reuse of the NormSession port number by multiple NORM instances or sessions is enabled. + ''' + libnorm.NormSetRxPortReuse(self, enable, + rxBindAddr.encode('utf-8') if rxBindAddr else None, + senderAddr.encode('utf-8') if senderAddr else None, + senderPort) + + def setMulticastInterface(self, iface:str): libnorm.NormSetMulticastInterface(self, iface.encode('utf-8')) - - def setSSM(self, srcAddr): + + def setSSM(self, srcAddr:str): libnorm.NormSetSSM(self, srcAddr.encode('utf-8')) - def setTTL(self, ttl): - libnorm.NormSetTTL(self, ttl) + def setTTL(self, ttl) -> bool: + return libnorm.NormSetTTL(self, ttl) - def setTOS(self, tos): - libnorm.NormSetTOS(self, tos) + def setTOS(self, tos) -> bool: + return libnorm.NormSetTOS(self, tos) - def setLoopback(self, loop): - libnorm.NormSetLoopback(self, loop) + def setLoopback(self, loopbackEnable:bool=False): + libnorm.NormSetLoopback(self, loopbackEnable) + + def setMulticastLoopback(self, enable:bool): + libnorm.NormSetMulticastLoopback(self, enable) + + def setTxLoss(self, percent:float): + libnorm.NormSetTxLoss(percent) + + def setRxLoss(self, percent:float): + libnorm.NormSetRxLoss(percent) - def getReportInterval(self): + def getReportInterval(self) -> int: return libnorm.NormGetReportInterval(self) - def setReportInterval(self, interval): + def setReportInterval(self, interval:int): libnorm.NormSetReportInterval(self, interval) ## Sender functions - def startSender(self, sessionId, bufferSpace, segmentSize, blockSize, numParity, fecId=0): - libnorm.NormStartSender(self, sessionId, bufferSpace, segmentSize, blockSize, numParity, fecId) + def startSender(self, sessionId:int, bufferSpace:int, segmentSize:int, blockSize:int, numParity:int, fecId=0) -> bool: + return libnorm.NormStartSender(self, sessionId, bufferSpace, segmentSize, blockSize, numParity, fecId) def stopSender(self): libnorm.NormStopSender(self) - def setTxRate(self, rate): - libnorm.NormSetTxRate(self, rate) + def setTxRate(self, txRate:float): + libnorm.NormSetTxRate(self, txRate) - def setTxSocketBuffer(self, size): + def getTxRate(self) -> float: + return libnorm.NormGetTxRate(self) + + def setTxSocketBuffer(self, size:int): libnorm.NormSetTxSocketBuffer(self, size) - def setCongestionControl(self, ccEnable, adjustRate=True): + def setCongestionControl(self, ccEnable:bool, adjustRate:bool=True): + ''' + must called before startSender + ''' libnorm.NormSetCongestionControl(self, ccEnable, adjustRate) - + def setEcnSupport(self, ecnEnable, ignoreLoss=False, tolerateLoss=False): libnorm.NormSetEcnSupport(self, ecnEnable, ignoreLoss, tolerateLoss) - + def setFlowControl(self, flowControlFactor): libnorm.NormSetFlowControl(self, flowControlFactor) - def setTxRateBounds(self, rateMin, rateMax): + def setTxRateBounds(self, rateMin, rateMax) -> bool: + ''' + If both rateMin and rateMax are greater than or equal to zero, + but (rateMax < rateMin), the rate bounds will remain unset or unchanged and the function will return false. + ''' + if rateMin > rateMax: + return False libnorm.NormSetTxRateBounds(self, rateMin, rateMax) + return True - def setTxCacheBounds(self, sizeMax, countMin, countMax): + def setTxCacheBounds(self, sizeMax:int, countMin:int, countMax:int): libnorm.NormSetTxCacheBounds(self, sizeMax, countMin, countMax) - def setAutoParity(self, parity): + def setAutoParity(self, parity:int): libnorm.NormSetAutoParity(self, parity) - def getGrttEstimate(self): + def getGrttEstimate(self) -> float: return libnorm.NormGetGrttEstimate(self) - def setGrttEstimate(self, grttMax): - libnorm.NormSetGrttEstimate(self, grtt) + def setGrttEstimate(self, grtt:float): + return libnorm.NormSetGrttEstimate(self, grtt) - def setGrttMax(self, max): - libnorm.NormSetGrttMax(self, grttMax) + def setGrttMax(self, grttMax:float): + return libnorm.NormSetGrttMax(self, grttMax) - def setGrttProbingMode(self, mode): - libnorm.NormSetGrttProbingMode(self, mode) + def setGrttProbingMode(self, probingMode:c.ProbingMode): + libnorm.NormSetGrttProbingMode(self, probingMode.value) - def setGrttProbingInterval(self, intervalMin, intervalMax): + def setGrttProbingInterval(self, intervalMin:int, intervalMax:int): libnorm.NormSetGrttProbingInterval(self, intervalMin, intervalMax) - def setBackoffFactor(self, factor): + def setBackoffFactor(self, factor:int): libnorm.NormSetBackoffFactor(self, factor) - def setGroupSize(self, size): + def setGroupSize(self, size:int): libnorm.NormSetGroupSize(self, size) - def fileEnqueue(self, filename, info=""): + def setTxRobustFactor(self, robustFactor:int): + libnorm.NormSetTxRobustFactor(self, robustFactor) + + def fileEnqueue(self, filename:str, info:bytes=b""): # TBD - allow for case of info being None? - result = libnorm.NormFileEnqueue(self, filename.encode('utf-8'), info.encode('utf-8'), len(info)) + result = libnorm.NormFileEnqueue(self, filename.encode(self.locale_encoding), info, len(info)) if ctypes.c_void_p.in_dll(libnorm, "NORM_OBJECT_INVALID") == result: return None; # enqueue not successful due to flow control or sender cache limit - else + else: # Put a reference of the object in our instance "_objects" cache to avoid creation # of duplicative Python NORM Object during event notification - obj = self._instance._objects[self._estruct.object] = Object(result) + obj = self._instance._objects[result] = Object(result) return obj - def dataEnqueue(self, data, info=""): + def dataEnqueue(self, data:bytes, info:bytes=b""): # TBD - allow for case of info being None? - result = libnorm.NormDataEnqueue(self, data.encode('utf-8'), len(data), info.encode('utf-8'), len(info)) + result = libnorm.NormDataEnqueue(self, data, len(data), info, len(info)) if ctypes.c_void_p.in_dll(libnorm, "NORM_OBJECT_INVALID") == result: return None; # enqueue not successful due to flow control or sender cache limit - else + else: # Put a reference of the object in our instance "_objects" cache to avoid creation # of duplicative Python NORM Object during event notification - obj = self._instance._objects[self._estruct.object] = Object(result) + obj = self._instance._objects[result] = Object(result) return obj - def streamOpen(self, bufferSize, info=""): + def streamOpen(self, bufferSize:int, info=b""): # TBD - allow for case of info being None? - result = libnorm.NormStreamOpen(self, bufferSize, info.encode('utf-8'), len(info)) + result = libnorm.NormStreamOpen(self, bufferSize, info, len(info)) if ctypes.c_void_p.in_dll(libnorm, "NORM_OBJECT_INVALID") == result: return None; # stream open/enqueue not successful due to flow control or sender cache limit - else + else: # Put a reference of the object in our instance "_objects" cache to avoid creation # of duplicative Python NORM Object during event notification - obj = self._instance._objects[self._estruct.object] = Object(result) + obj = self._instance._objects[result] = Object(result) return obj def requeueObject(self, normObject): libnorm.NormRequeueObject(self, normObject) - def sendCommand(self, cmdBuffer, robust=False): - return libnorm.NormSendCommand(self, cmdBuffer.encode('utf-8'), len(cmdBuffer), robust) - + def sendCommand(self, cmdBuffer:bytes, robust:bool=False) -> bool: + return libnorm.NormSendCommand(self, cmdBuffer, len(cmdBuffer), robust) + def cancelCommand(self): libnorm.NormCancelCommand(self) - def setWatermark(self, normObject, overrideFlush=False): - libnorm.NormSetWatermark(self, normObject, overrideFlush) - - def resetWatermark(self): - libnorm.NormResetWatermark(self) - - def cancelWatermark(self): + def setWatermark(self, normObject:Object, overrideFlush=False) -> bool: + return libnorm.NormSetWatermark(self, normObject, overrideFlush) + + def resetWatermark(self) -> bool: + return libnorm.NormResetWatermark(self) + + def cancelWatermark(self) -> None: libnorm.NormCancelWatermark(self) - def addAckingNode(self, nodeId): - libnorm.NormAddAckingNode(self, nodeId) + def addAckingNode(self, nodeId:int) -> bool: + return libnorm.NormAddAckingNode(self, nodeId) - def removeAckingNode(self, nodeId): + def removeAckingNode(self, nodeId:int) -> None: libnorm.NormRemoveAckingNode(self, nodeId) + + def setAutoAckingNodes(self, trackingStatus:c.TrackingStatus) -> None: + libnorm.NormSetAutoAckingNodes(self, trackingStatus.value) + + def getNextAckingNode(self, nodeID:int=c.NORM_NODE_NONE) -> (bool, int,int): + ''' + bool NormGetNextAckingNode(NormSessionHandle sessionHandle, + NormNodeId* nodeId, + NormAckingStatus* ackingStatus DEFAULT(0)); + ''' + ackingStatus = 0 + nodeIdBytes = ctypes.c_uint(ctypes.sizeof(nodeId)) + ackingStatusBytes = ctypes.c_uint(ctypes.sizeof(ackingStatus)) + + isSuccess = libnorm.NormGetNextAckingNode(self, ctypes.byref(nodeIdBytes),ctypes.byref(ackingStatusBytes) ) + return (isSuccess, nodeIdBytes.value, c.AckingStatus(ackingStatusBytes.value) if isSuccess else c.AckingStatus.INVALID) - def getAckingStatus(self, nodeId=c.NORM_NODE_ANY): - return libnorm.NormGetAckingStatus(self, nodeId) + def getAckingStatus(self, nodeId:int=c.NORM_NODE_ANY) -> c.AckingStatus: + return c.AckingStatus( libnorm.NormGetAckingStatus(self, nodeId) ) ## Receiver functions - def startReceiver(self, bufferSpace): - libnorm.NormStartReceiver(self, bufferSpace) + def startReceiver(self, bufferSpace:int) -> bool: + ''' + The bufferSpace parameter is used to set a limit on the amount of bufferSpace allocated by the receiver per active NormSender within the session. + The appropriate bufferSpace to use is a function of expected network delay*bandwidth product and packet loss characteristics. + ''' + return libnorm.NormStartReceiver(self, bufferSpace) - def stopReceiver(self): + def stopReceiver(self) -> None: """This will be called automatically if the receiver is active""" libnorm.NormStopReceiver(self) - def setRxCacheLimit(self, count): + def setRxCacheLimit(self, count:int): libnorm.NormSetRxCacheLimit(self, count) - - def setRxSocketBuffer(self, size): + def setRxSocketBuffer(self, size:int): libnorm.NormSetRxSocketBuffer(self, size) - def setSilentReceiver(self, silent, maxDelay=None): + def setSilentReceiver(self, silent:bool, maxDelay:Optional[int]=None): if maxDelay == None: maxDelay = -1 libnorm.NormSetSilentReceiver(self, silent, maxDelay) - def setDefaultUnicastNack(self, mode): - libnorm.NormSetDefaultUnicastNack(self, mode) - - def setDefaultSyncPolicy(self, policy): - libnorm.NormSetDefaultSyncPolicy(self, policy) + def setDefaultUnicastNack(self, enable:bool): + libnorm.NormSetDefaultUnicastNack(self, enable) - def setDefaultNackingMode(self, mode): - libnorm.NormSetDefaultNackingMode(self, mode) + def setDefaultSyncPolicy(self, policy:c.SyncPolicy): + libnorm.NormSetDefaultSyncPolicy(self, policy.value) - def setDefaultRepairBoundary(self, boundary): - libnorm.NormSetDefaultRepairBoundary(self, boundary) - + def setDefaultNackingMode(self, mode:c.NackingMode): + libnorm.NormSetDefaultNackingMode(self, mode.value) + + def setDefaultRepairBoundary(self, boundary:c.RepairBoundary): + libnorm.NormSetDefaultRepairBoundary(self, boundary.value) + + def setDefaultRxRobustFactor(self, rxRobustFactor:int): + libnorm.NormSetDefaultRxRobustFactor(rxRobustFactor) def setMessageTrace(self, state): libnorm.NormSetMessageTrace(self, state) ## Properties - nodeId = property(getNodeId) - grtt = property(getGrttEstimate, setGrttEstimate) - userData = property(getUserData, setUserData) - reportInterval = property(getReportInterval, setReportInterval) + nodeId:int = property(getNodeId) + grtt:float = property(getGrttEstimate, setGrttEstimate) + userData:str = property(getUserData, setUserData) + reportInterval:int = property(getReportInterval, setReportInterval) ## Private functions def __del__(self): @@ -233,6 +293,8 @@ class Session(object): return self._session def __cmp__(self, other): + def cmp(a, b): + return (a > b) - (a < b) try: return cmp(self._as_parameter_, other._as_parameter) except AttributeError: