Python Sink

· Python ·

TOC

• Sending email
• Environments
• Libs
  • asyncio
  • os
  • datetime
  • logging
  • schedule and threading
  • collections
  • string
  • enum
  • random
  • pickle
  • aiogram
  • matplotlib
  • wx
  • numPy
    • Datatypes
    • Arrays
  • PyTorch
  • TiffFile

Sending email

Tuned for gmail with starttls

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib

sender_email = "example@gmail.com"
receiver_email = "example@gmail.com"
acc_token = "email_password_or_token"
message = MIMEMultipart("alternative")
message["Subject"] = mail_sub
message["From"] = sender_email
message["To"] = receiver_email
message.attach(MIMEText(mail_body, "html"))

with smtplib.SMTP("smtp.gmail.com", 587) as server:
	server.starttls() 
    server.login(sender_email, acc_token)
    server.sendmail(sender_email, receiver_email, message.as_string())
    server.quit()

Environments

Source

rem create
python -m venv tutorial-env

rem create other version
py -3.11 -m venv F:\Keerah\Projects2\Dev\segment\segment\env

rem activate
tutorial-env\Scripts\activate.bat

rem deactivate
deactivate

rem install package 
python -m pip install novas

rem install specific version
python -m pip install requests==2.6.0

rem upgrade
python -m pip install --upgrade requests

rem info about the package
python -m pip show requests

rem install requirements (from the current folder)
python -m pip install -r requirements.txt
pip3 install -r requirements.txt

to select VENV in VSCode
Ctrl - Shift - P
Python: select interpreter

rem upgrade ENV python version

py --list

py -3.7 -m venv env_path

Libs

asyncio

import asyncio
import logging

# adding async event to event loop from sync code
def message_post_event(chatid: int, text: str, *args):
    logging.debug('Adding message post task to the main loop')
    quiet = True if not args else args[0]
    bot_loop = runner.get_loop()
    tasks = bot_loop.create_task(message_post(chatid, text, quiet))

async def main():
    message_post_event(admin_id, '👋 Bot is on duty 👋', True)
    await dp.start_polling(bot)

runner = asyncio.Runner()

if __name__ == '__main__':
	 runner.run(main())

os

os.system('title New window')        # cmd window title
filename=os.path.split(filepath)[1]  # file name from path

dosname = filename.replace('/', '\\')
os.system(f'copy "{dosname}" "{dosname}.bak"') # os rename file

datetime

import datetime

now_to_str = dt.datetime.strftime(dt.datetime.now(), '<date format>')
time_delta = dt.datetime.now() - dt.datetime.strptime('<string date>', '<date format>')
print(delta.seconds)

logging

import logging

logging.basicConfig(level=logging.INFO)

logging.error('hey this code is not something to run')

schedule and threading

# shows how to schedule events in a separate thread

import schedule
import threading
import logging

def schedule_thread():
    schedule_logger = logging.getLogger('schedule')
    schedule_logger.setLevel(level=logging.INFO)
    schedule.every(td.store_period).minutes.do(some_function1)
    schedule.every(td.tick_interval).minutes.do(some_function2)
    schedule.every().hour.at(":00").do(some_function3)
    
    logging.warning('Scheduler thread started')
    while _is_running:
        schedule.run_pending()
        time.sleep(1)

    schedule.clear()
    logging.error('Scheduler thread stopped')
    return

if __name__ == '__main__':
    schedule_job = threading.Timer(5.0, schedule_thread)
    schedule_job.start()

collections

from collections import OrderedDict

def hash_text(text: str)-> str:
    return "".join(OrderedDict.fromkeys(text))

string

import string

def contains_hashed(text: str, items_list: list) -> bool:
    stripped = text.translate(str.maketrans('','', string.punctuation))
    hashed = hash_text(stripped)
    return any(item in hashed for item in items_list)

enum

· Enum Docs · Enum HowTo ·

from enum import Enum

