Добавления драйвера

InterfaceL.py

@@ -1,4 +1,5 @@
 from idlelib.colorizer import color_config
+from lib2to3.pgen2.driver import Driver
 from logging import disable
 from math import acosh
 from tkinter import Tk, Label, Entry, Button, Spinbox, StringVar, BOTTOM, LEFT, SOLID, EW, END
@@ -13,6 +14,9 @@ import matplotlib.pyplot as plt
 import matplotlib
 import matplotlib.animation as animation
 import numpy as np
+from wrapper import GammaDevice
+import threading
+from typing import List
 import time
 import os
 from matplotlib.animation import FuncAnimation
@@ -49,6 +53,17 @@ class Interface():
     windowq.pack()
     window = ScrollableFrame(windowq)
     window.pack(side='top', anchor='sw')
+    t_list: List[float] = []
+    c_list: List[float] = []
+    adc_list: List[float] = []
+    znach = {
+        'temperature': t_list,
+        'counts': c_list,
+        'adc': adc_list}
+    portID = ''  # нужно будет добавить порт
+    GD = GammaDevice(portID)
+    t = threading.Thread(target=GD.update_loop)
+    t.start()
 
     def __init__(self):
         pass
@@ -340,6 +355,17 @@ class Interface():
                                        textvariable=self.spinbox_var2TEM, width=7)
         self.grafspinTmaxTEM.grid(column=0, row=9)
 
+    def Driver(self):
+        if Interface.GD.updated:
+            Interface.znach['temperature'].append(Interface.GD.temperature)
+            Interface.znach['counts'].append(Interface.GD.counts)
+            Interface.znach['adc'].append(Interface.GD.adc)
+            Interface.GD.updated = False
+        else:
+            pass
+        Interface.window.after(1000, self.Driver)
+
+
     def start(self):
         self.mennuB()
         self.create_widgets_S()
@@ -348,7 +374,7 @@ class Interface():
         self.ADC_create_Figure()
         self.TEM_create_Figure()
         self.block_control()
-
+        self.Driver()
         Interface.window.mainloop()
 
     def clicked_file(self):

__init__.py

@@ -0,0 +1,3 @@
+from .wrapper import *
+
+__all__ = ["GammaDevice"]

