Python WAS-110 Multicast Upgrader

multicast_upgrader.py

#!/usr/bin/env python3

__author__ = "Visual Studio"
__credits__ = ["djGrrr", "Visual Studio"]
__license__ = "MIT"
__version__ = "1.0.0.0"

# Credit: https://gist.github.com/djGrrr/4413dacee0df28429306e9fdf4624e02

from signal import *
from socket import *
from math import ceil
from time import sleep
from struct import pack
from pathlib import Path
from argparse import ArgumentParser

IMG_FILE = "multicast_upgrade.img"

# SRC_ADDR = "192.168.1.2"
# SRC_PORT = 0
DST_ADDR = "192.168.1.1"
DST_PORT = 13456

BLOCK_SIZE = 1436

def signal_handler(sig, frame):
    print('Exiting...')
    exit(0)

def main() -> int:
    signal(SIGINT, signal_handler)

    parser = ArgumentParser(description="A script to recover WAS-110 modules")
    parser.add_argument("--path", type=str, default=IMG_FILE, help="The path to the image to use")
    parser.add_argument("--addr", type=str, default=DST_ADDR, help="The module's address")
    parser.add_argument("--port", type=int, default=DST_PORT, help="The module's UDP port")
    parser.add_argument("--delay", type=float, default=0.0005, help="Delay per packet in seconds")
    args = parser.parse_args()

    p = Path(args.path)
    if not p.exists():
        print("Image file doesn't exist!")
        return 1
    size = p.stat().st_size

    sock = socket(AF_INET, SOCK_DGRAM)

    pkts = ceil(size / BLOCK_SIZE)
    if pkts > 65535:
        print("Image file is too large!")
        return 2

    with p.open("rb") as f:
        while True:
            for i in range(pkts):
                o = i + 1;
                if o % 100 == 0 or o == pkts:
                    print(f"{o}/{pkts}")

                blk = f.read(BLOCK_SIZE)

                if not blk:
                    break

                hdr = pack("!4I", i + 1, BLOCK_SIZE, 1, size)
                hdr += b"bfw-HGW"
                hdr += (b"\x00" * 13)
                pkt = hdr + blk
                try:
                    sock.sendto(pkt, (args.addr, args.port))
                except:
                    pass

                sleep(args.delay)
            f.seek(0)

    return 0

if __name__ == "__main__":
    exit(main())