Forem: Viet Hoang

Integrate Copilot feature into your React applications using CopilotKit

Viet Hoang — Fri, 03 May 2024 11:45:52 +0000

Introduction

In the ever-evolving landscape of web development, enhancing user experience is necessary. One innovative way to achieve this is by integrating an AI copilot into your application. Inspired by tools like GitHub Copilot, which assists developers by suggesting code snippets and entire functions in real-time, AI copilots can dramatically streamline user interactions within an app.

What is an AI Copilot?

An AI copilot acts as an intelligent assistant within your application, guiding the user through tasks, offering suggestions, and even performing actions autonomously. They leverage AI technologies to understand and anticipate needs based on user input and context.

Introducing CopilotKit

CopilotKit is an open-source AI platform designed specifically for React applications. It equips developers with the tools needed to seamlessly integrate powerful AI functionalities into their apps.

Capabilities of CopilotKit

CopilotKit allows developers to:

Embed a contextual chat feature: Engage users directly by answering queries and performing tasks based on conversational inputs.
Enable action-driven assistance: Allow the copilot to execute specific actions within the web environment, enhancing interactivity and efficiency.
Offer dynamic content recommendations: By analyzing user inputs, CopilotKit can suggest relevant content, significantly improving the user experience by catering to individual needs and preferences.

By integrate CopilotKit into your React project, you not only make your application smarter but also significantly enhance the overall user interaction, making it more engaging and responsive.

Let’s dive into how you can integrate CopilotKit and transform your React application into a more intelligent platform.

Prerequisites

To fully understand this tutorial, you need to have a basic understanding of React or Next.js.

You also need to create OpenAI API key to use ChatGPT models. Currently, CopilotKit only supports OpenAI API. I hope they will provide more models in the future to provide developers with more flexibility and choice.

Set up and Package Installation

First, we need to install these dependencies into the project

@copilotkit/react-core: CopilotKit frontend package with react hooks for providing app-state and actions to the copilot (AI functionalities)
@copilotkit/react-ui: CopilotKit frontend package for the chatbot sidebar UI
@copilotkit/react-textarea: CopilotKit frontend package for AI-assisted text-editing in the presentation speaker notes.
@copilotkit/backend: CopilotKit backend package

Install these packages:

npm i @copilotkit/react-core @copilotkit/react-ui @copilotkit/react-textarea @copilotkit/backend

Example: Enhancing a Mindmap Builder with CopilotKit

About the example project

The application we will focus on is a Mindmap builder. This tool enables users to create their mindmaps, allowing them to add and manage content within each node. By integrating an AI copilot, we aim to enrich the user experience by streamlining interactions and providing intelligent content suggestions.

Here is the demo of the application’s main flow.

Tech stack of the project:

NextJS
Typescript
Shadcn UI
State Management: Zustand
Reactflow: for drawing mindmap

If you want to know more about this tech stack, you can check out my template project: https://github.com/ngviethoang/next-shadcn-template

Because of the limitations of this post, I will leave the implementation of this application to another post. Let me know if you are interested.

You can check the project source code here: https://github.com/ngviethoang/MindmapCopilot

In the upcoming sections, I will guide you through the integration of the main features of CopilotKit into the project. This will include how to embed a contextual chat feature and facilitate action-driven assistance within the Mindmap application.

Copilot chatbot sidebar

Setup backend

This project uses NextJS App Router, you can check here for more information.

Add the file src/app/api/copilotkit/openai/route.ts

import { CopilotBackend, OpenAIAdapter } from '@copilotkit/backend';

export const runtime = 'edge';

export async function POST(req: Request): Promise<Response> {
  const copilotKit = new CopilotBackend();

  return copilotKit.response(req, new OpenAIAdapter({ model: 'gpt-4-turbo' }));
}

Here I am using gpt-4-turbo, the latest model, you can also change this to another model.

To use ChatGPT models, of course we need to add OpenAI API key. Let’s create .env file and enter your key.

OPENAI_API_KEY=sk-<your-api-key>

You can get your key here.

Setup frontend

First, let’s wrap CopilotKit component outside in the layout component in file src/app/layout.tsx

The url props here in CopilotKit is the backend URL that we set up above.

To use the chatbot sidebar component, let’s add CopilotSidebar component into the page you want. Here I add it directly to the layout component.

export default function RootLayout({
  children,
}: {
  children: React.ReactNode;
}) {
  return (
    <html lang="en">
      <body className={inter.className}>
        <ThemeProvider
          attribute="class"
          defaultTheme="system"
          enableSystem
          disableTransitionOnChange={false}
        >
          <CopilotKit url="/api/copilotkit/openai">
            <CopilotSidebar
              labels={{
                initial:
                  'Welcome to the Mindmap Copilot app! How can I help you?',
              }}
            >
              {children}
            </CopilotSidebar>
          </CopilotKit>
          <Analytics />
        </ThemeProvider>
      </body>
    </html>
  );
}

The initial label is the first message that the chatbot use to welcome user. You can change it as you like.

Provide context to chatbot

To provide more information about the application for the chatbot, we use the useMakeCopilotReadable hook.

useCopilotReadable is a React hook that provides app-state and other information to the Copilot. Optionally, the hook can also handle hierarchical state within your application, passing these parent-child relationships to the Copilot.

In this project I want provide the chatbot with the context of the current nodes of the mindmap that user is creating. So the chatbot can have enough information to answer user and perform actions based on it.

Let’s add this hook into the mindmap component in file src/components/mindmap/index.tsx

useMakeCopilotReadable(JSON.stringify(nodes));

You can also use useMakeCopilotDocumentReadable hook for another document type like Google Docs. Read the docs here.

Demo:

Copilot action

useCopilotAction is a React hook that lets you integrate actionable functions in the Copilot chat. The Copilot can then call these actions to trigger actions in your application, allowing for interactive and dynamic behavior controlled by the Copilot.

Let’s take an example of an action in this app: the copilot will add the nodes that user asks it automatically in any nodes.

Let’s add this hook into the mindmap component in file src/components/mindmap/index.tsx

useCopilotAction({
  name: 'addNodes', // The name of the action
  description: 'Add some nodes to the mindmap', // A description of the action. This is used to instruct the Copilot on how to use the action.
  parameters: [
    {
      name: 'nodes', // The name of the parameter
      type: 'object[]', // The type of the argument.
      description: 'The nodes to add', // A description of the argument. This is used to instruct the Copilot on what this argument is used for.
      attributes: [  // define the attributes of the complex object
        {
          name: 'label',
          type: 'string',
          description: 'The name of the node',
        },
        {
          name: 'parentId',
          type: 'string',
          description: 'The id of the parent node based on provided nodes',
        },
      ],
    },
  ],
  handler: (props) => { // The handler of the action
    const { nodes } = props;
    addNodesFromCopilot(nodes);
  },
});

