python版本的海萤物联网ntp服务上线

开源

介绍

基于Python编写，在海萤物联网提供校时服务。

本机地址是：

0x2141000000000405

服务

服务号|服务 —|— 1|读取时间1 2|读取时间2.返回的是结构体

读取时间服务1

CON请求：空或者带符号的1个字节。

当CON请求为空时，则默认为读取的是北京时间（时区8）。

也可以带1个字节表示时区号。这个字节是有符号的int8。

小技巧，可以使用0x100减去正值即负值。比如8对应的无符号数是0x100-8=248。

ACK应答：当前时间的字符串

当前时间字符串的格式：2006-01-02 15:04:05 -0700 MST

读取时间服务2.返回的是结构体

CON请求：格式与读取时间服务1一致

ACK应答：

struct {
  // 时区
  uint8 TimeZone
  uint16 Year
  uint8 Month
  uint8 Day
  uint8 Hour
  uint8 Minute
  uint8 Second
  // 星期
  uint8 Weekday
}
  

自定义错误码

错误码|含义 —|— 0x40|内部错误 0x41|接收格式错误

示例

读取时间

resp, err = tziot.call(pipe, 0x2141000000000005, 1, 1000, bytearray())
print("err:", err, "time:", resp)
  

输出：

err: 0 time: b'2021-03-22 10:06:09 +0800 MST'

读取时区为2的时间

resp, err = tziot.call(pipe, 0x2141000000000005, 1, 1000, bytearray([2]))
print("err:", err, "time:", resp)
  

输出：

err: 0 time: b'2021-03-22 04:07:20 +0200 MST'

读取时区为-6的时间

resp, err = tziot.call(pipe, 0x2141000000000005, 1, 1000, bytearray([0x100 - 6]))
print("err:", err, "time:", resp)
  

输出：

err: 0 time: b'2021-03-21 20:08:14 -0600 MST'

读取结构体格式时间

import tziot
import sbc


class AckRidGetTime(sbc.LEFormat):
    _fields_ = [
        # (字段名, c类型)
        # 时区
        ('time_zone', sbc.c_uint8),
        ('year', sbc.c_int16),
        ('month', sbc.c_uint8),
        ('day', sbc.c_uint8),
        ('hour', sbc.c_uint8),
        ('minute', sbc.c_uint8),
        ('second', sbc.c_uint8),
        # 星期
        ('weekday', sbc.c_uint8),
    ]


def main():
    pipe = tziot.bind_pipe_net(0x2141000000000401, pwd, '192.168.1.119', 12021)
    while not tziot.is_conn():
        pass
    resp, err = tziot.call(pipe, 0x2141000000000005, 2, 3000, bytearray())
    if err != 0:
        return

    ack = AckRidGetTime()
    if not ack.bytearray_to_struct(resp):
        return
    print(ack.time_zone, ack.year, ack.month, ack.day, ack.hour, ack.minute, ack.second, ack.weekday)


if __name__ == '__main__':
    main()
  

输出：

8 2021 4 2 11 43 7 5

源码

最新源码请查看仓库。

main.py

"""
Copyright 2021-2021 The jdh99 Authors. All rights reserved.
网络校时服务
Authors: jdh99 <jdh821@163.com>
"""

import config

import tziot
import dcompy as dcom
import lagan
import sbc

from datetime import datetime, timedelta

TAG = 'ntp'

# 应用错误码
# 内部错误
ERROR_CODE_INTERNAL_ERROR = 0x40
# 接收格式错误
ERROR_CODE_RX_FORMAT = 0x41

# rid号
# 读取时间.返回的是字符串
RID_GET_TIME1 = 1
# 读取时间.返回的是结构体
RID_GET_TIME2 = 2


class AckRidGetTime2(sbc.LEFormat):
    _fields_ = [
        # (字段名, c类型)
        # 时区
        ('time_zone', sbc.c_uint8),
        ('year', sbc.c_int16),
        ('month', sbc.c_uint8),
        ('day', sbc.c_uint8),
        ('hour', sbc.c_uint8),
        ('minute', sbc.c_uint8),
        ('second', sbc.c_uint8),
        # 星期
        ('weekday', sbc.c_uint8),
    ]


def main():
    config.init()

    lagan.load(0)
    lagan.set_filter_level(lagan.LEVEL_INFO)
    lagan.enable_color(True)
    dcom.set_filter_level(lagan.LEVEL_WARN)

    tziot.bind_pipe_net(config.LOCAL_IA, config.local_pwd, config.LOCAL_IP, config.LOCAL_PORT)
    tziot.register(RID_GET_TIME1, ntp_service1)
    tziot.register(RID_GET_TIME2, ntp_service2)


def ntp_service1(pipe: int, src_ia: int, req: bytearray) -> (bytearray, int):
    """校时服务.返回值是应答和错误码.错误码为0表示回调成功,否则是错误码"""
    ip, port = dcom.pipe_to_addr(pipe)

    if len(req) == 0:
        time_zone = 8
    elif len(req) == 1:
        time_zone = req[0]
        if time_zone >= 0x80:
            time_zone = -(0x100 - time_zone)
    else:
        lagan.warn(TAG, "ip:%s port:%d ia:0x%x ntp failed.len is wrong:%d", ip, port, src_ia, len(req))
        return None, ERROR_CODE_RX_FORMAT

    now = datetime.utcnow() + timedelta(hours=time_zone)
    if time_zone >= 0:
        s = '%04d-%02d-%02d %02d:%02d:%02d +%02d00 MST' % (now.year, now.month, now.day, now.hour, now.minute,
                                                            now.second, time_zone)
    else:
        s = '%04d-%02d-%02d %02d:%02d:%02d -%02d00 MST' % (now.year, now.month, now.day, now.hour, now.minute,
                                                            now.second, -time_zone)
    lagan.info(TAG, 'ip:%s port:%d ntp time:%s', ip, port, s)
    return tziot.str_to_bytearray(s), 0


def ntp_service2(pipe: int, src_ia: int, req: bytearray) -> (bytearray, int):
    """校时服务.返回值是应答和错误码.错误码为0表示回调成功,否则是错误码"""
    ip, port = dcom.pipe_to_addr(pipe)

    if len(req) == 0:
        time_zone = 8
    elif len(req) == 1:
        time_zone = req[0]
        if time_zone >= 0x80:
            time_zone = -(0x100 - time_zone)
    else:
        lagan.warn(TAG, "ip:%s port:%d ia:0x%x ntp failed.len is wrong:%d", ip, port, src_ia, len(req))
        return None, ERROR_CODE_RX_FORMAT

    now = datetime.utcnow() + timedelta(hours=time_zone)
    t = AckRidGetTime2()
    t.time_zone = time_zone
    t.year = now.year
    t.month = now.month
    t.day = now.day
    t.hour = now.hour
    t.minute = now.minute
    t.second = now.second
    t.weekday = now.isoweekday()
    lagan.info(TAG, 'ip:%s port:%d ntp time:%04d-%02d-%02d %02d:%02d:%02d +%02d00 MST', ip, port, t.year, t.month,
               t.day, t.hour, t.minute, t.second, t.time_zone)
    return t.struct_to_bytearray(), 0


if __name__ == '__main__':
    main()
  

config.py

"""
Copyright 2021-2021 The jdh99 Authors. All rights reserved.
配置文件
Authors: jdh99 <jdh821@163.com>
"""

# 系统参数
LOCAL_IA = 0x2141000000000005
LOCAL_IP = '0.0.0.0'
LOCAL_PORT = 12931

local_pwd = ''


def init():
    global local_pwd
    print('please input password:')
    local_pwd = input()
  