Upload 3 files

env.env

@@ -0,0 +1 @@
+API_ENDPOINT= https://226454676c90.ngrok-free.app/api/process-annotation

model.py

@@ -0,0 +1,80 @@
+import torch
+import torch.nn as nn
+import torchvision.models as models
+import torchvision.transforms as transforms
+from PIL import Image
+import time
+
+class AE(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.encoder = nn.Sequential(
+            nn.Linear(2048, 512), nn.ReLU(),
+            nn.Linear(512, 128)
+        )
+        self.decoder = nn.Sequential(
+            nn.Linear(128, 512), nn.ReLU(),
+            nn.Linear(512, 2048)
+        )
+    def forward(self, x):
+        return self.decoder(self.encoder(x))
+
+resnet = models.resnet50(weights=models.ResNet50_Weights.DEFAULT)
+resnet.fc = nn.Identity()
+resnet.eval()
+
+autoencoder = AE()
+
+def model_option(view, category):
+    if view == "crl":
+        if category == "abdomen":
+            autoencoder.load_state_dict(torch.load('models/abdomen_autoencoder-0.0058.pth'))
+        elif category == "body":
+            autoencoder.load_state_dict(torch.load('models/body_autoencoder-0.0060.pth'))
+        elif category == "diencephalon":
+            autoencoder.load_state_dict(torch.load('models/diencephalon_autoencoder-0.0050.pth'))
+        elif category == "gsac":
+            autoencoder.load_state_dict(torch.load('models/gestation_sac_autoencoder-0.0044.pth'))
+        elif category == "head":
+            autoencoder.load_state_dict(torch.load('models/head_autoencoder-0.0077.pth'))
+        elif category == "lv":
+            autoencoder.load_state_dict(torch.load('models/lateral_ventricle_autoencoder-0.0045.pth'))
+        elif category == "mx":
+            autoencoder.load_state_dict(torch.load('models/maxilla_autoencoder-0.0054.pth'))
+        elif category == "mds":
+            autoencoder.load_state_dict(torch.load('models/mds_mandible_autoencoder-0.0039.pth'))
+        elif category == "mls":
+            autoencoder.load_state_dict(torch.load('models/mls_mandible_ventricle_autoencoder-0.0047.pth'))
+        elif category == "nb":
+            autoencoder.load_state_dict(torch.load('models/nasal_bone_autoencoder-0.0026.pth'))
+        elif category == "ntaps":
+            autoencoder.load_state_dict(torch.load('models/ntaps_autoencoder-0.0032.pth'))
+        elif category == "rbp":
+            autoencoder.load_state_dict(torch.load('models/rhombencephalon_autoencoder-0.0044.pth'))
+        elif category == "thorax":
+            autoencoder.load_state_dict(torch.load('models/thorax_autoencoder-0.0058.pth'))
+    #elif view == "nt":
+    autoencoder.eval()
+    
+transform = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize([0.485, 0.456, 0.406],
+                         [0.229, 0.224, 0.225])
+])
+
+def predict(cropped, view, category):
+    model_option(view, category)
+    img = cropped.convert("RGB")
+    img_tensor = transform(img).unsqueeze(0)
+
+    with torch.no_grad():
+        feat = resnet(img_tensor).squeeze().numpy()
+
+    input_tensor = torch.tensor(feat).float().unsqueeze(0)
+    with torch.no_grad():
+        recon = autoencoder(input_tensor)
+
+    error = nn.functional.mse_loss(recon, input_tensor).item()
+
+    return error

telegram-bot.py

