solar-monitor/src/data-writer.py

import cv2
import pytesseract
import time
from datetime import datetime
from pytesseract import Output

import cv2
import pytesseract
from pytesseract import Output
import numpy as np
import re
import os
from influxdb import InfluxDBClient
client = InfluxDBClient(host="localhost", port=8086, username="influxdb", password="influxdbTSGAMES", timeout=120_000)
client.create_database("influxdb")
client.switch_database('influxdb')
class SolarMonitor:
    camera = None
    def __init__(self, test = False):
        self.test = test
        if not test:
            self.initCamera()
        
    def parse(self, img): 
        custom_oem=r'--oem 3 --psm 7'
        # https://github.com/adrianlazaro8/Tesseract_sevenSegmentsLetsGoDigital
        data = pytesseract.image_to_data(img, lang='lets', config=custom_oem, output_type=Output.DICT)
        #data = pytesseract.image_to_data(img, lang='letsgodigital', config=custom_oem, output_type=Output.DICT)
        print(data)
        results = []
        for i in range(len(data['text'])):
            text = data['text'][i].strip('.,-_')
            text = re.sub('[^0-9]', '', text)
            if text:
                results.append(text)
        if len(results) == 2:
            results = list(map(lambda x: int(x) / 10.,results ))
            if results[0] < 80 and results[1] < 900:
                return results
        return None
    def thresholding(self, image):
        image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        
        pixel_values = image.reshape((-1, 1))
        pixel_values = np.float32(pixel_values)
        print(pixel_values.shape)
        # define stopping criteria
        criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 100, 0.2)
        # number of clusters (K)
        k = 3
        _, labels, (centers) = cv2.kmeans(pixel_values, k, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS)
        # convert back to 8 bit values
        centers = np.uint8(centers)

        # flatten the labels array
        labels = labels.flatten()
        # convert all pixels to the color of the centroids
        segmented_image = centers[labels.flatten()]
        segmented_image = segmented_image.reshape(image.shape)
        
        # disable only the cluster number 2 (turn the pixel into black)
        masked_image = np.copy(image)
        # convert to the shape of a vector of pixel values
        masked_image = masked_image.reshape((-1, 1))
        # color (i.e cluster) to disable
        cluster = 2
        masked_image[labels != cluster] = [0]
        # convert back to original shape
        masked_image = masked_image.reshape(image.shape)
        
        return masked_image
    
        
        image = cv2.medianBlur(image,3)
        #image = np.invert(image)
        return cv2.threshold(image, 140, 255, cv2.THRESH_BINARY)[1]
        return image
    def thresholding2(self, image):
        image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        # Taking a matrix of size 5 as the kernel
        kernel = np.ones((3, 3), np.uint8)
        
        # The first parameter is the original image,
        # kernel is the matrix with which image is
        # convolved and third parameter is the number
        # of iterations, which will determine how much
        # you want to erode/dilate a given image.
        image = cv2.convertScaleAbs(image, alpha=10.0, beta=-500)
        #image = cv2.dilate(image, kernel, iterations=1)
        #image = cv2.erode(image, kernel, iterations=1)
        #image = cv2.convertScaleAbs(image, alpha=2.0, beta=-50)
        #image = (255 - image)
        image = cv2.medianBlur(image,3)
        #image = np.invert(image)
        return image
    
    def crop(self, image):
        #top=384
        #left=210
        #height= 60
        #width=230
        #return image[top : (top + height) , left: (left + width)]
        pt_A = [225, 393]
        pt_B = [223, 432]
        pt_C = [420, 426]
        pt_D = [422, 387]
        width_AD = np.sqrt(((pt_A[0] - pt_D[0]) ** 2) + ((pt_A[1] - pt_D[1]) ** 2))
        width_BC = np.sqrt(((pt_B[0] - pt_C[0]) ** 2) + ((pt_B[1] - pt_C[1]) ** 2))
        maxWidth = max(int(width_AD), int(width_BC))


        height_AB = np.sqrt(((pt_A[0] - pt_B[0]) ** 2) + ((pt_A[1] - pt_B[1]) ** 2))
        height_CD = np.sqrt(((pt_C[0] - pt_D[0]) ** 2) + ((pt_C[1] - pt_D[1]) ** 2))
        maxHeight = max(int(height_AB), int(height_CD))
        input_pts = np.float32([pt_A, pt_B, pt_C, pt_D])
        output_pts = np.float32([[0, 0],
                        [0, maxHeight - 1],
                        [maxWidth - 1, maxHeight - 1],
                        [maxWidth - 1, 0]])
        M = cv2.getPerspectiveTransform(input_pts,output_pts)
        image = cv2.warpPerspective(image,M,(maxWidth, maxHeight),flags=cv2.INTER_LINEAR)
        return image

    def initCamera(self):
        from picamera import PiCamera
        self.camera = PiCamera(
            resolution=(800, 608)
        )
        self.camera.awb_mode = 'off'
        self.camera.awb_gains = (1.5, 2.0)
        self.camera.shutter_speed = 10000
        self.camera.iso = 800
        self.camera.rotation = 180

    def capture(self): 
        from picamera.array import PiRGBArray
        rawCapture = PiRGBArray(self.camera)
        file = "images/" + str(datetime.now()) + ".jpg"
        self.camera.capture(rawCapture, format="bgr")
        img = rawCapture.array
        img = self.crop(img)
        print("Captured.")        
        return [file, img]

    def writeData(self, results): 
        client.write_points([
            {
                "measurement": "solar",
                "tags": {
                    "type": "U"
                },
                "fields": {
                    "value": results[0]
                }
            },
                                {
                "measurement": "solar",
                "tags": {
                    "type": "W"
                },
                "fields": {
                    "value": results[1]
                }
            }
        ], time_precision='ms')
        
    def run(self):
        if self.test:
            for file in os.listdir('tests'):
                img = cv2.imread('tests/' + file)
                img = self.thresholding2(img)
                results = self.parse(img)
                print(file)
                print(results)
                if results:
                    self.writeData(results)
                cv2.imwrite(file + '_r.jpg', img)
        else:
            [file, img] = solar.capture()
            img = self.thresholding2(img)
            results = self.parse(img)
            print(results)
            if results:
                self.writeData(results)
                img = cv2.putText(img, str(results), (75, 22), cv2.FONT_HERSHEY_SIMPLEX, 0.25, (255), 1, cv2.LINE_AA)
            cv2.imwrite(file + '_r.jpg', img, [cv2.IMWRITE_JPEG_QUALITY, 35])
            time.sleep(5)

solar = SolarMonitor(test = False)
while True:
    solar.run()