manyi/script/hardware_test/common_test.py

# -*- coding: utf-8 -*-
# !/usr/bin/env python

import os, sys
import uuid
import time

import simplejson as json
from django.conf import settings



PROJECT_ROOT = os.path.join(os.path.abspath(os.path.split(os.path.realpath(__file__))[0] + "/.."), '..')
sys.path.insert(0, PROJECT_ROOT)

os.environ.update({"DJANGO_SETTINGS_MODULE": "configs.testing"})

import django

django.setup()

from apps.web.core.networking import MessageSender
from apps.web.core.mqtt_client import MqttClient
from apps.web.device.models import Device

IMEI = '867435053812616'
server = '211.159.224.10'
port = 1884

event_received = False

mqttc = MqttClient(client_id = 'webapp_' + str(uuid.uuid1()))


def on_connect(client, userdata, flags, rc):
    # 订阅设备的120响应事件，用于检查
    device_topic = 'server/{}/120'.format(IMEI)
    client.subscribe(device_topic, qos = 0)

    device_topic = 'server/{}/110'.format(IMEI)
    client.subscribe(device_topic, qos = 0)

    device_topic = 'server/{}/100'.format(IMEI)
    client.subscribe(device_topic, qos = 0)

    device_topic = 'server/{}/301'.format(IMEI)
    client.subscribe(device_topic, qos = 0)

    device_topic = 'server/{}/307'.format(IMEI)
    client.subscribe(device_topic, qos = 0)


def on_message(mqttc, obj, msg):
    msgDict = json.loads(bytes.decode(msg.payload))
    print('received event msg = {}'.format(msgDict))
    event_received = True

    mqttc.loop_stop()
    mqttc.disconnect()
    mqttc.close()


try:
    mqttc.on_message = on_message
    mqttc.on_connect = on_connect

    mqttc.username_pw_set(settings.MQTT_USER, settings.MQTT_PSWD)
    mqttc.connect(server, port, 60)

    mqttc.loop_start()
finally:
    pass

CHECK_ROUNDS = 6
PWM_WID = 80
PWM_INTER = 600
PWM_IDLE = 0
UART_TIMEOUT = 20
CYCLE_SET = 120
DEBUG = True
ENABLE_PG = True

cmd_list = [
    {"cmd": 201},
    {"cmd": 201, "fields": ['soft_ver']},
    {"cmd": 201, "fields": ['signal']},
    # {"cmd":202, "addr_set":{"ip1":"211.159.224.10","port1":1883}},
    {"cmd": 202, "debug": DEBUG},
    {"cmd": 202, "cycle_set": CYCLE_SET},
    {"cmd": 202, "check_rounds": CHECK_ROUNDS},
    # {"cmd": 202, "keepalive_time": 320},
    {"cmd": 202, "coin_clear": True},
    {"cmd": 202, "clear_bls": True},
    # {"cmd": 202, "factory_set": True},
    # {"cmd": 202, "debounce": 25},
    # {"cmd": 202, "pg_inter": 62},
    {"cmd": 202, "enable_pg": ENABLE_PG},
    {"cmd": 202, "uart_timeout": UART_TIMEOUT},
    {"cmd": 202,
     "ntp_servers": ["ntp7.aliyun.com", "ntp6.aliyun.com", "ntp5.aliyun.com", "ntp4.aliyun.com", "ntp3.aliyun.com",
                     "ntp2.aliyun.com", "ntp1.aliyun.com", "ntp.ntsc.ac.cn"]},
    {"cmd": 202, "pulse_set": {"pwm_wid": PWM_WID, "pwm_inter": PWM_INTER, "pwm_idle": PWM_IDLE}},
    {"cmd": 202, "max_coin": 255},
    # {"cmd": 202, "ota_set": {"fw_url": "http://www.washpayer.com/uploaded/1SMART_BOX_3.2.4_Luat_V0020_8955_SSL.bin"}},
    # {"cmd": 202, "driver_set": {"driver_url": "http://121.43.232.118/uploaded/version/drivers/task/100210_uart_charge_dianchuan_v1.5/default.driver"}},

    {"cmd": 203, "app_pay": 8},
    {"cmd": 203, "app_pay": 12},
]

for cmd in cmd_list:
    payload = {'IMEI': IMEI}
    payload.update(cmd)

    dev = Device.get_dev(IMEI)

    result = MessageSender.send(dev, payload['cmd'], payload)
    if result['rst'] != 0:
        print 'error. cmd = {}'.format(str(cmd))
        exit(1)

result = MessageSender.send(dev, 201, {'cmd': 201})
if result['check_rounds'] != CHECK_ROUNDS:
    print("check_rounds update failure.")

if result['pwm_wid'] != PWM_WID:
    print("pwm_wid update failure.")

if result['cycle'] != CYCLE_SET:
    print("cycle_set update failure.")

if result['uart_timeout'] != UART_TIMEOUT:
    print("uart_timeout update failure.")

if result['pwm_inter'] != PWM_INTER:
    print("pwm_inter update failure.")

if result['debug'] != DEBUG:
    print("debug update failure.")

while True:
    if event_received == False:
        print('wait event to received.')
        time.sleep(5)
        continue
    else:
        exit(1)

print 'finished'