For handler function, we will pass the nodes props from ChatGPT response to a predefined function. The function addNodesFromCopilot is an Zustand action that add new nodes to the exact position in the mindmap.

You can use this method to add more actions in your app by letting the copilot to decide the props and when to call the funtion.

Demo:

Copilot textarea

Lastly, we want the copilot to add recommendation content to the textarea. To do this, we can use the CopilotTextarea component from the CopilotKit.

In this project, I created a custom component so it can be reused in multiple pages.

Check this file src/components/common/custom-copilot-textarea.tsx

import { CopilotTextarea } from '@copilotkit/react-textarea';
import { useRef, useState } from 'react';
import { cn } from '@/lib/utils';

interface CustomCopilotTextareaProps {
  placeholder: string;
  textareaPurpose: string;
  maxTokens?: number;
  defaultValue?: string;
  debounceDuration?: number;
  onDebouncedUpdate?: (value: string) => void;
  className?: string;
}

export function CustomCopilotTextarea({
  placeholder,
  textareaPurpose,
  maxTokens,
  defaultValue,
  debounceDuration,
  onDebouncedUpdate,
  className,
}: CustomCopilotTextareaProps) {
  const [text, setText] = useState(defaultValue || '');

  // debounce the update
  let timeout = useRef<any>(null);

  const debouncedUpdate = (value: string) => {
    timeout.current && clearTimeout(timeout.current);
    timeout.current = setTimeout(() => {
      onDebouncedUpdate?.(value);
    }, debounceDuration || 750);
  };

  // update the value
  const handleValueChange = (value: string) => {
    setText(value);
    debouncedUpdate(value);
  };

  return (
    <CopilotTextarea
      className={cn('w-full h-full p-4 overflow-hidden', className)}
      value={text}
      onValueChange={handleValueChange}
      placeholder={placeholder}
      autosuggestionsConfig={{ // ChatGPT config
        textareaPurpose,
        chatApiConfigs: {
          suggestionsApiConfig: {
            forwardedParams: {
              max_tokens: maxTokens || 50,
              stop: ['.', '?', '!'],
            },
          },
        },
      }}
    />
  );
}

In file src/components/mindmap/node-detail/index.tsx , we add this component inside.

<CustomCopilotTextarea
  placeholder="Enter content here..."
  textareaPurpose={getNodeContentPurpose}
  defaultValue={selectedNode.data.content || ''}
  onDebouncedUpdate={(val) => {
    setData(
      nodes.map((node) => {
        if (node.id === selectedNode.id) {
          return {
            ...node,
            data: {
              ...node.data,
              content: val,
            },
          };
        }
        return node;
      }),
      edges
    );
  }}
/>

We provide the context to this textarea component by this function.

const getNodeContentPurpose = useMemo(() => {
  if (!selectedNode) {
    return '';
  }
  const parents = [];
  let curNode = selectedNode;
  while (curNode) {
    const parent = nodes.find((n) => n.id === curNode.parentNode);
    if (!parent) {
      break;
    }
    parents.push(parent);
    curNode = parent;
  }
  const parentsString = parents.map((node) => node.data.label).join(', ');
  return `Content of the node: ${selectedNode?.data.label} (${parentsString})`;
}, [selectedNode, nodes]);

Demo:

Additional features from CopilotKit

Some features are not included in this example project, however, I recommend you give it a try and see if it fits your needs.

CopilotTask: This class is used to execute one-off tasks, for example on button press. It can use the context available via useCopilotReadable and the actions provided by useCopilotAction, or you can provided your own context and actions.
In-App Agents (via LangChain and LangGraph): CopilotKit will route relevant state as input to the standalone skill chains (state can come from the frontend, backend, 3rd party integrations, or from the user). When the chain returns, the Copilot Engine funnels its output to in-app interaction as needed.

Conclusion

CopilotKit is an incredible tool that allows you to add AI Copilots to your products within minutes. Whether you're interested in AI chatbots and assistants or automating complex tasks, CopilotKit makes it easy.

You can find the source code for this tutorial on GitHub: https://github.com/ngviethoang/MindmapCopilot

Thank you for reading :)

Build a Telegram voice chatbot using ChatGPT API and Whisper

Viet Hoang — Thu, 30 Mar 2023 13:54:03 +0000

In this article, I will provide you with a step-by-step guide on how to create your own voice chatbot in Telegram. With this, you will be able to engage in conversations with your chatbot in a way that is natural and intuitive.

You can chat with a Telegram bot or send it a voice file, and it will send a response back along with a voice file as a reply. The conversation can continue until you choose to reset it.

You can check the source code here.

https://github.com/ngviethoang/telegram-voice-chatbot

This guide requires you to deploy on your own server with a public IP so you can set up webhooks for your Telegram bot.

Also, this is my personal bot built for both Messenger and Telegram, you can check it here.

https://github.com/ngviethoang/ai-chatbot

It also has Dall-E 2 integrated, and other models in Replicate. If you are curious about this, I will write another blog to talk about it.

Set up project

We will use Bottender - a framework for writing Telegram bot faster. It also supports Session for us to store past conversation messages and other data, so it’s more convenient to build a chatbot with conversational memory.

To create a project, run this command:

npx create-bottender-app telegram-bot

In this step, select Telegram platform.

You can check their documentation here for more details. https://bottender.js.org/docs/

We will need to add Express server for serving static files also. And add Typescript so it’s easier to develop and maintain later.

npm install body-parser express
npm install typescript ts-node nodemon --save-dev
// or yarn
yarn add body-parser express
yarn add typescript ts-node nodemon --dev

Update scripts in package.json file

{
    "scripts": {
        "build": "tsc",
    "dev": "nodemon --exec ts-node src/server.ts",
    "lint": "eslint . --ext=js,ts",
    "start": "tsc && node dist/server.js",
    "test": "jest"
  },
}

Add tsconfig.json file

{
  "include": ["src/**/*"],
  "exclude": ["**/__tests__", "**/*.{spec,test}.ts"],
  "compilerOptions": {
    "target": "es2016",
    "lib": ["es2017", "es2018", "es2019", "es2020", "esnext.asynciterable"],
    "module": "commonjs",
    "skipLibCheck": true,
    "moduleResolution": "node",
    "esModuleInterop": true,
    "strict": true,
    "forceConsistentCasingInFileNames": true,
    "resolveJsonModule": true,
    "isolatedModules": true,
    "rootDir": "./src",
    "outDir": "./dist",
    "types": ["node", "jest"]
  }
}

Install NPM packages

In this project, we will use OpenAI APIs like model gpt-3.5-turbo for chat completion and Whisper to transcribe text from audio.

With generating voice from text, I will use Azure service, so we will install package microsoft-cognitiveservices-speech-sdk. You guys can use any other services like Google, Amazon,… to serve this purpose.

