manyi/script/hardware_test/thread_test.py

# -*- 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'