Expert-level livestock management, animal health monitoring, breeding programs, and ranch management
Expert guidance for livestock management, animal health monitoring, breeding programs, feed optimization, and ranch operations.
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
from enum import Enum
class AnimalType(Enum):
CATTLE = "cattle"
SHEEP = "sheep"
GOAT = "goat"
PIG = "pig"
POULTRY = "poultry"
class HealthStatus(Enum):
HEALTHY = "healthy"
OBSERVATION = "observation"
SICK = "sick"
QUARANTINE = "quarantine"
DECEASED = "deceased"
@dataclass
class Animal:
animal_id: str
tag_number: str
type: AnimalType
breed: str
sex: str
birth_date: datetime
weight_kg: float
sire_id: Optional[str]
dam_id: Optional[str]
health_status: HealthStatus
location: str
vaccinations: List[dict]
treatments: List[dict]
@dataclass
class HealthRecord:
record_id: str
animal_id: str
date: datetime
type: str # 'vaccination', 'treatment', 'check-up'
diagnosis: Optional[str]
treatment: Optional[str]
veterinarian_id: Optional[str]
notes: str
follow_up_date: Optional[datetime]
class LivestockManagement:
"""Livestock management system"""
def __init__(self, db):
self.db = db
def register_animal(self, animal_data):
"""Register new animal in system"""
animal = Animal(**animal_data)
# Generate unique tag if not provided
if not animal.tag_number:
animal.tag_number = self.generate_tag_number(animal.type)
# Create initial health record
health_record = HealthRecord(
record_id=generate_id(),
animal_id=animal.animal_id,
date=datetime.now(),
type='registration',
diagnosis=None,
treatment=None,
veterinarian_id=None,
notes='Initial registration',
follow_up_date=None
)
self.db.save_animal(animal)
self.db.save_health_record(health_record)
return animal
def monitor_animal_health(self, animal_id):
"""Monitor individual animal health"""
animal = self.db.get_animal(animal_id)
sensor_data = self.get_sensor_data(animal_id)
health_indicators = {
'temperature': sensor_data.get('temperature'),
'activity_level': sensor_data.get('activity_score'),
'rumination_time': sensor_data.get('rumination_minutes'), # For ruminants
'feeding_behavior': self.analyze_feeding_pattern(animal_id),
'weight_change': self.calculate_weight_trend(animal_id)
}
# Detect health issues
alerts = []
if health_indicators['temperature'] > 39.5: # Cattle normal: 38.5-39.5°C
alerts.append({
'severity': 'high',
'issue': 'Elevated temperature - possible fever',
'recommendation': 'Veterinary examination recommended'
})
if health_indicators['activity_level'] < 0.5: # Below 50% of normal
alerts.append({
'severity': 'medium',
'issue': 'Reduced activity',
'recommendation': 'Monitor closely, check for injury or illness'
})
return {
'animal_id': animal_id,
'tag_number': animal.tag_number,
'health_indicators': health_indicators,
'alerts': alerts,
'health_score': self.calculate_health_score(health_indicators)
}
def schedule_vaccinations(self, herd_id):
"""Generate vaccination schedule for herd"""
animals = self.db.get_herd_animals(herd_id)
vaccination_schedule = []
for animal in animals:
# Check vaccination history
last_vaccinations = self.db.get_vaccinations(animal.animal_id)
# Required vaccinations based on animal type and age
required_vaccines = self.get_required_vaccines(animal)
for vaccine in required_vaccines:
last_admin = next(
(v for v in last_vaccinations if v['vaccine'] == vaccine['name']),
None
)
# Check if due
if not last_admin or self.is_vaccine_due(last_admin, vaccine):
vaccination_schedule.append({
'animal_id': animal.animal_id,
'tag_number': animal.tag_number,
'vaccine': vaccine['name'],
'due_date': self.calculate_vaccine_due_date(last_admin, vaccine),
'priority': vaccine['priority']
})
# Sort by priority and due date
vaccination_schedule.sort(key=lambda x: (x['priority'], x['due_date']))
return vaccination_schedule
class BreedingManagement:
"""Breeding program management"""
def select_breeding_pairs(self, herd_id, breeding_goals):
"""Select optimal breeding pairs"""
eligible_males = self.db.get_breeding_males(herd_id)
eligible_females = self.db.get_breeding_females(herd_id)
# Score each potential pairing
breeding_recommendations = []
for female in eligible_females:
scores = []
for male in eligible_males:
# Check genetic compatibility
if self.are_related(male, female, max_generations=3):
continue # Skip closely related animals
# Calculate breeding value
score = self.calculate_breeding_value(
male,
female,
breeding_goals
)
scores.append({
'male_id': male.animal_id,
'male_tag': male.tag_number,
'score': score,
'expected_traits': self.predict_offspring_traits(male, female)
})
# Get best male for this female
if scores:
best_match = max(scores, key=lambda x: x['score'])
breeding_recommendations.append({
'female_id': female.animal_id,
'female_tag': female.tag_number,
'recommended_male': best_match,
'optimal_breeding_date': self.calculate_optimal_breeding_date(female)
})
return breeding_recommendations
def calculate_breeding_value(self, male, female, goals):
"""Calculate breeding value for pair"""
score = 0
# Evaluate based on breeding goals
if 'milk_production' in goals:
score += (male.genetics['milk_yield'] + female.genetics['milk_yield']) * 0.3
if 'growth_rate' in goals:
score += (male.genetics['growth_rate'] + female.genetics['growth_rate']) * 0.3
if 'disease_resistance' in goals:
score += (male.genetics['disease_resistance'] + female.genetics['disease_resistance']) * 0.2
if 'fertility' in goals:
score += (male.fertility_score + female.fertility_score) * 0.2
return score
def track_pregnancy(self, animal_id):
"""Track pregnancy and predict due date"""
animal = self.db.get_animal(animal_id)
breeding_record = self.db.get_last_breeding(animal_id)
if not breeding_record:
return {'status': 'not_pregnant'}
# Check pregnancy status
pregnancy_check = self.db.get_latest_pregnancy_check(animal_id)
if pregnancy_check and pregnancy_check['confirmed']:
gestation_period = self.get_gestation_period(animal.type)
due_date = breeding_record['date'] + timedelta(days=gestation_period)
days_pregnant = (datetime.now() - breeding_record['date']).days
# Schedule checkups
checkup_schedule = self.generate_pregnancy_checkups(
breeding_record['date'],
due_date
)
return {
'status': 'pregnant',
'breeding_date': breeding_record['date'],
'due_date': due_date,
'days_pregnant': days_pregnant,
'days_remaining': (due_date - datetime.now()).days,
'checkup_schedule': checkup_schedule
}
return {'status': 'unknown', 'needs_pregnancy_check': True}
class FeedManagement:
"""Feed optimization and management"""
def calculate_feed_requirements(self, animal_id):
"""Calculate nutritional requirements"""
animal = self.db.get_animal(animal_id)
# Base requirements on:
# - Weight
# - Age
# - Production status (lactating, pregnant, growing)
# - Activity level
requirements = {
'dry_matter_kg': self.calculate_dm_requirement(animal),
'crude_protein_kg': self.calculate_protein_requirement(animal),
'energy_mcal': self.calculate_energy_requirement(animal),
'minerals': self.calculate_mineral_requirements(animal)
}
return requirements
def optimize_feed_ration(self, herd_id):
"""Optimize feed ration for herd"""
animals = self.db.get_herd_animals(herd_id)
available_feeds = self.db.get_available_feeds()
# Group animals by similar requirements
groups = self.group_animals_by_requirements(animals)
ration_plans = []
for group in groups:
# Linear programming for least-cost ration
avg_requirements = self.calculate_group_requirements(group['animals'])
optimal_ration = self.solve_ration_optimization(
avg_requirements,
available_feeds
)
ration_plans.append({
'group_id': group['id'],
'animal_count': len(group['animals']),
'ration': optimal_ration,
'daily_cost_per_animal': sum(
ingredient['amount'] * ingredient['cost_per_kg']
for ingredient in optimal_ration
),
'meets_requirements': True
})
return ration_plans
def monitor_feeding_behavior(self, animal_id):
"""Monitor feeding patterns"""
feeding_data = self.db.get_feeding_data(animal_id, days=7)
analysis = {
'avg_daily_intake_kg': np.mean([d['intake'] for d in feeding_data]),
'feeding_frequency': len(feeding_data) / 7,
'intake_variation': np.std([d['intake'] for d in feeding_data]),
'eating_time_minutes': np.mean([d['duration'] for d in feeding_data])
}
# Detect issues
if analysis['avg_daily_intake_kg'] < expected_intake * 0.8:
analysis['alert'] = 'Low feed intake - possible health issue'
return analysis
class HerdAnalytics:
"""Herd performance analytics"""
def analyze_herd_performance(self, herd_id):
"""Comprehensive herd performance analysis"""
animals = self.db.get_herd_animals(herd_id)
metrics = {
'total_animals': len(animals),
'avg_weight': np.mean([a.weight_kg for a in animals]),
'avg_age_months': np.mean([
(datetime.now() - a.birth_date).days / 30
for a in animals
]),
'health_status_distribution': self.get_health_distribution(animals),
'mortality_rate': self.calculate_mortality_rate(herd_id),
'reproduction_rate': self.calculate_reproduction_rate(herd_id),
'avg_daily_gain': self.calculate_avg_daily_gain(herd_id),
'feed_conversion_ratio': self.calculate_fcr(herd_id)
}
# Financial metrics
metrics['production_value'] = self.calculate_production_value(herd_id)
metrics['feed_cost'] = self.calculate_total_feed_cost(herd_id)
metrics['veterinary_cost'] = self.calculate_vet_costs(herd_id)
metrics['profit_per_animal'] = (
metrics['production_value'] -
metrics['feed_cost'] -
metrics['veterinary_cost']
) / metrics['total_animals']
return metrics
def predict_production(self, animal_id, days_ahead=30):
"""Predict animal production (milk, eggs, etc.)"""
animal = self.db.get_animal(animal_id)
historical_production = self.db.get_production_history(animal_id, days=90)
# Use time series model
forecast = self.production_model.forecast(
historical_production,
periods=days_ahead
)
return {
'animal_id': animal_id,
'forecast_period_days': days_ahead,
'predicted_production': forecast.tolist(),
'total_predicted': sum(forecast),
'confidence_interval': self.calculate_confidence_interval(forecast)
}
❌ Poor record keeping ❌ No biosecurity measures ❌ Reactive health management ❌ Inbreeding ❌ Over or underfeeding ❌ Ignoring animal behavior ❌ Manual data collection only