Also install other packages for helper function: axios util uuid

npm install openai gpt-3-encoder microsoft-cognitiveservices-speech-sdk axios util uuid
npm install @types/uuid --save-dev
// Or use yarn
yarn add openai gpt-3-encoder microsoft-cognitiveservices-speech-sdk axios util uuid
yarn add @types/uuid --dev

Telegram setup

Edit file bottender.config.js with Telegram channel enabled

module.exports = {
  channels: {
    telegram: {
      enabled: true,
      path: '/webhooks/telegram',
      accessToken: process.env.TELEGRAM_ACCESS_TOKEN,
    },
  },
};

Make sure to set the channels.telegram.enabled field to true.

Create a bot and generate access token

You can get a Telegram bot account and a bot token by sending the /newbot command to @BotFather on Telegram.

After you get your Telegram Bot Token, paste the value into the TELEGRAM_ACCESS_TOKEN field in your .env file:

TELEGRAM_ACCESS_TOKEN=<Your Telegram Bot Token>

Set up commands for bot

Run /setcommands in Botfather to create commands for our bot.

new - Clear old conversation and create a new one
voice - Set up voice for bot to speak
language - Set up whisper language

Set up Express server

Change the code in file index.js in the root directory with this code.

/* eslint-disable import/no-unresolved */
module.exports = require('./dist').default;

Create a server.ts file in src directory and copy this code into it.

import bodyParser from 'body-parser';
import express from 'express';
import { bottender } from 'bottender';

const app = bottender({
  dev: process.env.NODE_ENV !== 'production',
});

const port = Number(process.env.PORT) || 5000;

const handle = app.getRequestHandler();

app.prepare().then(() => {
  const server = express();

  server.use(
    bodyParser.json({
      verify: (req, _, buf) => {
        (req as any).rawBody = buf.toString();
      },
    })
  );

    server.use('/static', express.static('static'));

  server.get('/api', (req, res) => {
    res.json({ ok: true });
  });

  server.all('*', (req, res) => {
    return handle(req, res);
  });

  server.listen(port, () => {
    console.log(`> Ready on http://localhost:${port}`);
  });
});

As you can see, I have a static directory to store all static files. We will use this directory to store the voice files that we generate in the next steps.

Let’s create static directory in the root directory and voices directory inside it.

mkdir static
mkdir static/voices

Handling Telegram events

There are 3 events that we need to handle from Telegram

Commands
Text messages
Voice messages

We will go over each of these events in detail.

Delete old files index.js and index.test.js. Create a index.ts file inside src directory.

Let’s use router to route different events to each handler.

import { Action, TelegramContext } from 'bottender';
import { router, text } from 'bottender/router';

export default async function App(
  context: TelegramContext
): Promise<Action<any> | void> {
  if (context.event.voice) {
    return HandleVoice;
  }
  return router([
    text(/^[/.](?<command>\w+)(?:\s(?<content>.+))?/i, HandleCommand),
    text('*', HandleText),
  ])
};

Handling text messages

First, we will handle the user’s messages by sending them to ChatGPT API, send the response to the user then save these messages so we can continue this conversation.

async function HandleText(context: TelegramContext) {
  await context.sendChatAction(ChatAction.Typing);
  let { text, replyToMessage } = context.event;
    // Add reply message to text content
  const { text: replyText } = replyToMessage || {}
  if (replyText) {
    text += `\n${replyText}`
  }

  await handleChat(context, text)
}

Next, let’s write a function to handle chat completion with ChatGPT API.

const configuration = new Configuration({
  apiKey: process.env.OPENAI_API_KEY,
});
const openai = new OpenAIApi(configuration);

export const createCompletion = async (messages: ChatCompletionRequestMessage[], max_tokens?: number, temperature?: number) => {
  const response = await openai.createChatCompletion({
    model: "gpt-3.5-turbo",
    messages,
    max_tokens,
    temperature,
  });
  return response.data.choices;
};

export const createCompletionFromConversation = async (
  context: TelegramContext,
  messages: ChatCompletionRequestMessage[]) => {
  try {
    // limit response to avoid message length limit, you can change this if you want
    const response_max_tokens = 500
    const GPT3_MAX_TOKENS = 4096
    const max_tokens = Math.min(getTokens(messages) + response_max_tokens, GPT3_MAX_TOKENS)

    const response = await createCompletion(messages, max_tokens);
    return response[0].message?.content;
  } catch (e) {
    return null;
  }
};

We need to add OPENAI_API_KEY variable to our .env file. You can get your API key from here.

https://platform.openai.com/account/api-keys

OPENAI_API_KEY=<Your OpenAI API key>

You can also set the system role message with your own prompt. In this way, your bot will have its own character and can serve your own purposes, such as a personal trainer, advisor, or any character from movies, novels...

Now, we will send a response back to user as a message in Markdown format.

We also want this conversation to keep going when user sends a new message. So we will save these messages in database. Bottender supports this through Session state. You can check about it here.

https://bottender.js.org/docs/the-basics-session

export const handleChat = async (context: TelegramContext, text: string) => {
  const response = await createCompletionFromConversation(context, [
    ...context.state.context as any,
    { role: 'user', content: text },
  ]);
  if (!response) {
    await context.sendText(
      'Sorry! Please try again`'
    );
    return;
  }
  let content = response.trim()

  await context.sendMessage(content, { parseMode: ParseMode.Markdown });
  await handleTextToSpeech(context, content, getAzureVoiceName(context))

    // save current conversation in session
  context.setState({
    ...context.state,
    context: [
      ...context.state.context as any,
      { role: 'user', content: text },
      { role: 'assistant', content },
    ],
  });
}

You can also set up different session driver by edit the session.driver in the bottender.config.js file.

// bottender.config.js

module.exports = {
  session: {
    driver: 'memory',
    stores: {
      memory: {
        maxSize: 500,
      },
      file: {
        dirname: '.sessions',
      },
      redis: {
        port: 6379,
        host: '127.0.0.1',
        password: 'auth',
        db: 0,
      },
      mongo: {
        url: 'mongodb://localhost:27017',
        collectionName: 'sessions',
      },
    },
  },
};

Send bot’s response as voice

Next, we want to convert this response to voice file in Telegram and send to the user like the bot is talking to them. To do this, I will use Azure Speech Service.

You can set up Azure service by following the documentation here.

Text-to-speech quickstart - Speech service - Azure Cognitive Services | Microsoft Learn

After creating Speech resource, let’s set the environment variables in .env file.

AZURE_SPEECH_KEY=
AZURE_SPEECH_REGION=

It will convert the message from the bot to an audio file in ogg format.

import { SpeechConfig, AudioConfig, SpeechSynthesizer, ResultReason } from 'microsoft-cognitiveservices-speech-sdk'

