Capture and Save IP Camera in Python OpenCV

import cv2

#print("Before URL")
cap = cv2.VideoCapture('rtsp://[email protected]:554/cam/realmonitor?channel=1&subtype=1')
#print("After URL")

(grabbed, frame) = cap.read()
fshape = frame.shape
fheight = fshape[0]
fwidth = fshape[1]
fourcc = cv2.VideoWriter_fourcc(*'XVID')
out = cv2.VideoWriter('output.avi',fourcc, 20.0, (fwidth,fheight))

while True:

    #print('About to start the Read command')
    ret, frame = cap.read()
    #print('About to show frame of Video.')
    cv2.imshow("Capturing",frame)
    #print('Running..')

    out.write(frame)

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

References
https://pupli.net/2021/01/access-ip-camera-in-python-opencv/
https://stackoverflow.com/questions/29317262/opencv-video-saving-in-python

Access IP Camera in Python OpenCV

import cv2

#print("Before URL")
cap = cv2.VideoCapture('rtsp://admin:[email protected]/H264?ch=1&subtype=0')
#print("After URL")

while True:

    #print('About to start the Read command')
    ret, frame = cap.read()
    #print('About to show frame of Video.')
    cv2.imshow("Capturing",frame)
    #print('Running..')

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

References
https://stackoverflow.com/questions/49978705/access-ip-camera-in-python-opencv
https://dahuawiki.com/Remote_Access/RTSP_via_VLC

SQLite3 and multiple processes

First, avoid concurrent access to sqlite database files. Concurrency is one of sqlite’s weak points and if you have a highly concurrent application, consider using another database engine.

If you cannot avoid concurrency or drop sqlite, wrap your write transactions in BEGIN IMMEDIATE; … END;. The default transaction mode in sqlite is DEFERRED which means that a lock is acquired only on first actual write attempt. With IMMEDIATE transactions, the lock is acquired immediately, or you get SQLITE_BUSY immediately. When someone holds a lock to the database, other locking attempts will result in SQLITE_BUSY.

Dealing with SQLITE_BUSY is something you have to decide for yourself. For many applications, waiting for a second or two and then retrying works quite all right, giving up after n failed attempts. There are sqlite3 API helpers that make this easy, e.g. sqlite3_busy_handler() and sqlite3_busy_timeout() but it can be done manually as well.

You could also use OS level synchronization to acquire a mutex lock to the database, or use OS level inter-thread/inter-process messaging to signal when one thread is done accessing the database.

If you want to use Python’s automatic transaction handling, leave isolation_level at its default value, or set it to one of the three levels.

If you want to do your own transaction handling, you have to prevent Python from doing its own by setting isolation_level to None.

References
https://stackoverflow.com/questions/1063438/sqlite3-and-multiple-processes
https://stackoverflow.com/questions/41915603/python3-sqlite3-begin-immediate-error

Processing Incoming Request Data in Flask

from flask import request

Query Arguments

@app.route('/query-example')
def query_example():
    language = request.args.get('language') #if key doesn't exist, returns None

    return '''<h1>The language value is: {}</h1>'''.format(language)

Form Data

@app.route('/form-example', methods=['GET', 'POST']) #allow both GET and POST requests
def form_example():
    if request.method == 'POST':  #this block is only entered when the form is submitted
        language = request.form.get('language')
        framework = request.form['framework']

        return '''<h1>The language value is: {}</h1>
                  <h1>The framework value is: {}</h1>'''.format(language, framework)

    return '''<form method="POST">
                  Language: <input type="text" name="language"><br>
                  Framework: <input type="text" name="framework"><br>
                  <input type="submit" value="Submit"><br>
              </form>'''

JSON Data

