DEV Community: Talles L

Transcribing mp3 files locally with OpenAI's whisper model

Talles L — Sat, 17 May 2025 00:41:20 +0000

#!/usr/bin/env python3

from whisper import load_model  # pip install openai-whisper

model = load_model('small')
result = model.transcribe('my_audio.mp3', language='pt')

print(result['text'])

Creating a solution from the CLI

Talles L — Sat, 08 Mar 2025 21:49:04 +0000

For folks that are on Windows and Visual Studio (not Code), solution files (.sln) are still pretty useful.

Below follows the one-liner that I use to quickly create a new .csproj, .sln, and fire Visual Studio from the CLI.

Console application:

dotnet new console -n MyConsoleApp --use-program-main; cd MyConsoleApp; dotnet new sln; dotnet sln add MyConsoleApp.csproj; ii MyConsoleApp.sln;

Web API:

dotnet new webapi -n MyApi --use-program-main; cd MyApi; dotnet new sln; dotnet sln add MyApi.csproj; ii MyApi.sln;

Gemini manual function calling

Talles L — Wed, 19 Feb 2025 20:21:02 +0000

from logging import basicConfig, ERROR

from google.genai import Client
from google.genai.types import AutomaticFunctionCallingConfig, Content, FunctionResponse, \
    GenerateContentConfig, Part
from requests import get

basicConfig(level=ERROR)


def get_weather(latitude: str, longitude: str) -> str:
    response = get(f'https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current=temperature_2m')

    data = response.json()
    temperature = data['current']['temperature_2m']

    return str(temperature)


# process the given function call
def process_function_call(function_calls):

    # we can have more than one function call in a single response
    for function_call in function_calls:

        # if we don't know the function, we raise an error
        if function_call.name != 'get_weather':
            raise Exception(f'Unknown function call: {function_call.name}')

        # calling our weather API
        temperature = get_weather(**function_call.args)

        # bulding the object representing the call (this one was hard to figure it out!)
        function_response = FunctionResponse(name='get_weather', response={'result': temperature})
        part = Part(function_response=function_response)
        content = Content(parts=[part], role='function')

        # adding the result of the call to the chat history
        chat_history.append(content)


# Gemini client
client = Client()

# we are keeping track of the chat history ourselves (and not via "session" from google's library)
chat_history = []

response = None

while True:
    # if we got a function call
    if response and response.function_calls:
        process_function_call(response.function_calls)

    # else it's just a regular prompt, as user for input
    else:
        user_input = input('You: ').strip()

        # buliding the object representing the user prompt
        part = Part(text=user_input)
        content = Content(role='user', parts=[part])

        # adding the user prompt to the chat history
        chat_history.append(content)

    # calling Gemini passing the current chat history and registering our weather API (with automatic function calling disabled)
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=chat_history,
        config=GenerateContentConfig(
            tools=[get_weather],
            automatic_function_calling=AutomaticFunctionCallingConfig(disable=True),
        )
    )

    # getting the Gemini response content
    response_content = response.candidates[0].content

    # adding the Gemini response to the chat history, this time we don't need to build the object ourselves
    chat_history.append(response_content)

    # getting the Gemini reply text out of the response object
    response_text = response_content.parts[0].text if response_content.parts else None

    # printing the Gemini response (if any)
    if response_text:
        print('Gemini: ', response_text)

Running it:

You: hey what can you do for me
Gemini:  I can fetch weather information for a specific latitude and longitude. Just let me know the coordinates!

You: whats the weather in 43,-75 right now
Gemini:  The weather is -8.3 degrees.

Using google-genai package at version 1.2.0.

OpenAI tool calling example

Talles L — Tue, 11 Feb 2025 03:57:27 +0000

from json import loads
from signal import signal, SIGINT

from requests import get  # pip install requests
from openai import OpenAI  # pip install openai

# suppressing 'KeyboardInterrupt' message
signal(SIGINT, lambda _, __: exit())


def get_weather(latitude, longitude):
    response = get(f'https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current=temperature_2m')

    data = response.json()
    temperature = data['current']['temperature_2m']

    return str(temperature)