export const textToSpeech = async (text: string, outputFile: string, voiceName?: string) => {
  return new Promise((resolve, reject) => {
    // This example requires environment variables named "SPEECH_KEY" and "SPEECH_REGION"
    const speechConfig = SpeechConfig.fromSubscription(process.env.AZURE_SPEECH_KEY || '', process.env.AZURE_SPEECH_REGION || '');
    const audioConfig = AudioConfig.fromAudioFileOutput(outputFile);

    // The language of the voice that speaks.
    speechConfig.speechSynthesisVoiceName = voiceName || "en-US-JennyNeural";

    // Create the speech synthesizer.
    const synthesizer = new SpeechSynthesizer(speechConfig, audioConfig);

    synthesizer?.speakTextAsync(text,
      function (result) {
        if (result.reason === ResultReason.SynthesizingAudioCompleted) {
          // console.log("synthesis finished.");
        } else {
          console.error("Speech synthesis canceled, " + result.errorDetails +
            "\nDid you set the speech resource key and region values?");
        }
        synthesizer?.close();
        resolve(result);
      },
      function (err) {
        console.trace("err - " + err);
        synthesizer?.close();
        reject(err);
      });
  });
}

export const handleTextToSpeech = async (context: TelegramContext, message: string, voiceName?: string) => {
  try {
    await context.sendChatAction(ChatAction.Typing);

    // set random filename
    const fileId = uuidv4().replaceAll('-', '')
    const outputDir = `static/voices`
    const outputFile = `${outputDir}/voice_${fileId}.ogg`
    const encodedOutputFile = `${outputDir}/voice_${fileId}_encoded.ogg`

    const result = await textToSpeech(
      message || '',
      outputFile,
      voiceName || getAzureVoiceName(context)
    )
    await encodeOggWithOpus(outputFile, encodedOutputFile)

    const voiceUrl = `${process.env.PROD_API_URL}/${encodedOutputFile}`

    await context.sendVoice(voiceUrl)
  } catch (err) {
    console.trace("err - " + err);
  }
}

In order to send this audio file as a voice in Telegram, we must do a small step to encode this ogg file with opus. The detail in here. I figured one way to do this is by ffmpeg package.

Let’s install this package on our machine first. You can check how to install it here.

In Windows, run this command

choco install ffmpeg

In Linux, run this command

sudo apt install ffmpeg

Next, we will run this command in our JS code to convert ogg file to an encoded file.

import { exec } from 'child_process';
import { promisify } from 'util';

const asyncExec = promisify(exec);

export const encodeOggWithOpus = async (inputFile: string, outputFile: string) => {
  try {
    const { stdout, stderr } = await asyncExec(`ffmpeg -loglevel error -i ${inputFile} -c:a libopus -b:a 96K ${outputFile}`);
    // console.log(stdout);

    if (stderr) {
      console.error(stderr);
    }
  } catch (err) {
    console.error(err);
  }
}

Great, after converting this new file, we will send it to the user.

One thing to notice is that I send this file as an URL, so we will store these files in static directory we created earlier. You will need to set the full URL, so remember to insert your domain you use to run this bot.

const voiceUrl = `${process.env.PROD_API_URL}/${encodedOutputFile}`

await context.sendVoice(voiceUrl)

Set environment varible PROD_API_URL in .env file with your domain like: https://example.com

PROD_API_URL=<your api url>

Handling commands

Handling all commands by this code below.

async function HandleCommand(
  context: TelegramContext,
  {
    match: {
      groups: { command, content },
    },
  }: any
) {
  switch (command.toLowerCase()) {
    case 'new':
      await clearServiceData(context);
      break;
    case 'voice':
      await setAzureVoiceName(context, content)
      break;
    case 'language':
      await setWhisperLang(context, content)
      break;
    default:
      await context.sendText('Sorry! Command not found.');
      break;
  }
}

With /new command, we will simply clear conversation’s data from state.

export const clearServiceData = async (context: TelegramContext) => {
  context.setState({
    ...context.state,
    context: [],
  });
  await context.sendText('New conversation.');
};

With /voice command, we will save this option to settings state.

const getSettings = (context: TelegramContext): any => {
  return context.state.settings || {}
}

export const setSettings = async (context: TelegramContext, key: string, value: string) => {
  let newValue: any = value
  if (value === 'true') {
    newValue = true
  } else if (value === 'false') {
    newValue = false
  }
  context.setState({
    ...context.state,
    settings: {
      ...getSettings(context),
      [key]: newValue,
    },
  })
}

export const setAzureVoiceName = async (context: TelegramContext, voiceName: string) => {
  await setSettings(context, 'azureVoiceName', voiceName)
}

You can check available voices support here:

Language support - Speech service - Azure Cognitive Services | Microsoft Learn

With /language command, we will do as same as /voice command. We will use this to set the Whisper API’s language parameter.

export const setWhisperLang = async (context: TelegramContext, language: string) => {
  await setSettings(context, 'whisperLang', language)
}

Handling voice messages

When user sends voice file to our bot, we need to transcribe this file to text. In order to do this, we will use Whisper API to transcribe.

Handling voice event by this code below.

async function HandleVoice(context: TelegramContext) {
  await handleAudioForChat(context)
}

export const handleAudioForChat = async (context: TelegramContext) => {
  let transcription: any
  const fileUrl = await getFileUrl(context.event.voice.fileId)
  if (fileUrl) {
    transcription = await getTranscription(context, fileUrl)
  }
  if (!transcription) {
    await context.sendText(`Error getting transcription!`);
    return
  }

  await context.sendMessage(`_${transcription}_`, { parseMode: ParseMode.Markdown });

  await context.sendChatAction(ChatAction.Typing);
  await handleChat(context, transcription)
}

When we receive voice events from webhooks, we only receive the file id. So, in the next step, we will use Telegram API to get the full path of this voice file.

import axios from "axios"

export const getFileUrl = async (file_id: string) => {
  try {
    const response = await axios({
      method: 'GET',
      url: `https://api.telegram.org/bot${process.env.TELEGRAM_ACCESS_TOKEN}/getFile`,
      params: {
        file_id
      }
    })
    if (response.status !== 200) {
      console.error(response.data);
      return null;
    }

    const filePath = response.data.result.file_path;
    return `https://api.telegram.org/file/bot${process.env.TELEGRAM_ACCESS_TOKEN}/${filePath}`
  } catch (e) {
    console.error(e);
    return null;
  }
}

After receiving the file path, we will download it to static directory with oga format. However, in order to use whisper API, the audio file must be in mp3 format. So we need to convert it to mp3 by this command, also using ffmpeg.

const asyncExec = promisify(exec);

export const convertOggToMp3 = async (inputFile: string, outputFile: string) => {
  try {
    const { stdout, stderr } = await asyncExec(`ffmpeg -loglevel error -i ${inputFile} -c:a libmp3lame -q:a 2 ${outputFile}`);
    // console.log(stdout);

    if (stderr) {
      console.error(stderr);
    }
  } catch (err) {
    console.error(err);
  }
}

