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