tools = [{
    'type': 'function',
    'function': {
        'name': 'get_weather',
        'description': 'Get current temperature for provided coordinates in celsius.',
        'parameters': {
            'type': 'object',
            'properties': {
                'latitude': {'type': 'number'},
                'longitude': {'type': 'number'}
            },
            'required': ['latitude', 'longitude'],
            'additionalProperties': False
        },
        'strict': True
    }
}]

# OpenAI client and chat history with the first (system) message
client = OpenAI()
messages = [{'role': 'system', 'content': 'You are a weather assistant.'}]

while True:

    # sending message and getting a response back (chat completion)
    completion = client.chat.completions.create(model='gpt-4o-mini', messages=messages, tools=tools)

    # getting the first choice (is it possible to get more than one at a time?)
    choice = completion.choices[0]

    # appending the message to the conversation history
    messages.append(choice.message)

    # switch based on the finish reason
    match choice.finish_reason:

        # stop (weirdly) means we got a message response
        case 'stop':

            # asking for the user for a prompt
            print('\nChatGPT:', choice.message.content)
            prompt = input('\nUser: ').strip()

            # appending the user message to the conversation history
            messages.append({'role': 'user', 'content': prompt})

        # in case we got a tool call
        case 'tool_calls':

            # getting the first tool call (is it possible to get more than one at a time?)
            tool_call = choice.message.tool_calls[0]

            # function name and arguments
            function_name = tool_call.function.name
            arguments = loads(tool_call.function.arguments)

            match function_name:

                # calling the function and appending the result to conversation history
                case 'get_weather':
                    result = get_weather(**arguments)
                    messages.append({'role': 'tool', 'tool_call_id': tool_call.id, 'content': result})

                case unexpected_function:
                    raise Exception(f'Unexpected function call: {unexpected_function}')

        case unexpected_reason:
            raise Exception(f'Unexpected "finish_reason": {unexpected_reason}')

Running it:

ChatGPT: How can I assist you with the weather today?

User: whats the weather in ny right now

ChatGPT: The current temperature in New York is -2.9°C. If you need more information about the weather or forecast, just let me know!

Using openai package at version 1.63.2.

An Alternating Asynchronous Generator in Python

Talles L — Thu, 02 Jan 2025 23:24:30 +0000

from asyncio import run, sleep

class AlternatingGenerator:
    def __init__(self, gen1, gen2):
        self.gen1 = gen1
        self.gen2 = gen2
        self.next = gen1

    def __aiter__(self):
        return self

    async def __anext__(self):

        # if we are done we both generators
        if not self.gen1 and not self.gen2:
            raise StopAsyncIteration

        try:
            # saving the current generator into a variable for later
            current = self.next

            # if the other genrerator is not null, alternating to it for the next iteration
            if self.next is self.gen1 and self.gen2:
                self.next = self.gen2
            elif self.next is self.gen2 and self.gen1:
                self.next = self.gen1

            # calling the generator of the current iteration and returning the next result
            return await anext(current)

        except StopAsyncIteration:

            # unsetting the generator that just finished
            if current is self.gen1:
                self.gen1 = None
            else:
                self.gen2 = None

            # recursive call, trying again with the other generator
            return await self.__anext__()



async def three():
    for i in range(3, 31, 10):
        await sleep(0.25)
        yield i


async def five():
    for i in range(5, 101, 10):
        await sleep(0.25)
        yield i

async def main():
    gen = AlternatingGenerator(three(), five())

    try:  # or just use "async for item in gen:"
        while True:
            print(await anext(gen))
    except StopAsyncIteration:
        pass

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

$ python alternating_generator.py 
3
5
13
15
23
25
35
45
55
65
75
85
95

Parsing command-line arguments in C

Talles L — Sun, 29 Dec 2024 03:06:42 +0000

Parsing on my own:

#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

typedef struct {
    char* filepath;
    bool myflag;
    int myvalue;
} CliArgs;

static inline void error(const char* message) {
    fprintf(stderr, "%s\n", message);
    exit(1);
}