Now, let’s send this mp3 file to Whisper API to get the transcription.

const downloadsPath = './static/voices';

export const getTranscription = async (context: TelegramContext, url: string, language?: string) => {
  try {
    let filePath = await downloadFile(url, downloadsPath);
    if (filePath.endsWith('.oga')) {
      const newFilePath = filePath.replace('.oga', '.mp3')
      await convertOggToMp3(filePath, newFilePath)
      filePath = newFilePath
    }
    const response = await openai.createTranscription(
      fs.createReadStream(filePath) as any,
      'whisper-1',
      undefined, undefined, undefined,
      language,
    );
    return response.data.text
  } catch (e) {
    return null;
  }
}

After we get the transcription, our job is similar to the previous step, we will use the function handleChat like before to handle user’s message and send ChatGPT’s reply back to user.

export const handleAudioForChat = async (context: TelegramContext) => {
  let transcription: any
  const fileUrl = await getFileUrl(context.event.voice.fileId)
  if (fileUrl) {
    transcription = await getTranscription(context, fileUrl)
  }
  if (!transcription) {
    await context.sendText(`Error getting transcription!`);
    return
  }

  await context.sendMessage(`_${transcription}_`, { parseMode: ParseMode.Markdown });

  await context.sendChatAction(ChatAction.Typing);
  await handleChat(context, transcription)
}

You can see our bot also converts its reply to speech and responds back to the user.

You can also have a settings option here to disable text message responses and only send back voice to the user.

Deployment

Run Bottender on your server by the following command:

npm start
# or use yarn
yarn start

Set Up Webhook for Production

Run this command to set up Webhook for Telegram bot, supposed your URL to be https://example.com/webhooks/telegram

npx bottender telegram webhook set -w https://example.com/webhooks/telegram

Now you are ready to talk to your bot.

Send voice file and wait for the bot to respond.

Conclusion

I hope with this guide, you can build your own talking chatbot and have fun talking with it.

Hope you like it!

Build a writing assistant application using ChatGPT API and NextJS

Viet Hoang — Fri, 10 Feb 2023 04:19:09 +0000

Without a doubt, everyone is familiar with ChatGPT, a recently developed AI bot that can converse and answer questions about nearly any topic. Despite the ongoing debates about the accuracy of its data, we cannot deny the immense potential of this tool and AI in boosting human productivity in many fields, such as programming and marketing.

In this article, I will use its API to write a simple application to help users learn English, specifically to optimize IELTS Writing and Speaking essay.

Of course, a lot of code in this application was written by ChatGPT 😃

Because OpenAI has not yet opened its ChatGPT API to the public, we will use the Text Completion API to generate text similar to ChatGPT.

You can refer to it here.

OpenAI API

The features of this application include:

From essay types: IELTS Writing task 2 and user-inputted topics, the application provides suggestions and creates sample essays.
Edit errors, suggest sentences, explain the meaning of words, ... based on the text the user has entered and the essay topic.

You can check the source code of the project here.

https://github.com/ngviethoang/ai-writing-assistant

Demo application.

Writing Assistant

Prerequisite

Create a NextJS project

yarn create next-app --typescript

Install the dependencies: OpenAI client, ChakraUI (UI framework)

yarn add @chakra-ui/react @emotion/react @emotion/styled framer-motion 
yarn add openai

Register OpenAI API key

Create API Secret Key

Create a .env file in the project and save the secret key.

OPENAI_API_KEY=[INSERT YOUR KEY]

Add this .env file to the .gitignore file to avoid exposing keys when committing code.

Building prompt to interact with the API

To communicate with the Text Completion API, we need to use prompts. This is an important step to get the desired output. According to the terminology in NLP, it is called prompt engineering.

Here is an example of a prompt to create a template outline for an IELTS Writing task essay based on the given topic.

Act as an IELTS test taker with a band score of 8.0. Write an essay outline in response to the following IELTS Writing Task 2 question: [insert IELTS Writing Task 2 question]

Here we can define the parameters that can be passed up from the UI:

actor: an IELTS test taker with a band score of 8.0
question: IELTS Writing Task 2 question
content: The text entered by the user.

Create a getPrompt function to query the API based on the parameters actor, question, content.

const getPrompt = (topicType: string, promptType: string, topic: string, content: string) => {
  let actor, questionType
  switch (topicType) {
    case 'IELTS Writing':
      questionType = 'IELTS Writing Task 2'
      actor = 'an IELTS test taker with a band score of 8.0'
      break
    case 'IELTS Speaking':
      questionType = 'IELTS Speaking'
      actor = 'an IELTS test taker with a band score of 8.0'
      break
    default:
      questionType = ''
      actor = 'a person'
      break
  }

  switch (promptType) {
        case 'outline':
      return `Act as ${actor}. Write an essay outline in response to the following ${questionType} question: ${topic}`
    case 'support_arguments':
      return `Act as ${actor}. Given the following ${questionType} question, generate 3 arguments to support the statement: ${topic}`
    case 'oppose_arguments':
      return `Act as ${actor}. Given the following ${questionType} question, generate 3 arguments to oppose the statement: ${topic}`
    case 'sample_answer':
      return `Act as ${actor}. Write an essay in response to the following ${questionType} question with at least 250 words: ${topic}`

    case 'summarize':
      return `Act as a summarizer and summarize this essay: 
${content}`
        // ...

    default:
      return ''
  }
}

Creating an API Handler in NextJS

To create an API handler to process query results from Text Completion, create an API route in the pages/api/prompt.ts directory.

import type { NextApiRequest, NextApiResponse } from 'next'
import { Configuration, OpenAIApi } from 'openai';

const configuration = new Configuration({
  apiKey: process.env.OPENAI_API_KEY,
});
const openai = new OpenAIApi(configuration);

const isEmpty = (str: string) => !str.trim().length

export default async function handler(
  req: NextApiRequest,
  res: NextApiResponse<any>
) {
  if (!configuration.apiKey) {
    res.status(500).json({
      error: {
        message:
          'OpenAI API key not configured, please follow instructions in README.md',
      },
    });
    return;
  }

  const question = req.body.question || '';
  const topicType = req.body.topicType || '';
  const promptType = req.body.promptType || '';
  const content = req.body.content || '';

  if (isEmpty(question) || isEmpty(topicType) || isEmpty(promptType)) {
    res.status(400).json({
      error: {
        message: 'Invalid args',
      },
    });
    return;
  }

  const prompt = getPrompt(topicType, promptType, question, content)
  if (isEmpty(prompt)) {
    res.status(400).json({
      error: {
        message: 'Invalid prompt',
      },
    });
    return;
  }

  try {
    const completion = await openai.createCompletion({
      model: 'text-davinci-003',
      prompt,
      temperature: 0.5,
      max_tokens: 550,
    });
    res.status(200).json({ result: completion.data.choices[0].text });
  } catch (error: any) {
    if (error.response) {
      console.error(error.response.status, error.response.data);
      res.status(error.response.status).json(error.response.data);
    } else {
      console.error(`Error with OpenAI API request: ${error.message}`);
      res.status(500).json({
        error: {
          message: 'An error occurred during your request.',
        },
      });
    }
  }
}

