python

Resizing a image is not as easy as you think

I found a very interesting picture:

The size of this image is about 8MB although it’s blurring. Then I use below python code to try to resize it using different interpolation strategy:

import cv2

img = cv2.imread("/Users/robin/Downloads/ou1.png")

for inter, name in [(cv2.INTER_AREA, "area"), (cv2.INTER_CUBIC, "cubic"), (cv2.INTER_LINEAR, "linear")]:
    after = cv2.resize(img, (310, 310), inter)
    cv2.imwrite(f"{name}.bmp", after)

Only cv2.INTER_AREA works well.

Then let me try dart language:

import 'dart:io';
import 'package:image/image.dart' as img;

void main() async {
  // Load the image
  final file = File('/Users/robin/Downloads/ou1.png');
  final bytes = await file.readAsBytes();
  final image = img.decodeImage(bytes);

  if (image == null) {
    print("Failed to load image");
    return;
  }

  // Resize and save images with different methods
  for (var method in ['area', 'cubic', 'linear', 'average']) {
    img.Image resized;
    if (method == 'area' || method == 'linear') {
      // Use `copyResize` for these, `area` will behave similarly to default
      resized = img.copyResize(image, width: 310, height: 310);
    } else if (method == 'cubic') {
      // Simulate cubic interpolation (approximation)
      resized = img.copyResize(image, width: 310, height: 310, interpolation: img.Interpolation.cubic);
    } else if (method == 'average') {
      resized = img.copyResize(image, width: 310, height: 310, interpolation: img.Interpolation.average);
    } else {
      print("Unknown interpolation method: $method");
      continue;
    }

    // Save the resized image
    final output = File('$method.bmp');
    await output.writeAsBytes(img.encodePng(resized));
    print("Saved resized image using $method interpolation as $method.bmp");
  }
}

The result looks also terrible except just one strategy “average”:

So, better use “Interpolation.average” in dart language.

The experience of using Google Cloud’s Text-to-Speech AI

Just using the Python API of Text-to-Speech AI to transform a PDF file to mp3 audio, as the example:

from google.cloud import texttospeech
from PyPDF2 import PdfReader

client = texttospeech.TextToSpeechClient()

reader = PdfReader("xxx.pdf")

voice = texttospeech.VoiceSelectionParams(
    language_code="cmn-CN", name="cmn-CN-Wavenet-B", ssml_gender=texttospeech.SsmlVoiceGender.MALE
)

audio_config = texttospeech.AudioConfig(
    audio_encoding=texttospeech.AudioEncoding.MP3,
    speaking_rate=0.8,
)

text = ""
index = 1
# try first 10 pages
for page in reader.pages[:10]:
    text += page.extract_text()

print(len(text))
synthesis_input = texttospeech.SynthesisInput(text=text)
response = client.synthesize_speech(
    input=synthesis_input, voice=voice, audio_config=audio_config
)

with open("outout.mp3", "wb") as out:
    out.write(response.audio_content)
    print("Written")

Very simple, right? But it just reported an error:

google.api_core.exceptions.InvalidArgument: 400 Either `input.text` or `input.ssml` is longer than the limit of 5000 bytes. This limit is different from quotas. To fix, reduce the byte length of the characters in this request, or consider using the Long Audio API: https://cloud.google.com/text-to-speech/docs/create-audio-text-long-audio-synthesis.

It seems the request is too long. Let’s use the “Long Audio API”:

from google.cloud import texttospeech
from PyPDF2 import PdfReader

client = texttospeech.TextToSpeechLongAudioSynthesizeClient()

reader = PdfReader("xxx.pdf")

voice = texttospeech.VoiceSelectionParams(
    language_code="cmn-CN", name="cmn-CN-Wavenet-B", ssml_gender=texttospeech.SsmlVoiceGender.MALE
)

audio_config = texttospeech.AudioConfig(
    audio_encoding=texttospeech.AudioEncoding.LINEAR16,
    speaking_rate=0.8,
)

text = ""
index = 1
for page in reader.pages[:10]:
    text += page.extract_text()

print(len(text))
synthesis_input = texttospeech.SynthesisInput(text=text)
request = texttospeech.SynthesizeLongAudioRequest(
    parent="projects/robin-00000/locations/us",
    input=synthesis_input, voice=voice, audio_config=audio_config,
    output_gcs_uri="gs://robin_tts/xxx.mp3"
)