class Data_Item(Enum):
	LEADER =          1, 'l',  'All time activity',  'Ever',      'members'    
	LEADERHOUR =      2, 'h',  'Last hour activity', 'Last hour', 'members'
	LEADERDAY =       3, 'd',  'This day activity',  'This day',  'members'
	KARMA =           4, 'k',  'Karma',              '',          'members'
	NAME =            5, 'n',  'Member name',        '',          'members'

	def __init__(self, value: int, symbol: str, doc: str, period: str, units: str):
		self._value_ = value
		self.symbol = symbol
		self.doc = doc # or builtin self.__doc__
		self.period = period
		self.units = units

print(f'Item: {Data_Item.LEADER}')
print(f'Type: {type(Data_Item.LEADER)}')
print(f'Item name: {Data_Item.LEADERDAY.name}')
print(f'Item NAME name: {Data_Item.NAME.name}')
print(f'Item LEADER description: {Data_Item.LEADER.doc}')
print(f'Same porogrammatically: {Data_Item["LEADER"].doc}')

random

import random

random.seed()  # seed from current time
randrange(10)  # [0,10)

pickle

import pickle
import logging

with open(filename, 'wb') as f:
	pickle.dump(variable, f)
	f.close

try:
	with open(filename, 'rb') as f:
		stats = pickle.load(f)
		f.close
	except Exception as e:
		logging.error('Error loading db file with exeption:')
		logging.error(e)
		result = None
	finally:
		if result:
			logging.info('Data loaded')
		else:
			logging.error('Loading error')

aiogram

· Aiogram v3.x Docs ·

# Basic example of main functionality
# from running the bot in poll mode
# to command, message and inline keyboard callbacks handling
# also shows how to add async events from sync code to the bot event loop (using Runner)

from aiogram import Bot, Dispatcher, types
from aiogram import F
from aiogram.types import Message, ContentType, URLInputFile, FSInputFile
from aiogram.filters import Command, CommandObject
from aiogram.enums import ParseMode
from aiogram.filters import Filter
from aiogram.filters.callback_data import CallbackData
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram.utils.media_group import MediaGroupBuilder
from aiogram.utils.chat_action import ChatActionSender

from enum import Enum
import logging, asyncio
from enum import Enum

bot_token = os.getenv('BOT_TELEGRAM_TOKEN', '<bot token>')
flood_timeout = int(os.getenv('BOT_FLOOD_TIMEOUT', '30'))
destruction_timeout = int(os.getenv('BOT_DELETE_TIMEOUT', '60'))
whitelist_chats = os.getenv('BOT_ALLOWED_CHATS', '<chat id>, <chat id>') #
whitelist_chats: list = None if whitelist_chats == '' else [int(chat) for chat in whitelist_chats.split(',')]

_is_running = True
admin_id = <admin user_id int>

# Initialize bot and dispatcher
logging.basicConfig(level=logging.INFO)
runner = asyncio.Runner()
bot: Bot = Bot(token=bot_token)
dp: Dispatcher = Dispatcher()


class VoteCallback(CallbackData, prefix="voting"):
    action: str
    value: str


class Reaction(Enum):
    LIKE =       'Liked',      '👍'
    DISLIKE =    'Disliked',   '👎'
    VOTEUP =     'Voted up',   '🔼'
    VOTEDOWN =   'Voted down', '🔽'
    CANCEL =     'Cancelled',  '❌'

    def __init__(self, doc: str, symbol: str):
        self.doc = doc
        self.symbol = symbol


# define inline keyboard
def get_bot_keyboard():
    builder = InlineKeyboardBuilder()
    builder.button(text=Reaction.LIKE.symbol,    callback_data=VoteCallback(action='vote', value=Reaction.LIKE.name))
    builder.button(text=Reaction.DISLIKE.symbol, callback_data=VoteCallback(action='vote', value=Reaction.DISLIKE.name)) 
    builder.button(text=Reaction.CANCEL.symbol,  callback_data=VoteCallback(action='vote', value=Reaction.CANCEL.name))
    builder.adjust(3)
    return builder.as_markup()


# start command handler
@dp.message(F.content_type == ContentType.TEXT, Command(re.compile(Event.START.triggers[0])))
async def command_start(message: types.Message):
	await autodelete_message(message.chat.id, message.message_id, 0)
	async with ChatActionSender.typing(bot=bot, chat_id=message.chat.id):
		msg = await bot.send_message(
							message.chat.id,
							text='hello, dear <b>Username</b>',
							parse_mode=ParseMode.HTML)    
	await autodelete_message(meg.chat.id, msg.message_id, 0)
	


