現代の世界では、配達サービスはますます人気があり、需要が高まっているため、この分野での自動化の可能性は、ビジネスとユーザーの両方にとって大きなメリットになります。私たちのブログ上の以前の記事では、我々はのためのマシンビジョンとニューラルネットワークの使用について話認識店舗で商品の値札を、などのための認識構成部品を。この記事では、それほど野心的ではない（しかしそれほど興味深いものではない）タスクについて説明します。Telegramチャットボット、QRコード、およびSAP SQL AnywhereリレーショナルDBMSを使用して、注文のステータスに関する顧客への通知を自動化します。
, , . , , QR-, Telegram . , Telegram , . , Telegram - API. Github , . , Telegram QR- live- - , .
, - e-mail, - . , , («», « », « », « » ..), . , «» – , . - , - QR-.
, SQL Anywhere, , - (Raspberry Pi PiCam, ) . , . QR-, . , , QR- -, , , Telegram . , , .
, QR- , , . QR- (UPC, ), . , , QR- Telegram, QR-.
Python, AIOgram Telegram Bot API sqlanydb SQL Anywhere. -, QR- OpenCV NumPy. , :
· SQL Anywhere ( sqlanydb);
· -, QR-, ( OpenCV NumPy);
· Telegram ( AIOgram).
.
SQL Anywhere
SAP SQL Anywhere. – , . , .
. CLI- dbinit:
dbinit -dba admin,password123 -p 4k -z UTF8 -ze UTF8 -zn UTF8 orders.db
«admin» ( «password123»), 4 , UTF-8. «orders.db» SQL Central ( SQL Anywhere) . SQL-:
CREATE TABLE Orders (
-- ID of an order
id UNSIGNED INT PRIMARY KEY NOT NULL IDENTITY,
-- Product's name
product NVARCHAR(24) NOT NULL,
-- Product's model
model NVARCHAR(20),
-- Product's price (in Euros)
price DECIMAL(10,2) NOT NULL,
-- Amount of the product
amount UNSIGNED INT NOT NULL DEFAULT 1,
-- Weight of the product (in kilograms)
weight DECIMAL(8,3) NOT NULL,
-- Customer's first name
first_name NVARCHAR(16) NOT NULL,
-- Customer's last name
last_name NVARCHAR(20),
-- Customer's physical address
address NVARCHAR(48) NOT NULL,
-- Customer's Telegram ID
telegram_id UNSIGNED INT NOT NULL,
-- Customer's timezone
timezone NVARCHAR(16) DEFAULT 'UTC',
-- Customer's prefered locale
locale NVARCHAR(5) DEFAULT 'en_US'
);
:
sqlanydb: credentials ( admin UID password123 ) ( .env dotenv). Orders:
conn = sqlanydb.connect(uid=config.DB_UID, pwd=config.DB_PASSWORD)
curs = conn.cursor()
QR-
, :
cap = cv2.VideoCapture(0)
, UI , QR- QR- .
async def scan_qr(area: int = 300, color: int = 196, side: int = 240, lang: str = "en", debug: bool = False) -> None:
"""Main function that creates a screen with the capture, monitors the web-cam's stream, searches for a QR-code in a squared area and passes the decoded QR-code to the notify module.
Args:
[optional] area (int): Minimal area of a detected object to be consider a QR-code.
[optional] color (int): Minimal hue of gray of a detected object to be consider a QR-code.
[optional] side (int): Length of the side of a square to be drawn in the center of the screen.
[optional] lang (str): Language of a text to be written above the square.
[optional] debug (bool): Crops and outputs an image containing inside the square at potential detection.
"""
if (cap is None) or (not cap.isOpened()):
logger.critical("No video stream detected. "
"Make sure that you've got a webcam connected and enabled")
return
kernel = np.ones((2, 2), np.uint8)
square = create_square(cap.read()[1], side=side)
while cap.isOpened():
ret, frame = cap.read()
key = cv2.waitKey(1)
if not ret or square is None or ((key & 0xFF) in {27, ord("Q"), ord("q")}):
exit(1)
image = draw_bounds(frame, square, lang=lang)
detected, cropped = detect_inside_square(frame, square, kernel, area_min=area, color_lower=color, debug=debug)
if detected:
address = detect_qr(cropped)
if address:
logger.debug("Detected: \"{}\"", address)
await notify.start(address)
cv2.imshow("Live Capture", image)
await asyncio.sleep(0.1)
. , ( -) OpenCV. , , QR- ( ), (x, y) `side` create_square(). draw_bounds() , `lang`. , draw_bounds() - UI:
, , QR- detect_inside_square():
def detect_inside_square(frame: Any, square: np.ndarray, kernel: np.ndarray, area_min: int = 300, color_lower: int = 212, color_upper: int = 255, debug: bool = False) -> Tuple[bool, Any]:
"""Detects and analyzes contours and shapes on the frame. If the detected shape's area is >= :area_min:, its color hue is >= :color_lower and a rectangle that encloses the shape contains inside the square returns True and the cropped image of the frame.
Args:
frame (Union[Mat, UMat]): A frame of the webcam's captured stream.
square (np.ndarray): A numpy array of the square's (x,y)-coordinates on the frame.
kernel (np.ndarray): A kernel for the frame dilation and transformation (to detect contours of shapes in the frame).
[optional] area_min (int): Minimal area of a detected object to be consider a QR-code.
[optional] color_lower (int): Minimal hue of gray of a detected object to be consider a QR-code.
[optional] color_upper (int): Maximal hue of gray of a detected object to be consider a QR-code.
[optional] debug (bool): Crops and outputs an image containing inside the square at potential detection.
Returns:
A tuple where the first element is whether a potential shape has been detected inside the square or not.
If it was then the second element is the square-cropped image with the detected shape, None otherwise.
"""
filter_lower = np.array(color_lower, dtype="uint8")
filter_upper = np.array(color_upper, dtype="uint8")
mask = cv2.inRange(frame, filter_lower, filter_upper)
dilation = cv2.dilate(mask, kernel, iterations=3)
closing = cv2.morphologyEx(dilation, cv2.MORPH_GRADIENT, kernel)
closing = cv2.morphologyEx(dilation, cv2.MORPH_CLOSE, kernel)
closing = cv2.GaussianBlur(closing, (3, 3), 0)
edge = cv2.Canny(closing, 175, 250)
if debug:
cv2.imshow("Edges", edge)
contours, hierarchy = cv2.findContours(edge, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
for contour in contours:
area = cv2.contourArea(contour)
if area < area_min:
continue
rect = cv2.minAreaRect(contour)
box = cv2.boxPoints(rect)
box = np.int0(box)
rect = order_points(box)
cv2.drawContours(frame, [box], 0, (0, 0, 255), 1)
if contains_in_area(rect, square):
cropped = frame[square[0][1]:square[2][1], square[0][0]:square[2][0]]
if debug:
cv2.imshow("Cropped", cropped)
return (True, cropped)
return (False, None)
, , . , QR- . , , «» (). . , `area_min`, , . , CLI-. , , , . , , , – , :
def contains_in_area(rectangle: np.ndarray, square: np.ndarray) -> bool:
"""Checks whether a rectangle fully contains inside the area of a square.
Args:
rectangle (np.array): An ordered numpy array of a rectangle's coordinates.
square (np.array): An ordered numpy array of a square's coordinates.
Returns:
Whether the rectangle contains inside the square. Since the both arrays are ordered it's suffice
to check that the top-left and the bottom-right points of the rectangle are both in the square.
"""
if ((rectangle[0][0] < square[0][0]) or (rectangle[0][1] < square[0][1])) or (
(rectangle[2][0] > square[2][0]) or (rectangle[2][1] > square[2][1])
):
return False
return True
, , QR-, ( `cropped`) detect_qr() .
QR-, , (, QR- `address`), Telegram ID :
async def start(address: str, pause_success: int = 5, pause_fail: int = 1) -> None:
"""Checks whether the :address: string contains in the set of all different addresses saved in the table.
If it does, gets the record containing :address: in its "address" field.
Sends the record to the notification function.
Args:
address (str): The decoded address to check the table with.
[optional] pause_success (int): Time in seconds to standby for after the notification was sent.
[optional] pause_fail (int): Time in seconds to standby for after detecting an invalid QR-code.
"""
try:
query_addresses = "SELECT address FROM %s.%s;"
curs.execute(
query_addresses
% (
config.DB_UID,
config.DB_TABLE_NAME,
)
)
response_addresses = curs.fetchall()
addresses = set([res[0] for res in response_addresses])
if not (address in addresses):
logger.warning('Address "{}" not found among the available addresses. Skipping', address)
logger.info("Standing by for {} second(s)", pause_fail)
await asyncio.sleep(pause_fail)
return
query = "SELECT * FROM %s.%s WHERE address='%s';"
curs.execute(
query
% (
config.DB_UID,
config.DB_TABLE_NAME,
address,
)
)
response = curs.fetchone()
logger.debug('Got response for address "{}": "{}"', address, response)
except sqlanydb.Error:
logger.exception("Encountered an error while handling query to the database. See below for the details")
return
res_row = {}
for (i, field) in zip(range(len(response)), config.FIELDS):
res_row[field] = response[i]
await notify_user(res_row)
logger.info("Standing by for {} second(s)", pause_success)
await asyncio.sleep(pause_success)
async def notify_user(row: Dict[str, str]) -> None:
"""Sends a notification about the order contained in :row: to a user with a Telegram ID from :row:.
Args:
row (dict): A dict containing full record about the user's order.
"""
try:
user_id = row["telegram_id"]
timestamp = datetime.now(pytz.timezone(row["timezone"])).strftime("%d/%m/%Y %H:%M:%S %Z")
lang = row.get("locale", "en_US")
info = constants.MSG_NOTIFY_EN if lang.startswith("en") else constants.MSG_NOTIFY_RU
info = info.format(
first_name=row["first_name"],
timestamp=timestamp,
id=row["id"],
address=row["address"],
product=row["product"],
model=row["model"],
price=float(row["price"]),
amount=row["amount"],
weight=float(row["weight"])
).replace(".", "\.").replace("-", "\-")
except KeyError:
logger.exception("Got invalid query response. See below for the details")
try:
await bot.send_message(user_id, info)
logger.success("Order notification message has been successfully sent to user {}", user_id)
except CantParseEntities as ex:
logger.error("Notification failed. AIOgram couldn't properly parse the following text:\n"
"\"{}\"\n"
"Exception: {}",
info, ex)
except ChatNotFound:
logger.error("Notification failed. User {} hasn\'t started the bot yet", user_id)
except BotBlocked:
logger.error("Notification failed. User {} has blocked the bot", user_id)
except UserDeactivated:
logger.error("Notification failed. User {}\'s account has been deactivated", user_id)
except NetworkError:
logger.critical("Could not access https://api.telegram.org/. Check your internet connection")
, -,
timestamp = datetime.now(pytz.timezone(row["timezone"])).strftime("%d/%m/%Y %H:%M:%S %Z")
, -, , , , Telegram.
, sqlanydb, , sanitization SQL-. , . `start()` «», , `address` . , QR- () «’; DROP TABLE Orders;», .
QR-. , , .
Telegram
, . , Telegram (, , , AIOgram , ). , AIOgram - HTTP- . , - Telegram . , /lang:
@dp.message_handler(commands=["lang"])
async def cmd_lang(message: Message) -> None:
"""Handles the "/lang" command from a Telegram user. Allows the user to change the locale from the chosen one.
Outputs the message in the language that was initially chosen by the user.
Args:
message (Message): User's Telegram message that is sent to the bot.
"""
query = "SELECT locale FROM %s.%s WHERE telegram_id=%d;"
curs.execute(
query
% (
config.DB_UID,
config.DB_TABLE_NAME,
message.from_user.id,
)
)
(lang,) = curs.fetchone()
logger.debug('Got user\'s {} current language "{}"', message.from_user.id, lang)
str_lang = "Please choose your language\." if lang.startswith("en") else ", \."
btn_en = InlineKeyboardButton("?? English", callback_data="lang_en")
btn_ru = InlineKeyboardButton("?? ", callback_data="lang_ru")
inline_kb = InlineKeyboardMarkup().add(btn_en, btn_ru)
await bot.send_message(message.chat.id, str_lang, reply_markup=inline_kb)
logger.info("User {} called /lang", message.from_user.id)
@dp.callback_query_handler(lambda c: c.data.startswith("lang"))
async def set_lang(cb_query: CallbackQuery) -> None:
"""Handles the callback that sets the user preferred locale. Updates the locale in the table.
Args:
cb_query (CallbackQuery): User's Telegram callback query that is sent to the bot.
"""
lang = "en_US" if cb_query.data.endswith("en") else "ru_RU"
info = "Setting your language..." if lang.startswith("en") else " ..."
await bot.answer_callback_query(cb_query.id, text=info)
try:
query = "UPDATE %s.%s SET locale='%s' WHERE telegram_id=%d;"
curs.execute(
query
% (
config.DB_UID,
config.DB_TABLE_NAME,
lang,
cb_query.from_user.id,
)
)
logger.debug("Commiting the changes")
conn.commit()
except sqlanydb.Error as ex:
logger.exception(ex)
return
str_setlang = (
"Language is set to English\.\nCall /lang to change it\."
if lang.startswith("en")
else " \.\n /lang, \."
)
logger.info('User {} set the language to "{}"', cb_query.from_user.id, lang)
await bot.send_message(cb_query.from_user.id, str_setlang)
, AIOgram , ( ) . , , «» – ID Telegram, .
SAP. , , . - . . - , .
, . , – SAP Cloud Platform (SCP). , Conversational AI .
Github. README , . - , , , issue pull request. !
- , Co-Innovation Labs, SAP Labs CIS