operation = client.synthesize_long_audio(request=request)
result = operation.result(timeout=300)
print(result)

It couldn’t work still:

google.api_core.exceptions.InvalidArgument: 400 The long audio API does not support the language zh. Supported languages: en, es.

Okay. It doesn’t support the Chinese language. Then, what should I do if I want to translate a Chinese pdf to mp3? Convert them page by page into 500 mp3 files? This is terrible. Even for the short mp3 it generated, it definitely sounds like a machine, not a human.

Google has the state-of-the-art technology of deep learning but some of their products in the cloud are ridiculously hard to use (such as Vertex AI, and this Text-to-Speech).

After some searching (at least Google search is perfect as before), I found this NaturalReader. Surprisingly, it supports the Chinese language and the voice is as well as a real human. The only problem is it is very expensive for individual users.

Accelerate augmentation of bird audio

audiomentions is a very convenient library for my bird sound classification. As the code below:

from audiomentations import Compose, AddGaussianNoise, AddGaussianSNR, TimeStretch, PitchShift

        self.augment = Compose([
            AddGaussianNoise(min_amplitude=0.005, max_amplitude=0.015, p=poss),
            AddGaussianSNR(min_snr_in_db=5.0, max_snr_in_db=40.0, p=poss),
            TimeStretch(min_rate=0.8, max_rate=1.2, p=poss),
            PitchShift(min_semitones=-2, max_semitones=2, p=poss)
        ])

These four augmentation methods are enough for current training. But the PitchShift method will cost a lot of CPU resources therefore the GPU couldn’t run to full load and the CPU usage jumps to 100%.

Failed to find an audio augmentation library that uses GPU, I started to check the source code of “audiomentions” and noticed that it uses librosa as its implementation:

        try:
            pitch_shifted_samples = librosa.effects.pitch_shift(
                samples, sr=sample_rate, n_steps=self.parameters["num_semitones"]
            )
        except librosa.util.exceptions.ParameterError:

Then the code of “librosa” for “pitch_shift”:

def pitch_shift(
    y: np.ndarray,
    *,
    sr: float,
    n_steps: float,
    bins_per_octave: int = 12,
    res_type: str = "soxr_hq",
    scale: bool = False,
    **kwargs: Any,
) -> np.ndarray:

The default “res_type” for “pitch_shift” is “soxr_hq”. This is a slow resource. After changing “it”res_type” to “linear” in “audiomentions”, the CPU usage jumps back to 50% on my desktop and the GPU ramps up to 100% when training.

—— 2023.07.28 ——

Thanks for the correction from Iver.

After I run this test snippet:

import time
import librosa

sound, sr = librosa.load("./song/background/AirportAnnouncements_1.wav")

for resource in [None, "linear", "soxr_hq", "kaiser_best"]:
    begin = time.time()
    for _ in range(10):
        if resource:
            librosa.effects.pitch_shift(sound, sr=sr, n_steps=1, res_type=resource)
        else:
            librosa.effects.pitch_shift(sound, sr=sr, n_steps=1)
    if resource:
        print(f"{resource} time:", time.time() - begin)
    else:
        print("default time:", time.time() - begin)

and got the result

default time: 8.455572366714478
linear time: 3.3037502765655518
soxr_hq time: 3.3474862575531006
kaiser_best time: 8.467342615127563

Iver is right: the soxr_hq is as fast as linear. And the actual default res_type of librosa which I was using is kaiser_best.

Extract only birds sound from audio

This paper introduced a method to extract only segments with bird sound from an audio file. Since the paper didn’t give any code, I started to write it by myself.

Here is the Python implementation:

import cv2
import time
import torch
import librosa
import soundfile as sf
import numpy as np

from torchlibrosa.stft import LogmelFilterBank, Spectrogram

class CFG:
    n_fft = 2048
    hop_length = 512
    sample_rate = 32000
    n_mels = 64
    fmin = 150
    fmax = 150000