# non-text messages handler
@dp.message(~F.content_type.in_({ContentType.TEXT}))
async def msg_media_activity(message: types.Message):
	async with ChatActionSender.upload_video(bot=bot, chat_id=message.chat.id):
		gif_url = 'https://media.tenor.com/J7nq50zbllUAAAAd/dog-funny.gif'
		await message.reply_animation(URLInputFile(url=gif_url, filename=os.path.split(gif_url)[1]), reply_markup=get_bot_keyboard())


# text-messages handler
@dp.message((F.content_type == ContentType.TEXT) & (F.text))
async def msg_activity(message: types.Message):
	async with ChatActionSender.upload_voice(bot=bot, chat_id=message.chat.id):
		media_file = 'files/reply.mp3'
		await message.reply_voice(FSInputFile(media_file,filename=os.path.split(media_file)[1]), reply_markup=get_bot_keyboard())


# inline keyboard callback handler
@dp.callback_query(VoteCallback.filter(F.action == 'vote')) # VoteCallback.filter(F.action.in_({'vote', 'like'}))
async def callback_vote(query: types.CallbackQuery, callback_data: VoteCallback):
    callback_data_action   = callback_data.action
    callback_data_reaction = Reaction[callback_data.value]
    
    if callback_data_reaction in [Reaction.LIKE]
	    await query.answer('Thanks', show_alert=False, cache_time=1)
	elif callback_data_reaction in [Reaction.DISLIKE, Reaction.CANCEL]
		await query.answer('Got it', show_alert=False, cache_time=1)


# autodelete message func
async def autodelete_message(chat_id: int, message_id: int, seconds=0):
    await asyncio.sleep(seconds)
    await bot.delete_message(chat_id=chat_id, message_id=message_id)


# async event to insert from sync
async def message_post(chatid: int, text: str, quiet: bool):
    async with ChatActionSender.typing(bot=bot, chat_id=chatid):
        await bot.send_message(chat_id=chatid,
                               text=f'{text}',
                               parse_mode=ParseMode.HTML,
                               disable_notification=quiet)


# add async event from sync code
def message_post_event(chatid: int, text: str, *args):
    logging.debug('\n🪲 Adding message post task to the main loop')
    quiet = True if not args else args[0]
    bot_loop = runner.get_loop()
    tasks = bot_loop.create_task(message_post(chatid, text, quiet))
    # tasks.add_done_callback(lambda t: logging.debug('Message posted\n'))


async def main():
    message_post_event(admin_id, '👋 Bot is on duty 👋', True)
    await dp.start_polling(bot)

if __name__ == '__main__':
	runner.run(main())

#media group handling
import os
from aiogram.utils.media_group import MediaGroupBuilder

files = ['filepath1', 'filepath2']
media_group = MediaGroupBuilder(caption="Here is your images")
for file in files:
	media_group.add(type="photo", media=FSInputFile(file, filename=os.path.split(file)[1]))

await message.reply_media_group(media=media_group.build())

matplotlib

· Matplotlib Docs ·

from matplotlib import pyplot as plt
import matplotlib.dates as mdates
from matplotlib.ticker import AutoMinorLocator, MultipleLocator

def forward(x): return x**(1/5)
def inverse(x): return x**5
linestyle_str = ['solid', 'dashed', (0, (5, 1)), 'dashdot', (5, (10, 3)), 'dotted']

fig, ax = plt.subplots(facecolor=color_bg, layout='constrained')
fig.set_facecolor(color_bg)
fig.set_facecolor(color_bg)
fig.patch.set_facecolor(color_bg2)
ax.plot(values_x_list,
		values_y_list,
		label = 'Main plot label',
		linewidth=0.5,
		linestyle=linestyle_str[0],
		color=color_fg1,
		zorder=0.5,
		solid_joinstyle='round',
		solid_capstyle='round',
		dash_joinstyle='round',
		dash_capstyle='round')
