Monarch Instruments USB Temperature & Relative Humidity Probe python driver

monarch_th.py

# Copyright 2022 Jetperch LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import serial
import serial.tools.list_ports
import sys
import time

VID = 0x0403
PID = 0x6015


class TempHumidity:
    """Monarch Instruments USB Temperature & Relative Humidity Probe.

    See https://monarchinstrument.com/products/portable-usb-temperature-humidity-probe
    """
    def __init__(self):
        info = {}
        ports = [p for p in serial.tools.list_ports.comports() if (p.vid == VID and p.pid == PID)]
        if len(ports) != 1:
            raise RuntimeError('Could not find serial port')
        port = ports[0]
        self.serial = serial.Serial(port.device, 9600, timeout=2.0)
        product_id = self.command('@PI').decode('utf-8')
        revision = self.serial.readline().strip().decode('utf-8')
        serial_number = self.command('@SERNO', skip_status=True).decode('utf-8')
        print(f'{product_id}, {revision}, {serial_number}')

    def command(self, command, skip_status=None):
        if isinstance(command, str):
            command = command.encode('utf-8')
        self.serial.write(command + b'\r\n')
        reply = self.serial.readline().strip()
        if reply.endswith(b' :'):
            reply = reply[:-2]
        if reply != command:
            raise RuntimeError(f'unexpected reply: {command}, {reply}')
        if not skip_status:
            reply = self.serial.readline().strip()
            if reply != b'OK':
                raise RuntimeError(f'unexpected ok: {command}, {reply}')
        reply = self.serial.readline().strip()
        return reply

    def measure(self):
        s = self.command('@D0').decode('utf-8')
        t, h = [float(p.strip()) for p in s.split(',')]
        return t, h


def run_example():
    th = TempHumidity()
    try:
        while True:
            t_start = time.time()
            response = th.measure()
            t_end = time.time()
            print(f'{response}, took {t_end - t_start:.3f} seconds')
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == '__main__':
    sys.exit(run_example())