class SignalExtractor:
    def __init__(self):
        self.spectrogram_extractor = Spectrogram(
            n_fft=CFG.n_fft, hop_length=CFG.hop_length, win_length=CFG.n_fft, window="hann",
            center=True, pad_mode="reflect", freeze_parameters=True)
        # Logmel feature extractor
        self.logmel_extractor = LogmelFilterBank(sr=CFG.sample_rate, n_fft=CFG.n_fft,
            n_mels=CFG.n_mels, fmin=CFG.fmin, fmax=CFG.fmax, ref=1.0, amin=1e-10, top_db=None, freeze_parameters=True)
        self.factors = [2.0, 1.8, 1.6, 1.4, 1.2, 1.1]
        self.kernel_size = 15
        self.sn_threshold = 0.2

    def extract(self, input):
        x = torch.from_numpy(input)
        x = x[None, :].float()

        x = self.spectrogram_extractor(x)
        x = self.logmel_extractor(x)

        x = x.squeeze(0).squeeze(0)
        x = x.permute(1, 0).numpy()
        x = x - np.amin(x)

        for factor in self.factors:
            sound, sn_ratio = self._factor_extract(input, x, factor)
            if sn_ratio >= self.sn_threshold:
                break

        return sound, sn_ratio

    def _factor_extract(self, input, x, factor: float):
        rows, cols = x.shape
        row_median = np.median(x, axis=1)
        row_median_matrix = np.tile(row_median, (cols, 1)).T * factor
        col_median = np.median(x, axis=0)
        col_median_matrix = np.tile(col_median, (rows, 1)) * factor

        y = x > row_median_matrix
        z = x > col_median_matrix
        res = np.logical_and(y, z) + np.zeros(x.shape)

        kernel = np.ones((self.kernel_size, self.kernel_size), np.uint8)
        img = cv2.dilate(res, kernel, iterations=1)

        indicator = np.sum(img, axis=0)
        chunk_size = input.shape[0] // indicator.shape[0]
        sounds = []
        for index, chunk in enumerate(indicator):
            if chunk > 0:
                sounds.append(input[index*chunk_size:(index+1)*chunk_size])
        if len(sounds) <= 0:
            return None, 0.0
        sound = np.concatenate(sounds)
        return sound, sound.shape[0]/input.shape[0]

The implementation has some differences from the method in the paper:

I didn’t use erosion since dilation is good enough for picking up the bird-sound segment
three times bigger than median is too strict for most audio files, so I use an array of ratios. When the 2.0 ratio couldn’t pick up any bird sound, the code will automatically try a 1.8 ratio etc.
I used a big kernel (15, 15) for dilation since it works well in my samples

The original sample:

After extracteing only bird sounds:

Using pendulum in Python

pendulum is a prevalent python library in my company. For example, if I want to get the time of the previous Monday, it could be written:

pendulum.today("US/Pacific").previous(pendulum.MONDAY)
# return type "pendulum.datetime.DateTime`

But this will return a class of pendulum.datetime.DateTime. What if I wish Date instead of pendulum.datetime.DataTime? Actually, the pendulum.datetime.DateTime inherited from datetime.datetime so we can use .date():

pendulum.today("US/Pacific").previous(pendulum.MONDAY).date()
# return type "pendulum.datetime.Date`

How to get the number of CPU cores inside a container

We usually use the below Python code to get CPU cores:

from multiprocessing import cpu_count
print("CPU cores:", cpu_count())

But when the snippet running inside a docker container, it will just return the number of CPU cores for the physical machine the container runs on, not the actually --cpus (for docker) or CPU limit (for Kubernetes).

Then, how could we get the CPU cores set by docker argument or Kubernetes configuration for this container?

The only answer I could find is here. And the corresponding Python code written by me is:

def get_cpu_limit():
  with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us") as fp:
    cfs_quota_us = int(fp.read())
  with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us") as fp:
    cfs_period_us = int(fp.read())
  container_cpus = cfs_quota_us // cfs_period_us
  # For physical machine, the `cfs_quota_us` could be '-1'
  cpus = cpu_count() if container_cpus < 1 else container_cpus
  return cpus

A handy Python library

To find an easy way to print the runtime of a function, I found an interesting Python library: funcy.

We can directly use decorator ‘@print_durations()’ like this:

from funcy import print_durations
@print_durations()
def my_func1():
  ...

There are also some other interesting decorators:

