225 lines
8.8 KiB
Python
225 lines
8.8 KiB
Python
import asyncio
|
||
import socket
|
||
import time
|
||
import logging
|
||
import datetime
|
||
|
||
from aiogram import Router, F
|
||
from aiogram import Bot, Dispatcher, types
|
||
from aiogram.filters.command import Command
|
||
from aiogram.types import FSInputFile
|
||
from aiogram.fsm.context import FSMContext
|
||
from aiogram.fsm.storage.memory import MemoryStorage
|
||
from aiogram.fsm.state import StatesGroup, State
|
||
from aiogram.utils.chat_action import ChatActionSender
|
||
#from aiogram.types import BufferedInputFile
|
||
|
||
import configparser
|
||
from asyncpg_lite import DatabaseManager
|
||
|
||
from os import listdir
|
||
from os.path import isfile, join
|
||
|
||
import lib.OCer as OCer
|
||
import serial
|
||
|
||
router = Router()
|
||
|
||
async def get_user_list(db_url, deletion_password, table_name):
|
||
try:
|
||
db_manager = DatabaseManager(db_url=db_url, deletion_password=deletion_password)
|
||
data = []
|
||
async with db_manager as manager:
|
||
one_data = await manager.select_data(table_name=table_name)
|
||
for i in one_data:
|
||
data.append(i)
|
||
logging.info(f"Fetched {len(data)} records from the database.")
|
||
return data
|
||
|
||
except Exception as err:
|
||
logging.error(err)
|
||
pass
|
||
|
||
def init(Bot):
|
||
global config, bot, logger, some_open_ports, zxc
|
||
|
||
bot = Bot
|
||
#consts&confs
|
||
config = configparser.ConfigParser()
|
||
#bc_calculator = lomka.BCCCalculator()
|
||
some_open_ports = serial.tools.list_ports.comports()
|
||
|
||
#load config file
|
||
config.read('config.ini')
|
||
# device_hwid = config['DEFAULT']['device_hwid']
|
||
# baudrate = config['DEFAULT']['baudrate']
|
||
|
||
#log section
|
||
logging.basicConfig(
|
||
level=logging.INFO if config['DEFAULT']['debug'] == "false" else logging.DEBUG,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
handlers=[
|
||
logging.FileHandler('lastest.log'),
|
||
logging.StreamHandler()])
|
||
logger = logging.getLogger(socket.gethostname())
|
||
|
||
zxc = OCer.caser(device_hwid="PID=1A86:7523", baudrate=9600,
|
||
config=config, logger=logger)
|
||
zxc.set_controller_number(1)
|
||
zxc.set_port_device()
|
||
|
||
|
||
main_kb = [
|
||
[types.KeyboardButton(text="Выбрать контроллер")],
|
||
[types.KeyboardButton(text="Открыть секцию")],
|
||
[types.KeyboardButton(text="Статус секции")],
|
||
[types.KeyboardButton(text="Статус ВСЕХ секций")],
|
||
[types.KeyboardButton(text="Открыть ВСЕ секции")],
|
||
[types.KeyboardButton(text="Контакты")]]
|
||
|
||
keyboardos = types.ReplyKeyboardMarkup(keyboard=main_kb, resize_keyboard=True,
|
||
input_field_placeholder="amogus life420")
|
||
|
||
class Form(StatesGroup):
|
||
input_controller = State()
|
||
input_section = State()
|
||
input_status_section = State()
|
||
input_status_section_all = State()
|
||
input_approve = State()
|
||
|
||
@router.message(Command("start"))
|
||
async def cmd_start(message: types.Message):
|
||
|
||
keyboard = types.ReplyKeyboardMarkup(keyboard=main_kb, resize_keyboard=True,
|
||
input_field_placeholder="Выберите действие...", is_persistent=True)
|
||
|
||
await message.answer(f"{message.from_user.id}\n",
|
||
reply_markup=keyboard)
|
||
|
||
@router.message(F.text.lower() == "выбрать контроллер")
|
||
async def choose_controller(message: types.Message, state: FSMContext):
|
||
kb = [
|
||
[types.KeyboardButton(text="Домой")]
|
||
]
|
||
keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True,
|
||
input_field_placeholder="amogus life420")
|
||
|
||
await message.reply("Определите номер управляемого контроллера", reply_markup=keyboard)
|
||
|
||
await state.set_state(Form.input_controller)
|
||
|
||
@router.message(F.text, Form.input_controller)
|
||
async def set_controller_number(message: types.Message, state: FSMContext):
|
||
await state.update_data(confirm=message.text)
|
||
user_data = await state.get_data()
|
||
|
||
try:
|
||
await zxc.set_controller_capacity(int(user_data["confirm"]))
|
||
await message.reply(f"Контроллер ({user_data["confirm"]}) выбран успешно!",
|
||
reply_markup=keyboardos)
|
||
except Exception as e:
|
||
await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos)
|
||
await state.clear()
|
||
|
||
@router.message(F.text.lower() == "статус всех секций")
|
||
async def status_all_sections(message: types.Message, state: FSMContext):
|
||
|
||
cool_data = await zxc.READ_ALL_CELLS()
|
||
|
||
await message.reply(f"{cool_data}", reply_markup=keyboardos)
|
||
await state.clear()
|
||
|
||
@router.message(F.text.lower() == "открыть все секции")
|
||
async def open_all_sections(message: types.Message, state: FSMContext):
|
||
kb = [
|
||
[types.KeyboardButton(text="Продолжить")],
|
||
[types.KeyboardButton(text="Домой")]
|
||
]
|
||
keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True,
|
||
input_field_placeholder="amogus life420")
|
||
|
||
await message.reply("Данная операция откроет ВСЕ секции доступные контроллеру.\nПродолжить?", reply_markup=keyboard)
|
||
|
||
await state.set_state(Form.input_approve)
|
||
|
||
@router.message(F.text, Form.input_approve)
|
||
async def open_all_sections_confirm(message: types.Message, state: FSMContext):
|
||
await state.update_data(confirm=message.text)
|
||
user_data = await state.get_data()
|
||
|
||
if str(user_data["confirm"]).lower() == "продолжить":
|
||
try:
|
||
cool_data = await zxc.OPEN_ALL_CELLS_LEGACY(True)
|
||
await message.reply(f"Статус ячеек:\n{cool_data}",
|
||
reply_markup=keyboardos)
|
||
except Exception as e:
|
||
await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos)
|
||
await state.clear()
|
||
|
||
@router.message(F.text.lower() == "контакты")
|
||
async def contact_info(message: types.Message):
|
||
keyboard = types.ReplyKeyboardMarkup(keyboard=main_kb, resize_keyboard=True,
|
||
input_field_placeholder="amogus life420")
|
||
await message.reply("Настучать админу: \n" \
|
||
"Discord: Valo#1234 | @valoker\n" \
|
||
"GitHub: @Valoker\n" \
|
||
"Telegram: @Valoka", reply_markup=keyboard)
|
||
|
||
@router.message(F.text.lower() == "домой")
|
||
async def home(message: types.Message, state: FSMContext):
|
||
await message.answer(reply_markup=keyboardos,
|
||
text="Меню")
|
||
await state.clear()
|
||
|
||
@router.message(F.text.lower() == "открыть секцию")
|
||
async def open_section(message: types.Message, state: FSMContext):
|
||
kb = [
|
||
[types.KeyboardButton(text="Домой")]
|
||
]
|
||
keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True,
|
||
input_field_placeholder="amogus life420")
|
||
|
||
await message.reply("Отправьте номер секции для открытия..", reply_markup=keyboard)
|
||
await state.set_state(Form.input_section)
|
||
|
||
@router.message(F.text, Form.input_section)
|
||
async def open_section_number(message: types.Message, state: FSMContext):
|
||
await state.update_data(confirm=message.text)
|
||
user_data = await state.get_data()
|
||
|
||
try:
|
||
await zxc.open_cell(int(user_data["confirm"]))
|
||
await message.reply(f"Секция ({user_data["confirm"]}) контроллера " \
|
||
f"({zxc.controller_number}) открыта.",
|
||
reply_markup=keyboardos)
|
||
except Exception as e:
|
||
await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos)
|
||
await state.clear()
|
||
|
||
@router.message(F.text.lower() == "статус секции")
|
||
async def get_section_status(message: types.Message, state: FSMContext):
|
||
kb = [
|
||
[types.KeyboardButton(text="Домой")]
|
||
]
|
||
keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True,
|
||
input_field_placeholder="amogus life420")
|
||
|
||
await message.reply("Отправьте номер желаемой секции..", reply_markup=keyboard)
|
||
|
||
await state.set_state(Form.input_status_section)
|
||
|
||
@router.message(F.text, Form.input_status_section)
|
||
async def get_section_status_fin(message: types.Message, state: FSMContext):
|
||
await state.update_data(confirm=message.text)
|
||
user_data = await state.get_data()
|
||
|
||
try:
|
||
controller_data = await zxc.read_controller(cell_number=int(user_data["confirm"]))
|
||
await message.reply(f"Секция ({user_data["confirm"]}) контроллера " \
|
||
f"({zxc.controller_number}) со статусом: {controller_data[0]["CURRENT_STATUS"]}. \n{controller_data}",
|
||
reply_markup=keyboardos)
|
||
except Exception as e:
|
||
await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos)
|
||
await state.clear()
|
||
|