PyPI - Python Version GitHub

Xena OpenAutomation Chimera Core

XOA Chimera Core provides a set of high-level APIs to administer Xena's network impairment emulator, Chimera for test automation.

Introduction

The XOA Chimera Core is a Python library that provides a user-friendly and powerful interface for automating network impairment emulation tasks using Xena Networks Emulator, Chimera.

The XOA Chimera Core is designed to be easy to use and integrate with other automation tools and frameworks. It provides a comprehensive set of methods and classes for interacting with Chimera, including the ability to create and run complex impairment scenarios.

The XOA Chimera Core simplifies the process of automating network impairment emulation tasks. It provides a simple, yet powerful, interface for interacting with Chimera using the Python programming language. With the XOA Chimera Core, testing professionals can easily create and execute network impairment scenarios, analyze traffic, and perform detailed analysis of 5G ORAN, all while leveraging the power and flexibility of the Python programming language.

Overall, the XOA Chimera Core is a valuable tool for anyone looking to automate their network emulation tasks using Xena Chimera. With its simple, yet powerful, interface and support for the Python programming language, the XOA Chimera Core provides a set of flexible and extensible APIs for automating network emulation and improving the quality of network infrastructure.

Chimera is a network impairment emulator that makes it easy to introduce consistent, accurate, well-defined and repeatable impairments (e.g. packet manipulation, packet drop, latency and jitter) to traffic between DUTs in the lab. All rates from 10GE to 100GE are supported by a single 2-slot test module.

Chimera is ideal for NEMs who want to optimize Quality of Experience (QoE) for customers by ensuring equipment can handle acceptable levels of impairment. Service providers, enterprises, and government agencies can also use Chimera to validate Quality of Service (QoS) for voice, video, and data traffic being sent across their networks.

Chimera can be installed as a test module in a ValkyrieBay chassis for seamless integration with ValkyrieManager. This provides a simple way to impair traffic being generated by Valkyrie from within the same UI - a unique feature for Chimera. Or it is available in a compact chassis for use as a standalone solution for emulating impairment on traffic between two DUTs.

Chimera can also be controlled using XOA CLI, a simple and easy to use command-line-interface scripting API making test automation easy and fast. The result is a cost-effective and powerful network impairment emulator solution for benchmarking, stress testing/negative testing, "what-if" testing, and regression testing.

IMPORTANT

For a brief product overview, go to Chimera Fact Sheet.

For complete details of how to use Chimera, go to Chimera User Manual.

For detailed technical specifications, go to Chimera Data Sheet.

Documentation

The user documentation is hosted: Xena OpenAutomation Chimera Core Documentation

Installation

Install Using `pip`

Make sure Python pip is installed on you system. If you are using virtualenv, then pip is already installed into environments created by virtualenv, and using sudo is not needed. If you do not have pip installed, download this file: https://bootstrap.pypa.io/get-pip.py and run python get-pip.py.

To install the latest, use pip to install from pypi:

~/> pip install chimera-core

To upgrade to the latest, use pip to upgrade from pypi:

~/> pip install chimera-core --upgrade

Install From Source Code

Make sure these packages are installed wheel, setuptools on your system.

Install setuptools using pip:

~/> pip install wheel setuptools

To install source of python packages:

/xoa_driver> python setup.py install

To build .whl file for distribution:

/xoa_driver> python setup.py bdist_wheel

Quick Start

