weatherlink.py

# pyright: strict
import serial # type: ignore
import struct
import time
import binascii

from conversions import dewpoint_approximation, f2c, inHg2hPa, mph2kts

from datatypes import SensorImage

class Link:
    def __init__(self, dev: str = "/dev/ttyUSB0", baud: int = 19200):
        self._ser = serial.Serial(dev, baud, timeout = 2)
        self.wakeup()
        self.setTime()
        
        self.getModel()
        #self.getVersion()

    def wakeup(self):
        while self._ser.inWaiting():
            self._ser.read()
        
        n = 0
        while True:
            self._ser.write(b"\n") # pyright: ignore[reportUnknownMemberType]
            buf = self._ser.read(1)
            
            if buf != b'' and buf[0] == ord('\n'):
                break
            
            n += 1
            
            if n >= 3:
                raise Exception("Failed to wake up console")
        

    def checkAck(self):
        resp = b"\n"

        while resp == b"\n" or resp == b"\r":
            resp = self._ser.read()

        if resp == b'':
            raise Exception("Timeout waiting for ack")
        
        ack = struct.unpack("<B", resp)

        if ack[0] == 0x21:
            raise Exception("Response NOK!")
        
        if ack[0] == 0x18:
            raise Exception("CRC error!")

        if ack[0] != 0x06:
            print("ack: ", ack[0], "response: ")
            for c in resp:
                print(c, end=' ')
            while (self._ser.inWaiting()):
                self._ser.read()
            raise Exception("Response was not NOK, but not ACK either..")
        

    def getSensorImage(self):
        self.wakeup()
        self._ser.write(b"LOOP 1\n") # pyright: ignore[reportUnknownMemberType]

        self.checkAck()

        buf = self._ser.read(99)
        
        if buf[0:3] != b"LOO":
            raise Exception("Start of block was not LOO")
            
        wind_speed = mph2kts(struct.unpack_from("<B", buf, 14)[0])
        average_wind_speed = mph2kts(struct.unpack_from("<B", buf, 15)[0])
        wind_direction = struct.unpack("<H", buf[16:18])[0]
        indoor_temperature = f2c(float(struct.unpack("<H", buf[9:11])[0])/10)
        indoorRelativeHumidity = struct.unpack_from("<B", buf, 11)[0]
        outdoor_temperature = f2c(float(struct.unpack("<H", buf[12:14])[0])/10)
        outdoor_relative_humidity = struct.unpack_from("<B", buf, 33)[0]
        qfe = inHg2hPa(float(struct.unpack("<H", buf[7:9])[0])/1000.0)
        qfe_trend = struct.unpack_from("<b", buf, 3)[0]
        forecast = struct.unpack_from("<B", buf, 89)[0]
        
        rain_rate = float(struct.unpack("<H", buf[41:43])[0]) * 0.2
        rain_day = float(struct.unpack("<H", buf[50:52])[0])* 0.2
        
        OutdoorDewpoint = dewpoint_approximation(outdoor_temperature, outdoor_relative_humidity)
        
        img = SensorImage(
            wind_speed_current=wind_speed,
            wind_speed_average=average_wind_speed,
            wind_direction=wind_direction,
            indoor_temperature=indoor_temperature,
            indoor_relative_humidity=indoorRelativeHumidity,
            outdoor_pressure=qfe,
            outdoor_pressure_trend=qfe_trend,
            forecast=forecast,
            rain_rate=rain_rate,
            rain_daily_total=rain_day,
            outdoor_dewpoint=OutdoorDewpoint,
            outdoor_temperature=outdoor_temperature,
            outdoor_relative_humidity=outdoor_relative_humidity,
            unix_timestamp=time.time()
        )
        
        stationcrc = struct.unpack(">H", buf[97:100])[0]
        
        calc_crc = binascii.crc_hqx(buf[:97], 0)

        if (stationcrc != calc_crc):
            raise Exception("CRC error, image crc does not match our crc!")
                
        return img
    
    def getModel(self):
        self.wakeup()
        self._ser.write(str("WRD" + chr(0x12) + chr(0x4d) + "\n").encode()) # pyright: ignore[reportUnknownMemberType]
        self.checkAck()
        
        make = ord(self._ser.read())
        
        print("Talking to a", end=' ')
        
        if make == 17:
            print("Vantage Vue")
    
    def getVersion(self):
        self.wakeup()
        self._ser.write(b"NVER\n") # pyright: ignore[reportUnknownMemberType]
        
        print(self._ser.read(100))
        
        self._ser.write(b"VER\n") # pyright: ignore[reportUnknownMemberType]
        
        print(self._ser.read(100))
    
    def setTime(self):
        self.wakeup()
        self._ser.write(b"SETTIME\n") # pyright: ignore[reportUnknownMemberType]
        
        self.checkAck()
        
        t = time.localtime()
        
        timestr = struct.pack("<BBBBBB", t.tm_sec, t.tm_min, t.tm_hour, t.tm_mday, t.tm_mon, t.tm_year - 1900)
        crc16 = binascii.crc_hqx(timestr, 0)
        
        self._ser.write(timestr + struct.pack(">H", crc16)) # pyright: ignore[reportUnknownMemberType]
        self.checkAck()

if __name__ == "__main__":
    link = Link()
    print(link.getSensorImage())