@@ -0,0 +1,229 @@
+import logging
+import aiohttp
+import os,time
+
+from aiohttp import FormData
+from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
+
+from telegram.ext import (
+    Application, CommandHandler, MessageHandler,
+    filters, CallbackQueryHandler, ContextTypes
+)
+
+from dotenv import load_dotenv
+load_dotenv(dotenv_path='env.env')
+
+
+#config
+TOKEN = "8320924107:AAH505mhHkOxeY3aLk0GObIpO_KCtY9hhLM"
+API_ENDPOINT = os.getenv("API_ENDPOINT")
+logging.basicConfig(level=logging.INFO)
+
+#category mappings
+VIEW_CATEGORIES = {
+    "crl": {
+        "Maxilla": "mx",
+        "Mandible-MDS": "mds", 
+        "Mandible-MLS": "mls",
+        "Lateral ventricle": "lv",
+        "Head": "head",
+        "Gestational sac": "gsac",
+        "Thorax": "thorax",
+        "Abdomen": "ab",
+        "Body(Biparietal diameter)": "bd",
+        "Rhombencephalon": "rbp",
+        "Diencephalon": "dp",
+        "NTAPS": "ntaps",
+        "Nasal bone": "nb"
+    },
+    "nt": {
+        "Maxilla": "mx",
+        "Mandible-MDS": "mds", 
+        "Mandible-MLS": "mls",
+        "Lateral ventricle": "lv",
+        "Head": "head",
+        "Thorax": "thorax",
+        "Abdomen": "ab",
+        "Rhombencephalon": "rbp",
+        "Diencephalon": "dp",
+        "Nuchal translucency": "nt",
+        "NTAPS": "ntaps",
+        "Nasal bone": "nb"
+    }
+}
+
+#start
+async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    await update.message.reply_text("Send an ultrasound image (JPG/png only) to begin. See /instructions first")
+
+async def instructions(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    instruction_message = """
+📋 **Instructions**
+
+🔸 **Step 1:** Send a cropped ultrasound image of a structure you want to analyze. See all structures with /list command
+🔸 **Step 2:** Select the ultrasound view (CRL or NT)
+🔸 **Step 3:** Choose the anatomical category to analyze
+🔸 **Step 4:** Wait for the AI analysis results
+
+📌 **Important notes:**
+• Only JPG/PNG images are supported
+• Ensure the ultrasound image is clear and properly oriented
+• Results are for reference only - always consult a medical professional
+• Processing may take a few seconds
+• Stop bot with /stop
+
+💡 **Tips:**
+• Use high-quality, well-lit images for better accuracy
+• Make sure the anatomical structure is clearly visible
+• Different views (CRL/NT) have different category options
+
+🆘 **Need help?** Contact support if you encounter any issues @d3ikshr.
+    """
+    
+    await update.message.reply_text(instruction_message, parse_mode='Markdown')
+
+async def list_categories(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    list_message = """
+📋 **Available Categories by View**
+
+🔍 **CRL view categories:**
+• Maxilla • Mandible-MDS • Mandible-MLS
+• Lateral ventricle • Head • Gestational sac
+• Thorax • Abdomen • Body(Biparietal diameter)
+• Rhombencephalon • Diencephalon • NTAPS
+• Nasal bone
+
+🔍 **NT view categories:**
+• Maxilla • Mandible-MDS • Mandible-MLS
+• Lateral ventricle • Head • Thorax
+• Abdomen • Rhombencephalon • Diencephalon
+• Nuchal translucency • NTAPS • Nasal bone
+
+💡 **Note:** Categories will be shown automatically based on your selected view during analysis.
+    """
+    
+    await update.message.reply_text(list_message, parse_mode='Markdown')
+
+async def stop(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    await update.message.reply_text("🛑 Bot stopped for this chat. Use /start to begin again.")
+    # Clear user data
+    context.user_data.clear()
+
+# Receive image
+async def handle_image(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    photo = update.message.photo[-1]
+    file = await photo.get_file()
+    file_path = f"{update.message.from_user.id}_ultrasound.jpg"
+    await file.download_to_drive(file_path)
+    context.user_data["image_path"] = file_path
+    
+    # get view
+    buttons = [
+        [InlineKeyboardButton("CRL", callback_data="view:crl"),
+         InlineKeyboardButton("NT", callback_data="view:nt")]
+    ]
+    await update.message.reply_text(
+        "Select the ultrasound view:",
+        reply_markup=InlineKeyboardMarkup(buttons)
+    )
+
+#selcet view
+async def handle_view(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    query = update.callback_query
+    await query.answer()
+    view = query.data.split(":")[1]
+    context.user_data["selected_view"] = view
+    
+    #get categories for the selected view
+    categories = list(VIEW_CATEGORIES[view].keys())
+    buttons = [[InlineKeyboardButton(cat, callback_data=f"category:{cat}")]
+               for cat in categories]
+    await query.edit_message_text(
+        "Select anatomical category:",
+        reply_markup=InlineKeyboardMarkup(buttons)
+    )
+
+#get category then upload
+async def handle_category(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    query = update.callback_query
+    await query.answer()
+    category_display = query.data.split(":")[1]
+    context.user_data["selected_category"] = category_display
+    
+    image_path = context.user_data.get("image_path")
+    view = context.user_data.get("selected_view")
+    
+    if not image_path or not view:
+        await query.edit_message_text("Missing image or view.")
+        return
+
+    await query.edit_message_text("🔄 Processing image...")
+    time.sleep(3)
+    await query.edit_message_text("🏥 Building diagnosis, please wait...")
+
+    try:
+        #read the image file into memory first
+        with open(image_path, "rb") as f:
+            image_data = f.read()
+        
+        #mapping
+        category_value = VIEW_CATEGORIES[view][category_display]
+        
+        #create form data
+        form = FormData()
+        form.add_field("view", view)
+        form.add_field("category", category_value)
+        form.add_field("source", "telegram")
+        form.add_field("image", image_data, filename="image.jpg", content_type="image/jpeg")
+        
+        #send request 
+        async with aiohttp.ClientSession() as session:
+            async with session.post(API_ENDPOINT, data=form) as resp:
+                if resp.status == 200:
+                    result = await resp.json()
+                    
+                    message = f"""
+📊 **Analysis Results**
+
+🔍 **View:** {result.get('view', view).upper()}
+🏥 **Category:** {category_display}
+
+📈 **Confidence:** {result.get('confidence', 0):.2f}%
+⚠️ **Reconstruction error:** {result.get('error', 0):.5f}
+
+📋 **Status:** {result.get('comment', 'No comment')}
+
+🩺 **Diagnosis:** {result.get('diagnosis', 'No diagnosis')}*
+                    """
+                    
+                    await query.edit_message_text(message, parse_mode='Markdown')
+                     
+                else:
+                    error_text = await resp.text()
+                    await query.edit_message_text(f"❌ Upload failed. Status: {resp.status}\nError: {error_text}")
+                    
+    except Exception as e:
+        logging.error(f"Error processing request: {e}")
+        await query.edit_message_text(f"❌ Error: {str(e)}")
+        
+    finally:
+        try:
+            if os.path.exists(image_path):
+                os.remove(image_path)
+                logging.info(f"Cleaned up image file: {image_path}")
+        except Exception as e:
+            logging.error(f"Error cleaning up file: {e}")
+
+def main():
+    app = Application.builder().token(TOKEN).build()
+    app.add_handler(CommandHandler("start", start))
+    app.add_handler(CommandHandler("instructions", instructions))
+    app.add_handler(CommandHandler("list", list_categories))
+    app.add_handler(CommandHandler("stop", stop))
+    app.add_handler(MessageHandler(filters.PHOTO, handle_image))
+    app.add_handler(CallbackQueryHandler(handle_view, pattern="^view:"))
+    app.add_handler(CallbackQueryHandler(handle_category, pattern="^category:"))
+    app.run_polling()
+
+if __name__ == "__main__":
+    main()