CliArgs parse(int argc, char** argv) {
    CliArgs cli_args = { .filepath = 0, .myflag = false, .myvalue = 0 };

    if (argc == 1)
        error("No argument was passed.");

    for (int i = 1; i < argc; ++i) {
        if (!strcmp(argv[i], "--myflag")) {
            cli_args.myflag = true;
            continue;
        }

        if (!strcmp(argv[i], "--myvalue")) {
            ++i;

            if (i == argc)
                error("No value passed for --myvalue.");

            char* str_end;
            cli_args.myvalue = strtol(argv[i], &str_end, 10);

            if (*str_end)
                error("Invalid value for --myvalue.");

            continue;
        }

        if (cli_args.filepath)
            error("Only a single filepath can be given.");
        else
            cli_args.filepath = argv[i];
    }

    if (!cli_args.filepath)
        error("No filepath was passed.");

    return cli_args;
}

int main(int argc, char** argv) {
    CliArgs cli_args = parse(argc, argv);

    printf("%s was called:\n", argv[0]);

    printf(
        "%s is the filepath, myflag is %s and myvalue is %i.\n",
        cli_args.filepath,
        cli_args.myflag ? "true" : "false",
        cli_args.myvalue
    );

    return 0;
}

Running it:

$ gcc cli_args.c -o cli_args

$ ./cli_args 
No argument was passed.

$ ./cli_args foo.txt
./cli_args was called:
foo.txt is the filepath, myflag is false and myvalue is 0.

$ ./cli_args foo.txt --myvalue
No value passed for --myvalue.

$ ./cli_args foo.txt --myvalue bar
Invalid value for --myvalue.

$ ./cli_args foo.txt --myvalue 123
./cli_args was called:
foo.txt is the filepath, myflag is false and myvalue is 123.

$ ./cli_args foo.txt --myvalue 123 --myflag
./cli_args was called:
foo.txt is the filepath, myflag is true and myvalue is 123.

Quick debugging with gdb

Talles L — Sun, 29 Dec 2024 02:21:08 +0000

Here's my routine to debug quick stuff on the cli:

$ gcc myprogram.c -g -o myprogram

$ gdb -tui myprogram

(gdb) break main

(gdb) run

You will have the program running and stopped at main. After this, here are the commands I use the most:

Command	Short version	Description
`step`	`s`	Step in
`next`	`n`	Step over
`until 123`	`unt 123`	Run until line 123
`info locals`	`i lo`	Prints local variables
`print myvar`	`p myvar`	Prints myvar value once
`display myvar`	`disp myvar`	Displays myvar value on every execution
`undisplay myvar`	`und myvar`	Stops displaying myvar value
`set myvar = 0`	`s myvar = 0`	Sets the value of myvar to 0
`break 456`	`b 456`	Adds a breakpoint on line 456
`info breakpoints`	`i b`	Lists all breakpoints
`delete`	`d`	Delete all breakpoints
`list main`	`l main`	Go to function main
`list foo.c:789`	`l foo.c:789`	Go to file foo.c on line 789
`where`	`whe`	Prints where the execution is at right now
`kill`	`k`	Kills the current execution

Ctrl-l refreshes the screen if needed. focus cmd makes the arrow keys work on the command line, focus src makes it scroll the source code being displayed.

Reading UTF-8 char by char in C

Talles L — Sat, 28 Dec 2024 18:58:06 +0000

Using wchar_t didn't quite worked out in my tests, so handling it on my own:

#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

// https://stackoverflow.com/a/44776334
int8_t utf8_length(char c) {
    // 4-byte character (11110XXX)
    if ((c & 0b11111000) == 0b11110000)
        return 4;

    // 3-byte character (1110XXXX)
    if ((c & 0b11110000) == 0b11100000)
        return 3;

    // 2-byte character (110XXXXX)
    if ((c & 0b11100000) == 0b11000000)
        return 2;

    // 1-byte ASCII character (0XXXXXXX)
    if ((c & 0b10000000) == 0b00000000)
        return 1;

    // Probably a 10XXXXXXX continuation byte
    return -1;
}

