# Project export: Lucy This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: Cal Hacks 11.0 - Tagline: Military personnel will utilize Lucy to identify potential threats during field operations. - Devpost: https://devpost.com/software/lucy-x0wd86 - GitHub: https://github.com/Jinash-Rouniyar/SpotDog - Video: https://www.youtube.com/embed/Qpsvuxynwek?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Result: winner (Chroma: Build your project with Chroma in an AI application) - Team: 1 GitHub contributor(s) — Jinash-Rouniyar (21 commits) ## Devpost submission (written by the team) ### Inspiration We are tinkerers and builders who love getting our hands on new technologies. When we discovered that the Spot Robot Dog from Boston Dynamics was available to build our project upon, we devised different ideas about the real-world benefits of robot dogs. From a conversational companion to a navigational assistant, we bounced off different ideas and ultimately decided to use the Spot robot to detect explosives in the surrounding environment as we realized the immense amount of time and resources that are put into training real dogs to perform these dangerous yet important tasks. ### What it does Lucy uses the capabilities of Spot Robot Dog to help identify potentially threatening elements in a surrounding through computer vision and advanced wave sensing capabilities. A user can command the dog to inspect a certain geographic area and the dog autonomously walks around the entire area and flags objects that could be a potential threat. It captures both raw and thermal images of the given object in multiple frames, which are then stored on a vector database and can be searched through semantic search. This project is a simplified approach inspired by the research "Atomic Magnetometer Multisensor Array for rf Interference Mitigation and Unshielded Detection of Nuclear Quadrupole Resonance" (https://link.aps.org/accepted/10.1103/PhysRevApplied.6.064014). ### How we built it We've combined the capabilities of OpenCV with a thermal sensing camera to allow Spot Robot to identify and flag potentially threatening elements in a given surrounding. To simulate these elements in the surroundings, we built a simple Arduino application that emits light waves in irregular patterns. The robot dog operates independently through speech instructions, which are powered by DeepGram's Speech to Text and Llama-3-8b model hosted on the Groq platform. Furthermore, we've leveraged ChromDB's vector database to tokenize images that allow people to easily search through images, which are captured in the range of 20-40fps. ### Challenges we ran into The biggest challenge we encountered was executing and testing our code on Spot due to the unreliable internet connection. We also faced configuration issues, as some parts of modules were not supported and used an older version, leading to multiple errors during testing. Additionally, the limited space made it difficult to effectively run and test the code. ### Accomplishments we're proud of We are proud that we took on the challenge of working with something that we had never worked with before and even after many hiccups and obstacles we were able to convert our idea in our brains into a physical reality. ### What we learned We learned how to integrate and deploy our program onto Spot. We also learned that to work around the limitations of the technology and our experience working with them. ### What's next We want to integrate LiDar in our approach, providing more accurate results then cameras. We plan to experiment beyond light to include different wave forms, thus helping improve the reliability of the results. ## README (from the GitHub repository) **This image contains basic commands to work with camera, microphone and speaker at Spot's computer** Interaction with Spot is based on docker containers with code. You could fork this repo to test your own code. All basic functions are introduced in _main.py_. ## Image link Image link could be acquired in **packages** section of the main repository page. After fork, you might need to turn on github action to make it work. ## How to use this image ### Instal robot agent CLI ``` pipx install rn-cli ``` ### Create a user key ``` rn keys gen user.key ``` Share public key with Vataly to get access to robot ### Setup environmental variables ``` export AGENT_RPC=ws://104.131.170.157:8888 export OWNER_KEY=Okkb1brctXS040mWyDun5aCYrG7yIHUjnx/Rza7KDhI= export USER_KEY_PATH=user.key ``` ### You can check, if you see Spot ``` rn robots list ``` ### Send job to robot Create _job.json_, change image link to your version ``` { "image": "ghcr.io/otaberu/hackathon-spot-image:main", "container_name": "", "ports": [], "network_mode": "host", "volumes": [ { "key": "/dev/video0", "value": "/dev/video0" }, { "key": "/dev/snd", "value": "/dev/snd" } ], "privileged": true, "store_data": false, "env": [ "SDL_AUDIODRIVER=alsa", "AUDIODEV=hw:1,0", "AUDIO_INPUT_DEVICE=hw:2,0" ] } ``` Create job (after you got access from Vitaly) ``` rn jobs add job.json spot ``` If you change last line of Dockerfile to ```CMD ["/bin/sh"]```, you could access terminal with ``` rn jobs terminal spot JOB_ID ``` Do not forget to exit it with ```exit``` command ## Detected evidence (automated analysis) Indexed codebase: 8 recognized source files, 46 KB. - Python (language) — detected in the code ## Codebase structure (from repository index) ### Files (14 of 14) ``` .github/workflows/build.yml .gitignore azure_pronunciation.py chroma_data/chroma.sqlite3 Dockerfile groq_bot.py image_retreiver.py job.json main.py README.md requirements.txt spot_controller.py try.py user.key ``` ### Dependencies No dependency index available. ### Recent commits (newest first) - final_update - final_commit - final repo - updated - updated - updated - updated - updated - updated - updated - updated - updated - update - update - update - update - update - update - update - update ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` ��bosdyn-client >= 3.1 langchain langchain-openai azure.cognitiveservices.speech numpy ``` ### Dockerfile ``` FROM ghcr.io/merklebot/hackathon-amd-image:master as build ENV PYTHONDONTWRITEBYTECODE 1 ENV PYTHONUNBUFFERED 1 ARG TARGETPLATFORM ARG BUILDPLATFORM ARG TARGETOS ARG TARGETARCH ARG Version ARG GitCommit RUN echo "I am running on $BUILDPLATFORM, building for $TARGETPLATFORM" COPY requirements.txt requirements.txt RUN python3.8 -m pip install -r requirements.txt COPY . . CMD ["python3.8", "main.py"] ``` ### main.py ```python import cv2 import numpy as np from collections import deque from dataclasses import dataclass from spot_controller import SpotController from typing import List, Tuple from bosdyn.client import create_standard_sdk from bosdyn.client.robot import Robot from bosdyn.client.image import ImageClient import time import os import math from deepgram import Deepgram from groq_bot import groq_chain from langchain.schema.runnable import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser import numpy as np @dataclass class LED: position: Tuple[int, int] radius: int last_seen: int class SpotThreatDetector: def __init__(self, spot_controller): self.spot_controller = spot_controller self.image_client = self.spot_controller.robot.ensure_client(ImageClient.default_service_name) # Initialize threat detection parameters self.led_history = deque(maxlen=10) self.frame_count = 0 self.led_tracker = [] # Available camera sources self.camera_sources = [ "frontleft_fisheye_image", "frontright_fisheye_image", "left_fisheye_image", "right_fisheye_image", "back_fisheye_image" ] self.current_camera = self.camera_sources[0] def capture_frame(self): """Captures a frame from Spot's current camera""" try: image_responses = self.image_client.get_image_from_sources([self.current_camera]) if not image_responses: raise Exception("No image responses received") image_response = image_responses[0] image_data = image_response.shot.image.data # Convert to OpenCV format numpy_array = np.frombuffer(image_data, dtype=np.uint8) frame = cv2.imdecode(numpy_array, cv2.IMREAD_COLOR) if frame is None: raise Exception("Failed to decode image data") return frame except Exception as e: print(f"Error capturing frame: {str(e)}") return None def process_frame(self, frame): """Process frame for threat detection""" if frame is None: return None, False # Convert to HSV for better red detection hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) # Define range for red color lower_red1 = np.array([0, 100, 100]) upper_red1 = np.array([10, 255, 255]) lower_red2 = np.array([160, 100, 100]) upper_red2 = np.array([180, 255, 255]) # Create masks for red detection mask1 = cv2.inRange(hsv, lower_red1, upper_red1) mask2 = cv2.inRange(hsv, lower_red2, upper_red2) mask = cv2.bitwise_or(mask1, mask2) # Apply noise reduction mask = cv2.erode(mask, None, iterations=2) mask = cv2.dilate(mask, None, iterations=2) # Find contours contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) current_leds = [] for contour in contours: if cv2.contourArea(contour) > 50: ((x, y), radius) = cv2.minEnclosingCircle(contour) if radius > 5: current_leds.append(LED((int(x), int(y)), int(radius), self.frame_count)) self.led_history.append(current_leds) return self.detect_threat(frame) def detect_threat(self, frame): """Detect potential threats in the frame""" if len(self.led_history) < 5: return frame, False led_groups = [] for leds in self.led_history: if len(leds) >= 3: positions = [(led.position[0], led.position[1]) for led in leds] led_groups.append(positions) if len(led_groups) < 5: return frame, False threat_detected = False threat_box = None for i in range(len(led_groups[-1])): nearby_leds = [] center = led_groups[-1][i] for j in range(len(led_groups[-1])): if i != j: dx = center[0] - led_groups[-1][j][0] dy = center[1] - led_groups[-1][j][1] distance = np.sqrt(dx*dx + dy*dy) if distance < 100: nearby_leds.append(led_groups[-1][j]) if len(nearby_leds) >= 2: threat_detected = True all_points = [center] + nearby_leds min_x = min(p[0] for p in all_points) min_y = min(p[1] for p in all_points) max_x = max(p[0] for p in all_points) max_y = max(p[1] for p in all_points) padding = 20 threat_box = ( max(0, min_x - padding), max(0, min_y - padding), min(frame.shape[1], max_x + padding), min(frame.shape[0], max_y + padding) ) break if threat_detected and threat_box: cv2.rectangle(frame, (int(threat_box[0]), int(threat_box[1])), (int(threat_box[2]), int(threat_box[3])), (0, 0, 255), 2) cv2.putText(frame, "THREAT DETECTED", (int(threat_box[0]), int(threat_box[1] - 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) # Save frame when threat is detected timestamp = time.strftime("%Y%m%d-%H%M%S") cv2.imwrite(f"threat_detected_{timestamp}.jpg", frame) return frame, threat_detected def create_thermal_vision(self, frame): """Create [truncated — 7777 more characters] ``` ### image_retreiver.py ```python import dotenv from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from PIL import Image from transformers import AutoTokenizer, AutoFeatureExtractor import torch import os folder_path = "head_movement_frames" CHROMA_PATH = "Questions_db" dotenv.load_dotenv() class ImageEmbedder: def __init__(self, model_name="google/vit-base-patch16-224"): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.feature_extractor = AutoFeatureExtractor.from_pretrained(model_name) def create_embeddings(self, image_path): image = Image.open(image_path) inputs = self.feature_extractor(images=image, return_tensors="pt") with torch.no_grad(): embeddings = self.tokenizer.encode(inputs['pixel_values'].squeeze().tolist(), add_special_tokens=True) return embeddings class ImageDocument: def __init__(self, image_path, metadata=None): self.image_path = image_path self.metadata = metadata or {} image_embedder = ImageEmbedder() # Assume we have a list of image paths image_paths = [image_path for image_path in os.listdir(folder_path) if image_path.endswith(('.jpg', '.jpeg', '.png'))] # Replace with your actual image paths # Create image documents image_documents = [ImageDocument(image_path) for image_path in image_paths] # Create embeddings for each image embeddings = [image_embedder.create_embeddings(doc.image_path) for doc in image_documents] # Create Chroma vector store image_vector_db = Chroma.from_embeddings( embeddings, OpenAIEmbeddings(), persist_directory=CHROMA_PATH ) ``` ### groq_bot.py ```python import dotenv import os from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser from langchain_community.vectorstores import Chroma from langchain.schema.runnable import RunnablePassthrough from langchain.prompts import ( PromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate, ChatPromptTemplate, ) from langchain_groq import ChatGroq dotenv.load_dotenv() groq_api_key = os.getenv("GROQ_API_KEY") review_system_template_str = """ You are controlling a Spot robot equipped with a vision camera. Based on the surroundings that you see, provide one of the following instructions for the robot's movement: - "quit" if no action is required or if the session should stop - "up" if the path ahead is clear and the robot should move forward - "down" if the robot should reverse - "left" if the robot needs to turn left and move forward - "right" if the robot needs to turn right and move forward - "turn" if the robot should perform a smooth 360-degree turn - "scan" if the robot should move its head down and capture frames Please base your decision on the camera input. Provide only the instruction in your response. """ review_system_prompt = SystemMessagePromptTemplate( prompt=PromptTemplate( input_variables=[], template=review_system_template_str ) ) review_human_prompt = HumanMessagePromptTemplate( prompt=PromptTemplate( input_variables=["question"], template="{question}" ) ) messages = [review_system_prompt, review_human_prompt] review_prompt_template = ChatPromptTemplate( input_variables=["question"], messages=messages, ) output_parser = StrOutputParser() chat_model = ChatGroq( api_key=groq_api_key, model_name='llama3-groq-8b-8192-tool-use-preview', temperature = 0 ) groq_chain = ( {"question": RunnablePassthrough()} | review_prompt_template | chat_model | StrOutputParser() ) ``` ### spot_controller.py ```python import time import bosdyn.client from bosdyn.client.robot_command import RobotCommandClient, RobotCommandBuilder, blocking_stand # , blocking_sit from bosdyn.geometry import EulerZXY from bosdyn.api.spot import robot_command_pb2 as spot_command_pb2 from bosdyn.client.frame_helpers import ODOM_FRAME_NAME from bosdyn.api.basic_command_pb2 import RobotCommandFeedbackStatus from bosdyn.client.estop import EstopClient, EstopEndpoint, EstopKeepAlive from bosdyn.client.robot_state import RobotStateClient from bosdyn.client.frame_helpers import ODOM_FRAME_NAME, VISION_FRAME_NAME, BODY_FRAME_NAME, \ GRAV_ALIGNED_BODY_FRAME_NAME, get_se2_a_tform_b from bosdyn.client import math_helpers import traceback VELOCITY_CMD_DURATION = 0.5 class SpotController: def __init__(self, username, password, robot_ip): self.username = username self.password = password self.robot_ip = robot_ip sdk = bosdyn.client.create_standard_sdk('ControllingSDK') self.robot = sdk.create_robot(robot_ip) id_client = self.robot.ensure_client('robot-id') self.robot.authenticate(username, password) self.command_client = self.robot.ensure_client(RobotCommandClient.default_service_name) self.robot.logger.info("Authenticated") self._lease_client = None self._lease = None self._lease_keepalive = None self._estop_client = self.robot.ensure_client(EstopClient.default_service_name) self._estop_endpoint = EstopEndpoint(self._estop_client, 'GNClient', 9.0) self._estop_keepalive = None self.state_client = self.robot.ensure_client(RobotStateClient.default_service_name) def release_estop(self): self._estop_endpoint.force_simple_setup() self._estop_keepalive = EstopKeepAlive(self._estop_endpoint) def set_estop(self): if self._estop_keepalive: try: self._estop_keepalive.stop() except: self.robot.logger.error("Failed to set estop") traceback.print_exc() self._estop_keepalive.shutdown() self._estop_keepalive = None def lease_control(self): self._lease_client = self.robot.ensure_client('lease') self._lease = self._lease_client.take() self._lease_keepalive = bosdyn.client.lease.LeaseKeepAlive(self._lease_client, must_acquire=True) self.robot.logger.info("Lease acquired") def return_lease(self): self._lease_client.return_lease(self._lease) self._lease_keepalive.shutdown() self._lease_keepalive = None def __enter__(self): self.lease_control() self.release_estop() self.power_on_stand_up() return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type: self.robot.logger.error("Spot powered off with " + exc_val + " exception") self.power_off_sit_down() self.return_lease() self.set_estop() return True if exc_type else False def move_head_in_points(self, yaws, pitches, rolls, body_height=0, sleep_after_point_reached=0, timeout=3): for i in range(len(yaws)): footprint_r_body = EulerZXY(yaw=yaws[i], roll=rolls[i], pitch=pitches[i]) params = RobotCommandBuilder.mobility_params(footprint_R_body=footprint_r_body, body_height=body_height) blocking_stand(self.command_client, timeout_sec=timeout, update_frequency=0.02, params=params) self.robot.logger.info("Moved to yaw={} rolls={} pitch={}".format(yaws[i], rolls[i], pitches[i])) if sleep_after_point_reached: time.sleep(sleep_after_point_reached) def wait_until_action_complete(self, cmd_id, timeout=15): start_time = time.time() while time.time() - start_time < timeout: feedback = self.command_client.robot_command_feedback(cmd_id) mobility_feedback = feedback.feedback.synchronized_feedback.mobility_command_feedback if mobility_feedback.status != RobotCommandFeedbackStatus.STATUS_PROCESSING: print("Failed to reach the goal") return False traj_feedback = mobility_feedback.se2_trajectory_feedback if (traj_feedback.status == traj_feedback.STATUS_AT_GOAL and traj_feedback.body_movement_status == traj_feedback.BODY_STATUS_SETTLED): print("Arrived at the goal.") return True time.sleep(0.5) def move_to_goal(self, goal_x=0, goal_y=0): cmd = RobotCommandBuilder.synchro_trajectory_command_in_body_frame( goal_x_rt_body=goal_x, goal_y_rt_body=goal_y, goal_heading_rt_body=0, frame_tree_snapshot=self.robot.get_frame_tree_snapshot() ) # cmd = RobotCommandBuilder.synchro_se2_trajectory_point_command(goal_x=goal_x, goal_y=goal_y, goal_heading=0, # frame_name=GRAV_ALIGNED_BODY_FRAME_NAME) cmd_id = self.command_client.robot_command(lease=None, command=cmd, end_time_secs=time.time() + 10) self.wait_until_action_complete(cmd_id) self.robot.logger.info("Moved to x={} y={}".format(goal_x, goal_y)) def power_on_stand_up(self): self.robot.power_on(timeout_sec=20) assert self.robot.is_powered_on(), "Not powered on" self.robot.time_sync.wait_for_sync() blocking_stand(self.command_client, timeout_sec=10) def power_off_sit_down(self): self.move_head_in_points(yaws=[0], pitches=[0], rolls=[0]) self.robot.power_off(cut_immediately=False) def make_stance(self, x_offset, y_offset): state = self.state_client.get_robot_state() vo_T_body = get_se2_a_tform_b(state.kinematic_state.transforms_snapshot, VISION_FRAME_NAME, [truncated — 2131 more characters] ``` ### try.py ```python import os import time import subprocess import tempfile from spot_controller import SpotController import cv2 from math_bot import math_chain, question, answer_exp import azure.cognitiveservices.speech as speechsdk from azure_pronunciation import SpeechToTextManager import requests ROBOT_IP = "192.168.80.3" SPOT_USERNAME = "admin" SPOT_PASSWORD = "2zqa8dgw7lor" WAVE_OUTPUT_FILENAME = "aaaa.wav" LANG_CODE = "en-US" class RoomExplorationBot: def __init__(self, spot_username, spot_password, robot_ip): # Initialize Spot controller parameters self.spot_username = spot_username self.spot_password = spot_password self.robot_ip = robot_ip # Initialize speech components self.speech_config = speechsdk.SpeechConfig( subscription=os.environ.get('SPEECH_KEY'), region=os.environ.get('SPEECH_REGION') ) self.speech_synthesizer = speechsdk.SpeechSynthesizer(speech_config=self.speech_config) self.speech_to_text_manager = SpeechToTextManager() # Initialize object detection list self.detected_objects = [] def detect_objects(self, image): """ Detect objects in the captured image using OpenCV. Returns list of detected objects with their positions. """ gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) params = cv2.SimpleBlobDetector_Params() params.minThreshold = 10 params.maxThreshold = 200 params.filterByArea = True params.minArea = 1500 detector = cv2.SimpleBlobDetector_create(params) keypoints = detector.detect(gray) # Store detected objects with their locations objects = [(int(k.pt[0]), int(k.pt[1])) for k in keypoints] self.detected_objects.extend(objects) return keypoints def inspect_object(self, spot, sleep_time=1): """ Perform an inspection motion (head nod) when examining an object. """ try: # Nod down to inspect spot.move_head_in_points( yaws=[0], pitches=[0.5], # Look down rolls=[0], sleep_after_point_reached=sleep_time ) # Capture image while looking down self.capture_image() # Return to original position spot.move_head_in_points( yaws=[0], pitches=[0], rolls=[0], sleep_after_point_reached=sleep_time ) except Exception as e: print(f"Error during object inspection: {e}") spot.move_head_in_points(yaws=[0], pitches=[0], rolls=[0]) def capture_image(self): camera_capture = cv2.VideoCapture(0) rv, image = camera_capture.read() camera_capture.release() return rv, image def capture_and_process(self, spot, capture_interval=2): rv, image = self.capture_image() if rv: objects = self.detect_objects(image) if objects: print(f"Detected {len(objects)} objects - initiating inspection") self.inspect_object(spot) cv2.imwrite(f"exploration_frame_{time.time()}.jpg", image) time.sleep(capture_interval) def explore_room(self, spot, capture_interval=2): """ Execute room exploration pattern with object detection. """ print("Starting room exploration...") try: # 1. Enter room print("Phase 1: Entering room...") spot.move_to_goal(goal_x=1.5, goal_y=0) self.capture_and_process(spot, capture_interval) # 2. Scan the room from entrance print("Phase 2: Initial room scan...") scan_positions = [ (0.4, 0.3, 0), # Look right-up (0, 0.3, 0), # Look up (-0.4, 0.3, 0), # Look left-up (-0.4, 0, 0), # Look left (0, 0, 0), # Look center (0.4, 0, 0), # Look right ] for yaw, pitch, roll in scan_positions: spot.move_head_in_points( yaws=[yaw], pitches=[pitch], rolls=[roll], sleep_after_point_reached=1 ) self.capture_and_process(spot, capture_interval) # 3. Move in a square pattern print("Phase 3: Room exploration pattern...") movements = [ (0, 1), # Move left 1m (1, 0), # Move forward 1m (0, -1), # Move right 1m (-1, 0), # Move back 1m ] for dx, dy in movements: spot.move_to_goal(goal_x=dx, goal_y=dy) for yaw, pitch, roll in [(0.3, 0.2, 0), (0, 0.2, 0), (-0.3, 0.2, 0)]: spot.move_head_in_points( yaws=[yaw], pitches=[pitch], rolls=[roll], sleep_after_point_reached=1 ) self.capture_and_process(spot, capture_interval) # 4. Return to start print("Phase 4: Returning to start position...") spot.move_to_goal(goal_x=-1.5, goal_y=0) spot.move_head_in_points(yaws=[0], pitches=[0], rolls=[0], sleep_after_point_reached=1) except Exception as e: print(f"Error during room exploration: {e}") spot.move_head_in_points(yaws=[0], pitches=[0], rolls=[0]) raise return len(self.detected_objects) def stream_and_synthesize_response(self, text): """ Stream and sy [truncated — 2708 more characters] ``` ### azure_pronunciation.py ```python import time import azure.cognitiveservices.speech as speechsdk import os import numpy as np class SpeechToTextManager: azure_speechconfig = None azure_audioconfig = None azure_speechrecognizer = None def __init__(self): # Creates an instance of a speech config with specified subscription key and service region. # Replace with your own subscription key and service region (e.g., "westus"). try: self.azure_speechconfig = speechsdk.SpeechConfig(subscription=os.getenv('AZURE_TTS_KEY'), region=os.getenv('AZURE_TTS_REGION')) except TypeError: exit("Ooops! You forgot to set AZURE_TTS_KEY or AZURE_TTS_REGION in your environment!") self.stop_listening_flag = False self.accuracy_scores = [] self.pronunciation_scores = [] self.completeness_scores = [] self.fluency_scores = [] def stop_listening(self): self.stop_listening_flag = True def send_message(self,message,speaker=None): pass def speechtotext_from_mic(self,lang_code): self.azure_speechconfig.speech_recognition_language=lang_code self.azure_audioconfig = speechsdk.audio.AudioConfig(use_default_microphone=True) self.azure_speechrecognizer = speechsdk.SpeechRecognizer(speech_config=self.azure_speechconfig, audio_config=self.azure_audioconfig) print("Speak into your microphone.") speech_recognition_result = self.azure_speechrecognizer.recognize_once_async().get() text_result = speech_recognition_result.text if speech_recognition_result.reason == speechsdk.ResultReason.RecognizedSpeech: print("Recognized: {}".format(speech_recognition_result.text)) elif speech_recognition_result.reason == speechsdk.ResultReason.NoMatch: print("No speech could be recognized: {}".format(speech_recognition_result.no_match_details)) elif speech_recognition_result.reason == speechsdk.ResultReason.Canceled: cancellation_details = speech_recognition_result.cancellation_details print("Speech Recognition canceled: {}".format(cancellation_details.reason)) if cancellation_details.reason == speechsdk.CancellationReason.Error: print("Error details: {}".format(cancellation_details.error_details)) print("Did you set the speech resource key and region values?") print(f"We got the following text: {text_result}") return text_result def pronunciation_check(self, filename, reference_text,lang_code): """performs one-shot speech recognition with input from an audio file, showing detailed recognition results including word-level timing and pronunciation assessment """ # Ask for detailed recognition result # speech_config.output_format = speechsdk.OutputFormat.Detailed # If you also want word-level timing in the detailed recognition results, set the following. # Note that if you set the following, you can omit the previous line # "speech_config.output_format = speechsdk.OutputFormat.Detailed", # since word-level timing implies detailed recognition results. self.azure_speechconfig.speech_recognition_language=lang_code self.azure_audioconfig = speechsdk.AudioConfig(filename=filename) self.azure_speechrecognizer = speechsdk.SpeechRecognizer(speech_config=self.azure_speechconfig, audio_config=self.azure_audioconfig) try: self.azure_speechconfig.request_word_level_timestamps() except: print("Error in requesting word_level_timestamps ") print("Listening to the file \n") pronunciation_config = speechsdk.PronunciationAssessmentConfig( reference_text=reference_text, grading_system=speechsdk.PronunciationAssessmentGradingSystem.FivePoint, granularity=speechsdk.PronunciationAssessmentGranularity.Phoneme, enable_miscue=True) pronunciation_config.apply_to(self.azure_speechrecognizer) result = self.azure_speechrecognizer.recognize_once_async().get() # Check the result if result.reason == speechsdk.ResultReason.RecognizedSpeech: pronunciation_result = speechsdk.PronunciationAssessmentResult(result) print('Pronunciation assessment completed for: {}'.format(result.text)) # Store scores in arrays self.accuracy_scores.append(pronunciation_result.accuracy_score) self.pronunciation_scores.append(pronunciation_result.pronunciation_score) self.completeness_scores.append(pronunciation_result.completeness_score) self.fluency_scores.append(pronunciation_result.fluency_score) # print(' Word-level details:') # for idx, word in enumerate(pronunciation_result.words): # print(' {}: word: {}\taccuracy score: {}\terror type: {};'.format( # idx + 1, word.word, word.accuracy_score, word.error_type # )) elif result.reason == speechsdk.ResultReason.NoMatch: print("No speech could be recognized: {}".format(result.no_match_details)) elif result.reason == speechsdk.ResultReason.Canceled: cancellation_details = result.cancellation_details print("Speech Recognition canceled: {}".format(cancellation_details.reason)) if cancellation_details.reason == speechsdk.CancellationReason.Error: print("Error details: {}".format(cancellation_details.error_details)) def speechtotext_from_file(self, filename,lang_code): self.azure_speechconfig.speech_recognition_language= lang_code self.azure_audioconfig = speechsdk.AudioConfig(filename=filename) self.azure_speechrecognizer = speechsdk.SpeechRecognizer(speech_config=self.az [truncated — 4452 more characters] ``` ### .github/workflows/build.yml ```yaml name: Build image on: push: branches: [ main ] tags: - "*" workflow_dispatch: env: REGISTRY: ghcr.io IMAGE_NAME: ${{ github.repository }} jobs: build: runs-on: ubuntu-latest permissions: contents: read packages: write steps: - uses: actions/checkout@v3 - name: Set up QEMU uses: docker/setup-qemu-action@v2 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v2 - uses: docker/login-action@v2 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Extract metadata (tags, labels) for image builder uses: docker/metadata-action@v4 id: meta with: images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} - name: Build and push container image uses: docker/build-push-action@v3 with: context: . push: true platforms: linux/amd64 tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} ```