Livestock Management System

backend/api/v1/__init__.py

@@ -51,6 +51,7 @@
 from .carbon import carbon_bp
 from .logistics import smart_freight_bp
 from .carbon_v2 import carbon_v2_bp
+from .livestock import livestock_bp
 from .crop_advisory import advisory_bp
 from .irrigation_v2 import irrigation_v2_bp
 
@@ -110,6 +111,7 @@
 api_v1.register_blueprint(spatial_yield_bp, url_prefix="/spatial-yield")
 api_v1.register_blueprint(carbon_bp, url_prefix="/carbon")
 api_v1.register_blueprint(carbon_v2_bp, url_prefix="/carbon-v2")
+api_v1.register_blueprint(livestock_bp, url_prefix="/livestock")
 api_v1.register_blueprint(gov_schemes_bp, url_prefix="/government-schemes")
 api_v1.register_blueprint(advisory_bp, url_prefix="/crop-advisory")
 api_v1.register_blueprint(gews_bp, url_prefix='/gews')

backend/api/v1/livestock.py

@@ -0,0 +1,155 @@
+from flask import Blueprint, request, jsonify
+from backend.services.livestock_service import LivestockService
+from auth_utils import token_required
+
+livestock_bp = Blueprint("livestock", __name__)
+
+
+@livestock_bp.route("/animals", methods=["POST"])
+@token_required
+def add_animal(current_user):
+    data = request.get_json()
+    if not data or "farm_id" not in data or "animal_type" not in data:
+        return jsonify({"status": "error", "message": "Missing required fields"}), 400
+
+    data["user_id"] = current_user.id
+    animal = LivestockService.add_animal(data)
+
+    return jsonify(
+        {
+            "status": "success",
+            "message": "Animal added successfully",
+            "data": animal.to_dict(),
+        }
+    ), 201
+
+
+@livestock_bp.route("/animals/<int:farm_id>", methods=["GET"])
+@token_required
+def get_animals(current_user, farm_id):
+    animals = LivestockService.get_farm_animals(farm_id)
+
+    return jsonify({"status": "success", "data": [a.to_dict() for a in animals]})
+
+
+@livestock_bp.route("/health-record", methods=["POST"])
+@token_required
+def add_health_record(current_user):
+    data = request.get_json()
+    if not data or "animal_id" not in data or "health_status" not in data:
+        return jsonify({"status": "error", "message": "Missing required fields"}), 400
+
+    record = LivestockService.add_health_record(data)
+
+    return jsonify(
+        {
+            "status": "success",
+            "message": "Health record added successfully",
+            "data": record.to_dict(),
+        }
+    ), 201
+
+
+@livestock_bp.route("/health-history/<int:animal_id>", methods=["GET"])
+@token_required
+def get_health_history(current_user, animal_id):
+    history = LivestockService.get_animal_health_history(animal_id)
+
+    return jsonify({"status": "success", "data": [r.to_dict() for r in history]})
+
+
+@livestock_bp.route("/breeding", methods=["POST"])
+@token_required
+def record_breeding(current_user):
+    data = request.get_json()
+    if not data or "farm_id" not in data or "breeding_date" not in data:
+        return jsonify({"status": "error", "message": "Missing required fields"}), 400
+
+    breeding = LivestockService.record_breeding(data)
+
+    return jsonify(
+        {
+            "status": "success",
+            "message": "Breeding recorded successfully",
+            "data": breeding.to_dict(),
+        }
+    ), 201
+
+
+@livestock_bp.route("/breeding/<int:farm_id>", methods=["GET"])
+@token_required
+def get_breeding_records(current_user, farm_id):
+    records = LivestockService.get_farm_breeding_records(farm_id)
+
+    return jsonify({"status": "success", "data": [r.to_dict() for r in records]})
+
+
+@livestock_bp.route("/production", methods=["POST"])
+@token_required
+def record_production(current_user):
+    data = request.get_json()
+    if (
+        not data
+        or "animal_id" not in data
+        or "production_type" not in data
+        or "amount" not in data
+    ):
+        return jsonify({"status": "error", "message": "Missing required fields"}), 400
+
+    production = LivestockService.record_production(data)
+
+    return jsonify(
+        {
+            "status": "success",
+            "message": "Production recorded successfully",
+            "data": production.to_dict(),
+        }
+    ), 201
+
+
+@livestock_bp.route("/production/<int:animal_id>", methods=["GET"])
+@token_required
+def get_production(current_user, animal_id):
+    production = LivestockService.get_animal_production(animal_id)
+
+    return jsonify({"status": "success", "data": [p.to_dict() for p in production]})
+
+
+@livestock_bp.route("/feed-plan", methods=["POST"])
+@token_required
+def create_feed_plan(current_user):
+    data = request.get_json()
+    if (
+        not data
+        or "farm_id" not in data
+        or "animal_type" not in data
+        or "feed_type" not in data
+        or "daily_quantity" not in data
+    ):
+        return jsonify({"status": "error", "message": "Missing required fields"}), 400
+
+    plan = LivestockService.create_feed_plan(data)
+
+    return jsonify(
+        {
+            "status": "success",
+            "message": "Feed plan created successfully",
+            "data": plan.to_dict(),
+        }
+    ), 201
+
+
+@livestock_bp.route("/feed-plans/<int:farm_id>", methods=["GET"])
+@token_required
+def get_feed_plans(current_user, farm_id):
+    plans = LivestockService.get_farm_feed_plans(farm_id)
+
+    return jsonify({"status": "success", "data": [p.to_dict() for p in plans]})
+
+
+@livestock_bp.route("/summary/<int:farm_id>", methods=["GET"])
+@token_required
+def get_summary(current_user, farm_id):
+    summary = LivestockService.get_livestock_summary(farm_id)
+
+    return jsonify({"status": "success", "data": summary})

