summaryrefslogtreecommitdiffstats
path: root/kernel/sound/arm/pxa2xx-pcm-lib.c
blob: e9b98af6b52c83faecccd4cc1011286a75560849 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 */

#include <linux/slab.h>
#include <linux/module.h>
#include <linux/dma-mapping.h>
#include <linux/dmaengine.h>
#include <linux/dma/pxa-dma.h>

#include <sound/core.h>
#include <sound/pcm.h>
#include <sound/pcm_params.h>
#include <sound/pxa2xx-lib.h>
#include <sound/dmaengine_pcm.h>

#include "pxa2xx-pcm.h"

static const struct snd_pcm_hardware pxa2xx_pcm_hardware = {
	.info			= SNDRV_PCM_INFO_MMAP |
				  SNDRV_PCM_INFO_MMAP_VALID |
				  SNDRV_PCM_INFO_INTERLEAVED |
				  SNDRV_PCM_INFO_PAUSE |
				  SNDRV_PCM_INFO_RESUME,
	.formats		= SNDRV_PCM_FMTBIT_S16_LE |
					SNDRV_PCM_FMTBIT_S24_LE |
					SNDRV_PCM_FMTBIT_S32_LE,
	.period_bytes_min	= 32,
	.period_bytes_max	= 8192 - 32,
	.periods_min		= 1,
	.periods_max		= 256,
	.buffer_bytes_max	= 128 * 1024,
	.fifo_size		= 32,
};

int __pxa2xx_pcm_hw_params(struct snd_pcm_substream *substream,
				struct snd_pcm_hw_params *params)
{
	struct dma_chan *chan = snd_dmaengine_pcm_get_chan(substream);
	struct snd_soc_pcm_runtime *rtd = substream->private_data;
	struct snd_dmaengine_dai_dma_data *dma_params;
	struct dma_slave_config config;
	int ret;

	dma_params = snd_soc_dai_get_dma_data(rtd->cpu_dai, substream);
	if (!dma_params)
		return 0;

	ret = snd_hwparams_to_dma_slave_config(substream, params, &config);
	if (ret)
		return ret;

	snd_dmaengine_pcm_set_config_from_dai_data(substream,
			snd_soc_dai_get_dma_data(rtd->cpu_dai, substream),
			&config);

	ret = dmaengine_slave_config(chan, &config);
	if (ret)
		return ret;

	snd_pcm_set_runtime_buffer(substream, &substream->dma_buffer);

	return 0;
}
EXPORT_SYMBOL(__pxa2xx_pcm_hw_params);

int __pxa2xx_pcm_hw_free(struct snd_pcm_substream *substream)
{
	snd_pcm_set_runtime_buffer(substream, NULL);
	return 0;
}
EXPORT_SYMBOL(__pxa2xx_pcm_hw_free);

int pxa2xx_pcm_trigger(struct snd_pcm_substream *substream, int cmd)
{
	return snd_dmaengine_pcm_trigger(substream, cmd);
}
EXPORT_SYMBOL(pxa2xx_pcm_trigger);

snd_pcm_uframes_t
pxa2xx_pcm_pointer(struct snd_pcm_substream *substream)
{
	return snd_dmaengine_pcm_pointer(substream);
}
EXPORT_SYMBOL(pxa2xx_pcm_pointer);

int __pxa2xx_pcm_prepare(struct snd_pcm_substream *substream)
{
	return 0;
}
EXPORT_SYMBOL(__pxa2xx_pcm_prepare);

