Virtual Env
https://packaging.python.org/guides/installing-using-pip-and-virtualenv/
setup virtual env
$ python3 -m virtualenv env
switch env (goto the folder)
$ source env/bin/activate
Python Tips
https://packaging.python.org/guides/installing-using-pip-and-virtualenv/
$ python3 -m virtualenv env
$ source env/bin/activate
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
@scheduler.scheduled_job('cron', hour=16, minute=59, second=59)
def execute_daily_task():
pass # do something
scheduler.start() # start scheduler
import sys
import time
from subprocess import Popen, PIPE
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BackgroundScheduler(
{'apscheduler.job_defaults.max_instances':'2'}
)
blocker = BlockingScheduler()
theproc = None
def run_task():
# run long process task
print('run task', time.ctime())
global theproc
theproc = Popen([sys.executable, "LongProcessTask.py"], stdout=PIPE)
result = theproc.communicate()
print(result)
# run task every 30 mins
@scheduler.scheduled_job('interval', minutes=30)
def execute_restart():
print('restarting', time.ctime())
global theproc
# if old process exist, kill it
if theproc is not None:
print('kill old process')
theproc.kill()
# run task
run_task()
scheduler.add_job(run_task, 'date') # run the task now
scheduler.start()
# blocker prevent the program from exiting
blocker.start()
class ObjectHelper(object):
@staticmethod
def stringify(obj):
attrs = vars(obj)
return ', '.join("%s: %s" % item for item in attrs.items())
# plugin framework
def start(self):
"""Start tracker engine.
"""
if not self._loop.is_running():
LOGGER.info('Run event loop forever')
threading.Thread(target=self._loop.run_forever).start()
if not self._is_started:
self._loop.call_soon_threadsafe(
asyncio.ensure_future, self._start()
)
LOGGER.info("Request progress tracker started")
self._is_started = True
return
if self._is_paused:
self._is_paused = False
else:
LOGGER.warning('Request progress tracker already started')
https://stackoverflow.com/questions/29760402/converting-a-txt-file-to-an-image-in-python
import PIL
import PIL.Image
import PIL.ImageFont
import PIL.ImageOps
import PIL.ImageDraw
PIXEL_ON = 0 # PIL color to use for "on"
PIXEL_OFF = 255 # PIL color to use for "off"
def main():
image = text_image('content.txt')
image.show()
image.save('content.png')
def text_image(text_path, font_path=None):
"""Convert text file to a grayscale image with black characters on a white background.
arguments:
text_path - the content of this file will be converted to an image
font_path - path to a font file (for example impact.ttf)
"""
grayscale = 'L'
# parse the file into lines
with open(text_path) as text_file: # can throw FileNotFoundError
lines = tuple(l.rstrip() for l in text_file.readlines())
# choose a font (you can see more detail in my library on github)
large_font = 20 # get better resolution with larger size
font_path = font_path or 'cour.ttf' # Courier New. works in windows. linux may need more explicit path
try:
font = PIL.ImageFont.truetype(font_path, size=large_font)
except IOError:
font = PIL.ImageFont.load_default()
print('Could not use chosen font. Using default.')
# make the background image based on the combination of font and lines
pt2px = lambda pt: int(round(pt * 96.0 / 72)) # convert points to pixels
max_width_line = max(lines, key=lambda s: font.getsize(s)[0])
# max height is adjusted down because it's too large visually for spacing
test_string = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
max_height = pt2px(font.getsize(test_string)[1])
max_width = pt2px(font.getsize(max_width_line)[0])
height = max_height * len(lines) # perfect or a little oversized
width = int(round(max_width + 40)) # a little oversized
image = PIL.Image.new(grayscale, (width, height), color=PIXEL_OFF)
draw = PIL.ImageDraw.Draw(image)
# draw each line of text
vertical_position = 5
horizontal_position = 5
line_spacing = int(round(max_height * 0.8)) # reduced spacing seems better
for line in lines:
draw.text((horizontal_position, vertical_position),
line, fill=PIXEL_ON, font=font)
vertical_position += line_spacing
# crop the text
c_box = PIL.ImageOps.invert(image).getbbox()
image = image.crop(c_box)
return image
if __name__ == '__main__':
main()
https://developers.redhat.com/blog/2018/08/13/install-python3-rhel/
Locks have 2 states: locked and unlocked. 2 methods are used to manipulate them: acquire() and release()
RLock is a reentrant lock. acquire() can be called multiple times by the same thread without blocking. Keep in mind that release() needs to be called the same number of times to unlock the resource.
Condition is a synchronization mechanism where a thread waits for a specific condition and another thread signals that this condition has happened. Once the condition happened, the thread acquires the lock to get exclusive access to the shared resource.
Semaphore is based on an internal counter which is decremented each time acquire() is called and incremented each time release() is called. If the counter is equal to 0 then acquire() blocks. It is the Python implementation of the Dijkstra semaphore concept: P() and V(). Using a semaphore makes sense when you want to control access to a resource with limited capacity like a server.
Event a simple mechanism. A thread signals an event and the other thread(s) wait for it.