Need for UNI-T UT161D

99-ut161d.rules

@@ -0,0 +1 @@
+SUBSYSTEM=="usb", ATTR{idVendor}=="1a86", ATTR{idProduct}=="e429", MODE="0666" 

README

@@ -0,0 +1,23 @@
+Ohjelma toimii varmuudella vain Ubuntu Mate 2024.04 LTS:ssä, mutta linuxia osaava käyttäjä voi muokata sen toimimaan muissakin ympäristöissä.
+
+Tämä ohjelmakirjasto on tehty vibekoodaamalla. Tosin käytin AI:ta lähinnä hakukoneena.
+Lisenssi on GPL 3.0.
+
+Ohjelman tarkoituksena on luoda yksinkertainen rajapinta UNIT-T UT161D-mittarille. Ohjelma on tavallaan ajuri git:n unit_ut61eplus-projektille, siksi hakemiston on outo UNI-T UT161D:lle.
+
+Tämän asentaminen tapahtuu sekä root-käyttäjänä (install.sh) että normaalina käyttjänä (installVenvEnvironmentDoNotUseSudo.sh).
+
+Root-oikeuksia (install.sh-tiedosto) tarvitaan kirjastojen asentamiseen ja tarvittavien UDEV-sääntöjen määrittelemiseen.
+Tavallisella käyttäjällä asennetaan venv-virtuaaliympäristö.
+
+Kun käyttää päätteellä, niin ohjelman saa suoritettua venv-ympäristössä käskyparilla (pitää olla tämän projektin työhakemistossa):
+
+source venv/bin/activate ;python3 readUT161dDMM.py
+
+Kyseinen testiohjelma lukee mittarin nimen, laittaa näytön taustavalon joko päälle tai pois. Lukee kertaalleen mittarin lukeman.
+
+Isäntä projektin koodin saa siis näkyviin, kunhan on asentanut git:n ...
+git clone https://github.com/ljakob/unit_ut61eplus.git
+
+Kun olet ladannut, niin sen jälkeen kopioi tämän projektini tiedostot ko. asennushakemistoon, niin pääset kokeilemaan minun räpellystä. Olettaen, että omistat UNI-T UT161D-yleismittarin ja PC:n (esim. Raspberry Pi 4:n), jossa on Debian tai Ubuntu-variantti. (En tosin ole juuri tätä projektia kerennyt (vielä) testata Raspissa.)
+

README.en