backend/models/livestock.py

@@ -0,0 +1,178 @@
+from datetime import datetime
+from backend.extensions import db
+
+
+class LivestockAnimal(db.Model):
+    __tablename__ = "livestock_animals"
+
+    id = db.Column(db.Integer, primary_key=True)
+    farm_id = db.Column(db.Integer, db.ForeignKey("farms.id"), nullable=False)
+    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
+
+    animal_type = db.Column(db.String(50), nullable=False)
+    breed = db.Column(db.String(100))
+    tag_number = db.Column(db.String(100), unique=True)
+
+    date_of_birth = db.Column(db.Date)
+    gender = db.Column(db.String(20))
+    weight = db.Column(db.Float)
+
+    status = db.Column(db.String(50), default="healthy")
+    location = db.Column(db.String(200))
+
+    image = db.Column(db.String(500))
+
+    created_at = db.Column(db.DateTime, default=datetime.utcnow)
+    updated_at = db.Column(
+        db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
+    )
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "farm_id": self.farm_id,
+            "user_id": self.user_id,
+            "animal_type": self.animal_type,
+            "breed": self.breed,
+            "tag_number": self.tag_number,
+            "date_of_birth": self.date_of_birth.isoformat()
+            if self.date_of_birth
+            else None,
+            "gender": self.gender,
+            "weight": self.weight,
+            "status": self.status,
+            "location": self.location,
+            "image": self.image,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }
+
+
+class LivestockHealthRecord(db.Model):
+    __tablename__ = "livestock_health_records"
+
+    id = db.Column(db.Integer, primary_key=True)
+    animal_id = db.Column(
+        db.Integer, db.ForeignKey("livestock_animals.id"), nullable=False
+    )
+
+    checkup_date = db.Column(db.DateTime, default=datetime.utcnow)
+    health_status = db.Column(db.String(50), nullable=False)
+    symptoms = db.Column(db.Text)
+    diagnosis = db.Column(db.Text)
+    treatment = db.Column(db.Text)
+    veterinarian = db.Column(db.String(200))
+    notes = db.Column(db.Text)
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "animal_id": self.animal_id,
+            "checkup_date": self.checkup_date.isoformat(),
+            "health_status": self.health_status,
+            "symptoms": self.symptoms,
+            "diagnosis": self.diagnosis,
+            "treatment": self.treatment,
+            "veterinarian": self.veterinarian,
+            "notes": self.notes,
+        }
+
+
+class LivestockBreeding(db.Model):
+    __tablename__ = "livestock_breeding"
+
+    id = db.Column(db.Integer, primary_key=True)
+    farm_id = db.Column(db.Integer, db.ForeignKey("farms.id"), nullable=False)
+
+    male_animal_id = db.Column(db.Integer, db.ForeignKey("livestock_animals.id"))
+    female_animal_id = db.Column(db.Integer, db.ForeignKey("livestock_animals.id"))
+
+    breeding_date = db.Column(db.Date, nullable=False)
+    expected_birth_date = db.Column(db.Date)
+    actual_birth_date = db.Column(db.Date)
+
+    offspring_count = db.Column(db.Integer, default=0)
+    status = db.Column(db.String(50), default="in_progress")
+    notes = db.Column(db.Text)
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "farm_id": self.farm_id,
+            "male_animal_id": self.male_animal_id,
+            "female_animal_id": self.female_animal_id,
+            "breeding_date": self.breeding_date.isoformat()
+            if self.breeding_date
+            else None,
+            "expected_birth_date": self.expected_birth_date.isoformat()
+            if self.expected_birth_date
+            else None,
+            "actual_birth_date": self.actual_birth_date.isoformat()
+            if self.actual_birth_date
+            else None,
+            "offspring_count": self.offspring_count,
+            "status": self.status,
+            "notes": self.notes,
+        }
+
+
+class LivestockProduction(db.Model):
+    __tablename__ = "livestock_production"
+
+    id = db.Column(db.Integer, primary_key=True)
+    animal_id = db.Column(
+        db.Integer, db.ForeignKey("livestock_animals.id"), nullable=False
+    )
+
+    production_type = db.Column(db.String(50), nullable=False)
+    amount = db.Column(db.Float, nullable=False)
+    unit = db.Column(db.String(50))
+
+    record_date = db.Column(db.DateTime, default=datetime.utcnow)
+    notes = db.Column(db.Text)
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "animal_id": self.animal_id,
+            "production_type": self.production_type,
+            "amount": self.amount,
+            "unit": self.unit,
+            "record_date": self.record_date.isoformat(),
+            "notes": self.notes,
+        }
+
+
+class LivestockFeedPlan(db.Model):
+    __tablename__ = "livestock_feed_plans"
+
+    id = db.Column(db.Integer, primary_key=True)
+    farm_id = db.Column(db.Integer, db.ForeignKey("farms.id"), nullable=False)
+    animal_type = db.Column(db.String(50), nullable=False)
+
+    feed_type = db.Column(db.String(100), nullable=False)
+    daily_quantity = db.Column(db.Float, nullable=False)
+    unit = db.Column(db.String(50))
+
+    cost_per_unit = db.Column(db.Float)
+    feeding_frequency = db.Column(db.String(50))
+
+    start_date = db.Column(db.Date, nullable=False)
+    end_date = db.Column(db.Date)
+
+    notes = db.Column(db.Text)
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "farm_id": self.farm_id,
+            "animal_type": self.animal_type,
+            "feed_type": self.feed_type,
+            "daily_quantity": self.daily_quantity,
+            "unit": self.unit,
+            "cost_per_unit": self.cost_per_unit,
+            "feeding_frequency": self.feeding_frequency,
+            "start_date": self.start_date.isoformat() if self.start_date else None,
+            "end_date": self.end_date.isoformat() if self.end_date else None,
+            "notes": self.notes,
+        }