int __pxa2xx_pcm_open(struct snd_pcm_substream *substream)
{
	struct snd_soc_pcm_runtime *rtd = substream->private_data;
	struct snd_pcm_runtime *runtime = substream->runtime;
	struct snd_dmaengine_dai_dma_data *dma_params;
	int ret;

	runtime->hw = pxa2xx_pcm_hardware;

	dma_params = snd_soc_dai_get_dma_data(rtd->cpu_dai, substream);
	if (!dma_params)
		return 0;

	/*
	 * For mysterious reasons (and despite what the manual says)
	 * playback samples are lost if the DMA count is not a multiple
	 * of the DMA burst size.  Let's add a rule to enforce that.
	 */
	ret = snd_pcm_hw_constraint_step(runtime, 0,
		SNDRV_PCM_HW_PARAM_PERIOD_BYTES, 32);
	if (ret)
		return ret;

	ret = snd_pcm_hw_constraint_step(runtime, 0,
		SNDRV_PCM_HW_PARAM_BUFFER_BYTES, 32);
	if (ret)
		return ret;

	ret = snd_pcm_hw_constraint_integer(runtime,
					    SNDRV_PCM_HW_PARAM_PERIODS);
	if (ret < 0)
		return ret;

	return snd_dmaengine_pcm_open_request_chan(substream,
					pxad_filter_fn,
					dma_params->filter_data);
}
EXPORT_SYMBOL(__pxa2xx_pcm_open);

int __pxa2xx_pcm_close(struct snd_pcm_substream *substream)
{
	return snd_dmaengine_pcm_close_release_chan(substream);
}
EXPORT_SYMBOL(__pxa2xx_pcm_close);

int pxa2xx_pcm_mmap(struct snd_pcm_substream *substream,
	struct vm_area_struct *vma)
{
	struct snd_pcm_runtime *runtime = substream->runtime;
	return dma_mmap_writecombine(substream->pcm->card->dev, vma,
				     runtime->dma_area,
				     runtime->dma_addr,
				     runtime->dma_bytes);
}
EXPORT_SYMBOL(pxa2xx_pcm_mmap);

int pxa2xx_pcm_preallocate_dma_buffer(struct snd_pcm *pcm, int stream)
{
	struct snd_pcm_substream *substream = pcm->streams[stream].substream;
	struct snd_dma_buffer *buf = &substream->dma_buffer;
	size_t size = pxa2xx_pcm_hardware.buffer_bytes_max;
	buf->dev.type = SNDRV_DMA_TYPE_DEV;
	buf->dev.dev = pcm->card->dev;
	buf->private_data = NULL;
	buf->area = dma_alloc_writecombine(pcm->card->dev, size,
					   &buf->addr, GFP_KERNEL);
	if (!buf->area)
		return -ENOMEM;
	buf->bytes = size;
	return 0;
}
EXPORT_SYMBOL(pxa2xx_pcm_preallocate_dma_buffer);

void pxa2xx_pcm_free_dma_buffers(struct snd_pcm *pcm)
{
	struct snd_pcm_substream *substream;
	struct snd_dma_buffer *buf;
	int stream;

	for (stream = 0; stream < 2; stream++) {
		substream = pcm->streams[stream].substream;
		if (!substream)
			continue;
		buf = &substream->dma_buffer;
		if (!buf->area)
			continue;
		dma_free_writecombine(pcm->card->dev, buf->bytes,
				      buf->area, buf->addr);
		buf->area = NULL;
	}
}
EXPORT_SYMBOL(pxa2xx_pcm_free_dma_buffers);