void main ()
{

    const char* filepath = "example.txt";

    FILE* file = fopen(filepath, "r");

    if (!file) {
        perror(filepath);
        exit(1);
    }

    char c;

    for(;;) {

        c = getc(file);

        if (c == EOF)
            break;

        putc(c, stdout);

        int8_t length = utf8_length(c);

        while (--length) {
            c = getc(file);
            putc(c, stdout);
        }

        getchar();
    }

    fclose (file);
}

And here's my test file:

Hello, World! 🌍🚀
Hello
¡Hola!
Ça va?
你好
こんにちは
안녕하세요
©®™✓✗
😄😢😎🔥✨
€𐍈𒀭

RabbitMQ Refresher

Talles L — Tue, 24 Dec 2024 02:12:28 +0000

Exchanges & Queues

Messages are published to exchanges. Publishers only need to interact with them to deliver messages.

Messages are delivered to queues. Subscribers only need to interact with them to consume messages.

RabbitMQ binds exchanges to queues, decoupling receivers and consumers.

Message Routing

You have a lot of flexibility on routing messages (from exchanges to queues). The most straightforward one is the direct routing (which routes based on the given routing key), which can act as unicast or multicast:

Another useful one is the fanout (no routing key), which is pretty much a broadcast:

One Connection, Many Commands

Connection:

It’s TCP-based and assumes long-lived connections for efficiency (a new connection is not opened per protocol operation).
By default, RabbitMQ will listen on port 5672 on all available interfaces.

Channel:

A lightweight connection that shares a single TCP connection (multiplexing).
A channel only exists within a connection and never on its own.
It is very common to open a new channel per thread and not share channels between them.
A channel can be closed due to a protocol exception.

Virtual Hosts

Virtual hosts is loosely like a namespace:

Has a name.
Provides logical grouping and separation of resources.
Are created and deleted using rabbitmqctl or the HTTP API (no need to edit a configuration file).
Resource permissions in RabbitMQ are scope per virtual host.

AMQP 0.9.1

Counterintuitively, AMQP 1.0 is quite different from 0.9.1, sharing nothing at the wire level. Many of RabbitMQ’s core ideas, like exchanges and routing keys, only exist in 0.9.1.

Erlang

RabbitMQ is written in Erlang, which fits nicely with the problem it's tackling. Erlang applications are built of very lightweight (Erlang) processes that can be seen as “living” objects, with data encapsulation and message passing, but capable of changing behavior during runtime.

Mnesia vs Khepri

Mnesia is a distributed, soft real-time database management system included as part of Erlang’s standard library (Open Telecom Platform, OTP). It provides features such as atomic transactions, table replication, and a flexible schema, making it highly suitable for use cases demanding high availability and fault tolerance.

Khepri is a new storage backend for RabbitMQ metadata, designed to replace Mnesia. Built on the Raft protocol, it simplifies cluster management by automatically handling network partitions, similar to quorum queues, without requiring additional configuration.

Mnesia is the original storage of RabbitMQ and still the default as time of writing, despite the introduction of Khepri.

Web Interface

Rabbit offers a lovely web interface, that runs by default on, allowing directly on the web browser:

Message publishing.
Message consuming.
Monitoring usage.
Basic server statistics.

If you are running it locally, you can access on http://localhost:15672 and use guest as both username and password.

Permissions

An user has configure, write, and read permissions.
Configure: create, destroy, or alter resources (exchanges and queues).
Write: message publishing.
Read: message consuming.
Permissions are a regex, “^foo” gives permission on resources with names initiated by “foo”.
Permissions are per virtual host.

Some Useful Commands

sudo service rabbitmq-server status
sudo rabbitmq-plugins enable rabbitmq-management
sudo service rabbitmq-server restart
sudo rabbitmqctl add_user ccm-admin hare123
sudo rabbitmqctl set_user_tags ccm-admin administrator
sudo rabbitmqctl add_vhost ccm_dev_vhost
sudo rabbitmqctl set_permissions -p ccm-dev-host ccm-admin ".*" ".*" ".*"

Handling Python event loop shutdown without exceptions

