Compare commits

...

5 Commits

View File

@@ -5,11 +5,9 @@ from datetime import datetime, timedelta
from threading import Thread
import journal
# SUPPORTED_REGISTER_TYPES = ['D', 'M']
SUPPORTED_REGISTER_TYPES = ['D']
LOG_TYPE_ON_SCAN = 'on-scan'
LOG_TYPE_ON_CHANGE = 'on-change'
@@ -109,7 +107,6 @@ class Endpoint(Thread):
self.ranges = parse_ranges(config)
# база данных
self.connection = None
self.db_table = config['database']['db-table']
self.db_col_names = ''
for rn in self.ranges:
@@ -134,14 +131,21 @@ class Endpoint(Thread):
return values
def __read_last_db_regs(self):
with self.connection.cursor() as cursor:
cursor.execute(f"select {self.db_col_names} from `{self.db_table}`"
f"order by `{self.config['database']['db-datetime-col']}` desc limit 1;")
row = cursor.fetchall()
if len(row) != 0:
return row[0]
else:
return None
with connect(
host=self.config["database"]["host"],
port=self.config["database"]["port"],
user=self.config["database"]["db-user"],
password=self.config["database"]["db-password"],
database=self.config["database"]["db-name"],
) as connection:
with connection.cursor() as cursor:
cursor.execute(f"select {self.db_col_names} from `{self.db_table}`"
f"order by `{self.config['database']['db-datetime-col']}` desc limit 1;")
row = cursor.fetchall()
if len(row) != 0:
return row[0]
else:
return None
def __insert_db_row(self, values):
vals_str = ""
@@ -151,10 +155,17 @@ class Endpoint(Thread):
else:
vals_str += f"{v}"
query = f"insert into `{self.db_table}` ({self.db_col_names}) values ({vals_str});"
journal.log(f"Executing query: {query}")
with self.connection.cursor() as cursor:
cursor.execute(query)
with connect(
host=self.config["database"]["host"],
port=self.config["database"]["port"],
user=self.config["database"]["db-user"],
password=self.config["database"]["db-password"],
database=self.config["database"]["db-name"]
) as connection:
query = f"insert into `{self.db_table}` ({self.db_col_names}) values ({vals_str});"
journal.log(f"[{self.config['database']['db-user']}:{self.db_table}] Executing query: {query}")
with connection.cursor() as cursor:
cursor.execute(query)
def __endless_loop(self):
# if self.log_type == 'on-change':
@@ -168,40 +179,37 @@ class Endpoint(Thread):
last_row = None
while True:
# время следующего опроса
need_time = last_query + scan_rate
try:
row = self.__mb_read_all_regs()
while True:
try:
row = self.__mb_read_all_regs()
if self.log_type == LOG_TYPE_ON_SCAN:
if self.log_type == LOG_TYPE_ON_SCAN:
self.__insert_db_row(row)
last_row = row
else:
equal = True
if last_row is None:
equal = False
else:
for i in range(0, len(last_row)):
if last_row[i] != row[i]:
equal = False
break
if not equal:
self.__insert_db_row(row)
last_row = row
else:
equal = True
if last_row is None:
equal = False
else:
for i in range(0, len(last_row)):
if last_row[i] != row[i]:
equal = False
break
if not equal:
self.__insert_db_row(row)
last_row = row
last_query = need_time
break
except Exception as ex:
journal.log(f"Exception: {ex}")
time.sleep(0.1) # небольшая задержка, чтоб не спамить запросами
need_time = datetime.now() + scan_rate
except Exception as ex:
journal.log(f"[{self.config['database']['db-user']}:{self.db_table}] Exception: {ex}")
# вычислим время до следующего опроса и подождем
# время следующего опроса
need_time = last_query + scan_rate
curr_time = datetime.now()
if need_time > curr_time:
delta = need_time - curr_time
time.sleep(delta.seconds + (delta.microseconds / 1000000))
last_query = need_time
else:
last_query = datetime.now()
def run(self):
# запускаем клиента модбас
@@ -210,15 +218,6 @@ class Endpoint(Thread):
port=self.config["modbus"]["port"],
debug=self.config["modbus"]["debug"],
auto_open=True)
with connect(
host=self.config["database"]["host"],
port=self.config["database"]["port"],
user=self.config["database"]["db-user"],
password=self.config["database"]["db-password"],
database=self.config["database"]["db-name"]
) as _connection:
self.connection = _connection
# запуск бесконечного цикла
self.__endless_loop()
# запуск бесконечного цикла
self.__endless_loop()