tools/pkt_gen/xena/XenaDriver.py

   1 # Copyright 2016 Red Hat Inc & Xena Networks.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #   http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 # This is a port of code by Xena and Flavios ported to python 3 compatibility.
  16 # Credit given to Xena and Flavio for providing most of the logic of this code.
  17 # The code has changes for PEP 8 and python 3 conversion. Added Stat classes
  18 # for better scaling of future requirements. Also added calculation functions
  19 # for line rate to align within VSPerf project.
  20 # Flavios xena libraries available at https://github.com/fleitner/XenaPythonLib
  21
  22 # Contributors:
  23 #   Flavio Leitner, Red Hat Inc.
  24 #   Dan Amzulescu, Xena Networks
  25 #   Christian Trautman, Red Hat Inc.
  26
  27 """
  28 Xena Socket API Driver module for communicating directly with Xena system
  29 through socket commands and returning different statistics.
  30 """
  31 import locale
  32 import logging
  33 import socket
  34 import struct
  35 import sys
  36 import threading
  37 import time
  38
  39 # Xena Socket Commands
  40 CMD_CLEAR_RX_STATS = 'pr_clear'
  41 CMD_CLEAR_TX_STATS = 'pt_clear'
  42 CMD_COMMENT = ';'
  43 CMD_CREATE_STREAM = 'ps_create'
  44 CMD_DELETE_STREAM = 'ps_delete'
  45 CMD_GET_PORT_SPEED = 'p_speed ?'
  46 CMD_GET_PORT_SPEED_REDUCTION = 'p_speedreduction ?'
  47 CMD_GET_RX_STATS_PER_TID = 'pr_tpldtraffic'
  48 CMD_GET_STREAM_DATA = 'pt_stream'
  49 CMD_GET_STREAMS_PER_PORT = 'ps_indices'
  50 CMD_GET_TID_PER_STREAM = 'ps_tpldid'
  51 CMD_GET_TX_STATS_PER_STREAM = 'pt_stream'
  52 CMD_GET_RX_STATS = 'pr_all ?'
  53 CMD_GET_TX_STATS = 'pt_all ?'
  54 CMD_INTERFRAME_GAP = 'p_interframegap'
  55 CMD_LOGIN = 'c_logon'
  56 CMD_LOGOFF = 'c_logoff'
  57 CMD_OWNER = 'c_owner'
  58 CMD_PORT = ';Port:'
  59 CMD_PORT_IP = 'p_ipaddress'
  60 CMD_RESERVE = 'p_reservation reserve'
  61 CMD_RELEASE = 'p_reservation release'
  62 CMD_RELINQUISH = 'p_reservation relinquish'
  63 CMD_RESET = 'p_reset'
  64 CMD_SET_PORT_ARP_REPLY = 'p_arpreply'
  65 CMD_SET_PORT_ARP_V6_REPLY = 'p_arpv6reply'
  66 CMD_SET_PORT_PING_REPLY = 'p_pingreply'
  67 CMD_SET_PORT_PING_V6_REPLY = 'p_pingv6reply'
  68 CMD_SET_PORT_TIME_LIMIT = 'p_txtimelimit'
  69 CMD_SET_STREAM_HEADER_PROTOCOL = 'ps_headerprotocol'
  70 CMD_SET_STREAM_ON_OFF = 'ps_enable'
  71 CMD_SET_STREAM_PACKET_HEADER = 'ps_packetheader'
  72 CMD_SET_STREAM_PACKET_LENGTH = 'ps_packetlength'
  73 CMD_SET_STREAM_PACKET_LIMIT = 'ps_packetlimit'
  74 CMD_SET_STREAM_PACKET_PAYLOAD = 'ps_payload'
  75 CMD_SET_STREAM_RATE_FRACTION = 'ps_ratefraction'
  76 CMD_SET_STREAM_TEST_PAYLOAD_ID = 'ps_tpldid'
  77 CMD_SET_TPLD_MODE = 'p_tpldmode'
  78 CMD_START_TRAFFIC = 'p_traffic on'
  79 CMD_STOP_TRAFFIC = 'p_traffic off'
  80 CMD_STREAM_MODIFIER = 'ps_modifier'
  81 CMD_STREAM_MODIFIER_COUNT = 'ps_modifiercount'
  82 CMD_STREAM_MODIFIER_RANGE = 'ps_modifierrange'
  83 CMD_VERSION = 'c_versionno ?'
  84
  85 _LOCALE = locale.getlocale()[1]
  86 _LOGGER = logging.getLogger(__name__)
  87
  88
  89 class SimpleSocket(object):
  90     """
  91     Socket class
  92     """
  93     def __init__(self, hostname, port=5025, timeout=1):
  94         """Constructor
  95         :param hostname: hostname or ip as string
  96         :param port: port number to use for socket as int
  97         :param timeout: socket timeout as int
  98         :return: SimpleSocket object
  99         """
 100         self.hostname = hostname
 101         try:
 102             self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 103             self.sock.settimeout(timeout)
 104             self.sock.connect((hostname, port))
 105         except socket.error as msg:
 106             _LOGGER.error(
 107                 "Cannot connect to Xena Socket at %s", hostname)
 108             _LOGGER.error("Exception : %s", msg)
 109             sys.exit(1)
 110
 111     def __del__(self):
 112         """Deconstructor
 113         :return:
 114         """
 115         self.sock.close()
 116
 117     def ask(self, cmd):
 118         """ Send the command over the socket
 119         :param cmd: cmd as string
 120         :return: byte utf encoded return value from socket
 121         """
 122         cmd += '\n'
 123         try:
 124             self.sock.send(cmd.encode('utf-8'))
 125             return self.sock.recv(1024)
 126         except OSError:
 127             return ''
 128
 129     def read_reply(self):
 130         """ Get the response from the socket
 131         :return: Return the reply
 132         """
 133         reply = self.sock.recv(1024)
 134         if reply.find("---^".encode('utf-8')) != -1:
 135             # read again the syntax error msg
 136             reply = self.sock.recv(1024)
 137         return reply
 138
 139     def send_command(self, cmd):
 140         """ Send the command specified over the socket
 141         :param cmd: Command to send as string
 142         :return: None
 143         """
 144         cmd += '\n'
 145         self.sock.send(cmd.encode('utf-8'))
 146
 147     def set_keep_alive(self):
 148         """ Set the keep alive for the socket
 149         :return: None
 150         """
 151         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 152
 153
 154 class KeepAliveThread(threading.Thread):
 155     """
 156     Keep alive socket class
 157     """
 158     message = ''
 159
 160     def __init__(self, connection, interval=10):
 161         """ Constructor
 162         :param connection: Socket for keep alive
 163         :param interval: interval in seconds to send keep alive
 164         :return: KeepAliveThread object
 165         """
 166         threading.Thread.__init__(self)
 167         self.connection = connection
 168         self.interval = interval
 169         self.finished = threading.Event()
 170         self.setDaemon(True)
 171         _LOGGER.debug(
 172             'Xena Socket keep alive thread initiated, interval ' +
 173             '{} seconds'.format(self.interval))
 174
 175     def stop(self):
 176         """ Thread stop. See python thread docs for more info
 177         :return: None
 178         """
 179         self.finished.set()
 180         self.join()
 181
 182     def run(self):
 183         """ Thread start. See python thread docs for more info
 184         :return: None
 185         """
 186         while not self.finished.isSet():
 187             self.finished.wait(self.interval)
 188             self.connection.ask(self.message)
 189
 190
 191 class XenaSocketDriver(SimpleSocket):
 192     """
 193     Xena socket class
 194     """
 195     reply_ok = '<OK>'
 196
 197     def __init__(self, hostname, port=22611):
 198         """ Constructor
 199         :param hostname: Hostname or ip as string
 200         :param port: port to use as int
 201         :return: XenaSocketDriver object
 202         """
 203         SimpleSocket.__init__(self, hostname=hostname, port=port)
 204         SimpleSocket.set_keep_alive(self)
 205         self.access_semaphor = threading.Semaphore(1)
 206
 207     def ask(self, cmd):
 208         """ Send the command over the socket in a thread safe manner
 209         :param cmd: Command to send
 210         :return: reply from socket
 211         """
 212         self.access_semaphor.acquire()
 213         reply = SimpleSocket.ask(self, cmd)
 214         self.access_semaphor.release()
 215         return reply
 216
 217     def ask_verify(self, cmd):
 218         """ Send the command over the socket in a thread safe manner and
 219         verify the response is good.
 220         :param cmd: Command to send
 221         :return: Boolean True if command response is good, False otherwise
 222         """
 223         resp = self.ask(cmd).decode(_LOCALE).strip('\n')
 224         _LOGGER.info('[ask_verify] %s', resp)
 225         if resp == self.reply_ok:
 226             return True
 227         return False
 228
 229     def disconnect(self):
 230         """
 231         Close the socket connection
 232         :return: None
 233         """
 234         self.sock.close()
 235
 236     def send_command(self, cmd):
 237         """ Send the command over the socket with no return
 238         :param cmd: Command to send
 239         :return: None
 240         """
 241         self.access_semaphor.acquire()
 242         SimpleSocket.send_command(self, cmd)
 243         self.access_semaphor.release()
 244
 245     def send_query_replies(self, cmd):
 246         """ Send the command over the socket and wait for all replies and return
 247         the lines as a list
 248         :param cmd: Command to send
 249         :return: Response from command as list
 250         """
 251         # send the command followed by cmd SYNC to find out
 252         # when the last reply arrives.
 253         self.send_command(cmd.strip('\n'))
 254         self.send_command('SYNC')
 255         replies = []
 256         self.access_semaphor.acquire()
 257         msg = SimpleSocket.read_reply(self).decode(_LOCALE)
 258         msgleft = ''
 259         while True:
 260             if '\n' in msg:
 261                 (reply, msgleft) = msg.split('\n', 1)
 262                 # check for syntax problems
 263                 if reply.rfind('Syntax') != -1:
 264                     self.access_semaphor.release()
 265                     return []
 266
 267                 if reply.rfind('<SYNC>') == 0:
 268
 269                     self.access_semaphor.release()
 270                     return replies
 271
 272                 replies.append(reply + '\n')
 273                 msg = msgleft
 274             else:
 275                 # more bytes to come
 276                 msgnew = SimpleSocket.read_reply(self).decode(_LOCALE)
 277                 msg = msgleft + msgnew
 278
 279
 280 class XenaManager(object):
 281     """
 282     Manager class for port and socket functions
 283     """
 284     def __init__(self, socketDriver, user='', password='xena'):
 285         """Constructor
 286
 287         Establish a connection to Xena using a ``driver`` with the ``password``
 288         supplied.
 289
 290         Attributes:
 291         :param socketDriver: XenaSocketDriver connection object
 292         :param password: Password to the Xena traffic generator
 293         :returns: XenaManager object
 294         """
 295         self.driver = socketDriver
 296         self.ports = list()
 297         self.keep_alive_thread = KeepAliveThread(self.driver)
 298
 299         if self.logon(password):
 300             _LOGGER.info('Connected to Xena at %s', self.driver.hostname)
 301         else:
 302             _LOGGER.error('Failed to logon to Xena at %s', self.driver.hostname)
 303             return
 304
 305         self.set_owner(user)
 306
 307     def disconnect(self):
 308         """ Release ports and disconnect from chassis.
 309         """
 310         for module_port in self.ports:
 311             module_port.release_port()
 312         self.ports = []
 313         self.logoff()
 314         self.keep_alive_thread.stop()
 315
 316     def add_module_port(self, module, port):
 317         """Factory for Xena Ports
 318
 319         :param module: String or int of module
 320         :param port: String or int of port
 321         :return: XenaPort object if success, None if port already added
 322         """
 323         xenaport = XenaPort(self, module, port)
 324         if xenaport in self.ports:
 325             return None
 326         else:
 327             self.ports.append(xenaport)
 328             return xenaport
 329
 330     def get_module_port(self, module, port):
 331         """Return the Xena Port object if available
 332         :param module: module number as int or str
 333         :param port: port number as int or str
 334         :return: XenaPort object or None if not found
 335         """
 336         for por in self.ports:
 337             if por.port == str(port) and por.module == str(module):
 338                 return por
 339         return None
 340
 341     def get_version(self):
 342         """
 343         Get the version from the chassis
 344         :return: versions of server and driver as string
 345         """
 346         res = self.driver.ask(make_manager_command(
 347             CMD_VERSION, '')).decode(_LOCALE)
 348         res = res.rstrip('\n').split()
 349         return "Server: {} Driver: {}".format(res[1], res[2])
 350
 351     def logoff(self):
 352         """
 353         Logoff from the Xena chassis
 354         :return: Boolean True if response OK, False if error.
 355         """
 356         return self.driver.ask_verify(make_manager_command(CMD_LOGOFF))
 357
 358     def logon(self, password):
 359         """Login to the Xena traffic generator using the ``password`` supplied.
 360
 361         :param password: string of password
 362         :return: Boolean True if response OK, False if error.
 363         """
 364         self.keep_alive_thread.start()
 365         return self.driver.ask_verify(make_manager_command(CMD_LOGIN, password))
 366
 367     def set_owner(self, username):
 368         """Set the ports owner.
 369         :return: Boolean True if response OK, False if error.
 370         """
 371         return self.driver.ask_verify(make_manager_command(CMD_OWNER, username))
 372
 373
 374 class XenaPort(object):
 375     """
 376     Xena Port emulator class
 377     """
 378     def __init__(self, manager, module, port):
 379         """Constructor
 380
 381         :param manager: XenaManager object
 382         :param module: Module as string or int of module to use
 383         :param port: Port as string or int of port to use
 384         :return: XenaPort object
 385         """
 386         self._manager = manager
 387         self._module = str(module)
 388         self._port = str(port)
 389         self._streams = list()
 390
 391     @property
 392     def manager(self):
 393         """Property for manager attribute
 394         :return: manager object
 395         """
 396         return self._manager
 397
 398     @property
 399     def module(self):
 400         """Property for module attribute
 401         :return: module value as string
 402         """
 403         return self._module
 404
 405     @property
 406     def port(self):
 407         """Property for port attribute
 408         :return: port value as string
 409         """
 410         return self._port
 411
 412     def port_string(self):
 413         """String builder with attributes
 414         :return: String of module port for command sequence
 415         """
 416         stringify = "{}/{}".format(self._module, self._port)
 417         return stringify
 418
 419     def add_stream(self):
 420         """Add a stream to the port.
 421         :return: XenaStream object, None if failure
 422         """
 423         identifier = len(self._streams)
 424         stream = XenaStream(self, identifier)
 425         if self._manager.driver.ask_verify(make_stream_command(
 426                 CMD_CREATE_STREAM, '', stream)):
 427             self._streams.append(stream)
 428             return stream
 429         else:
 430             _LOGGER.error("Error during stream creation")
 431             return None
 432
 433     def clear_stats(self, rx_clear=True, tx_clear=True):
 434         """Clear the port stats
 435
 436         :param rx_clear: Boolean if rx stats are to be cleared
 437         :param tx_clear: Boolean if tx stats are to be cleared
 438         :return: Boolean True if response OK, False if error.
 439         """
 440         command = make_port_command(CMD_CLEAR_RX_STATS, self)
 441         res1 = self._manager.driver.ask_verify(command) if rx_clear else True
 442         command = make_port_command(CMD_CLEAR_TX_STATS, self)
 443         res2 = self._manager.driver.ask_verify(command) if tx_clear else True
 444         return all([res1, res2])
 445
 446     def get_effective_speed(self):
 447         """
 448         Get the effective speed on the port
 449         :return: effective speed as float
 450         """
 451         port_speed = self.get_port_speed()
 452         reduction = self.get_port_speed_reduction()
 453         effective_speed = port_speed * (1.0 - reduction / 1000000.0)
 454         return effective_speed
 455
 456     def get_inter_frame_gap(self):
 457         """
 458         Get the interframe gap and return it as string
 459         :return: integer of interframe gap
 460         """
 461         command = make_port_command(CMD_INTERFRAME_GAP + '?', self)
 462         res = self._manager.driver.ask(command).decode(_LOCALE)
 463         res = int(res.rstrip('\n').split(' ')[-1])
 464         return res
 465
 466     def get_port_speed(self):
 467         """
 468         Get the port speed as bits from port and return it as a int.
 469         :return: Int of port speed
 470         """
 471         command = make_port_command(CMD_GET_PORT_SPEED, self)
 472         res = self._manager.driver.ask(command).decode(_LOCALE)
 473         port_speed = res.split(' ')[-1].rstrip('\n')
 474         return int(port_speed) * 1000000
 475
 476     def get_port_speed_reduction(self):
 477         """
 478         Get the port speed reduction value as int
 479         :return: Integer of port speed reduction value
 480         """
 481         command = make_port_command(CMD_GET_PORT_SPEED_REDUCTION, self)
 482         res = self._manager.driver.ask(command).decode(_LOCALE)
 483         res = int(res.rstrip('\n').split(' ')[-1])
 484         return res
 485
 486     def get_rx_stats(self):
 487         """Get the rx stats and return the data as a dict.
 488         :return: Receive stats as dictionary
 489         """
 490         command = make_port_command(CMD_GET_RX_STATS, self)
 491         rx_data = self._manager.driver.send_query_replies(command)
 492         data = XenaRXStats(rx_data, time.time())
 493         return data
 494
 495     def get_tx_stats(self):
 496         """Get the tx stats and return the data as a dict.
 497         :return: Receive stats as dictionary
 498         """
 499         command = make_port_command(CMD_GET_TX_STATS, self)
 500         tx_data = self._manager.driver.send_query_replies(command)
 501         data = XenaTXStats(tx_data, time.time())
 502         return data
 503
 504     def micro_tpld_disable(self):
 505         """Disable micro TPLD and return to standard payload size
 506         :return: Boolean if response OK, False if error
 507         """
 508         command = make_port_command(CMD_SET_TPLD_MODE + ' normal', self)
 509         return self._manager.driver.ask_verify(command)
 510
 511     def micro_tpld_enable(self):
 512         """Enable micro TPLD 6 byte payloads.
 513         :Return Boolean if response OK, False if error
 514         """
 515         command = make_port_command(CMD_SET_TPLD_MODE + ' micro', self)
 516         return self._manager.driver.ask_verify(command)
 517
 518     def release_port(self):
 519         """Release the port
 520         :return: Boolean True if response OK, False if error.
 521         """
 522         command = make_port_command(CMD_RELEASE, self)
 523         return self._manager.driver.ask_verify(command)
 524
 525     def reserve_port(self):
 526         """Reserve the port
 527         :return: Boolean True if response OK, False if error.
 528         """
 529         command = make_port_command(CMD_RESERVE, self)
 530         return self._manager.driver.ask_verify(command)
 531
 532     def reset_port(self):
 533         """Reset the port
 534         :return: Boolean True if response OK, False if error.
 535         """
 536         command = make_port_command(CMD_RESET, self)
 537         return self._manager.driver.ask_verify(command)
 538
 539     def set_port_arp_reply(self, on=True, v6=False):
 540         """
 541         Set the port arpreply value
 542         :param on: Enable or disable the arp reply on the port
 543         :param v6: set the value on the ip v6, disabled will set at ip v4
 544         :return: Boolean True if response OK, False if error
 545         """
 546         command = make_port_command('{} {}'.format(
 547             CMD_SET_PORT_ARP_V6_REPLY if v6 else CMD_SET_PORT_ARP_REPLY,
 548             "on" if on else "off"), self)
 549         return self._manager.driver.ask_verify(command)
 550
 551     def set_port_ping_reply(self, on=True, v6=False):
 552         """
 553         Set the port ping reply value
 554         :param on: Enable or disable the ping reply on the port
 555         :param v6: set the value on the ip v6, disabled will set at ip v4
 556         :return: Boolean True if response OK, False if error
 557         """
 558         command = make_port_command('{} {}'.format(
 559             CMD_SET_PORT_PING_V6_REPLY if v6 else CMD_SET_PORT_PING_REPLY,
 560             "on" if on else "off"), self)
 561         return self._manager.driver.ask_verify(command)
 562
 563     def set_port_ip(self, ip_addr, cidr, gateway, wild='255'):
 564         """
 565         Set the port ip address of the specific port
 566         :param ip_addr: IP address to set to port
 567         :param cidr: cidr number for the subnet
 568         :param gateway: Gateway ip for port
 569         :param wild: wildcard used for ARP and PING replies
 570         :return: Boolean True if response OK, False if error
 571         """
 572         # convert the cidr to a dot notation subnet address
 573         subnet = socket.inet_ntoa(
 574             struct.pack(">I", (0xffffffff << (32 - cidr)) & 0xffffffff))
 575
 576         command = make_port_command('{} {} {} {} 0.0.0.{}'.format(
 577             CMD_PORT_IP, ip_addr, subnet, gateway, wild), self)
 578         return self._manager.driver.ask_verify(command)
 579
 580     def set_port_time_limit(self, micro_seconds):
 581         """Set the port time limit in ms
 582         :param micro_seconds: ms for port time limit
 583         :return: Boolean True if response OK, False if error.
 584         """
 585         command = make_port_command('{} {}'.format(
 586             CMD_SET_PORT_TIME_LIMIT, micro_seconds), self)
 587         return self._manager.driver.ask_verify(command)
 588
 589     def traffic_off(self):
 590         """Start traffic
 591         :return: Boolean True if response OK, False if error.
 592         """
 593         command = make_port_command(CMD_STOP_TRAFFIC, self)
 594         return self._manager.driver.ask_verify(command)
 595
 596     def traffic_on(self):
 597         """Stop traffic
 598         :return: Boolean True if response OK, False if error.
 599         """
 600         command = make_port_command(CMD_START_TRAFFIC, self)
 601         return self._manager.driver.ask_verify(command)
 602
 603
 604 class XenaStream(object):
 605     """
 606     Xena stream emulator class
 607     """
 608     def __init__(self, xenaPort, streamID):
 609         """Constructor
 610
 611         :param xenaPort: XenaPort object
 612         :param streamID: Stream ID as int or string
 613         :return: XenaStream object
 614         """
 615         self._xena_port = xenaPort
 616         self._stream_id = str(streamID)
 617         self._manager = self._xena_port.manager
 618         self._header_protocol = None
 619
 620     @property
 621     def xena_port(self):
 622         """Property for port attribute
 623         :return: XenaPort object
 624         """
 625         return self._xena_port
 626
 627     @property
 628     def stream_id(self):
 629         """Property for streamID attribute
 630         :return: streamID value as string
 631         """
 632         return self._stream_id
 633
 634     def enable_multistream(self, flows, layer):
 635         """
 636         Basic implementation of multi stream. Enable multi stream by setting
 637         modifiers on the stream
 638         :param flows: Numbers of flows or end range
 639         :param layer: layer to enable multi stream as str. Acceptable values
 640         are L2, L3, or L4
 641         :return: True if success False otherwise
 642         """
 643         if not self._header_protocol:
 644             raise RuntimeError(
 645                 "Please set a protocol header before calling this method.")
 646
 647         # byte offsets for setting the modifier
 648         offsets = {
 649             'L2': [0, 6],
 650             'L3': [32, 36] if 'VLAN' in self._header_protocol else [28, 32],
 651             'L4': [38, 40] if 'VLAN' in self._header_protocol else [34, 36]
 652         }
 653
 654         responses = list()
 655         if layer in offsets.keys() and flows > 0:
 656             command = make_port_command(
 657                 CMD_STREAM_MODIFIER_COUNT + ' [{}]'.format(self._stream_id) +
 658                 ' 2', self._xena_port)
 659             responses.append(self._manager.driver.ask_verify(command))
 660             command = make_port_command(
 661                 CMD_STREAM_MODIFIER + ' [{},0] {} 0xFFFF0000 INC 1'.format(
 662                     self._stream_id, offsets[layer][0]), self._xena_port)
 663             responses.append(self._manager.driver.ask_verify(command))
 664             command = make_port_command(
 665                 CMD_STREAM_MODIFIER_RANGE + ' [{},0] 0 1 {}'.format(
 666                     self._stream_id, flows), self._xena_port)
 667             responses.append(self._manager.driver.ask_verify(command))
 668             command = make_port_command(
 669                 CMD_STREAM_MODIFIER + ' [{},1] {} 0xFFFF0000 INC 1'.format(
 670                     self._stream_id, offsets[layer][1]), self._xena_port)
 671             responses.append(self._manager.driver.ask_verify(command))
 672             command = make_port_command(
 673                 CMD_STREAM_MODIFIER_RANGE + ' [{},1] 0 1 {}'.format(
 674                     self._stream_id, flows), self._xena_port)
 675             responses.append(self._manager.driver.ask_verify(command))
 676             return all(responses)  # return True if they all worked
 677         elif flows < 1:
 678             _LOGGER.warning(
 679                 'No flows specified in enable multistream. Bypassing...')
 680             return False
 681         else:
 682             raise NotImplementedError(
 683                 "Non-implemented stream layer in method enable multistream ",
 684                 "layer=", layer)
 685
 686     def get_stream_data(self):
 687         """
 688         Get the response for stream data
 689         :return: String of response for stream data info
 690         """
 691         command = make_stream_command(CMD_GET_STREAM_DATA, '?', self)
 692         res = self._manager.driver.ask(command).decode(_LOCALE)
 693         return res
 694
 695     def set_header_protocol(self, protocol_header):
 696         """Set the header info for the packet header hex.
 697         If the packet header contains just Ethernet and IP info then call this
 698         method with ETHERNET IP as the protocol header.
 699
 700         :param protocol_header: protocol header argument
 701         :return: Boolean True if success, False if error
 702         """
 703         command = make_stream_command(
 704             CMD_SET_STREAM_HEADER_PROTOCOL,
 705             protocol_header, self)
 706         if self._manager.driver.ask_verify(command):
 707             self._header_protocol = protocol_header
 708             return True
 709         else:
 710             return False
 711
 712     def set_off(self):
 713         """Set the stream to off
 714         :return: Boolean True if success, False if error
 715         """
 716         return self._manager.driver.ask_verify(make_stream_command(
 717             CMD_SET_STREAM_ON_OFF, 'off', self))
 718
 719     def set_on(self):
 720         """Set the stream to on
 721         :return: Boolean True if success, False if error
 722         """
 723         return self._manager.driver.ask_verify(make_stream_command(
 724             CMD_SET_STREAM_ON_OFF, 'on', self))
 725
 726     def set_packet_header(self, header):
 727         """Set the stream packet header
 728
 729         :param header: packet header as hex bytes
 730         :return: Boolean True if success, False if error
 731         """
 732         return self._manager.driver.ask_verify(make_stream_command(
 733             CMD_SET_STREAM_PACKET_HEADER, header, self))
 734
 735     def set_packet_length(self, pattern_type, minimum, maximum):
 736         """Set the pattern length with min and max values based on the pattern
 737         type supplied
 738
 739         :param pattern_type: String of pattern type, valid entries [ fixed,
 740          butterfly, random, mix, incrementing ]
 741         :param minimum: integer of minimum byte value
 742         :param maximum: integer of maximum byte value
 743         :return: Boolean True if success, False if error
 744         """
 745         return self._manager.driver.ask_verify(make_stream_command(
 746             CMD_SET_STREAM_PACKET_LENGTH, '{} {} {}'.format(
 747                 pattern_type, minimum, maximum), self))
 748
 749     def set_packet_limit(self, limit):
 750         """Set the packet limit
 751
 752         :param limit: number of packets that will be sent, use -1 to disable
 753         :return: Boolean True if success, False if error
 754         """
 755         return self._manager.driver.ask_verify(make_stream_command(
 756             CMD_SET_STREAM_PACKET_LIMIT, limit, self))
 757
 758     def set_packet_payload(self, payload_type, hex_value):
 759         """Set the payload to the hex value based on the payload type
 760
 761         :param payload_type: string of the payload type, valid entries [ pattern,
 762          incrementing, prbs ]
 763         :param hex_value: hex string of valid hex
 764         :return: Boolean True if success, False if error
 765         """
 766         return self._manager.driver.ask_verify(make_stream_command(
 767             CMD_SET_STREAM_PACKET_PAYLOAD, '{} {}'.format(
 768                 payload_type, hex_value), self))
 769
 770     def set_rate_fraction(self, fraction):
 771         """Set the rate fraction
 772
 773         :param fraction: fraction for the stream
 774         :return: Boolean True if success, False if error
 775         """
 776         return self._manager.driver.ask_verify(make_stream_command(
 777             CMD_SET_STREAM_RATE_FRACTION, fraction, self))
 778
 779     def set_payload_id(self, identifier):
 780         """ Set the test payload ID
 781         :param identifier: ID as int or string
 782         :return: Boolean True if success, False if error
 783         """
 784         return self._manager.driver.ask_verify(make_stream_command(
 785             CMD_SET_STREAM_TEST_PAYLOAD_ID, identifier, self))
 786
 787
 788 class XenaRXStats(object):
 789     """
 790     Receive stat class
 791     """
 792     def __init__(self, stats, epoc):
 793         """ Constructor
 794         :param stats: Stats from pr all command as list
 795         :param epoc: Current time in epoc
 796         :return: XenaRXStats object
 797         """
 798         self._stats = stats
 799         self._time = epoc
 800         self.data = self.parse_stats()
 801         self.preamble = 8
 802
 803     @staticmethod
 804     def _pack_stats(param, start, fields=None):
 805         """ Pack up the list of stats in a dictionary
 806         :param param: The list of params to process
 807         :param start: What element to start at
 808         :param fields: The field names to pack as keys
 809         :return: Dictionary of data where fields match up to the params
 810         """
 811         if not fields:
 812             fields = ['bps', 'pps', 'bytes', 'packets']
 813         data = {}
 814         i = 0
 815         for column in fields:
 816             data[column] = int(param[start + i])
 817             i += 1
 818
 819         return data
 820
 821     @staticmethod
 822     def _pack_tplds_stats(param, start):
 823         """ Pack up the tplds stats
 824         :param param: List of params to pack
 825         :param start: What element to start at
 826         :return: Dictionary of stats
 827         """
 828         data = {}
 829         i = 0
 830         for val in range(start, len(param) - start):
 831             data[i] = int(param[val])
 832             i += 1
 833         return data
 834
 835     def _pack_rxextra_stats(self, param, start):
 836         """ Pack up the extra stats
 837         :param param: List of params to pack
 838         :param start: What element to start at
 839         :return: Dictionary of stats
 840         """
 841         fields = ['fcserrors', 'pauseframes', 'arprequests', 'arpreplies',
 842                   'pingrequests', 'pingreplies', 'gapcount', 'gapduration']
 843         return self._pack_stats(param, start, fields)
 844
 845     def _pack_tplderrors_stats(self, param, start):
 846         """ Pack up tlpd errors
 847         :param param: List of params to pack
 848         :param start: What element to start at
 849         :return: Dictionary of stats
 850         """
 851         fields = ['dummy', 'seq', 'mis', 'pld']
 852         return self._pack_stats(param, start, fields)
 853
 854     def _pack_tpldlatency_stats(self, param, start):
 855         """ Pack up the tpld latency stats
 856         :param param: List of params to pack
 857         :param start: What element to start at
 858         :return: Dictionary of stats
 859         """
 860         fields = ['min', 'avg', 'max', '1sec']
 861         return self._pack_stats(param, start, fields)
 862
 863     def _pack_tpldjitter_stats(self, param, start):
 864         """ Pack up the tpld jitter stats
 865         :param param: List of params to pack
 866         :param start: What element to start at
 867         :return: Dictionary of stats
 868         """
 869         fields = ['min', 'avg', 'max', '1sec']
 870         return self._pack_stats(param, start, fields)
 871
 872     @property
 873     def time(self):
 874         """
 875         :return: Time as String of epoc of when stats were collected
 876         """
 877         return self._time
 878
 879     def parse_stats(self):
 880         """ Parse the stats from pr all command
 881         :return: Dictionary of all stats
 882         """
 883         statdict = {}
 884         for line in self._stats:
 885             param = line.split()
 886             if param[1] == 'PR_TOTAL':
 887                 statdict['pr_total'] = self._pack_stats(param, 2)
 888             elif param[1] == 'PR_NOTPLD':
 889                 statdict['pr_notpld'] = self._pack_stats(param, 2,)
 890             elif param[1] == 'PR_EXTRA':
 891                 statdict['pr_extra'] = self._pack_rxextra_stats(param, 2)
 892             elif param[1] == 'PT_STREAM':
 893                 entry_id = "pt_stream_%s" % param[2].strip('[]')
 894                 statdict[entry_id] = self._pack_stats(param, 3)
 895             elif param[1] == 'PR_TPLDS':
 896                 tid_list = self._pack_tplds_stats(param, 2)
 897                 if len(tid_list):
 898                     statdict['pr_tplds'] = tid_list
 899             elif param[1] == 'PR_TPLDTRAFFIC':
 900                 if 'pr_tpldstraffic' in statdict:
 901                     data = statdict['pr_tpldstraffic']
 902                 else:
 903                     data = {}
 904                 entry_id = param[2].strip('[]')
 905                 data[entry_id] = self._pack_stats(param, 3)
 906                 statdict['pr_tpldstraffic'] = data
 907             elif param[1] == 'PR_TPLDERRORS':
 908                 if 'pr_tplderrors' in statdict:
 909                     data = statdict['pr_tplderrors']
 910                 else:
 911                     data = {}
 912                 entry_id = param[2].strip('[]')
 913                 data[entry_id] = self._pack_tplderrors_stats(param, 3)
 914                 statdict['pr_tplderrors'] = data
 915             elif param[1] == 'PR_TPLDLATENCY':
 916                 if 'pr_tpldlatency' in statdict:
 917                     data = statdict['pr_tpldlatency']
 918                 else:
 919                     data = {}
 920                 entry_id = param[2].strip('[]')
 921                 data[entry_id] = self._pack_tpldlatency_stats(param, 3)
 922                 statdict['pr_tpldlatency'] = data
 923             elif param[1] == 'PR_TPLDJITTER':
 924                 if 'pr_tpldjitter' in statdict:
 925                     data = statdict['pr_tpldjitter']
 926                 else:
 927                     data = {}
 928                 entry_id = param[2].strip('[]')
 929                 data[entry_id] = self._pack_tpldjitter_stats(param, 3)
 930                 statdict['pr_pldjitter'] = data
 931             elif param[1] == 'PR_FILTER':
 932                 if 'pr_filter' in statdict:
 933                     data = statdict['pr_filter']
 934                 else:
 935                     data = {}
 936                 entry_id = param[2].strip('[]')
 937                 data[entry_id] = self._pack_stats(param, 3)
 938                 statdict['pr_filter'] = data
 939             elif param[1] == 'P_RECEIVESYNC':
 940                 if param[2] == 'IN_SYNC':
 941                     statdict['p_receivesync'] = {'IN SYNC': 'True'}
 942                 else:
 943                     statdict['p_receivesync'] = {'IN SYNC': 'False'}
 944             else:
 945                 logging.warning("XenaPort: unknown stats: %s", param[1])
 946
 947         mydict = statdict
 948         return mydict
 949
 950
 951 class XenaTXStats(object):
 952     """
 953     Xena transmit stat class
 954     """
 955     def __init__(self, stats, epoc):
 956         """ Constructor
 957         :param stats: Stats from pt all command as list
 958         :param epoc: Current time in epoc
 959         :return: XenaTXStats object
 960         """
 961         self._stats = stats
 962         self._time = epoc
 963         self._ptstreamkeys = list()
 964         self.data = self.parse_stats()
 965         self.preamble = 8
 966
 967     @staticmethod
 968     def _pack_stats(params, start, fields=None):
 969         """ Pack up the list of stats in a dictionary
 970         :param params: The list of params to process
 971         :param start: What element to start at
 972         :param fields: The field names to pack as keys
 973         :return: Dictionary of data where fields match up to the params
 974         """
 975         if not fields:
 976             fields = ['bps', 'pps', 'bytes', 'packets']
 977         data = {}
 978         i = 0
 979         for column in fields:
 980             data[column] = int(params[start + i])
 981             i += 1
 982
 983         return data
 984
 985     def _pack_txextra_stats(self, params, start):
 986         """ Pack up the tx extra stats
 987         :param params: List of params to pack
 988         :param start: What element to start at
 989         :return: Dictionary of stats
 990         """
 991         fields = ['arprequests', 'arpreplies', 'pingrequests', 'pingreplies',
 992                   'injectedfcs', 'injectedseq', 'injectedmis', 'injectedint',
 993                   'injectedtid', 'training']
 994         return self._pack_stats(params, start, fields)
 995
 996     @property
 997     def pt_stream_keys(self):
 998         """
 999         :return: Return a list of pt_stream_x stream key ids
1000         """
1001         return self._ptstreamkeys
1002
1003     @property
1004     def time(self):
1005         """
1006         :return: Time as String of epoc of when stats were collected
1007         """
1008         return self._time
1009
1010     def parse_stats(self):
1011         """ Parse the stats from pr all command
1012         :return: Dictionary of all stats
1013         """
1014         statdict = {}
1015         for line in self._stats:
1016             param = line.split()
1017             if param[1] == 'PT_TOTAL':
1018                 statdict['pt_total'] = self._pack_stats(param, 2)
1019             elif param[1] == 'PT_NOTPLD':
1020                 statdict['pt_notpld'] = self._pack_stats(param, 2,)
1021             elif param[1] == 'PT_EXTRA':
1022                 statdict['pt_extra'] = self._pack_txextra_stats(param, 2)
1023             elif param[1] == 'PT_STREAM':
1024                 entry_id = "pt_stream_%s" % param[2].strip('[]')
1025                 self._ptstreamkeys.append(entry_id)
1026                 statdict[entry_id] = self._pack_stats(param, 3)
1027             else:
1028                 logging.warning("XenaPort: unknown stats: %s", param[1])
1029         mydict = statdict
1030         return mydict
1031
1032 def aggregate_stats(stat1, stat2):
1033     """
1034     Judge whether stat1 and stat2 both have same key, if both have same key,
1035     call the aggregate fuction, else use the stat1's value
1036     """
1037     newstat = dict()
1038     for keys in stat1.keys():
1039         if keys in stat2 and isinstance(stat1[keys], dict):
1040             newstat[keys] = aggregate(stat1[keys], stat2[keys])
1041         else:
1042             newstat[keys] = stat1[keys]
1043     return newstat
1044
1045 def aggregate(stat1, stat2):
1046     """
1047     Recursive function to aggregate two sets of statistics. This is used when
1048     bi directional traffic is done and statistics need to be calculated based
1049     on two sets of statistics.
1050     :param stat1: One set of dictionary stats from RX or TX stats
1051     :param stat2: Second set of dictionary stats from RX or TX stats
1052     :return: stats for data entry in RX or TX Stats instance
1053     """
1054     newstat = dict()
1055     for (keys1, keys2) in zip(stat1.keys(), stat2.keys()):
1056         if isinstance(stat1[keys1], dict):
1057             newstat[keys1] = aggregate(stat1[keys1], stat2[keys2])
1058         else:
1059             if not isinstance(stat1[keys1], int) and not isinstance(
1060                     [keys1], float):
1061                 # its some value we don't need to aggregate
1062                 return stat1[keys1]
1063             # for latency stats do the appropriate calculation
1064             if keys1 == 'max':
1065                 newstat[keys1] = max(stat1[keys1], stat2[keys2])
1066             elif keys1 == 'min':
1067                 newstat[keys1] = min(stat1[keys1], stat2[keys2])
1068             elif keys1 == 'avg':
1069                 newstat[keys1] = (stat1[keys1] + stat2[keys2]) / 2
1070             else:
1071                 newstat[keys1] = (stat1[keys1] + stat2[keys2])
1072     return newstat
1073
1074
1075 def line_percentage(port, stats, time_active, packet_size):
1076     """
1077     Calculate the line percentage rate from the duration, port object and stat
1078     object.
1079     :param port: XenaPort object
1080     :param stats: Xena RXStat or TXStat object
1081     :param time_active: time the stream was active in secs as int
1082     :param packet_size: packet size as int
1083     :return: line percentage as float
1084     """
1085     # this is ugly, but its prettier than calling the get method 3 times...
1086     try:
1087         packets = stats.data['pr_total']['packets']
1088     except KeyError:
1089         try:
1090             packets = stats.data['pt_total']['packets']
1091         except KeyError:
1092             _LOGGER.error(
1093                 'Could not calculate line rate because packet stat not found.')
1094             return 0
1095     ifg = port.get_inter_frame_gap()
1096     pps = packets_per_second(packets, time_active)
1097     l2br = l2_bit_rate(packet_size, stats.preamble, pps)
1098     l1br = l1_bit_rate(l2br, pps, ifg, stats.preamble)
1099     return 100.0 * l1br / port.get_effective_speed()
1100
1101
1102 def l2_bit_rate(packet_size, preamble, pps):
1103     """
1104     Return the l2 bit rate
1105     :param packet_size: packet size on the line in bytes
1106     :param preamble: preamble size of the packet header in bytes
1107     :param pps: packets per second
1108     :return: l2 bit rate as float
1109     """
1110     return (packet_size * preamble) * pps
1111
1112
1113 def l1_bit_rate(l2br, pps, ifg, preamble):
1114     """
1115     Return the l1 bit rate
1116     :param l2br: l2 bit rate int bits per second
1117     :param pps: packets per second
1118     :param ifg: the inter frame gap
1119     :param preamble: preamble size of the packet header in bytes
1120     :return: l1 bit rate as float
1121     """
1122     return l2br + (pps * ifg * preamble)
1123
1124
1125 def make_manager_command(cmd, argument=None):
1126     """ String builder for Xena socket commands
1127
1128     :param cmd: Command to send
1129     :param argument: Arguments for command to send
1130     :return: String of command
1131     """
1132     command = '{} "{}"'.format(cmd, argument) if argument else cmd
1133     _LOGGER.info("[Command Sent] : %s", command)
1134     return command
1135
1136
1137 def make_port_command(cmd, xena_port):
1138     """ String builder for Xena port commands
1139
1140     :param cmd: Command to send
1141     :param xena_port: XenaPort object
1142     :return: String of command
1143     """
1144     command = "{} {}".format(xena_port.port_string(), cmd)
1145     _LOGGER.info("[Command Sent] : %s", command)
1146     return command
1147
1148
1149 def make_stream_command(cmd, args, xena_stream):
1150     """ String builder for Xena port commands
1151
1152     :param cmd: Command to send
1153     :param xena_stream: XenaStream object
1154     :return: String of command
1155     """
1156     command = "{} {} [{}] {}".format(xena_stream.xena_port.port_string(), cmd,
1157                                      xena_stream.stream_id, args)
1158     _LOGGER.info("[Command Sent] : %s", command)
1159     return command
1160
1161
1162 def packets_per_second(packets, time_in_sec):
1163     """
1164     Return the pps as float
1165     :param packets: total packets
1166     :param time_in_sec: time in seconds
1167     :return: float of pps
1168     """
1169     return packets / time_in_sec