25except ModuleNotFoundError:
27 "Error: ns3 Python module not found;"
28 " Python bindings may not be enabled"
29 " or your PYTHONPATH might not be properly configured"
33UINT32_MAX = 0xFFFFFFFF
48 """! Test schedule now
49 @param self this object
53 def callback(args: ns.cppyy.gbl.std.vector) ->
None:
54 """! Callback function
61 ns.Simulator.Destroy()
66 EventImpl* pythonMakeEvent(void (*f)(std::vector<std::string>), std::vector<std::string> l)
68 return MakeEvent(f, l);
72 event = ns.cppyy.gbl.pythonMakeEvent(callback, sys.argv)
73 ns.Simulator.ScheduleNow(event)
76 self.assertEqual(self._cb_time.GetSeconds(), 0.0)
80 @param self this object
84 def callback(args: ns.cppyy.gbl.std.vector):
85 """! Callback function
92 ns.Simulator.Destroy()
97 EventImpl* pythonMakeEvent2(void (*f)(std::vector<std::string>), std::vector<std::string> l)
99 return MakeEvent(f, l);
103 event = ns.cppyy.gbl.pythonMakeEvent2(callback, sys.argv)
104 ns.Simulator.Schedule(ns.Seconds(123), event)
107 self.assertEqual(self._cb_time.GetSeconds(), 123.0)
110 """! Test schedule destroy
111 @param self this object
115 def callback(args: ns.cppyy.gbl.std.vector):
116 """! Callback function
123 ns.Simulator.Destroy()
126 ns.cppyy.cppdef(
"void null(){ return; }")
127 ns.Simulator.Schedule(ns.Seconds(123), ns.cppyy.gbl.null)
130 EventImpl* pythonMakeEvent3(void (*f)(std::vector<std::string>), std::vector<std::string> l)
132 return MakeEvent(f, l);
136 event = ns.cppyy.gbl.pythonMakeEvent3(callback, sys.argv)
137 ns.Simulator.ScheduleDestroy(event)
139 ns.Simulator.Destroy()
141 self.assertEqual(self._cb_time.GetSeconds(), 123.0)
144 """! Test schedule with context
145 @param self this object
149 def callback(context, args: ns.cppyy.gbl.std.vector):
151 @param context the context
152 @param args the arguments
159 ns.Simulator.Destroy()
165 EventImpl* pythonMakeEvent4(void (*f)(uint32_t, std::vector<std::string>), uint32_t context, std::vector<std::string> l)
167 return MakeEvent(f, context, l);
171 event = ns.cppyy.gbl.pythonMakeEvent4(callback, 54321, sys.argv)
172 ns.Simulator.ScheduleWithContext(54321, ns.Seconds(123), event)
176 self.assertEqual(self._cb_time.GetSeconds(), 123.0)
179 """! Test time comparison
180 @param self this object
183 self.assertTrue(ns.Seconds(123) == ns.Seconds(123))
184 self.assertTrue(ns.Seconds(123) >= ns.Seconds(123))
185 self.assertTrue(ns.Seconds(123) <= ns.Seconds(123))
186 self.assertTrue(ns.Seconds(124) > ns.Seconds(123))
187 self.assertTrue(ns.Seconds(123) < ns.Seconds(124))
190 """! Test numeric operations
191 @param self this object
194 self.assertEqual(ns.Seconds(10) + ns.Seconds(5), ns.Seconds(15))
195 self.assertEqual(ns.Seconds(10) - ns.Seconds(5), ns.Seconds(5))
197 v1 = ns.int64x64_t(5.0) * ns.int64x64_t(10)
198 self.assertEqual(v1, ns.int64x64_t(50))
201 """! Test configuration
202 @param self this object
205 ns.Config.SetDefault("ns3::OnOffApplication::PacketSize", ns.UintegerValue(123))
213 nc = ns.NodeContainer(1)
215 internet = ns.InternetStackHelper()
216 internet.Install(node)
219 def python_rx_callback(socket) -> None:
224 Callback<void,ns3::Ptr<ns3::Socket> > make_rx_callback_test_socket(void(*func)(Ptr<Socket>))
226 return MakeCallback(func);
231 sink = ns.Socket.CreateSocket(node, ns.TypeId.LookupByName("ns3::UdpSocketFactory"))
232 sink.Bind(ns.InetSocketAddress(ns.Ipv4Address.GetAny(), 80).ConvertTo())
233 sink.SetRecvCallback(ns.cppyy.gbl.make_rx_callback_test_socket(python_rx_callback))
235 source = ns.Socket.CreateSocket(node, ns.TypeId.LookupByName(
"ns3::UdpSocketFactory"))
239 ns.InetSocketAddress(ns.Ipv4Address(
"127.0.0.1"), 80).ConvertTo(),
250 """! Test attributes function
251 @param self this object
255 queue = ns.CreateObject[ns.DropTailQueue[ns.Packet]]()
256 queueSizeValue = ns.QueueSizeValue(ns.QueueSize(
"500p"))
257 queue.SetAttribute(
"MaxSize", queueSizeValue)
259 limit = ns.QueueSizeValue()
260 queue.GetAttribute(
"MaxSize", limit)
261 self.assertEqual(limit.Get(), ns.QueueSize(
"500p"))
264 mobility = ns.CreateObject[ns.RandomWaypointMobilityModel]()
265 ptr = ns.PointerValue()
266 mobility.GetAttribute(
"PositionAllocator", ptr)
267 self.assertEqual(ptr.GetObject(), ns.Ptr[
"Object"](ns.cppyy.nullptr))
269 pos = ns.ListPositionAllocator()
271 mobility.SetAttribute(
"PositionAllocator", ptr)
273 ptr2 = ns.PointerValue()
274 mobility.GetAttribute(
"PositionAllocator", ptr2)
275 self.assertNotEqual(ptr.GetObject(), ns.Ptr[
"Object"](ns.cppyy.nullptr))
278 del queue, mobility, ptr, ptr2
282 @param self this object
285 csma = ns.CreateObject[ns.CsmaNetDevice]()
286 channel = ns.CreateObject[ns.CsmaChannel]()
289 c1 = csma.GetChannel()
290 c2 = csma.GetChannel()
292 self.assertEqual(c1, c2)
299 @param self this object
302 ok, typeId1 = ns.LookupByNameFailSafe("ns3::UdpSocketFactory")
304 self.assertEqual(typeId1.GetName(),
"ns3::UdpSocketFactory")
306 ok, typeId1 = ns.LookupByNameFailSafe(
"ns3::__InvalidTypeName__")
310 """! Test command line
311 @param self this object
314 from ctypes
import c_bool, c_char_p, c_double, c_int, create_string_buffer
318 test3 = c_double(3.1415)
320 test4Buffer = create_string_buffer(b
"this is a test option", BUFFLEN)
321 test4 = c_char_p(test4Buffer.raw)
323 cmd = ns.CommandLine(__file__)
324 cmd.AddValue(
"Test1",
"this is a test option", test1)
325 cmd.AddValue(
"Test2",
"this is a test option", test2)
326 cmd.AddValue[
"double"](
"Test3",
"this is a test option", test3)
327 cmd.AddValue(
"Test4",
"this is a test option", test4, BUFFLEN)
329 cmd.Parse([
"python"])
330 self.assertEqual(test1.value,
True)
331 self.assertEqual(test2.value, 42)
332 self.assertEqual(test3.value, 3.1415)
333 self.assertEqual(test4.value, b
"this is a test option")
335 cmd.Parse([
"python",
"--Test1=false",
"--Test2=0",
"--Test3=0.0"])
336 self.assertEqual(test1.value,
False)
337 self.assertEqual(test2.value, 0)
338 self.assertEqual(test3.value, 0.0)
340 cmd.Parse([
"python",
"--Test4=new_string"])
341 self.assertEqual(test4.value, b
"new_string")
345 @param self this object
350 class MyNode(ns.Node):
351 def GetLocalTime(self) -> ns.Time:
352 return ns.Seconds(10)
355 forced_local_time = node.GetLocalTime()
356 self.assertEqual(forced_local_time, ns.Seconds(10))
360 """! Test python-based application
361 @param self this object
364 ns.Simulator.Destroy()
366 nodes = ns.NodeContainer()
369 pointToPoint = ns.PointToPointHelper()
370 pointToPoint.SetDeviceAttribute("DataRate", ns.StringValue(
"5Mbps"))
371 pointToPoint.SetChannelAttribute(
"Delay", ns.StringValue(
"2ms"))
373 devices = pointToPoint.Install(nodes)
375 stack = ns.InternetStackHelper()
378 address = ns.Ipv4AddressHelper()
379 address.SetBase(ns.Ipv4Address(
"10.1.1.0"), ns.Ipv4Mask(
"255.255.255.0"))
381 interfaces = address.Assign(devices)
387 Callback<void,Ptr<Socket> > make_rx_callback(void(*func)(Ptr<Socket>))
389 return MakeCallback(func);
391 EventImpl* pythonMakeEventSend(void (*f)(Ptr<Socket>, Ptr<Packet>, Address&), Ptr<Socket> socket, Ptr<Packet> packet, Address address)
393 return MakeEvent(f, socket, packet, address);
400 class EchoServer(ns.Application):
403 socketToInstanceDict = {}
405 def __init__(self, node: ns.Node, port=ECHO_PORT):
406 """! Constructor needs to call first the constructor to Application (super class)
407 @param self this object
408 @param node node where this application will be executed
409 @param port port to listen
419 node, ns.TypeId.LookupByName(
"ns3::UdpSocketFactory")
422 ns.InetSocketAddress(ns.Ipv4Address.GetAny(), self.
port).ConvertTo()
424 self.
m_socket.SetRecvCallback(ns.make_rx_callback(EchoServer._Receive))
425 EchoServer.socketToInstanceDict[self.
m_socket] = self
429 @param self this object
432 del EchoServer.socketToInstanceDict[self.m_socket]
434 def Send(self, packet: ns.Packet, address: ns.Address) ->
None:
435 """! Function to send a packet to an address
436 @param self this object
437 @param packet packet to send
438 @param address destination address
441 self.m_socket.SendTo(packet, 0, address)
442 if EchoServer.LOGGING:
443 inetAddress = ns.InetSocketAddress.ConvertFrom(address)
445 "At time +{s}s server sent {b} bytes from {ip} port {port}".format(
446 s=ns.Simulator.Now().GetSeconds(),
447 b=packet.__deref__().GetSize(),
448 ip=inetAddress.GetIpv4(),
449 port=inetAddress.GetPort(),
456 """! Function to receive a packet from an address
457 @param self this object
460 address = ns.Address()
461 packet = self.m_socket.RecvFrom(address)
462 if EchoServer.LOGGING:
463 inetAddress = ns.InetSocketAddress.ConvertFrom(address)
465 "At time +{s}s server received {b} bytes from {ip} port {port}".format(
466 s=ns.Simulator.Now().GetSeconds(),
467 b=packet.__deref__().GetSize(),
468 ip=inetAddress.GetIpv4(),
469 port=inetAddress.GetPort(),
474 event = ns.pythonMakeEventSend(EchoServer._Send, self.
m_socket, packet, address)
475 ns.Simulator.Schedule(ns.Seconds(1), event)
478 def _Send(socket: ns.Socket, packet: ns.Packet, address: ns.Address):
479 """! Static send function, which matches the output socket
480 to the EchoServer instance to call the instance Send function
481 @param socket socket
from the instance that should send the packet
482 @param packet packet to send
483 @param address destination address
486 instance = EchoServer.socketToInstanceDict[socket]
487 instance.Send(packet, address)
490 def _Receive(socket: ns.Socket) ->
None:
491 """! Static receive function, which matches the input socket
492 to the EchoServer instance to call the instance Receive function
493 @param socket socket
from the instance that should receive the packet
496 instance = EchoServer.socketToInstanceDict[socket]
499 echoServer = EchoServer(nodes.Get(1))
500 nodes.Get(1).AddApplication(echoServer)
502 serverApps = ns.ApplicationContainer()
503 serverApps.Add(echoServer)
504 serverApps.Start(ns.Seconds(1.0))
505 serverApps.Stop(ns.Seconds(10.0))
507 address = interfaces.GetAddress(1).ConvertTo()
508 echoClient = ns.UdpEchoClientHelper(address, EchoServer.ECHO_PORT)
509 echoClient.SetAttribute("MaxPackets", ns.UintegerValue(10))
510 echoClient.SetAttribute(
"Interval", ns.TimeValue(ns.Seconds(1.0)))
511 echoClient.SetAttribute(
"PacketSize", ns.UintegerValue(101))
513 clientApps = echoClient.Install(nodes.Get(0))
514 clientApps.Start(ns.Seconds(2.0))
515 clientApps.Stop(ns.Seconds(10.0))
518 ns.Simulator.Destroy()
521if __name__ ==
"__main__":
522 unittest.main(verbosity=1, failfast=
True)
def testScheduleDestroy(self)
Test schedule destroy.
port
Listen port for the server.
def testCommandLine(self)
Test command line.
def testTimeNumericOperations(self)
Test numeric operations.
def testEchoServerApplication(self)
Test python-based application.
__python_owns__
EchoServer application class.
def testScheduleNow(self)
Test schedule now.
def testTypeId(self)
Test type ID.
def testSubclass(self)
Test subclass.
def testSchedule(self)
Test schedule.
def testScheduleWithContext(self)
Test schedule with context.
def testAttributes(self)
Test attributes function.
def testSocket(self)
Test socket.
def testTimeComparison(self)
Test time comparison.
def testConfig(self)
Test configuration.
m_socket
Socket used by the server to listen to port.
_received_packet
received packet
def testIdentity(self)
Test identify.
static void Send(Ptr< NetDevice > dev, int level, std::string emuMode)