The parameters in the Text Completion API are used.

model: Use the latest and most capable model, text-davinci-003, among the GPT-3 models.
prompt: The query built in the previous step.
temperature: Determine the stability of the results, the higher the temperature the more diverse the model results.
max_tokens: The maximum number of tokens returned, the number of tokens returned per prompt can be limited to reduce costs.

Frontend code

Next is the frontend part of the application, we will write the basic components such as

Text editor to input questions, and content of the article.
The buttons are used to call the corresponding API for functions such as creating an article outline, creating a sample article, correcting grammar, commenting, etc.
Display component for returning results from the API.

Create components and layout for the page using ChakraUI.

import { Box, Button, Heading, HStack, Select, Spinner, Text, Textarea, Tooltip, useToast, VStack } from '@chakra-ui/react';
import { useState } from 'react';

const topicTypes = ['IELTS Writing', 'IELTS Speaking'];

const Writing = () => {
  const [topicType, setTopicType] = useState(topicTypes[0]);
  const [question, setQuestion] = useState('');
  const [content, setContent] = useState('');
  const [selectedContent, setSelectedContent] = useState('');

  return (
    <div style={{ position: 'relative' }}>
      <VStack spacing={5} padding={5}>
        <VStack w={'100%'} spacing={2} alignItems="flex-start">
          <HStack alignItems="flex-start" w="100%" gap={2}>
            <Text>AI Type: </Text>
            <Select
              size={'sm'}
              w={40}
              value={topicType}
              onChange={(e) => setTopicType(e.target.value)}
            >
              {topicTypes.map((type) => (
                <option key={type} value={type}>
                  {type}
                </option>
              ))}
            </Select>
          </HStack>
          <HStack alignItems="flex-start" w="100%" gap={2}>
            <Text>Question: </Text>
            <Textarea
              value={question}
              onChange={(e) => setQuestion(e.target.value)}
            />
          </HStack>
        </VStack>

        <HStack spacing={5} alignItems="flex-start" w="100%">
          <VStack w="100%">
            <Textarea
              rows={20}
              value={content}
              onChange={(e) => setContent(e.target.value)}
              onSelect={(e: any) => {
                // Save the selected text to get API suggestions for these words.
                e.preventDefault();
                const { selectionStart, selectionEnd }: any = e.target;
                const selectedText = content.slice(selectionStart, selectionEnd);
                setSelectedContent(selectedText);
              }}
            />
          </VStack>
          {/* Render buttons and suggest results. */}
          <VStack alignItems="flex-start" w="100%"></VStack>
        </HStack>
      </VStack>
    </div>
  );
};

export default Writing;

Render buttons to generate prompts and suggestion results from the API.

const generateButtons = [
  { name: 'Outline', promptType: 'outline', tooltip: 'Write an essay outline' },
  {
    name: 'Supportive arguments',
    promptType: 'support_arguments',
    tooltip: 'generate 3 arguments to support the statement',
  },
  {
    name: 'Opposite arguments',
    promptType: 'oppose_arguments',
    tooltip: 'generate 3 arguments to oppose the statement',
  },
  // ... full list button in source code
];

const vocabButtons = [
  {
    name: 'Dictionary',
    promptType: 'dictionary',
    tooltip:
      'Explain the meaning of the word and give me an example of how to use it in real life',
  },
  { name: 'Synonyms', promptType: 'synonyms', tooltip: 'Give me 5 synonyms' },
  { name: 'Antonyms', promptType: 'antonyms', tooltip: 'Give me 5 antonyms' },
];

const [result, setResult] = useState({ title: '', content: '' });

const renderButtons = (buttons: any[], color: string, content: string, isDisabled: boolean) => {
  return (
    <HStack gap={1} wrap="wrap" alignItems="flex-start">
      {buttons.map((btn, i) => (
        <Tooltip key={i} hasArrow label={btn.tooltip}>
          <Button
            colorScheme={color}
            variant="outline"
            size="sm"
            isDisabled={isDisabled}
            onClick={async () => {
              setSelectContent();
              const resultContent = await queryPrompt(btn.promptType, content);
              if (resultContent) {
                setResult({ title: btn.name, content: resultContent });
              }
            }}
          >
            {btn.name}
          </Button>
        </Tooltip>
      ))}
    </HStack>
  );
};

return (
    // ...
    <VStack alignItems="flex-start" w="100%">
      {renderButtons(generateButtons, 'blue', content, false)}
        <Text fontSize="sm">For selection text: </Text>
        {/* Only enable the buttons when the text content is selected. */}
    {renderButtons(contentButtons, 'teal', selectedContent, !selectedContent )}

        {!!result.title && (
        <VStack alignItems="flex-start">
          <Heading size="md">{result.title}</Heading>
          <pre style={{ whiteSpace: 'pre-wrap', fontFamily: 'inherit' }}>
            {result.content}
          </pre>
        </VStack>
      )}
    </VStack>
    // ...
)

The result returned from the API.

This GPT API has a high cost, so we can use caching to save the results of previous queries.

Call the /api/prompt API when clicking the buttons above to display the suggestion results.

const toast = useToast();

const [loadingPrompt, setLoadingPrompt] = useState(false);

const queryPrompt = async (promptType: string, content: string) => {
  setLoadingPrompt(true);
  const response = await fetch('/api/prompt', {
    method: 'POST',
    headers: {
      Accept: 'application/json',
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ topicType, promptType, question, content }),
  });
  const data = await response.json();
  setLoadingPrompt(false);
  if (!response.ok) {
    toast({
      title: 'Error',
      description: data?.error?.message,
      status: 'error',
      duration: 9000,
      isClosable: true,
    });
    return '';
  }
  return (data.result || '').trim();
};

Run the application.

npm run dev

Application Interface

Conclusion

Through building this application, I hope you have gained an understanding of how to integrate AI into your application to serve other use cases such as chatbot, tutoring, etc.

See you in the next post!

Detect body gestures to control RPG game using Mediapipe

Viet Hoang — Fri, 03 Feb 2023 13:46:24 +0000

If you have ever played Nintendo Switch games, you are likely familiar with the game Ring Fit Adventure, an exercise-based video game that encourages physical activity. Or Mario Tennis and its swing mode, which allows players to swing the Joy-Con like a tennis racket to play the game. Physical activity and fun gameplay make these games an excellent choice for anyone seeking a unique and entertaining way to stay fit.