@memoize
def ip_to_city(ip):
  ...

Python code for the sequence partition problem (using NumPy)

Imagine we have an array of numbers [9, 8, 7, 1, 2, 3, 4, 5, 6]. What’s the best solution to split it into 3 partitions with the most “average sum”, which means they have minimum differences for their sums. (Remember that we can’t change the order of this array)

For example, we can split this array into three partitions like this [9, 8, 7] [1, 2, 3, 4] [5, 6]. Then the sums of the three partitions are 24, 10, 11. This solution is not the best since the sums have big differences.

The algorithm and source code for this problem is in the book “The Algorithm Design Manual“. And below is the python code of my implementation:

# Find the most average way for partitions
import numpy as np
def maximum_partition(sequence, M, nr_partitions, sum_array):
    for n in range(2, len(sequence) + 1):
        for k in range(2, nr_partitions + 1):
            array = []
            for i in range(1, n + 1):
                select = max(M[i][k - 1], sum_array[n - 1] - sum_array[i - 1])
                array.append(select)
            M[n][k] = min(array)
    return M[len(sequence)][nr_partitions]
def init_matrix(sequence, nr_partitions, M, sum_array):
    for index in range(len(sequence)):
        sum_array.append(sum(sequence[: index + 1]))
    for k in range(1, nr_partitions + 1):
        M[1][k] = sequence[0]
    for n in range(1, len(sequence) + 1):
        M[n][1] = sum(sequence[:n])
if __name__ == "__main__":
    # The sequence and the number of partitions
    sequence = [9, 8, 7, 1, 2, 3, 4, 5, 6]
    partitions = 3
    # init
    M = np.zeros((len(sequence) + 1, partitions + 1), dtype=int)
    sum_array = []
    init_matrix(sequence, partitions, M, sum_array)
    # call the main function
    range_sum_max = maximum_partition(sequence, M, partitions, sum_array)
    print("Sum of the maximum range:", range_sum_max)
    # split the sequence by using maximum sum of one range
    current_sum = 0
    for index in range(len(sequence)):
        if (current_sum + sequence[index]) > range_sum_max:
            print("| ", end="")
            current_sum = 0
        current_sum += sequence[index]
        print(sequence[index], end=" ")
    print("\r")

The code is unbelievable simple (just less than 50 lines). And this is the power and charm of dynamic programming.

Take care of the comma (in Python)

Think about the result of this snippet:

def concat(a, b):
    return a + "_" + b
left = "hello",
right = "world"
print(concat(left, right))

Should be “hello_world”, right?

But the actual result is an error:

Traceback (most recent call last):
  File "test.py", line 7, in <module>
    print(concat(left, right))
  File "test.py", line 2, in concat
    return a + "_" + b
TypeError: can only concatenate tuple (not "str") to tuple

and the error seems hard to understand.

As a matter of fact, the reason for this error is we have just put a comma after “hello” so the value of variable “left” is a tuple (“hello”,) instead of a string “hello”

Even for this shortcode sample, it is hard to find the reason, not mention in the real program to find out the root. For the first time in recent 3 years, I began to miss the strong-type programming language like C…

An error about multiprocessing of Python

Our python program reported errors when running a new dataset:

[77 rows x 4 columns]]'. Reason: 'error("'i' format requires -2147483648 <= number <= 2147483647",)'
multiprocessing.pool.MaybeEncodingError: Error sending result: '[                          id  ... email_send_date
    raise self._value
  File "/usr/local/lib/python3.6/multiprocessing/pool.py", line 644, in get
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/local/lib/python3.6/multiprocessing/pool.py", line 266, in map
    dfs = pool.map(partial(pd.read_parquet, **kwargs), file_list)

Then I found this issue in the python community quickly https://bugs.python.org/issue17560 https://bugs.python.org/issue17560. Seems the reason is the multiprocessing mechanism in python only support 32bit to encode object length. And this problem existed even up to Python-3.8

The solution is just using multithreads instead of multiprocessing

Previous code:

with Pool(processes=n_jobs) as pool:
  pool.map(...)

Solution code:

with ThreadPoolExecutor(max_workers=n_jobs) as pool:
  pool.map(...)

“area”	“cubic”	“linear”

“area”	“cubic”	“linear”	“average”