MODULE_AUTHOR("Nicolas Pitre");
MODULE_DESCRIPTION("Intel PXA2xx sound library");
MODULE_LICENSE("GPL");
pan class="n">config.read(self.cfp) self.stub = None self.dut_check = 0 self.tgen_check = 0 def get_mode(self): """read the mode for the client""" return self.config.get('Mode', 'mode') def get_deploy_channel_info(self): """get the channel data""" return (self.config.get('DeployServer', 'ip'), self.config.get('DeployServer', 'port')) def get_test_channel_info(self): """get the channel for tgen""" return (self.config.get('TestServer', 'ip'), self.config.get('TestServer', 'port')) def create_stub(self, channel): """create stub to talk to controller""" self.stub = vsperf_pb2_grpc.ControllerStub(channel) def host_connect(self): """provice dut-host credential to controller""" global DUT_CHECK hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) connect_reply = self.stub.HostConnect(hostinfo) DUT_CHECK = 1 print(connect_reply.message) def tgen_connect(self): """provide tgen-host credential to controller""" global TGEN_CHECK tgeninfo = vsperf_pb2.HostInfo(ip=self.config.get('TGen', 'ip'), uname=self.config.get('TGen', 'uname'), pwd=self.config.get('TGen', 'pwd')) connect_reply = self.stub.TGenHostConnect(tgeninfo) TGEN_CHECK = 1 print(connect_reply.message) def host_connect_both(self): """provice dut-host credential to controller""" global DUT_CHECK hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) connect_reply = self.stub.HostConnect(hostinfo) client = VsperfClient() client.automatically_test_dut_connect() DUT_CHECK = 1 print(connect_reply.message) def tgen_connect_both(self): """provide tgen-host credential to controller""" global TGEN_CHECK tgeninfo = vsperf_pb2.HostInfo(ip=self.config.get('TGen', 'ip'), uname=self.config.get('TGen', 'uname'), pwd=self.config.get('TGen', 'pwd')) connect_reply = self.stub.TGenHostConnect(tgeninfo) TGEN_CHECK = 1 client = VsperfClient() client.automatically_test_tgen_connect() print(connect_reply.message) @classmethod def automatically_test_dut_connect(cls): """handle automatic connection with tgen""" client = VsperfClient() ip_add, port = client.get_test_channel_info() channel = grpc.insecure_channel(ip_add + ':' + port) client.create_stub(channel) client.host_testcontrol_connect() @classmethod def automatically_test_tgen_connect(cls): """handle automatic connection with tgen""" client = VsperfClient() ip_add, port = client.get_test_channel_info() channel = grpc.insecure_channel(ip_add + ':' + port) client.create_stub(channel) client.tgen_testcontrol_connect() def exit_section(self): """exit""" @classmethod def section_execute(cls, menuitems, client, ip_add, port): """it will use to enter into sub-option""" channel = grpc.insecure_channel(ip_add + ':' + port) while True: client.create_stub(channel) while True: # os.system('clear') print(colorize(HEADER, 'blue')) print(colorize('version 0.1\n', 'pink')) for item in menuitems: print(colorize("[" + str(menuitems.index(item)) + "]", 'green') + list(item.keys())[0]) choice = input(">> ") try: if int(choice) < 0: raise ValueError if (int(choice) >= 0) and (int(choice) < (len(menuitems) - 1)): list(menuitems[int(choice)].values())[0]() else: break except (ValueError, IndexError): pass break @classmethod def get_user_trex_conf_location(cls): """Ask user for t-rex configuration location""" while True: filename_1 = str(input("Provide correct location for your t-rex configuration " \ "file where trex_cfg.yaml exist\n" \ "***************** Make Sure You Choose Correct" \ " File for Upload*******************\n" \ "Provide location: \n")) user_file = Path("{}".format(filename_1.strip())) if user_file.is_file(): break else: print("**************File Does Not Exist*****************\n") continue return filename_1 def upload_tgen_config(self): """t-rex config file as a chunk to controller""" if TGEN_CHECK == 0: return print("TGen-Host is not Connected [!]" \ "\nMake sure to establish connection with TGen-Host.") default_location = self.config.get('ConfFile', 'tgenpath') if not default_location: filename = self.get_user_trex_conf_location() else: user_preference = str(input("Use location specified in vsperfclient.conf?[Y/N] :")) while True: if 'y' in user_preference.lower().strip(): filename = self.config.get('ConfFile', 'tgenpath') user_file = Path("{}".format(filename.strip())) if user_file.is_file(): break else: print("**************File Does Not Exist*****************\n") user_preference = 'n' continue elif 'n' in user_preference.lower().strip(): filename = self.get_user_trex_conf_location() break else: print("Invalid Input") user_preference = str(input("Use location specified in vsperfclient.conf?" \ "[Y/N] : ")) continue filename = filename.strip() chunks = self.get_file_chunks_1(filename) upload_status = self.stub.TGenUploadConfigFile(chunks) print(upload_status.Message) def vsperf_install(self): """vsperf install on dut-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) install_reply = self.stub.VsperfInstall(hostinfo) print(install_reply.message) def collectd_install(self): """collectd install on dut-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) install_reply = self.stub.CollectdInstall(hostinfo) print(install_reply.message) def tgen_install(self): """install t-rex on Tgen host""" tgeninfo = vsperf_pb2.HostInfo(ip=self.config.get('TGen', 'ip'), uname=self.config.get('TGen', 'uname'), pwd=self.config.get('TGen', 'pwd')) install_reply = self.stub.TGenInstall(tgeninfo) print(install_reply.message) @classmethod def get_user_conf_location(cls): """get user input for test configuration file""" while True: filename_1 = str(input("Provide correct location for your test configuration " \ "file where it exist\n" \ "***************** Make Sure You Choose Correct" \ " Test File for Upload*******************\n" \ "Provide location: \n")) user_file = Path("{}".format(filename_1.strip())) if user_file.is_file(): break else: print("**************File Does Not Exist*****************\n") continue return filename_1 def upload_config(self): """transfer config file as a chunk to controller""" if DUT_CHECK == 0: return print("DUT-Host is not Connected [!]" \ "\nMake sure to establish connection with DUT-Host.") default_location = self.config.get('ConfFile', 'path') if not default_location: filename = self.get_user_conf_location() else: user_preference = str(input("Use location specified in vsperfclient.conf?[Y/N] :")) while True: if 'y' in user_preference.lower().strip(): filename = self.config.get('ConfFile', 'path') user_file = Path("{}".format(filename.strip())) if user_file.is_file(): break else: print("**************File Does Not Exist*****************\n") user_preference = 'n' continue elif 'n' in user_preference.lower().strip(): filename = self.get_user_conf_location() break else: print("Invalid Input") user_preference = str(input("Use location specified in vsperfclient.conf?" \ "[Y/N] : ")) continue filename = filename.strip() upload_param = self.get_file_chunks(filename) upload_status = self.stub.UploadConfigFile(upload_param) print(upload_status.Message) def start_test(self): """start test parameter, test config file and test name""" test_control = vsperf_pb2.ControlVsperf(testtype=self.config.get('Testcase', 'test'), \ conffile=self.config.get('Testcase', 'conffile')) control_reply = self.stub.StartTest(test_control) print(control_reply.message) def start_tgen(self): """start t-rex traffic generetor on tgen-host""" tgen_control = vsperf_pb2.ControlTGen(params=self.config.get('TGen', 'params')) control_reply = self.stub.StartTGen(tgen_control) print(control_reply.message) @classmethod def get_file_chunks(cls, filename): """convert file into chunk to stream between client and controller with filename""" with open(filename, 'rb') as f_1: while True: file_path = filename file_path_list = file_path.split("/") test_filename = file_path_list[(len(file_path_list)-1)] piece = f_1.read(CHUNK_SIZE) if not piece: return None return vsperf_pb2.ConfFileTest(Content=piece, Filename=test_filename) @classmethod def get_file_chunks_1(cls, filename): """Convert file into chunks""" with open(filename, 'rb') as f: while True: piece = f.read(CHUNK_SIZE) if len(piece) == 0: return yield vsperf_pb2.ConfFile(Content=piece) def test_status(self): """check the test_status""" test_check = vsperf_pb2.StatusQuery( testtype=self.config.get('Testcase', 'test')) check_result_reply = self.stub.TestStatus(test_check) print(check_result_reply.message) def vsperf_terminate(self): """after running test terminate vsperf on dut host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) termination_reply = self.stub.TerminateVsperf(hostinfo) print(termination_reply.message) def start_beats(self): """start beats on dut-host before running test""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.StartBeats(hostinfo) print(status_reply.message) def remove_vsperf(self): """remove vsperf from dut-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.RemoveVsperf(hostinfo) print(status_reply.message) def remove_result_folder(self): """remove resutl folder from dut-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.RemoveResultFolder(hostinfo) print(status_reply.message) def remove_config_files(self): """remove all config files""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.RemoveUploadedConfig(hostinfo) print(status_reply.message) def remove_collectd(self): """remove collectd from dut-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.RemoveCollectd(hostinfo) print(status_reply.message) def remove_everything(self): """remove everything from dut host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.RemoveEverything(hostinfo) print(status_reply.message) def sanity_nic_check(self): """nic is available on tgen host check""" tgeninfo = vsperf_pb2.HostInfo(ip=self.config.get('TGen', 'ip'), uname=self.config.get('TGen', 'uname'), pwd=self.config.get('TGen', 'pwd')) status_reply = self.stub.SanityNICCheck(tgeninfo) print(status_reply.message) def sanity_collectd_check(self): """check collecd properly running""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.SanityCollectdCheck(hostinfo) print(status_reply.message) def cpu_allocation_check(self): """check cpu allocation""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.SanityCPUAllocationCheck(hostinfo) print(status_reply.message) def sanity_vnf_path(self): """vnf path available on dut host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.SanityVNFpath(hostinfo) print(status_reply.message) def sanity_vsperf_check(self): """check vsperf correctly installed""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.SanityVSPERFCheck(hostinfo) print(status_reply.message) def sanity_dut_tgen_conn_check(self): """check the connection between dut-host and tgen-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.SanityTgenConnDUTCheck(hostinfo) print(status_reply.message) def dut_test_availability(self): """dut-host is free for test check""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.DUTvsperfTestAvailability(hostinfo) print(status_reply.message) def get_test_conf_from_dut(self): """get the vsperf test config file from dut host for user to check""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) status_reply = self.stub.GetVSPERFConffromDUT(hostinfo) print(status_reply.message) def dut_hugepage_config(self): """setup hugepages on dut-host""" configparam = vsperf_pb2.HugepConf(hpmax=self.config.get('HugepageConfig', 'HpMax'), \ hprequested=self.config.get('HugepageConfig',\ 'HpRequested')) config_status_reply = self.stub.DutHugepageConfig(configparam) print(config_status_reply.message) @classmethod def get_user_collectd_conf_location(cls): """get collectd configuration file location from user""" while True: filename_1 = str(input("Provide correct location for your collectd configuration " \ "file where collectd.conf exist\n" \ "***************** Make Sure You Choose Correct" \ " File for Upload*******************\n" \ "Provide location: \n")) user_file = Path("{}".format(filename_1.strip())) if user_file.is_file(): break else: print("**************File Does Not Exist*****************\n") continue return filename_1 def host_testcontrol_connect(self): """provice dut-host credential to test controller""" global DUT_CHECK hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) self.stub.HostConnect(hostinfo) def tgen_testcontrol_connect(self): """provide tgen-host credential to test controller""" global TGEN_CHECK tgeninfo = vsperf_pb2.HostInfo(ip=self.config.get('TGen', 'ip'), uname=self.config.get('TGen', 'uname'), pwd=self.config.get('TGen', 'pwd')) self.stub.TGenHostConnect(tgeninfo) def upload_collectd_config(self): """collectd config file chunks forwarded to controller""" if DUT_CHECK == 0: return print("DUT-Host is not Connected [!]" \ "\nMake sure to establish connection with DUT-Host.") default_location = self.config.get('ConfFile', 'collectdpath') if not default_location: filename = self.get_user_collectd_conf_location() else: user_preference = str(input("Use location specified in vsperfclient.conf?[Y/N] :")) while True: if 'y' in user_preference.lower().strip(): filename = self.config.get('ConfFile', 'collectdpath') user_file = Path("{}".format(filename.strip())) if user_file.is_file(): break else: print("**************File Does Not Exist*****************\n") user_preference = 'n' continue elif 'n' in user_preference.lower().strip(): filename = self.get_user_collectd_conf_location() break else: print("Invalid Input") user_preference = str(input("Use location specified in vsperfclient.conf?" \ "[Y/N] : ")) continue filename = filename.strip() chunks = self.get_file_chunks_1(filename) upload_status = self.stub.CollectdUploadConfig(chunks) print(upload_status.Message) def dut_check_dependecies(self): """check_dependecies on dut-host""" hostinfo = vsperf_pb2.HostInfo(ip=self.config.get('Host', 'ip'), uname=self.config.get('Host', 'uname'), pwd=self.config.get('Host', 'pwd')) check_reply = self.stub.CheckDependecies(hostinfo) print(check_reply.message) @classmethod def establish_connection_both(cls): """ This Function use to establish connection for vsperf to both the deploy server \ and testcontrol server """ client = VsperfClient() ip_add, port = client.get_deploy_channel_info() print("Establish connection for vsperf") menuitems_connection = [ {"Connect to DUT Host": client.host_connect_both}, {"Connect to TGen Host": client.tgen_connect_both}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_connection, client, ip_add, port) @classmethod def establish_connection_deploy(cls): """ This Function use to establish connection for vsperf to either using the dploy or using the testcontrol server """ client = VsperfClient() ip_add, port = client.get_deploy_channel_info() print("Establish connection for vsperf") menuitems_connection = [ {"Connect to DUT Host": client.host_connect}, {"Connect to TGen Host": client.tgen_connect}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_connection, client, ip_add, port) @classmethod def establish_connection_test(cls): """ This Function use to establish connection for vsperf to either using the dploy or using the testcontrol server """ client = VsperfClient() ip_add, port = client.get_test_channel_info() print("Establish connection for vsperf") menuitems_connection = [ {"Connect to DUT Host": client.host_connect}, {"Connect to TGen Host": client.tgen_connect}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_connection, client, ip_add, port) @classmethod def vsperf_setup(cls): """setup sub-options""" client = VsperfClient() ip_add, port = client.get_deploy_channel_info() print("Prerequisites Installation for VSPERF") menuitems_setup = [ {"Install VSPERF": client.vsperf_install}, {"Install TGen ": client.tgen_install}, {"Install Collectd": client.collectd_install}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) @classmethod def upload_config_files(cls): """all the upload sub-options""" client = VsperfClient() ip_add, port = client.get_deploy_channel_info() menuitems_setup = [ {"Upload TGen Configuration File": client.upload_tgen_config}, {"Upload Collectd Configuration File": client.upload_collectd_config}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) @classmethod def manage_sysparam_config(cls): """manage system parameter on dut host before run test""" client = VsperfClient() ip_add, port = client.get_deploy_channel_info() menuitems_setup = [ {"DUT-Host hugepages configuration": client.dut_hugepage_config}, {"Check VSPERF Dependencies on DUT-Host": client.dut_check_dependecies}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) @classmethod def test_status_check(cls): """after running test , test status related sub-options""" client = VsperfClient() ip_add, port = client.get_test_channel_info() menuitems_setup = [ {"Test status": client.test_status}, {"Get Test Configuration file from DUT-host": client.get_test_conf_from_dut}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) @classmethod def sanity_check_options(cls): """all sanity check sub-options""" client = VsperfClient() ip_add, port = client.get_test_channel_info() menuitems_setup = [ {"Check installed VSPERF": client.sanity_vsperf_check}, {"Check Test Config's VNF path is available on DUT-Host": client.sanity_vnf_path}, {"Check NIC PCIs is available on Traffic Generator": client.sanity_nic_check}, {"Check CPU allocation on DUT-Host": client.cpu_allocation_check}, {"Check installed Collectd": client.sanity_collectd_check}, {"Check Connection between DUT-Host and Traffic Generator Host": client.sanity_dut_tgen_conn_check}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) @classmethod def run_test(cls): """run test sub-options""" print("**Before user Run Tests we highly recommend user to perform Sanity Checks.......") client = VsperfClient() ip_add, port = client.get_test_channel_info() menuitems_setup = [ {"Upload Test Configuration File": client.upload_config}, {"Perform Sanity Checks before running tests": client.sanity_check_options}, {"Check if DUT-HOST is available": client.dut_test_availability}, {"Start TGen ": client.start_tgen}, {"Start Beats": client.start_beats}, {"Start Test": client.start_test}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) @classmethod def clean_up(cls): """clean-up sub-options""" print( "*******************************************************************\n\n\ IF you are performing Test on IntelPOD 12 - Node 4, Be careful during removal\n\n\ *******************************************************************") client = VsperfClient() ip_add, port = client.get_test_channel_info() menuitems_setup = [ {"Remove VSPERF": client.remove_vsperf}, {"Terminate VSPERF": client.vsperf_terminate}, {"Remove Results from DUT-Host": client.remove_result_folder}, {"Remove Uploaded Configuration File": client.remove_config_files}, {"Remove Collectd": client.remove_collectd}, {"Remove Everything": client.remove_everything}, {"Return to Previous Menu": client.exit_section} ] client.section_execute(menuitems_setup, client, ip_add, port) def run(): """It will run the actul primary options""" client = VsperfClient() client_mode = client.get_mode() print(client_mode) if "deploy" in client_mode.lower().strip(): menuitems = [ {"Establish Connections": client.establish_connection_deploy}, {"Installation": client.vsperf_setup}, {"Upload Configuration Files": client.upload_config_files}, {"Manage DUT-System Configuration": client.manage_sysparam_config}, {"Exit": sys.exit} ] #ip_add, port = client.get_channel_info() #channel = grpc.insecure_channel(ip_add + ':' + port) while True: # os.system('clear') print(colorize(HEADER, 'blue')) print(colorize('version 0.1\n', 'pink')) for item in menuitems: print(colorize("[" + str(menuitems.index(item)) + "]", 'green') + list(item.keys())[0]) choice = input(">> ") try: if int(choice) < 0: raise ValueError list(menuitems[int(choice)].values())[0]() except (ValueError, IndexError): pass elif "test" in client_mode.lower().strip(): menuitems = [ {"Establish Connections": client.establish_connection_test}, {"Run Test": client.run_test}, {"Test Status": client.test_status_check}, {"Clean-Up": client.clean_up}, {"Exit": sys.exit} ] #ip_add, port = client.get_channel_info() #channel = grpc.insecure_channel(ip_add + ':' + port) while True: # os.system('clear') print(colorize(HEADER, 'blue')) print(colorize('version 0.1\n', 'pink')) for item in menuitems: print(colorize("[" + str(menuitems.index(item)) + "]", 'green') + list(item.keys())[0]) choice = input(">> ") try: if int(choice) < 0: raise ValueError list(menuitems[int(choice)].values())[0]() except (ValueError, IndexError): pass elif "together" in client_mode.lower().strip(): menuitems = [ {"Establish Connections": client.establish_connection_both}, {"Installation": client.vsperf_setup}, {"Upload Configuration Files": client.upload_config_files}, {"Manage DUT-System Configuration": client.manage_sysparam_config}, {"Run Test": client.run_test}, {"Test Status": client.test_status_check}, {"Clean-Up": client.clean_up}, {"Exit": sys.exit} ] #ip_add, port = client.get_channel_info() #channel = grpc.insecure_channel(ip_add + ':' + port) while True: # os.system('clear') print(colorize(HEADER, 'blue')) print(colorize('version 0.1\n', 'pink')) for item in menuitems: print(colorize("[" + str(menuitems.index(item)) + "]", 'green') + list(item.keys())[0]) choice = input(">> ") try: if int(choice) < 0: raise ValueError list(menuitems[int(choice)].values())[0]() except (ValueError, IndexError): pass else: print("You have not defined client mode in vsperfclient.conf [!]") if __name__ == '__main__': run()