So this post will show you how to create an app that converts your body gestures into real-time game controls. It can't yet replace all the functions of a keyboard or controller, but we'll strive to improve it in future versions.

You can view the demo video here. I am utilizing this application to play The Legend of Zelda: Breath of the Wild.

This also has a function to control the steering wheel, allowing you to use it to control a racing game. You can view the demo video here.

Here is the full code of this project.
https://github.com/ngviethoang/body-gesture-to-keyboard-control

This post below will explain how this project works.

Outline

First, let's break this down into smaller problems. It will be easier to comprehend and address this issue.

Detect human pose from the camera: We will apply the Mediapipe pose solution here to get the pose in real-time
Detect pre-defined body gestures: Define body gestures from pose estimation output to detect when the conditions for the gesture are met.
Trigger keyboard events corresponding to each gesture.

We will attempt to solve each problem and then incorporate them into the app.

Pose detection using Mediapipe

In this step, I will use PyQt to build the app and add mediapipe solutions to it.

First, install the packages that we need to use in this project, like PySide6, mediapipe, opencv, numpy.

pip install PySide6 opencv-python mediapipe numpy

Next, create a file to run the app.

import sys
from PySide6.QtCore import Qt, Slot
from PySide6.QtGui import QImage, QPixmap
from PySide6.QtWidgets import (
    QApplication,
    QComboBox,
    QHBoxLayout,
    QLabel,
    QMainWindow,
    QCheckBox,
    QVBoxLayout,
    QWidget,
    QFormLayout,
    QSlider,
    QPushButton,
)

class Window(QMainWindow):
    def __init__(self):
        super().__init__()
        # Title and dimensions
        self.setWindowTitle("Pose Detection")
        self.setGeometry(100, 100, 900, 650)

if __name__ == "__main__":
    app = QApplication()
    w = Window()
    w.show()
    sys.exit(app.exec())

To run the camera and extract images from it, we need to run an independent thread. This thread will read the image output from the camera. We can then use Mediapipe to process and display this image.

Create a class that inherits from PySide6's QThread. This class should be used to run the pose detection process.

class Cv2Thread(QThread):
        def __init__(self, parent=None):
        QThread.__init__(self, parent)

        def run(self):
                pass

In this run() function, we will read an image from the camera and run MediaPipe's pose estimation.

You can see the mediapipe solution for Python here.

Here is the code to run the pose estimation. It also draws the landmarks for the detected pose.

The results variable is the output of the pose detection. It contains the body's landmarks, which we can use to detect gestures.

You can read the comment for more information.

def run(self):
    self.cap = cv2.VideoCapture(0)
    with mp_pose.Pose() as pose:
        while self.cap.isOpened():
            success, image = self.cap.read()
            if not success:
                print("Ignoring empty camera frame.")
                # If loading a video, use 'break' instead of 'continue'.
                continue

            # To improve performance, optionally mark the image as not writeable to
            # pass by reference.
            # Recolor image to RGB
            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
            image.flags.writeable = False

            # Make detection
            results = pose.process(image)

            # Recolor back to BGR
            image.flags.writeable = True
            image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)

            # Draw landmark annotation on the image.
            mp_drawing.draw_landmarks(
                image,
                results.pose_landmarks,
                mp_pose.POSE_CONNECTIONS,
                landmark_drawing_spec=mp_drawing_styles.get_default_pose_landmarks_style(),
            )

            # Reading the image in RGB to display it
            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)

            # Creating and scaling QImage
            h, w, ch = image.shape
            image = QImage(image.data, w, h, ch * w, QImage.Format_RGB888)
            image = image.scaled(640, 480, Qt.KeepAspectRatio)

            if cv2.waitKey(5) & 0xFF == 27:
                break

    sys.exit(-1)

The processor can now be run on the thread, but we need the app to display the processed image. In PyQt, we have the Signal to send data to the main window.

class Cv2Thread(QThread):
        update_frame = Signal(QImage)

In the run function, add the emit function.

def run(self):
    self.cap = cv2.VideoCapture(0)
    with mp_pose.Pose() as pose:
        while self.cap.isOpened():
            success, image = self.cap.read()

                        ...

            # Creating and scaling QImage
            h, w, ch = image.shape
            image = QImage(image.data, w, h, ch * w, QImage.Format_RGB888)
            image = image.scaled(640, 480, Qt.KeepAspectRatio)

                        # Emit signal
            self.update_frame.emit(image)

            if cv2.waitKey(5) & 0xFF == 27:
                break

    sys.exit(-1)

Now back to Window class, update the __init__ function to create the thread and receive the image data.

After receiving the image, we set it to the label and show it in the app.

class Window(QMainWindow):
    def __init__(self):
        super().__init__()
        # Title and dimensions
        self.setWindowTitle("Pose Detection")
        self.setGeometry(100, 100, 900, 650)

        self.cv2_thread = Cv2Thread(self)
        self.cv2_thread.finished.connect(self.close)
                # Receive data from thread
        self.cv2_thread.update_frame.connect(self.setImage)

    @Slot(QImage)
    def setImage(self, image):
        self.camera_label.setPixmap(QPixmap.fromImage(image))

Now let’s run the app and see the result. Don’t forget to activate the virtualenv if it’s used.

python window.py

Now you can see pose landmarks marked in the camera.

Detect body gestures

Using the coordinates of these landmarks, we will attempt to detect body gestures and emit events when they are identified.

The results contain 33 pose landmarks, as shown in the image below. We can use these landmarks to detect gestures.

Another solution is to use machine learning (ML) to train the dataset for each pose. As I'm new to this field, we will skip this part for now and revisit it later to enhance our solution.

You can check another solution here: Pose Classification

In this post, I will use a simple method for gesture detection.

Each gesture has different angles between the landmarks. For example, when we’re standing, the angle between hip, knee and ankle is close to 180 degrees. Also when we curl our hands, the angle between shoulder, elbow and wrist landmarks is less than 45 degrees.

So, in the first step, we will get all landmarks that we need and calculate the angles between them for further calculation.

First, create a helper function for getting the right landmark.

def get_landmark_coordinates(landmarks, landmark):
    value = landmarks[landmark.value]
    return [
        value.x,
        value.y,
        value.z,
        value.visibility,
    ]

Then, extract the landmarks from results.

import mediapipe as mp

mp_pose = mp.solutions.pose

pose_landmarks = results.pose_landmarks.landmark

# Get coordinates
left_shoulder = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.LEFT_SHOULDER
)
right_shoulder = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.RIGHT_SHOULDER
)

left_elbow = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.LEFT_ELBOW
)
right_elbow = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.RIGHT_ELBOW
)

left_wrist = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.LEFT_WRIST
)
right_wrist = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.RIGHT_WRIST
)

