fota.py

import hashlib
import socket
import struct
import sys

PACKET_START = 0x00000001
PACKET_FAMILY1 = 0x03E8
PACKET_FAMILY2 = 0x07D0

# packet types
PACKET_PING = 0x0001
PACKET_GETTIME = 0x0002
PACKET_RDY_UPGRADE = 0x0005
PACKET_UPLOAD_FIRMWARE = 0x0006
PACKET_RESTART = 0x0008


class BlackVue(object):
    def __init__(self, host):
        self.debug = True

        self.host = host
        self.port = 9771

        self.conn = None

    def log(self, msg):
        if self.debug:
            print("DEBUG: {}".format(msg))

    def connect(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((self.host, self.port))
        self.conn = s

    def close(self):
        self.conn.close()

    def _send(self, msg):
        self.conn.send(msg)

    def _read(self):
        # read first part of packet, which is 12
        # 4 for packet start, 2 packet family, 2 packet type, 4 packet length
        header = self.conn.recv(12)
        (_, _, _, length) = struct.unpack(">lhhl", header)
        self.log("receiving {} bytes of data".format(length))

        return self.conn.recv(length)

    def getTime(self):
        packet = struct.pack(">lhhl", PACKET_START, PACKET_FAMILY1, PACKET_GETTIME, 0)
        self._send(packet)
        return self._read()

    def ping(self):
        packet = struct.pack(">lhhl", PACKET_START, PACKET_FAMILY1, PACKET_PING, 0)
        self._send(packet)
        return self._read()

    def restart(self):
        packet = struct.pack(">lhhl", PACKET_START, PACKET_FAMILY1, PACKET_RESTART, 0)
        self._send(packet)
        return self._read()

    def prepareUpgrade(self):
        packet = struct.pack(
            ">lhhl", PACKET_START, PACKET_FAMILY1, PACKET_RDY_UPGRADE, 0
        )
        self._send(packet)
        return self._read()

    def uploadFirmware(self, firmware_path):
        firmware = None
        with open(firmware_path, "rb") as file:
            firmware = file.read()
        md5hash = hashlib.md5(firmware).digest()

        packet = (
            struct.pack(
                ">lhhl",
                PACKET_START,
                PACKET_FAMILY2,
                PACKET_UPLOAD_FIRMWARE,
                len(md5hash) + len(firmware),
            )
            + md5hash
            + firmware
        )
        self.log("Firmware: {}".format(firmware_path))
        self.log("Size: {}".format(len(firmware)))
        self.log("MD5: {}".format(md5hash.hex()))
        self.log("Length: {}".format(len(md5hash) + len(firmware)))
        self._send(packet)
        return self._read()


if len(sys.argv) < 2:
    print("Usage python fota.py <host> <ping|restart|upgrade> <?firmware-path>")
    exit()

host = sys.argv[1]
x = BlackVue(host)
x.connect()

action = sys.argv[2]
if action == "ping":
    print(x.ping())
elif action == "restart":
    print(x.restart())
elif action == "upgrade":
    if len(sys.argv) != 4:
        print("Usage python fota.py <host> upgrade <path-to-tgz>")
        exit()

    path = sys.argv[3]
    print(x.prepareUpgrade())
    x.uploadFirmware(path)

x.close()