Experimental Physics and
Industrial Control System

Matt Newville via Tech-talk <tech-talk at aps.anl.gov> · Mon, 13 Nov 2023 09:53:11 -0600

#!/usr/bin/env python
import requests
import time
import json
import xml.etree.ElementTree as ET
from epics import get_pv

class ESensor:
    """ read Esensor data, push to Epics PVs"""
    def __init__(self, label, url, prefix, format="xml"):
        self.label = label
        self.url = url
        self.prefix = prefix
        self.format = format

        self.tempf_pv = get_pv(f"{prefix:s}:TempF.VAL")
        self.tempc_pv = get_pv(f"{prefix:s}:TempC.VAL")
        self.humid_pv = get_pv(f"{prefix:s}:Humid.VAL")
        self.timestamp_pv = get_pv(f"{prefix:s}:TimeStamp.VAL")

        self.tattr = "tm0"
        self.tunit = "tun0"
        self.hattr = "hu0"
        self.reader = ET.fromstring
        if format == "json":
            self.tattr = "tmp"
            self.hattr = "hum"
            self.tunit = "tun"
            self.reader = json.loads

    def getval(self, source, attr, force_float=True):
        if self.format == "xml":
            try:
                val = source.find(attr).text
            except:
                val = "-1"
        elif self.format == "json":
            try:
                val = source.get(attr, -1.0)
            except:
                val = "-1"
        if force_float:
            try:
                val = float(val)
            except:
                val = -1.0
        return val

    def read(self):
        resp = requests.get(self.url, timeout=0.25)
        tdat = self.reader(resp.content.decode("utf-8"))

        tunt = self.getval(tdat, self.tunit, force_float=False)
        tval = self.getval(tdat, self.tattr)
        hval = self.getval(tdat, self.hattr)
        return tval, tunt, hval

    def update(self):
        tval, tunit, hval = self.read()
        if tunit in (0, '0', 'F'):
            tval_f = tval
            tval_c = (tval - 32)*5.0/9.0
        else:
            tval_c = tval
            tval_f = (tval_c * 9.0/5) + 32
        tval_f = float("%.3f" % tval_f)
        tval_c = float("%.3f" % tval_c)
        hval   = float("%.3f" % hval)

        if self.tempf_pv.connected:
            self.tempf_pv.put(tval_f)
            self.tempc_pv.put(tval_c)
            self.humid_pv.put(hval)
            self.timestamp_pv.put(time.ctime())
            print(time.ctime(), self.label, tval_c, hval)

#
if __name__ == '__main__':

    esensors = [ESensor("13-ID-E", "http://164.54.160.168/sensorLogUpdate.xml";, "13IDE:Esensor", format="xml"),
                ESensor("13-ID-C", "http://10.54.160.76/status.json";, "13IDC:Esensor", format="json"),
                ]

    while True:
        try:
            [e.update() for e in esensors]
        except:
            pass
        time.sleep(0.5)

Experimental Physics and Industrial Control System

Experimental Physics and
Industrial Control System