ax2 = ax.twinx()
ax2.plot(values_x_list,
		values_y_list,
		label='Line name',
		linewidth=0.5,
		linestyle=linestyle_str[1],
		color=color_fg2,
		zorder=1,
		solid_joinstyle='round',
		solid_capstyle='round',
		dash_joinstyle='round',
		dash_capstyle='round')

fig.suptitle('Main plots', color=color_fg)
fig.set_facecolor(color_bg)
fig.patch.set_facecolor(color_bg2)
ax.set_facecolor(color_bg)
ax.set_ylabel('Cumulative Karma', color=color_fg3, fontsize='x-small')
ax.set_xlabel('Date', color=color_fg, fontsize='x-small')
locatorx1maj = mdates.AutoDateLocator()
locatorx1min = mdates.AutoDateLocator()
locatorx1min.intervald[DAILY] = [1]
formatter1 = mdates.ConciseDateFormatter(locatorx1maj)
ax.xaxis.set_major_locator(locatorx1maj)
ax.xaxis.set_minor_locator(locatorx1min)
ax.xaxis.set_major_formatter(formatter1)
ax.tick_params(axis='both', which='both', color=color_fg2)
ax.xaxis.grid(visible=True, which='both', color=color_bg2, linestyle='solid', linewidth=0.2, zorder=0)

ax2.set_facecolor(color_bg)
ax2.set_ylabel('Secondary plots', color=color_fg, fontsize='x-small')
ax2.spines[['bottom','top', 'left', 'right']].set_color(color_bg)
ax2.set_yscale('function', functions=(forward, inverse))
ax2.yaxis.set_major_locator(MultipleLocator(100))
ax2.tick_params(axis='both', which='both', color=color_fg2)
for label in (ax2.get_xticklabels()+ax2.get_yticklabels()+ax.get_xticklabels()):
	label.set_fontsize(6)
	label.set_color(color_fg)
for label in ax.get_yticklabels():
	label.set_fontsize(6)
	label.set_color(color_fg3)
ax2.yaxis.grid(visible=True, which='major', color=color_bg2, linestyle='solid', linewidth=0.2, zorder=0)
fig.legend(loc='outside right', ncols=1, fontsize='xx-small', labelcolor=color_fg, facecolor=color_bg, framealpha=0) 
plt.savefig(plotfile, dpi=fig_dpi) 
plt.close()

wx

wx docs
wx wiki
useful wx wiki examples
wx widgets

numPy

Datatypes

Docs

Advanced types for structured arrays

Arrays

slicing np arrays

operations

import numpy as np

# fills
np.ones(shape=(x,y,z)) 
np.zeroes(shape=()) 
nparray.fill(value) # in place method
nparray[:] = value
nparray.repeat(value=Any, shape=(), axis=0)
nparray.arrange(minval, maxval, stepval)

# extract
np.unique(array, axis=0)
np.diagonal(array, offset=0)

# fill wit random 8-bit vectors of range(0,256)
np.random.randint(mival=0, maxval=256, shape=(256, 256, 4), 'uint8')

# concatenate with a fill of random color and alpha of 0.55
np.concatenate([np.random.random(3), [0.55]]) 

# reshape
nparray.reshape(shape=())
nparray.flatten()                    # copy of the origian array, = npayrray.reshape(nparray.size())
nparray.ravel()                      # view of the original array
nparray.swapaxes(array, axis, axis)
nparray.transpose(axis, axis)        # same as swapaxes(array, 0, 1)
nparray.T                            # alias for transpose()
np.stack(array)                      # Join a sequence of arrays along a new axis

# conversion
nparray.tolist(array)
nparray.tofile('myarray.txt', sep=';')

# math
nparray.sum(axis=0)
nparray.prod()
nparray.mean()
nparray.min()
nparray.argmin()     # index of
nparray.max()
nparray.argmax()     # index of
nparray.ptp()        # = nparray.max() - nparray.()min
nparray.size()

a + b           # element-wise sum of matrices
a % b           # element-wise modulo
a // b          # element-wise floor division
np.floor(a/b)   # includes conversion to float
a * b           # element-wise multiplication

np.matmul(a, b) # matrix multiplication, dot product
np.dot(a, b)    # same, dot product
a @ b           # same, dot product

PyTorch

Start locally instructions

TiffFile

Example of multi-layered tiff using psdtags