wrapper.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+import serial
+from typing import List, Dict, Never, Iterator, Optional
+from enum import StrEnum
+import warnings
+import time
+import random
+
+
+class GammaDevice:
+    # Определение типов периферийных устройств
+    class PeripheryTypes(StrEnum):
+        ADC = "A"  # Аналого-цифровой преобразователь
+        DAC = "D"  # Цифро-аналоговый преобразователь
+        TEMP = "C"  # Температура
+        COUNTER = "T"  # Счетчик
+        PORTB = "B"  # Порт B
+
+    # Класс для хранения данных ЦАП
+    class DACStorage:
+        parent: GammaDevice
+        __channels: Dict[int, float]  # Частный атрибут для хранения каналов
+
+        def __init__(self, parent: GammaDevice):
+            self.parent = parent  # Ссылка на родительский объект GammaDevice
+            self.__channels = {1: 0, 2: 0, 3: 0, 4: 0}  # Инициализация каналов ЦАП
+
+        def __getitem__(self, index: int) -> float:
+            return self.__channels[index]  # Получение значения канала
+
+        def __setitem__(self, index: int, value: float):
+            if index in (1, 2):
+                self.__channels[index] = (int(value * 100) % 330) / 100  # Обработка значения для каналов 1 и 2
+                self.parent._GammaDevice__send(
+                    f"*D{index}W{f"{self.__channels[index] :04.2f}".replace(".", "")}#")  # Отправка команды на устройство
+            else:
+                self.__channels[index] = int(value) % 256  # Обработка значения для других каналов
+                self.parent._GammaDevice__send(
+                    f"*D{index}W{self.channels[index] :03d}#")  # Отправка команды на устройство
+
+        def __iter__(self) -> Iterator[int]:
+            return self.__channels.__iter__()  # Итератор по каналам
+
+    comport: Optional[serial.Serial]  # Последовательный порт, может быть None
+    message: str  # Сообщение для отправки/получения
+
+    __periphery: Dict[PeripheryTypes, Dict[int, int | float] | DACStorage]  # Словарь для периферийных устройств
+    update_pending: bool  # Флаг ожидания обновления
+    loop_started: bool  # Флаг запуска цикла
+    exposition_time: float  # Время экспозиции для цикла обновления
+
+    def __init__(self, portID: str):
+        try:
+            self.comport = serial.Serial(portID, baudrate=230400)  # Инициализация последовательного порта
+        except serial.SerialException as e:
+            raise e  # Обработка исключения при инициализации порта
+
+        self.message = ""  # Инициализация сообщения
+        self.__periphery = {
+            GammaDevice.PeripheryTypes.ADC: {1: 0},  # Инициализация АЦП
+            GammaDevice.PeripheryTypes.DAC: GammaDevice.DACStorage(self),  # Инициализация ЦАП
+            GammaDevice.PeripheryTypes.TEMP: {1: 0},  # Инициализация температуры
+            GammaDevice.PeripheryTypes.COUNTER: {2: 0},  # Инициализация счетчика
+            GammaDevice.PeripheryTypes.PORTB: {1: 0}  # Инициализация порта B
+        }
+        self.update_pending = False  # Инициализация флага ожидания обновления
+        self.loop_started = False  # Инициализация флага запуска цикла
+        self.error_occured = False  # Инициализация флага ошибки
+        self.exposition_time = 0.1  # Инициализация времени экспозиции
+
+    @property
+    def temperature(self) -> float:
+        return self.__periphery[GammaDevice.PeripheryTypes.TEMP][1]  # Получение значения температуры
+
+    @temperature.setter
+    def temperature(self, value: float) -> Never:
+        raise IOError(
+            "Temperature cannot be set from userspace")  # Запрет установки температуры из пользовательского пространства
+
+    @property
+    def counts(self) -> int:
+        return int(self.__periphery[GammaDevice.PeripheryTypes.COUNTER][2])  # Получение значения счетчика
+
+    @counts.setter
+    def counts(self, value: int):
+        if value:
+            warnings.warn("Setting counts to non-zero values from userspace is not recommended",
+                          category=RuntimeWarning)  # Предупреждение при установке счетчика
+        self.__periphery[GammaDevice.PeripheryTypes.COUNTER][2] = value % 1000
+        self.__send(f"*T2W{self.__periphery[GammaDevice.PeripheryTypes.COUNTER][2]}#")  # Отправка команды на устройство
+
+    @property
+    def portB(self) -> bool:
+        return True if self.__periphery[GammaDevice.PeripheryTypes.PORTB][1] else False  # Получение значения порта B
+
+    @portB.setter
+    def portB(self, value: bool):
+        self.__periphery[GammaDevice.PeripheryTypes.PORTB][1] = 1 if value else 0
+        self.__send(f"*B1W{self.__periphery[GammaDevice.PeripheryTypes.PORTB][1]}00#")  # Отправка команды на устройство
+
+    @property
+    def adc(self) -> float:
+        return self.__periphery[GammaDevice.PeripheryTypes.ADC][1]  # Получение значения АЦП
+
+    @adc.setter
+    def adc(self, value: float) -> Never:
+        raise IOError(
+            "Temperature cannot be set from userspace")  # Запрет установки АЦП из пользовательского пространства
+
+    @property
+    def dac(self) -> GammaDevice.DACStorage:
+        result = self.__periphery[GammaDevice.PeripheryTypes.DAC]
+        assert isinstance(result, GammaDevice.DACStorage)
+        return result  # Получение объекта DACStorage
+
+    @dac.setter
+    def dac(self, value: GammaDevice.DACStorage) -> Never:
+        raise IOError(
+            "DACStorage cannot be recreated from userspace")  # Запрет пересоздания DACStorage из пользовательского пространства
+
+    def update(self, updatedac: bool = False):
+        if updatedac:
+            self.__send("*D1R000#*D2R000#*D3R000#*D4R000#")  # Отправка команд для обновления ЦАП
+        else:
+            self.__send("*C1R000#*T2R000#*B1R000#")  # Отправка команд для обновления других устройств
+
+        if self.comport is None:
+            self.__periphery[GammaDevice.PeripheryTypes.COUNTER][2] = random.randint(1000,
+                                                                                     2000)  # Установка случайного значения счетчика, если последовательный порт не инициализирован
+
+    def __send(self, message: str):
+        if self.comport:
+            self.comport.write(
+                (message + "0" * (33 - len(message))).encode("ascii"))  # Отправка сообщения на устройство
+
+    def __update_values(self, log=False) -> int:
+        commands_parsed = 0
+
+        if self.comport:
+            self.message += self.comport.read_until(b"#").decode("utf8")  # Чтение сообщения до символа '#'
+
+            while "#" in self.message:
+                args = self.message.split("#")
+                command, self.message = args[0], "#".join(args[1:])
+                if "*" in command:
+                    command = command.replace("*", "")
+                    mode, deviceID, payload = command[0], command[1], command[2:]
+                    if int(deviceID) not in self.__periphery[GammaDevice.PeripheryTypes(mode)]:
+                        continue
+                    match GammaDevice.PeripheryTypes(mode):
+                        case GammaDevice.PeripheryTypes.TEMP:
+                            self.__periphery[GammaDevice.PeripheryTypes.TEMP][int(deviceID)] = int(
+                                payload) / 10  # Обновление значения температуры
+                        case GammaDevice.PeripheryTypes.COUNTER:
+                            self.__periphery[GammaDevice.PeripheryTypes.COUNTER][2] = int(
+                                payload)  # Обновление значения счетчика
+                        case GammaDevice.PeripheryTypes.PORTB:
+                            self.__periphery[GammaDevice.PeripheryTypes.PORTB][int(deviceID)] = int(
+                                payload[0])  # Обновление значения порта B
+                        case GammaDevice.PeripheryTypes.DAC:
+                            if int(deviceID) in (1, 2):
+                                value = int(payload[0:3]) / 100  # Обновление значения каналов 1 и 2 ЦАП
+                            else:
+                                value = int(payload[0:3])  # Обновление значения других каналов ЦАП
+                            self.__periphery[GammaDevice.PeripheryTypes.DAC]._DACStorage__channels[
+                                int(deviceID)] = value  # Обновление значения в __channels
+                        case GammaDevice.PeripheryTypes.ADC:
+                            self.__periphery[GammaDevice.PeripheryTypes.ADC][int(deviceID)] = int(
+                                payload[0:3]) / 100  # Обновление значения АЦП
+                    self.update_pending = True  # Установка флага ожидания обновления
+                if log:
+                    print(f"Parsed: {command}")
+                commands_parsed += 1
+        else:
+            self.__periphery[GammaDevice.PeripheryTypes.COUNTER][2] = random.randint(1000,
+                                                                                     7000)  # Установка случайного значения счетчика, если последовательный порт не инициализирован
+            self.update_pending = True  # Установка флага ожидания обновления
+
+        return commands_parsed
+
+    def read_loop(self) -> Never:
+        while True:
+            try:
+                self.__update_values()
+            except serial.SerialException as e:
+                self.error_occured = True  # Установка флага ошибки
+            except Exception as e:
+                print(e)
+            time.sleep(self.exposition_time / 10)  # Небольшая пауза перед следующей итерацией
+
+    def update_loop(self) -> Never:
+        self.loop_started = True  # Установка флага запуска цикла
+        while True:
+            if self.loop_started and not self.error_occured:
+                try:
+                    self.update()
+                except serial.SerialException as e:
+                    self.error_occured = True  # Установка флага ошибки
+                except Exception as e:
+                    print(e)
+
+            time.sleep(self.exposition_time)  # Пауза перед следующей итерацией