left_hip = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.LEFT_HIP
)
right_hip = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.RIGHT_HIP
)

left_knee = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.LEFT_KNEE
)
right_knee = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.RIGHT_KNEE
)

left_ankle = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.LEFT_ANKLE
)
right_ankle = get_landmark_coordinates(
    pose_landmarks, mp_pose.PoseLandmark.RIGHT_ANKLE
)

...

To calculate the angle between three points in a two-dimensional coordinate system, use the following example code:

import numpy as np

# calculate angle between line ab and bc
def calculate_angle(a, b, c):
    a = np.array(a)  # First
    b = np.array(b)  # Mid
    c = np.array(c)  # End

    radians = np.arctan2(c[1] - b[1], c[0] - b[0]) - np.arctan2(a[1] - b[1], a[0] - b[0])
    angle = np.abs(radians * 180.0 / np.pi)

    if angle > 180.0:
        angle = 360 - angle

    return angle

We can now use this function to calculate the angles of the body's landmarks.

left_shoulder_angle = calculate_angle(left_elbow, left_shoulder, left_hip)
right_shoulder_angle = calculate_angle(
    right_elbow, right_shoulder, right_hip
)

left_elbow_angle = calculate_angle(left_shoulder, left_elbow, left_wrist)
right_elbow_angle = calculate_angle(
    right_shoulder, right_elbow, right_wrist
)

left_hip_angle = calculate_angle(left_shoulder, left_hip, left_knee)
right_hip_angle = calculate_angle(right_shoulder, right_hip, right_knee)

left_knee_angle = calculate_angle(left_hip, left_knee, left_ankle)
right_knee_angle = calculate_angle(right_hip, right_knee, right_ankle)

left_hip_knee_angle = calculate_angle(right_hip, left_hip, left_knee)
right_hip_knee_angle = calculate_angle(left_hip, right_hip, right_knee)

We can use these angles to monitor our body's states when we move and emit events like curling our hands or walking, if detected.

We need an object to store the body's current states, such as walking and swinging hands, so that we can predict the current gesture.

Predict body gesture

I will use the squat exercise as an example in this post.

First, create a class called LegState to manage the state of the body when we receive the angles above.

Set the property squat = False to keep track of the current state.

If the angle of both the left and right knee are smaller than a certain threshold (I set it to 155 degrees here), we change the property squat to True and emit the event. Otherwise, the property is set to False.

class LegsState:

    KNEE_UP_MAX_ANGLE = 155

    def __init__(self):
        self.squat = False

    def update(
        self,
        left_hip,
        right_hip,
        left_knee,
        right_knee,
        left_ankle,
        right_ankle,
        left_hip_angle,
        right_hip_angle,
        left_knee_angle,
        right_knee_angle,
    ):
        if (
            left_knee_angle < self.KNEE_UP_MAX_ANGLE
            and right_knee_angle < self.KNEE_UP_MAX_ANGLE
        ):
            if not self.squat:
                self.squat = True
                                print('squat')
                # emit event squat here 
        else:
            self.squat = False

Trigger keyboard events

The next step is to create a new class to receive events such as squats, walking,… and turn them into keyboard events.

First, let's create a class called Command for pressing and releasing the correct key for each event.

In the add_command function, we pass a command_key_mappings dictionary to retrieve the key associated with a command. If no key is found, the function terminates.

When a key is pressed, it must be released after a period of 100 milliseconds to 1 second or more. However, the event queue can become overloaded with numerous events in a single second, so we need a way to prevent pressing the same key that is already pressed and to release the key after a certain amount of time.

We will use a timer to separately release the currently pressed key. We also need to remember the most recent pressed key to avoid duplication.

from datetime import datetime
from pynput.keyboard import Controller
from threading import Timer

class CommandProcessor:
    def __init__(self):
        self.keyboard = Controller()
        self.pressing_key = None
        self.pressing_timer = None

    def release_previous_key(self):
        if self.pressing_key:
            previous_key = self.pressing_key["key"]
            # print(f"releasing {previous_key}")
            self.keyboard.release(previous_key)
            self.pressing_key = None

    def add_command(
        self,
        command,
        keyboard_enabled: bool,
        command_key_mappings: dict,
        pressing_timer_interval: float,
    ):
        now = datetime.now()

        if keyboard_enabled:
            if command in command_key_mappings:
                key = command_key_mappings[command]
                # get current pressing key
                previous_key = None
                if self.pressing_key:
                    previous_key = self.pressing_key["key"]

                # clear old timer
                if self.pressing_timer and self.pressing_timer.is_alive():
                    # print("cancel timer")
                    self.pressing_timer.cancel()

                # new action
                if previous_key != key:
                    self.release_previous_key()
                    if key:
                        print("pressing", key)
                        self.keyboard.press(key)

                if key:
                    # create new timer
                    self.pressing_timer = Timer(
                        pressing_timer_interval,
                        self.release_previous_key,
                    )
                    self.pressing_timer.start()

                    self.pressing_key = dict(key=key, time=now)

Now, let's create a class called Events to receive all events from the Command class above and run the add_command method.

In the Events class, we can create multiple Command properties to press keys simultaneously. We can route the commands to the appropriate processor based on the command name, as shown in the code below.

class Events:
    def __init__(
        self,
        keyboard_enabled,
        cross_cmd_enabled,
        pressing_timer_interval,
        d1_pressing_timer_interval,
        d2_pressing_timer_interval,
        command_key_mappings,
    ):
        self.keyboard_enabled = keyboard_enabled
        self.cross_cmd_enabled = cross_cmd_enabled
        self.command_key_mappings = command_key_mappings
        self.pressing_timer_interval = pressing_timer_interval
        self.d1_pressing_timer_interval = d1_pressing_timer_interval
        self.d2_pressing_timer_interval = d2_pressing_timer_interval

        self.cmd_process = CommandProcessor()

        # process cmd related to direction (left, right)
        self.d1_cmd_process = CommandProcessor()  # walk
        self.d2_cmd_process = CommandProcessor()  # tilt face

    # Add command to pipeline
    def add(self, command):
        # Split command by type name
        if "walk" in command or "d1" in command:
            self.d1_cmd_process.add_command(
                command,
                self.keyboard_enabled,
                self.command_key_mappings,
                self.d1_pressing_timer_interval,
            )
        elif "face" in command or "d2" in command:
            self.d2_cmd_process.add_command(
                command,
                self.keyboard_enabled,
                self.command_key_mappings,
                self.d2_pressing_timer_interval,
            )
        else:
            self.cmd_process.add_command(
                command,
                self.keyboard_enabled,
                self.command_key_mappings,
                self.pressing_timer_interval,
            )

The full code of my implementation can be found in the body directory.

Conclusion

This project is still simple and could be improved more. I would appreciate your opinion or any feedback you may have.

Hope you like it 👋