Typescript で sharp による画像の各ピクセルの色を変更

Typescript で画像処理モジュールの sharp を使って、画像の各ピクセルの色を変更する方法のメモ。

■インストール

npm install sharp
npm i --save @types/sharp

■プログラム
.raw().toBuffer() で画像の各ピクセルのデータを参照できるようにした後で、各ピクセルの色を補色に変更し、変更後のデータで画像を生成してファイルに保存しています。

import * as fs from 'fs';
import sharp from 'sharp';

function change_color(data: any) {
  const size = data.length;

  for (let idx = 0; idx < size; idx += 4) {
    let r = data[idx];
    let g = data[idx+1];
    let b = data[idx+2];

    // 補色に変更
    let max_min = Math.max(r, g, b) + Math.min(r, g, b);
    data[idx] = max_min - r;
    data[idx+1] = max_min - g;
    data[idx+2] = max_min - b;
  }
}

(async () => {
  const img_in_path: string = 'test1.png';
  const img_out_path: string = 'conv1.png';

  const img =
    await await sharp(img_in_path)
      .raw()
      .toBuffer({resolveWithObject: true});

  change_color(img.data);

  await sharp(new Uint8ClampedArray(img.data.buffer),
              {raw: {width: img.info.width,
                     height: img.info.height,
                     channels: img.info.channels}})
    .toFile(img_out_path);
})();

MySQL の Connection Pool

python の MySQL ライブラリでは、コネクションプールにコネクションがなくなると queue の例外が発生するため、例外を発生させずにブロックして待ち続けるか、タイムアウトさせるかを指定できるようにしてみました。

ブロックするには MySQLTimeoutConnectionPool(block=True) を指定します。
また、タイムアウトさせるには MySQLTimeoutConnectionPool(block=True, timeout=秒数) を指定します。

mysql-connector-python のプログラムを流用しています。
https://github.com/mysql/mysql-connector-python/blob/master/lib/mysql/connector/pooling.py

■ライブラリ

# -*- coding:utf-8 -*-
#
# MySQL Timeout Connection Pool
#

import threading
import queue
from mysql.connector.pooling import MySQLConnectionPool
from mysql.connector.pooling import PooledMySQLConnection
from mysql.connector.errors import PoolError


class MySQLTimeoutConnectionPool(MySQLConnectionPool):
    def __init__(self,
                 block=True, timeout=0,
                 pool_size=5, pool_name=None,
                 pool_reset_session=True,
                 **kwargs):
        self.block = block
        self.timeout = timeout
        super().__init__(pool_size=pool_size, pool_name=pool_name,
                         pool_reset_session=pool_reset_session,
                         **kwargs)


    def get_connection(self):
        with threading.RLock():
            try:
                cnx = self._cnx_queue.get(block=self.block,
                                          timeout=self.timeout)
            except queue.Empty as err:
                raise PoolError("Failed getting connection; pool exhausted") from err

            if (
                not cnx.is_connected()
                or self._config_version != cnx.pool_config_version
            ):
                cnx.config(**self._cnx_config)
                try:
                    cnx.reconnect()
                except InterfaceError:
                    self._queue_connection(cnx)
                    raise
                cnx.pool_config_version = self._config_version

            return PooledMySQLConnection(self, cnx)

■テスト用プログラム
コネクションを取得できない場合、5秒でタイムアウトさせます。

# -*- coding:utf-8 -*-

import sys
import time
sys.path.append('../../lib')
from mysql_timeout_connection_pool import MySQLTimeoutConnectionPool

config = {
    'host': '127.0.0.1',
    'port': 3306,
    'database': 'test1',
    'user': 'root',
    'password': None,
    'pool_size': 2,
    'block': True,
    'timeout': 5,
}

def main():
    conn_pool = MySQLTimeoutConnectionPool(
        host=config['host'],
        port=config['port'],
        database=config['database'],
        user=config['user'],
        password=config['password'],
        pool_size=config['pool_size'],
        block=config['block'],
        timeout=config['timeout']
    )

    try:
        conn1 = conn_pool.get_connection()
        print(f"conn1: {time.time()}")
        conn2 = conn_pool.get_connection()
        print(f"conn2: {time.time()}")
        conn3 = conn_pool.get_connection()
        print(f"conn3: {time.time()}")
    except Exception as e:
        print(f"exception: {time.time()}")
        print(e)

    return 0

if __name__ == '__main__':
    res = main()
    exit(res)

■実行結果
2つめのコネクションを取得し、5秒後にタイムアウトしています。

$ python test1.py
conn1: 1672927735.0839279
conn2: 1672927735.0845752
exception: 1672927740.0888727
Failed getting connection; pool exhausted