Talles L — Sat, 21 Dec 2024 23:57:47 +0000

#!/usr/bin/env python3

from asyncio import gather, new_event_loop, sleep, Event, set_event_loop
from signal import SIGINT, SIGTERM

def shutdown_signaled():
    print('Shutdown requested.')
    shutdown.set()

async def small_work(shutdown):
    while not shutdown.is_set():
        await sleep(0.5)
        print('Small work is done!')
    print('Exited small work.')

async def big_work(shutdown):
    while not shutdown.is_set():
        await sleep(5)
        print('Big work is done!!!')
    print('Exited big work.')

# Create the shutdown event
shutdown = Event()

# Set up a new event loop and make it the current loop
event_loop = new_event_loop()
set_event_loop(event_loop)

# Set up signal handlers
event_loop.add_signal_handler(SIGINT, shutdown_signaled)
event_loop.add_signal_handler(SIGTERM, shutdown_signaled)

# Combine the tasks
combined_tasks = gather(small_work(shutdown), big_work(shutdown))

# Run the tasks and block until they complete
event_loop.run_until_complete(combined_tasks)

Previewing a .npy file

Talles L — Tue, 17 Dec 2024 03:43:30 +0000

Here's a little script I made to preview some of my (huge) NumPy's .npy files:

#! /usr/bin/env python3

from sys import argv, getsizeof
from numpy import load

data = load(argv[1])

size = getsizeof(data)
size_kb = size / 1024
size_mb = size / (1024 ** 2)

print(f'Shape: {data.shape}')
print(f'Data type: {data.dtype}')
print(f'Size: {data.size}')
print(f'Occupied RAM: {size} bytes ({size_kb:.2f} KB, {size_mb:.2f} MB)')
print()
print('Content preview:')
print(data[:200])
print()

Output example:

Shape: (7468541, 1024)
Data type: float64
Size: 7647785984
Occupied RAM: 61182288000 bytes (59748328.12 KB, 58347.98 MB)

Content preview:
[[-0.07646164 -0.1739306  -0.5888805  ...  0.01423606 -0.9392316
   0.09138256]
 [-0.11803283 -0.27169943 -0.6343062  ... -0.09977917 -1.1889461
   0.4786728 ]
 [ 0.33316666 -0.25544062 -0.06951106 ...  0.21167792 -0.5084344
   0.18797423]
 ...
 [-0.5917888  -0.24519566  0.0250344  ...  0.10975855 -0.77993053
  -0.19755107]
 [-0.07159433 -0.3512457  -0.23779947 ...  0.32987532 -0.7357785
   0.2488086 ]
 [ 0.01552454 -0.5029288  -0.08170918 ... -0.5175352  -0.5811472
  -0.0504258 ]]

Monitoring Queries on PostgreSQL

Talles L — Sun, 01 Dec 2024 20:07:31 +0000

I'm used to Redis, and there is a wonderful command called MONITOR that shows the commands being executed by the server in real-time. Here's how I'm achieving the same with Postgres.

Enabling logging_collector

First, check if logging_collector is on:

show logging_collector;

If it's showing off, you have to enable it by changing Postgres configuration file.

To check the config filepath:

show config_file;

Here's mine:

/etc/postgresql/17/main/postgresql.conf

There, locate this line:

#logging_collector = off

And change to:

logging_collector = on

Restart the server:

sudo systemctl restart postgresql@17-main.service

Enabling log_statement

Now, on the desired database, check if log_statement is enabled:

show log_statement;

If it says none, you can change to all with the following:

alter database <your-database-name> set log_statement = 'all';

This enables logging only for the desired database. This setting persists even in a service restart.

Checking the logs

You can run those to get the log filepath:

show data_directory;
show log_directory;
show log_filename;

Running them together in a single command:

select pg_catalog.current_setting('data_directory') || '/' || pg_catalog.current_setting('log_directory') || '/' || pg_catalog.current_setting('log_filename');

Which gave me:

/var/lib/postgresql/17/main/log/postgresql-%Y-%m-%d_%H%M%S.log

Now just tail -f this file :)