@app.route('/json-example', methods=['POST']) #GET requests will be blocked
def json_example():
    req_data = request.get_json()

    language = req_data['language']
    framework = req_data['framework']
    python_version = req_data['version_info']['python'] #two keys are needed because of the nested object
    example = req_data['examples'][0] #an index is needed because of the array
    boolean_test = req_data['boolean_test']

    return '''
           The language value is: {}
           The framework value is: {}
           The Python version is: {}
           The item at index 0 in the example list is: {}
           The boolean value is: {}'''.format(language, framework, python_version, example, boolean_test)

References
https://www.digitalocean.com/community/tutorials/processing-incoming-request-data-in-flask

Return a JSON response using Flask in Python

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/api/get-json')
def hello():
    return jsonify(hello='world') # Returns HTTP Response with {"hello": "world"}
person = {'name': 'Alice', 'birth-year': 1986}
return jsonify(person)
@app.route("/")
def index():
    return flask.jsonify(response_value_1=1,response_value_2="value")

References
https://riptutorial.com/flask/example/5831/return-a-json-response-from-flask-api
https://www.kite.com/python/answers/how-to-return-a-json-response-using-flask-in-python

Make a PyQT window bring to the front

from win32gui import SetWindowPos
import win32con

SetWindowPos(window.winId(),
             win32con.HWND_TOPMOST, # = always on top. only reliable way to bring it to the front on windows
             0, 0, 0, 0,
             win32con.SWP_NOMOVE | win32con.SWP_NOSIZE | win32con.SWP_SHOWWINDOW)
SetWindowPos(window.winId(),
             win32con.HWND_NOTOPMOST, # disable the always on top, but leave window at its top position
             0, 0, 0, 0,
             win32con.SWP_NOMOVE | win32con.SWP_NOSIZE | win32con.SWP_SHOWWINDOW)
window.raise_()
window.show()
window.activateWindow()

References
https://stackoverflow.com/questions/12118939/how-to-make-a-pyqt4-window-jump-to-the-front

How to send emoji in Python telegram bot

client.send_message(chat, '😉')
client.send_message(chat, '\U0001F609')

If you prefer to use text replacements in your code, install the emoji package:

import emoji
client.send_message(chat, emoji.emojize(':wink:'))

References
https://stackoverflow.com/questions/57549311/python-telegram-telethon-how-to-send-emoji
https://pypi.org/project/emoji/
https://apps.timwhitlock.info/emoji/tables/unicode
https://www.webfx.com/tools/emoji-cheat-sheet/

PyInstaller-built Windows EXE fails with multiprocessing

Enabling --onedir multiprocessing support on Windows (does NOT work with --onefile on Windows, but otherwise safe on all platforms/configurations):

if __name__ == '__main__':
    # On Windows calling this function is necessary.
    multiprocessing.freeze_support()

Enabling --onefile multiprocessing support on Windows (safe on all platforms/configurations, compatible with --onedir):

import multiprocessing.forking
import os
import sys

class _Popen(multiprocessing.forking.Popen):
    def __init__(self, *args, **kw):
        if hasattr(sys, 'frozen'):
            # We have to set original _MEIPASS2 value from sys._MEIPASS
            # to get --onefile mode working.
            os.putenv('_MEIPASS2', sys._MEIPASS)
        try:
            super(_Popen, self).__init__(*args, **kw)
        finally:
            if hasattr(sys, 'frozen'):
                # On some platforms (e.g. AIX) 'os.unsetenv()' is not
                # available. In those cases we cannot delete the variable
                # but only set it to the empty string. The bootloader
                # can handle this case.
                if hasattr(os, 'unsetenv'):
                    os.unsetenv('_MEIPASS2')
                else:
                    os.putenv('_MEIPASS2', '')

class Process(multiprocessing.Process):
    _Popen = _Popen

# ...

if __name__ == '__main__':
    # On Windows calling this function is necessary.
    multiprocessing.freeze_support()

    # Use your new Process class instead of multiprocessing.Process

References
https://stackoverflow.com/questions/24944558/pyinstaller-built-windows-exe-fails-with-multiprocessing