# -*- coding: utf-8 -*- # !/usr/bin/env python import getopt import os import sys import threading import time PROJECT_ROOT = os.path.join(os.path.abspath(os.path.split(os.path.realpath(__file__))[0] + "/.."), '..') sys.path.insert(0, PROJECT_ROOT) try: options, args = getopt.getopt(sys.argv[1:], 'e:i:c:', ['env=', 'imei=', 'code=']) except getopt.GetoptError as e: print(str(e)) sys.exit() system_env = 'testing' imei = '' for name, value in options: if name in ('-e', '--env'): system_env = value if name in ('-i', '--imei'): imei = value if name in ('-c', '--code'): code = value if not imei: print 'imei is null.' sys.exit(2) import os os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'configs.{env}'.format(env = system_env)) from script.base import init_env init_env(interactive = False) from apps.web.device.models import Device from apps.web.core.helpers import ActionDeviceBuilder from apps.web.core.exceptions import ServiceException class Tester(threading.Thread): def __init__(self, sleepTime, runTimes, imei, func, **args): super(Tester, self).__init__() dev = Device.get_dev(imei) self._smartBox = ActionDeviceBuilder.create_action_device(dev, typeCode = code) self._func = func self._args = args self._runTimes = runTimes self._sleepTime = sleepTime def run(self): count = 0 while count < self._runTimes: count += 1 try: result = eval('self._smartBox.%s(**self._args)' % self._func) # result = self._smartBox.get_port_status_from_dev() print result if self._sleepTime: time.sleep(self._sleepTime) except ServiceException, e: if u'充电桩正在忙' in e.result.get('description'): print('the device is busuy. error = %s' % e) except Exception, e: print(str(e)) # ii = 0 # for ii in range(0, 10): # Tester(0, 1, imei, 'get_port_info', line = ii % 10).start() # ii = 0 # for ii in range(2): # Tester(0,10000000000,logicalCode,'get_IC_coin_power_config').start() # ii = 0 # for ii in range(2): # Tester(0,10000000000,logicalCode,'get_freemode_volume_andsoon_config').start() # ii = 0 # for ii in range(2): # Tester(0,10000000000,logicalCode,'get_fullstop_cardrefund').start() # Tester(0,10000000000,logicalCode,'set_IC_coin_power_config',infoDict={'maxPower':1000,'icMoney':1,'time1':5,'time2':10,'time3':16}).start() # Tester(0,10000000000,logicalCode,'set_card_time_config',infoDict={'card1Time':5,'card2Time':10,'card3Time':16}).start() # Tester(0,10000000000,logicalCode,'set_coin_card_enable',infoDict={'putCoins':True,'icCard':True}).start() # Tester(0,10000000000,logicalCode,'set_freemode_volume_config',infoDict={'chargeFree':False,'volume':5}).start() # Tester(0,10000000000,logicalCode,'set_fuchong_config',infoDict={'fuchongPower':200,'fuchongTime':1}).start() # Tester(0,10000000000,logicalCode,'set_fullstop_cardrefund',infoDict={'autoStop':True,'cardRefund':True}).start() # Tester(0,10000000000,logicalCode,'set_gear_conf',infoDict={'power1':100,'power1ratio':800,'power2':200,'power2ratio':900,'power3':300,'power3ratio':1000,'power4':400,'power4ratio':1200,'power5':500,'power5ratio':1500}).start() # Tester(30, 10, imei, 'get_port_status_from_dev').start() # time.sleep(1) # Tester(200, 1, imei, 'get_dev_consume_count').start() # time.sleep(1) # # Tester(200, 1, imei, 'get_dev_setting').start() # time.sleep(1) # for ii in range(0, 10): Tester(1, 1, imei, 'start_device', package = {'time': 1, 'unit': u'分钟', 'coins': 1, 'price': 1}, openId = 'aaaaaaaaaaaaaa', attachParas = {'chargeIndex': ii + 1}).start() # # for ii in range(10): # Tester(100, 1, imei, 'stop_charging_port', port = 10 - ii).start() # time.sleep(5) # # time.sleep(10) # dev = Device.get_dev_by_l(logicalCode) # box = ActionDeviceBuilder.create_action_device(dev) # box.get_port_status_from_dev() print 'finished'