Get Python pip if not already installed (Download https://bootstrap.pypa.io/get-pip.py): python get-pip.py
Install the latest chimera-core: pip install chimera-core -U

Write Python code to manage with Xena testers:

import asyncio
from functools import partial
from typing import List
from ipaddress import IPv4Address, IPv6Address
from loguru import logger

from chimera_core.controller import MainController
from chimera_core.types import distributions, enums, dataset

CHASSIS_IP = "10.20.1.252"
USERNAME = "chimera-core"
MODULE_IDX = 11
PORT_IDX = 0
FLOW_IDX = 1

async def my_awesome_func(stop_event: asyncio.Event):

    # create credential object
    credentials = dataset.Credentials(
        product=dataset.EProductType.VALKYRIE,
        host=CHASSIS_IP)

    #----------------------------------------------
    # 1. Connect to Valkyrie chassis and select port
    # ---------------------------------------------
    # region Connect to Valkyrie chassis and select port

    # create chimera core controller object
    controller = await MainController()

    # add chimera emulator into the chassis inventory and get the ID
    tester_id = await controller.add_tester(credentials=credentials)

    # create tester object
    tester = await controller.use_tester(tester_id, username=USERNAME, reserve=False, debug=False)

    # create module object
    module = await tester.use_module(module_id=MODULE_IDX, reserve=False)

    # free the module in case it is reserved by others
    await module.free(should_free_sub_resources=True)
    await module.reserve()

    # create port object and reserver the port
    port = await tester.use_port(module_id=MODULE_IDX, port_id=PORT_IDX, reserve=False)

    # reserve the port
    await port.reserve()

    # reset port
    await port.reset()

    # endregion

    #----------------------------------------------
    # 2. Configure Chimera port
    # ---------------------------------------------
    # region Configure Chimera port
    port_config = await port.config.get()
    port_config.comment = "My Chimera Port"

    port_config.set_autoneg_on()
    port_config.set_autoneg_off()

    port_config.set_fcs_error_mode_discard()
    port_config.set_fcs_error_mode_pass()

    port_config.set_link_flap(enable=enums.OnOff.ON, duration=100, period=1000, repetition=0)
    port_config.set_link_flap_off()

    port_config.set_pma_error_pulse(enable=enums.OnOff.ON, duration=100, period=1000, repetition=0, coeff=100, exp=-4)
    port_config.set_pma_error_pulse_off()

    port_config.set_impairment_off()
    port_config.set_impairment_on()

    await port.config.set(port_config)

    # endregion

    #----------------------------------------------
    # 3. Configure flow's basic filter on a port
    # ---------------------------------------------
    # region Flow configuration + basic filter on a port

    # Configure flow properties
    flow = port.flows[FLOW_IDX]
    flow_config = await flow.get()
    flow_config.comment = "On VLAN 111"
    await flow.set(config=flow_config)

    # Initialize shadow filter on the flow
    shadow_filter = flow.shadow_filter
    await shadow_filter.init()
    await shadow_filter.clear()

    # Configure shadow filter to BASIC mode
    basic_filter = await shadow_filter.use_basic_mode()
    basic_filter_config = await basic_filter.get()

    await basic_filter.set(basic_filter_config)
    await shadow_filter.enable()
    await shadow_filter.apply()

    # Configure flow properties
    flow = port.flows[FLOW_IDX]
    flow_config = await flow.get()
    flow_config.comment = "On VLAN 111"
    await flow.set(config=flow_config)

    # Initialize shadow filter on the flow
    shadow_filter = flow.shadow_filter
    await shadow_filter.init()
    await shadow_filter.clear()

    # Gaussian distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.Gaussian(mean=1, sd=1)
    dist.continuous()

    # Set distribution and start impairment Latency & Jitter
    latency_jitter_config = await flow.latency_jitter.get()
    latency_jitter_config.set_distribution(dist)
    await flow.latency_jitter.start(latency_jitter_config)
    await flow.latency_jitter.stop(latency_jitter_config)

    #------------------
    # Ethernet subfilter
    #------------------
    # Use and configure basic-mode shadow filter's Ethernet subfilter
    ethernet_subfilter = basic_filter_config.layer_2.use_ethernet()
    ethernet_subfilter.exclude()
    ethernet_subfilter.include()
    ethernet_subfilter.src_addr.on(value=dataset.Hex("AAAAAAAAAAAA"), mask=dataset.Hex("FFFFFFFFFFFF"))
    ethernet_subfilter.dest_addr.on(value=dataset.Hex("BBBBBBBBBBBB"), mask=dataset.Hex("FFFFFFFFFFFF"))

    #------------------
    # Layer 2+ subfilter
    #------------------
    # Not use basic-mode shadow filter's Layer 2+ subfilter
    layer_2_plus_subfilter = basic_filter_config.layer_2_plus.use_none()

    # Use and configure basic-mode shadow filter's Layer2+ subfilter (One VLAN tag)
    layer_2_plus_subfilter = basic_filter_config.layer_2_plus.use_1_vlan_tag()
    layer_2_plus_subfilter.off()
    layer_2_plus_subfilter.exclude()
    layer_2_plus_subfilter.include()
    layer_2_plus_subfilter.tag_inner.on(value=1234, mask=dataset.Hex("FFF"))
    layer_2_plus_subfilter.pcp_inner.on(value=3, mask=dataset.Hex("7"))

    # Use and configure basic-mode shadow filter's Layer2+ subfilter (Two VLAN tag)
    layer_2_plus_subfilter = basic_filter_config.layer_2_plus.use_2_vlan_tags()
    layer_2_plus_subfilter.off()
    layer_2_plus_subfilter.exclude()
    layer_2_plus_subfilter.include()
    layer_2_plus_subfilter.tag_inner.on(value=1234, mask=dataset.Hex("FFF"))
    layer_2_plus_subfilter.pcp_inner.on(value=3, mask=dataset.Hex("7"))
    layer_2_plus_subfilter.tag_outer.on(value=2345, mask=dataset.Hex("FFF"))
    layer_2_plus_subfilter.pcp_outer.on(value=0, mask=dataset.Hex("7"))

    # Use and configure basic-mode shadow filter's Layer2+ subfilter (MPLS)
    layer_2_plus_subfilter = basic_filter_config.layer_2_plus.use_mpls()
    layer_2_plus_subfilter.off()
    layer_2_plus_subfilter.exclude()
    layer_2_plus_subfilter.include()
    layer_2_plus_subfilter.label.on(value=1000, mask=dataset.Hex("FFFFF"))
    layer_2_plus_subfilter.toc.on(value=0, mask=dataset.Hex("7"))

    #------------------
    # Layer 3 subfilter
    #------------------
    # Not use basic-mode shadow filter's Layer 3 subfilter
    layer_3_subfilter = basic_filter_config.layer_3.use_none()

    # Use and configure basic-mode shadow filter's Layer 3 subfilter (IPv4)
    layer_3_subfilter = basic_filter_config.layer_3.use_ipv4()
    layer_3_subfilter.off()
    layer_3_subfilter.exclude()
    layer_3_subfilter.include()
    layer_3_subfilter.src_addr.on(value=IPv4Address("10.0.0.2"), mask=dataset.Hex("FFFFFFFF"))
    layer_3_subfilter.dest_addr.on(value=IPv4Address("11.0.0.2"), mask=dataset.Hex("FFFFFFFF"))
    layer_3_subfilter.dscp.on(value=0, mask=dataset.Hex("FC"))

    # Use and configure basic-mode shadow filter's Layer 3 subfilter (IPv6)
    layer_3_subfilter = basic_filter_config.layer_3.use_ipv6()
    layer_3_subfilter.exclude()
    layer_3_subfilter.include()
    layer_3_subfilter.src_addr.on(value=IPv6Address("2001::2"), mask=dataset.Hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF"))
    layer_3_subfilter.dest_addr.on(value=IPv6Address("2002::2"), mask=dataset.Hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF"))

    #------------------
    # Layer 4 subfilter
    #------------------
    # Not use basic-mode shadow filter's Layer 4 subfilter
    layer_4_subfilter = basic_filter_config.layer_4.use_none()

    # Use and configure basic-mode shadow filter's Layer 4 subfilter (TCP)
    layer_4_subfilter = basic_filter_config.layer_4.use_tcp()
    layer_4_subfilter.off()
    layer_4_subfilter.exclude()
    layer_4_subfilter.include()
    layer_4_subfilter.src_port.on(value=1234, mask=dataset.Hex("FFFF"))
    layer_4_subfilter.dest_port.on(value=80, mask=dataset.Hex("FFFF"))

    # Use and configure basic-mode shadow filter's Layer 4 subfilter (UDP)
    layer_4_subfilter = basic_filter_config.layer_4.use_udp()
    layer_4_subfilter.off()
    layer_4_subfilter.exclude()
    layer_4_subfilter.include()
    layer_4_subfilter.src_port.on(value=1234, mask=dataset.Hex("FFFF"))
    layer_4_subfilter.dest_port.on(value=80, mask=dataset.Hex("FFFF"))

    #------------------
    # Layer Xena subfilter
    #------------------
    # Not use basic-mode shadow filter's Layer Xena subfilter
    layer_xena_subfilter = basic_filter_config.layer_xena.use_none()

    # Use and configure basic-mode shadow filter's Layer 4 subfilter (TCP)
    layer_xena_subfilter = basic_filter_config.layer_xena.use_tpld()
    layer_xena_subfilter.exclude()
    layer_xena_subfilter.include()
    layer_xena_subfilter[0].on(tpld_id=2)       
    layer_xena_subfilter[0].off()
    layer_xena_subfilter[1].on(tpld_id=4)       
    layer_xena_subfilter[1].off()
    layer_xena_subfilter[2].on(tpld_id=6)       
    layer_xena_subfilter[2].off()
    layer_xena_subfilter[3].on(tpld_id=8)       
    layer_xena_subfilter[3].off()
    layer_xena_subfilter[4].on(tpld_id=10)       
    layer_xena_subfilter[4].off()
    layer_xena_subfilter[5].on(tpld_id=20)       
    layer_xena_subfilter[5].off()
    layer_xena_subfilter[6].on(tpld_id=40)       
    layer_xena_subfilter[6].off()
    layer_xena_subfilter[7].on(tpld_id=60)       
    layer_xena_subfilter[7].off()
    layer_xena_subfilter[8].on(tpld_id=80)       
    layer_xena_subfilter[8].off()
    layer_xena_subfilter[9].on(tpld_id=100)       
    layer_xena_subfilter[9].off()
    layer_xena_subfilter[10].on(tpld_id=102)       
    layer_xena_subfilter[10].off()
    layer_xena_subfilter[11].on(tpld_id=104)       
    layer_xena_subfilter[11].off()
    layer_xena_subfilter[12].on(tpld_id=106)       
    layer_xena_subfilter[12].off()
    layer_xena_subfilter[13].on(tpld_id=108)       
    layer_xena_subfilter[13].off()
    layer_xena_subfilter[14].on(tpld_id=110)       
    layer_xena_subfilter[14].off()
    layer_xena_subfilter[15].on(tpld_id=200)       
    layer_xena_subfilter[15].off()

    #------------------
    # Layer Any subfilter
    #------------------
    # Not use basic-mode shadow filter's Layer Any subfilter
    layer_any_subfilter = basic_filter_config.layer_any.use_none()

    # Use and configure basic-mode shadow filter's Layer 4 subfilter (TCP)
    layer_any_subfilter = basic_filter_config.layer_any.use_any_field()
    layer_any_subfilter.off()
    layer_any_subfilter.exclude()
    layer_any_subfilter.include()
    layer_any_subfilter.on(position=0, value=dataset.Hex("112233445566"), mask=dataset.Hex("112233445566"))

    # Enable and apply the basic filter settings
    await basic_filter.set(basic_filter_config)
    await shadow_filter.enable()
    await shadow_filter.apply()

    # endregion

    #----------------------------------------------
    # 4. Configure flow's extended filter on a port
    # ---------------------------------------------
    # region Flow configuration + extended filter on a port

    # Configure flow properties
    flow = port.flows[FLOW_IDX]
    flow_config = await flow.get()
    flow_config.comment = "On VLAN 111"
    await flow.set(config=flow_config)

    # Initialize shadow filter on the flow
    shadow_filter = flow.shadow_filter
    await shadow_filter.init()
    await shadow_filter.clear()

    # Configure shadow filter to EXTENDED mode
    extended_filter = await shadow_filter.use_extended_mode()
    extended_filter_config = await extended_filter.get()

    ethernet = dataset.ProtocolSegement(
        protocol_type=dataset.ProtocolOption.ETHERNET,
        value='000001111',
        mask='1110000',
    )
    ipv4_1 = dataset.ProtocolSegement(
        protocol_type=dataset.ProtocolOption.IP,
        value='000001111',
        mask='1110000',
    )
    ipv4_2 = dataset.ProtocolSegement(
        protocol_type=dataset.ProtocolOption.IP,
        value='000001111',
        mask='1110000',
    )
    extended_filter_config.protocol_segments = (ethernet, ipv4_1, ipv4_2)

    await extended_filter.set(extended_filter_config)
    await shadow_filter.enable()
    await shadow_filter.apply()

    # endregion

    #----------------------------------------------
    # 5. Configure impairment - Drop
    # ---------------------------------------------
    # region Configure impairment - Drop

    # Fixed Burst distribution for impairment Drop
    dist = distributions.drop.FixedBurst(burst_size=5)
    dist.repeat(period=5)
    dist.one_shot()

    # Random Burst distribution for impairment Drop
    dist = distributions.drop.RandomBurst(minimum=1, maximum=10, probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Fixed Rate distribution for impairment Drop
    dist = distributions.drop.FixedRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Bit Error Rate distribution for impairment Drop
    dist = distributions.drop.BitErrorRate(coefficient=1, exponent=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Random Rate distribution for impairment Drop
    dist = distributions.drop.RandomRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gilbert Elliot distribution for impairment Drop
    dist = distributions.drop.GilbertElliot(good_state_impair_prob=0, good_state_trans_prob=0, bad_state_impair_prob=0, bad_state_trans_prob=0)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Uniform distribution for impairment Drop
    dist = distributions.drop.Uniform(minimum=1, maximum=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gaussian distribution for impairment Drop
    dist = distributions.drop.Gaussian(mean=1, sd=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Poisson distribution for impairment Drop
    dist = distributions.drop.Poisson(lamda=9)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gamma distribution for impairment Drop
    dist = distributions.drop.Gamma(shape=1, scale=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Custom distribution for impairment Drop
    data_x=[0, 1] * 256
    custom_distribution = await port.custom_distributions.add(
        linear=False,
        entry_count = len(data_x),
        data_x=data_x,
        comment="Example Custom Distribution"
    )
    dist = distributions.drop.Custom(custom_distribution=custom_distribution)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Set distribution and start impairment Drop
    drop_config = await flow.drop.get()
    drop_config.set_distribution(dist)
    await flow.drop.start(drop_config)
    await flow.drop.stop(drop_config)

    # endregion

    #----------------------------------------------
    # 6. Configure impairment - Misordering
    # ---------------------------------------------
    # region Configure impairment - Misordering

    # Fixed Burst distribution for impairment Misordering
    dist = distributions.misordering.FixedBurst(burst_size=1)
    dist.repeat(period=5)
    dist.one_shot()

    # Fixed Rate distribution for impairment Misordering
    dist = distributions.misordering.FixedRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Set distribution and start impairment Misordering
    misordering_config = await flow.misordering.get()
    misordering_config.depth = 1
    misordering_config.set_distribution(dist)
    await flow.misordering.start(misordering_config)
    await flow.misordering.stop(misordering_config)

    # endregion

    #----------------------------------------------
    # 7. Configure impairment - Latency & Jitter
    # ---------------------------------------------
    # region Configure impairment - Latency & Jitter

    # Fixed Burst distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.ConstantDelay(delay=100)

    # Random Burst distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.AccumulateBurst(burst_delay=1300)
    dist.repeat(period=1)
    dist.one_shot()

    # Step distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.Step(min=1300, max=77000)
    dist.continuous()

    # Uniform distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.Uniform(minimum=1, maximum=1)
    dist.continuous()

    # Gaussian distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.Gaussian(mean=1, sd=1)
    dist.continuous()

    # Poisson distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.Poisson(lamda=9)
    dist.continuous()

    # Gamma distribution for impairment Latency & Jitter
    dist = distributions.latency_jitter.Gamma(shape=1, scale=1)
    dist.continuous()

    # Custom distribution for impairment Latency & Jitter
    data_x=[0, 1] * 256
    custom_distribution = await port.custom_distributions.add(
        linear=False,
        entry_count = len(data_x),
        data_x=data_x,
        comment="Example Custom Distribution"
    )
    dist = distributions.latency_jitter.Custom(custom_distribution=custom_distribution)
    dist.continuous()

    # Set distribution and start impairment Latency & Jitter
    latency_jitter_config = await flow.latency_jitter.get()
    latency_jitter_config.set_distribution(dist)
    await flow.latency_jitter.start(latency_jitter_config)
    await flow.latency_jitter.stop(latency_jitter_config)

    # endregion

    #----------------------------------------------
    # 8. Configure impairment - Duplication
    # ---------------------------------------------
    # region Configure impairment - Duplication

    # Fixed Burst distribution for impairment Duplication
    dist = distributions.duplication.FixedBurst(burst_size=5)
    dist.repeat(period=5)
    dist.one_shot()

    # Random Burst distribution for impairment Duplication
    dist = distributions.duplication.RandomBurst(minimum=1, maximum=10, probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Fixed Rate distribution for impairment Duplication
    dist = distributions.duplication.FixedRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Bit Error Rate distribution for impairment Duplication
    dist = distributions.duplication.BitErrorRate(coefficient=1, exponent=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Random Rate distribution for impairment Duplication
    dist = distributions.duplication.RandomRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gilbert Elliot distribution for impairment Duplication
    dist = distributions.duplication.GilbertElliot(good_state_impair_prob=0, good_state_trans_prob=0, bad_state_impair_prob=0, bad_state_trans_prob=0)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Uniform distribution for impairment Duplication
    dist = distributions.duplication.Uniform(minimum=1, maximum=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gaussian distribution for impairment Duplication
    dist = distributions.duplication.Gaussian(mean=1, sd=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Poisson distribution for impairment Duplication
    dist = distributions.duplication.Poisson(lamda=9)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gamma distribution for impairment Duplication
    dist = distributions.duplication.Gamma(shape=1, scale=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Custom distribution for impairment Duplication
    data_x=[0, 1] * 256
    custom_distribution = await port.custom_distributions.add(
        linear=False,
        entry_count = len(data_x),
        data_x=data_x,
        comment="Example Custom Distribution"
    )
    dist = distributions.duplication.Custom(custom_distribution=custom_distribution)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Set distribution and start impairment Duplication
    duplication_config = await flow.duplication.get()
    duplication_config.set_distribution(dist)
    await flow.duplication.start(duplication_config)
    await flow.duplication.stop(duplication_config)

    # endregion

    #----------------------------------------------
    # 9. Configure impairment - Corruption
    # ---------------------------------------------
    # region Configure impairment - Corruption

    # Fixed Burst distribution for impairment Corruption
    dist = distributions.corruption.FixedBurst(burst_size=5)
    dist.repeat(period=5)
    dist.one_shot()

    # Random Burst distribution for impairment Corruption
    dist = distributions.corruption.RandomBurst(minimum=1, maximum=10, probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Fixed Rate distribution for impairment Corruption
    dist = distributions.corruption.FixedRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Bit Error Rate distribution for impairment Corruption
    dist = distributions.corruption.BitErrorRate(coefficient=1, exponent=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Random Rate distribution for impairment Corruption
    dist = distributions.corruption.RandomRate(probability=10_000)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gilbert Elliot distribution for impairment Corruption
    dist = distributions.corruption.GilbertElliot(good_state_impair_prob=0, good_state_trans_prob=0, bad_state_impair_prob=0, bad_state_trans_prob=0)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Uniform distribution for impairment Corruption
    dist = distributions.corruption.Uniform(minimum=1, maximum=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gaussian distribution for impairment Corruption
    dist = distributions.corruption.Gaussian(mean=1, sd=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Poisson distribution for impairment Corruption
    dist = distributions.corruption.Poisson(lamda=9)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Gamma distribution for impairment Corruption
    dist = distributions.corruption.Gamma(shape=1, scale=1)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Custom distribution for impairment Corruption
    data_x=[0, 1] * 256
    custom_distribution = await port.custom_distributions.add(
        linear=False,
        entry_count = len(data_x),
        data_x=data_x,
        comment="Example Custom Distribution"
    )
    dist = distributions.corruption.Custom(custom_distribution=custom_distribution)
    dist.repeat_pattern(duration=1, period=1)
    dist.continuous()

    # Set distribution and start impairment Corruption
    corruption_config = await flow.corruption.get()
    corruption_config.corruption_type = enums.CorruptionType.ETH
    corruption_config.corruption_type = enums.CorruptionType.IP
    corruption_config.corruption_type = enums.CorruptionType.TCP
    corruption_config.corruption_type = enums.CorruptionType.UDP
    corruption_config.set_distribution(dist)
    await flow.corruption.start(corruption_config)
    await flow.corruption.stop(corruption_config)

    # endregion

    #----------------------------------------------
    # 10. Configure bandwidth control - Policer
    # ---------------------------------------------
    # region Configure bandwidth control - Policer

    # Set and start bandwidth control Policer
    policer_config = await flow.policer.get()
    policer_config.set_control_mode(mode=enums.PolicerMode.L1)
    policer_config.set_control_on_l1()
    policer_config.set_control_mode(mode=enums.PolicerMode.L2)
    policer_config.set_control_on_l2()
    policer_config.mode = enums.PolicerMode.L1
    policer_config.mode = enums.PolicerMode.L2
    policer_config.cir = 10_000
    policer_config.cbs = 1_000
    policer_config.set_on_off(on_off=enums.OnOff.ON)
    policer_config.set_on()
    policer_config.set_on_off(on_off=enums.OnOff.OFF)
    policer_config.set_off()
    await flow.policer.start(policer_config)
    await flow.policer.stop(policer_config)

    # endregion

    #----------------------------------------------
    # 11. Configure bandwidth control - Shaper
    # ---------------------------------------------
    # region Configure bandwidth control - Shaper

    # Set and start bandwidth control Shaper
    shaper_config = await flow.shaper.get()
    shaper_config.set_control_mode(mode=enums.PolicerMode.L1)
    shaper_config.set_control_on_l1()
    shaper_config.set_control_mode(mode=enums.PolicerMode.L2)
    shaper_config.set_control_on_l2()
    shaper_config.mode = enums.PolicerMode.L1
    shaper_config.mode = enums.PolicerMode.L2
    shaper_config.cir = 10_000
    shaper_config.cbs = 1_000
    shaper_config.buffer_size = 1_000
    shaper_config.set_on_off(on_off=enums.OnOff.ON)
    shaper_config.set_on()
    shaper_config.set_on_off(on_off=enums.OnOff.OFF)
    shaper_config.set_off()
    await flow.shaper.start(shaper_config)
    await flow.shaper.stop(shaper_config)

    # endregion

    #----------------------------------------------
    # 12. Flow statistics
    # ---------------------------------------------
    # region Flow Statistics

    rx_total = await flow.statistics.rx.total.get()
    rx_total.byte_count
    rx_total.packet_count
    rx_total.l2_bps
    rx_total.pps

    tx_total = await flow.statistics.tx.total.get()
    tx_total.byte_count
    tx_total.packet_count
    tx_total.l2_bps
    tx_total.pps

    flow_drop_total = await flow.statistics.total.dropped.get()
    flow_drop_total.pkt_drop_count_total
    flow_drop_total.pkt_drop_count_programmed
    flow_drop_total.pkt_drop_count_bandwidth
    flow_drop_total.pkt_drop_count_other
    flow_drop_total.pkt_drop_ratio_total
    flow_drop_total.pkt_drop_ratio_programmed
    flow_drop_total.pkt_drop_ratio_bandwidth
    flow_drop_total.pkt_drop_ratio_other

    flow_corrupted_total = await flow.statistics.total.corrupted.get()
    flow_corrupted_total.fcs_corrupted_pkt_count
    flow_corrupted_total.fcs_corrupted_pkt_ratio
    flow_corrupted_total.ip_corrupted_pkt_count
    flow_corrupted_total.ip_corrupted_pkt_ratio
    flow_corrupted_total.tcp_corrupted_pkt_count
    flow_corrupted_total.tcp_corrupted_pkt_ratio
    flow_corrupted_total.total_corrupted_pkt_count
    flow_corrupted_total.total_corrupted_pkt_ratio
    flow_corrupted_total.udp_corrupted_pkt_count
    flow_corrupted_total.udp_corrupted_pkt_ratio

    flow_delayed_total = await flow.statistics.total.delayed.get()
    flow_delayed_total.pkt_count
    flow_delayed_total.ratio

    flow_jittered_total = await flow.statistics.total.jittered.get()
    flow_jittered_total.pkt_count
    flow_jittered_total.ratio

    flow_duplicated_total = await flow.statistics.total.duplicated.get()
    flow_duplicated_total.pkt_count
    flow_duplicated_total.ratio

    flow_misordered_total = await flow.statistics.total.misordered.get()
    flow_misordered_total.pkt_count
    flow_misordered_total.ratio

    await flow.statistics.tx.clear.set()
    await flow.statistics.rx.clear.set()
    await flow.statistics.clear.set()

    # endregion

    #----------------------------------------------
    # 13. Port statistics
    # ---------------------------------------------
    # region Port Statistics
    port_drop = await port.config.statistics.dropped.get()
    port_drop.pkt_drop_count_total
    port_drop.pkt_drop_count_programmed
    port_drop.pkt_drop_count_bandwidth
    port_drop.pkt_drop_count_other
    port_drop.pkt_drop_ratio_total
    port_drop.pkt_drop_ratio_programmed
    port_drop.pkt_drop_ratio_bandwidth
    port_drop.pkt_drop_ratio_other

    port_corrupted = await port.config.statistics.corrupted.get()
    port_corrupted.fcs_corrupted_pkt_count
    port_corrupted.fcs_corrupted_pkt_ratio
    port_corrupted.ip_corrupted_pkt_count
    port_corrupted.ip_corrupted_pkt_ratio
    port_corrupted.tcp_corrupted_pkt_count
    port_corrupted.tcp_corrupted_pkt_ratio
    port_corrupted.total_corrupted_pkt_count
    port_corrupted.total_corrupted_pkt_ratio
    port_corrupted.udp_corrupted_pkt_count
    port_corrupted.udp_corrupted_pkt_ratio

    port_delayed = await port.config.statistics.delayed.get()
    port_delayed.pkt_count
    port_delayed.ratio

    port_jittered = await port.config.statistics.jittered.get()
    port_jittered.pkt_count
    port_jittered.ratio

    port_duplicated = await port.config.statistics.duplicated.get()
    port_duplicated.pkt_count
    port_duplicated.ratio

    port_misordered = await port.config.statistics.misordered.get()
    port_misordered.pkt_count
    port_misordered.ratio

    await port.config.statistics.clear.set()

    # endregion

async def main() -> None:
    stop_event = asyncio.Event()
    await my_awesome_func(stop_event=stop_event)

if __name__ == "__main__":
    asyncio.run(main())

FOR TESTING BEYOND THE STANDARD.

xenanetworks / open-automation-chimera-core

readme

Xena OpenAutomation Chimera Core

Introduction

Documentation

Installation

Install Using `pip`

Install From Source Code

Quick Start

xenanetworks / open-automation-chimera-core

readme

Xena OpenAutomation Chimera Core

Introduction

Documentation

Installation

Install Using pip

Install From Source Code

Quick Start

Install Using `pip`