@@ -0,0 +1,24 @@
+The program works for sure only in Ubuntu Mate 2024.04 LTS, but a user who knows Linux can modify it to work in other distros as well.
+
+This Program library was made by vibecoding. However, I mainly used AI as a search engine.
+License is GPL 3.0.
+
+Ohjelman löytää on yksinkertainen rajapinta UNIT-T UT161D-mittarille. The program is kind of a driver for git's unit_ut61eplus project, that's why the directory is strange for UNI-T UT161D.
+
+This can install as root user by install.sh-bash command.
+
+Root privileges (install.sh file) are required to install the libraries and define the necessary UDEV rules.
+User privileges (installVenvEnvironmentDoNotUseSudo.sh) are requred to install the venv-environment.
+
+When using a terminal, the program can be executed in the venv environment with a pair of commands (must be in the working directory of this project):
+
+source venv/bin/activate ;python3 readUT161dDMM.py
+
+The test program in question reads the name of the meter, turns the screen backlight on or off. Reads the meter reading once.
+
+So the host can view the project's code as long as he has installed git...
+git clone https://github.com/ljakob/unit_ut61eplus.git
+
+After you have downloaded, then copy the files of this project to to the installation directory, and you can try out my code. Assuming you own a UNI-T UT161D multimeter and a PC (eg Raspberry Pi 4) running Debian or Ubuntu variant. (However, I haven't (yet) gotten around to testing this particular project in Raspberry Pi.)
+
+

install.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/bash
+echo "Install needed packets"
+sudo apt install libusb-1.0-0-dev python3-pip python3-venv
+echo "Add UDEV-rules for UNI-T UT161D-multimeter"
+sudo cp ./99-ut161d.rules /etc/udev/rules.d/99-ut161d.rules
+echo "Update UDEV-rules"
+sudo udevadm control --reload-rules
+sudo udevadm trigger

readUT161dDMM.py

@@ -0,0 +1,22 @@
+
+import logging
+from ut161d import UT161D
+
+log = logging.getLogger(__name__)
+
+logging.basicConfig(level=logging.DEBUG)
+
+#dmm = UT61EPLUS()
+dmm = UT161D()
+name=dmm.getName()
+print("name=",name)
+print(f'name={dmm.getName()}')
+dmm.sendCommand('lamp')
+m = dmm.takeMeasurement()
+log.info('measurent=%s', m)
+#
+for i in range(500):
+    mesure=dmm.takeMeasurement()
+    print(mesure.display_decimal)
+#mesure=dmm.takeMeasurement()
+#log.info('maasurement=%s',mesure)

requirements.txt

@@ -1 +1 @@
-hidapi
+hidapi pyusb

ut161d/__init__.py

@@ -0,0 +1 @@
+from .ut161d import UT161D

ut161d/ut161d.py

@@ -0,0 +1,214 @@
+import binascii
+import logging
+import decimal
+import usb.core
+import usb.util
+import string
+import time
+from time import sleep
+
+from ut61eplus import UT61EPLUS
+
+log = logging.getLogger(__name__)
+
+class UT161D(UT61EPLUS):
+    #class UT161D:
+    UT161D_VID = 0x1a86
+    UT161D_PID = 0xe429
+
+    def __init__(self):
+        #VID = 0x1a86
+        #PID = 0xe429
+
+        """
+        print("UT161D.__init__")
+        #self.
+        dev = usb.core.find(idVendor=self.UT161D_VID, idProduct=self.UT161D_PID)
+        #if self.dev is None:
+        if dev is None:
+            print("Device not found. Is it connected?")
+            exit(1)
+        print("USB device is found.")
+
+        # Try unplug kernel-driver
+        try:
+            #if self.dev.is_kernel_driver_active(0):
+            if dev.is_kernel_driver_active(0):
+                print("Unplug kernel-driver.")
+                #self.dev.deatch_kernel_driver(0)
+                dev.deatch_kernel_driver(0)
+        except Exception as e:
+            print("Kernel-driver can't unplug:",e)
+
+        # Set configuration
+        #self.
+        dev.set_configuration()
+        #cfg = self.dev.get_active_configuration()
+        cfg = dev.get_active_configuration()
+        intf = cfg[(0,0)]
+
+        # Find IN-endpoint
+        self.ep_in = usb.util.find_descriptor(intf,
+            custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN)
+
+        # Find OUT-endpoint
+        self.ep_out = usb.util.find_descriptor( intf,
+            custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT)
+
+        print(f"IN endpoint: {hex(self.ep_in.bEndpointAddress)}")
+        print(f"OUT endpoint: {hex(self.ep_out.bEndpointAddress)}")
+        """
+
+
+        log.debug("Etsin laitetta...")
+
+        self.dev = usb.core.find(idVendor=self.UT161D_VID, idProduct=self.UT161D_PID)
+        if self.dev is None:
+            print("Device not found. Is it connected?")
+            exit(1)
+        log.debug("USB device is found.")
+
+        # Yritä irrottaa kernel-ajuri
+        try:
+            if self.dev.is_kernel_driver_active(0):
+                print("Unplug kernel-driver.")
+                self.dev.detach_kernel_driver(0)
+        except Exception as e:
+            print("Kernel-ajuria ei voitu irrottaa:", e)
+
+        # Aseta konfiguraatio
+        self.dev.set_configuration()
+
+        cfg = self.dev.get_active_configuration()
+        intf = cfg[(0, 0)]
+
+        # Etsi IN-endpoint
+        self.ep_in = usb.util.find_descriptor(
+            intf,
+            custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN
+        )
+
+        # Etsi OUT-endpoint
+        self.ep_out = usb.util.find_descriptor(
+            intf,
+            custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT
+        )
+        log.debug('IN endpoint: %02x',self.ep_in.bEndpointAddress)
+        log.debug('OUT endpoint: %02x',self.ep_out.bEndpointAddress)
+        #self._write(self._SEQUENCE_GET_NAME)
+        #self._write(self._SEQUENCE_SEND_DATA)
+
+    def _write(self, b: bytes):
+        buf = []
+        buf.append(len(b))
+        buf += b
+        while len(buf)<64:
+            buf.append(0)
+        #print(len(buf))
+        #for d in buf:
+        #    print(hex(d),end=" ")
+        #exit()
+        self.writebuf=buf
+        self.dev.write(self.ep_out.bEndpointAddress,buf) # Toivottavasti toimii. On lisätty 20260720
+        #try:
+        #    self.dev.read(self.ep_in.bEndpointAddress, self.ep_in.wMaxPacketSize,timeout=1000)
+        #except:
+        #    pass
+
+    def _readResponse(self) -> bytes:
+        self.readData=[]
+        state=0
+        buf:bytes = None
+        index:int = None
+        sum:int = 0
+
+        toomanyTimeOuts=0
+        #print("TULI VARMUUDELLA")
+        while True:
+            try:
+                x = self.dev.read(self.ep_in.bEndpointAddress, self.ep_in.wMaxPacketSize,timeout=1000)
+                #print("read meni onnistuneesti läpi!!!",x)
+                toomanyTimeOuts=0
+
+                #for d in x:
+                #    print(hex(d),end=" ")
+                #print()
+                #for c in x:
+                #    print(chr(c),end="'")
+                #print()
+                #print("\n\n\n")
+                #exit()
+                self.readData=x
+                #return True
+                b:int
+                for b in x[1:]:
+                    #print(hex(b))
+                    if state < 3 or index +2 < len(buf): # sum all bytes except last 2
+                        sum += b
+                    if state == 0 and b == 0xAB:
+                        state = 1
+                    elif state == 1 and b == 0xcd:
+                        state = 2
+                    elif state == 2:
+                        buf = bytearray(b)
+                        index = 0
+                        state = 3
+                    elif state == 3:
+                        buf[index] = b
+                        index += 1
+                        if index == len(buf):
+                            recevied_sum = (buf[-2] << 8) + buf[-1]
+                            log.debug('calculated sum=%04x expected sum=%04x',sum,recevied_sum)
+                            if sum != recevied_sum:
+                                log.warning('checksum mismatch')
+                                #print("Tarkistussumma heittää")
+                                # exit()
+                                return None
+                            #print("Lukeminen onnistui !!!")
+                            return buf[:-2] # drop last 2 bytes at end with checksum
+                    else:
+                        log.warning('unexpected byte %02x in state %i',b,state)
+                        exit(1)
+            except usb.core.USBTimeoutError:
+                toomanyTimeOuts+=1
+                log.debug("%i, Ei data (timeout), yritän uudelleen...",toomanyTimeOuts)
+                if toomanyTimeOuts==10:
+                    print("Yritetty riittävästi, joten lopetamme ohjelman suorituksen.")
+                    exit()
+                return False
+                #self.dev.write(self.ep_out.bEndpointAddress,self.writebuf)
+            pass
+
+    def getName(self):
+        #self._write(self._SEQUENCE_SEND_DATA)
+        #print("write meni\n")
+        #self._write(self._SEQUENCE_GET_NAME)
+        #name=self._readResponse()
+        #self._write(self._SEQUENCE_GET_NAME)
+        #name=self._readResponse()
+
+
+        name=False
+        while name==False:
+            #self._write(self._SEQUENCE_SEND_DATA)
+            self._write(self._SEQUENCE_GET_NAME)
+            name = self._readResponse()
+            if name!=False:
+                if name[0]<32 or name[0]>ord('z'):
+                    #print("Failas")
+                    name=False
+        #print("read meni\n")
+        #print(self.readData,"\n")
+        #print()
+        #print("TULI VARMUUDELLA TÄNNE!!!!")
+        #print(name.decode('ASCII'))
+        #exit()
+
+        return name.decode('ASCII')
+
+
+
+
+
+
+

ut161d/ut61eplus

@@ -0,0 +1 @@
+../ut61eplus