pranit144's picture
Upload 56 files
429a26d verified
from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
from datetime import datetime, date
from werkzeug.security import generate_password_hash, check_password_hash
import json
# Initialize the SQLAlchemy db instance for the app to import
db = SQLAlchemy()
class Farmer(UserMixin, db.Model):
__tablename__ = 'farmers'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(200), nullable=False)
age = db.Column(db.Integer, nullable=True)
gender = db.Column(db.String(20), nullable=True)
aadhaar_id = db.Column(db.String(50), unique=True, nullable=False)
contact_number = db.Column(db.String(50), nullable=True)
address = db.Column(db.Text, nullable=True)
password_hash = db.Column(db.String(255), nullable=False)
telegram_chat_id = db.Column(db.String(100), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
farms = db.relationship('Farm', backref='owner', lazy=True)
def set_password(self, password: str):
self.password_hash = generate_password_hash(password)
def check_password(self, password: str) -> bool:
if not self.password_hash:
return False
return check_password_hash(self.password_hash, password)
def __repr__(self):
return f'<Farmer {self.id} {self.name}>'
def as_dict(self):
return {
'id': self.id,
'name': self.name,
'age': self.age,
'gender': self.gender,
'aadhaar_id': self.aadhaar_id,
'contact_number': self.contact_number,
'address': self.address,
'telegram_chat_id': self.telegram_chat_id,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class Farm(db.Model):
__tablename__ = 'farms'
id = db.Column(db.Integer, primary_key=True)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
farm_name = db.Column(db.String(200), nullable=False)
farm_size = db.Column(db.Float, nullable=True) # acres
# Multi-sector support
farm_type = db.Column(db.String(50), default='crop') # crop, dairy, poultry, livestock, fishery, mixed
# Crop farming fields
irrigation_type = db.Column(db.String(100), nullable=True)
crop_types = db.Column(db.Text, nullable=True) # comma-separated names for quick display
crop_details = db.Column(db.Text, nullable=True) # JSON string for detailed crop info
# Livestock/Dairy/Poultry fields
livestock_types = db.Column(db.Text, nullable=True) # JSON string: cows, buffaloes, goats, etc.
livestock_count = db.Column(db.Integer, nullable=True) # Total number of animals
housing_type = db.Column(db.String(100), nullable=True) # shed, barn, free-range, cages
feeding_system = db.Column(db.String(100), nullable=True) # automatic, manual, grazing, mixed
breed_info = db.Column(db.Text, nullable=True) # JSON string for breed details
# Common fields
latitude = db.Column(db.Float, nullable=True)
longitude = db.Column(db.Float, nullable=True)
field_coordinates = db.Column(db.Text, nullable=True) # JSON string of coordinates
weather_alerts_enabled = db.Column(db.Boolean, default=True, nullable=False) # Enable/disable weather alerts
created_at = db.Column(db.DateTime, default=datetime.utcnow)
soil_data = db.relationship('SoilData', backref='farm', lazy=True)
def set_crop_types(self, types_list):
try:
if isinstance(types_list, (list, tuple)):
self.crop_types = ','.join([str(t).strip() for t in types_list if t])
else:
self.crop_types = str(types_list)
except Exception:
self.crop_types = None
def get_crop_types(self):
if not self.crop_types:
return []
return [t.strip() for t in self.crop_types.split(',') if t.strip()]
def set_crop_details(self, details):
try:
self.crop_details = json.dumps(details or [])
except Exception:
self.crop_details = json.dumps([])
def get_crop_details(self):
try:
return json.loads(self.crop_details) if self.crop_details else []
except Exception:
return []
def set_livestock_types(self, livestock_data):
"""Set livestock types and details as JSON"""
try:
self.livestock_types = json.dumps(livestock_data or [])
except Exception:
self.livestock_types = json.dumps([])
def get_livestock_types(self):
"""Get livestock types and details as list"""
try:
return json.loads(self.livestock_types) if self.livestock_types else []
except Exception:
return []
def set_breed_info(self, breed_data):
"""Set breed information as JSON"""
try:
self.breed_info = json.dumps(breed_data or {})
except Exception:
self.breed_info = json.dumps({})
def get_breed_info(self):
"""Get breed information as dictionary"""
try:
return json.loads(self.breed_info) if self.breed_info else {}
except Exception:
return {}
def get_farm_type_display(self):
"""Get user-friendly farm type display name"""
type_mapping = {
'crop': 'Crop Farming',
'dairy': 'Dairy Farming',
'poultry': 'Poultry Farming',
'livestock': 'Livestock Farming',
'fishery': 'Fish Farming',
'mixed': 'Mixed Farming'
}
return type_mapping.get(self.farm_type, 'Crop Farming')
def __repr__(self):
return f'<Farm {self.id} {self.farm_name}>'
def as_dict(self):
return {
'id': self.id,
'farmer_id': self.farmer_id,
'farm_name': self.farm_name,
'farm_size': self.farm_size,
'farm_type': self.farm_type,
'farm_type_display': self.get_farm_type_display(),
'irrigation_type': self.irrigation_type,
'latitude': self.latitude,
'longitude': self.longitude,
'field_coordinates': json.loads(self.field_coordinates) if self.field_coordinates else None,
'crop_types': self.get_crop_types(),
'crop_details': self.get_crop_details(),
'livestock_types': self.get_livestock_types(),
'livestock_count': self.livestock_count,
'housing_type': self.housing_type,
'feeding_system': self.feeding_system,
'breed_info': self.get_breed_info(),
'weather_alerts_enabled': self.weather_alerts_enabled,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class SoilData(db.Model):
__tablename__ = 'soil_data'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
soil_type = db.Column(db.String(200), nullable=True)
ph_level = db.Column(db.Float, nullable=True)
nitrogen_level = db.Column(db.Float, nullable=True)
phosphorus_level = db.Column(db.Float, nullable=True)
potassium_level = db.Column(db.Float, nullable=True)
moisture_percentage = db.Column(db.Float, nullable=True)
# NOTE: some deployments may have older schema without recorded_at column.
# Keep as optional attribute access in as_dict to avoid OperationalError.
def __repr__(self):
return f'<SoilData {self.id} farm={self.farm_id}>'
def as_dict(self):
rec_at = getattr(self, 'recorded_at', None)
return {
'id': self.id,
'farm_id': self.farm_id,
'soil_type': self.soil_type,
'ph_level': self.ph_level,
'nitrogen_level': self.nitrogen_level,
'phosphorus_level': self.phosphorus_level,
'potassium_level': self.potassium_level,
'moisture_percentage': self.moisture_percentage,
'recorded_at': rec_at.isoformat() if rec_at else None
}
class FarmingActivity(db.Model):
__tablename__ = 'farming_activities'
id = db.Column(db.Integer, primary_key=True)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=True)
activity_type = db.Column(db.String(200), nullable=True)
scheduled_date = db.Column(db.DateTime, nullable=True)
notes = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def __repr__(self):
return f'<FarmingActivity {self.id} {self.activity_type}>'
def as_dict(self):
return {
'id': self.id,
'farmer_id': self.farmer_id,
'farm_id': self.farm_id,
'activity_type': self.activity_type,
'scheduled_date': self.scheduled_date.isoformat() if self.scheduled_date else None,
'notes': self.notes,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class WeatherData(db.Model):
__tablename__ = 'weather_data'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=True)
date = db.Column(db.Date, default=date.today)
data = db.Column(db.Text, nullable=True) # JSON blob of weather info
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def get_data(self):
try:
return json.loads(self.data) if self.data else {}
except Exception:
return {}
def __repr__(self):
return f'<WeatherData {self.id} farm={self.farm_id} date={self.date}>'
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'date': self.date.isoformat() if self.date else None,
'data': self.get_data(),
'created_at': self.created_at.isoformat() if self.created_at else None
}
class DailyAdvisory(db.Model):
__tablename__ = 'daily_advisories'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
date = db.Column(db.Date, nullable=False)
# Updated columns to match the existing database schema
task_to_do = db.Column(db.Text, nullable=True)
task_to_avoid = db.Column(db.Text, nullable=True)
reason_explanation = db.Column(db.Text, nullable=True)
crop_stage = db.Column(db.String(50), nullable=True)
weather_context = db.Column(db.Text, nullable=True) # JSON blob
gemini_response = db.Column(db.Text, nullable=True) # raw AI response JSON
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def __repr__(self):
return f'<DailyAdvisory {self.id} farm={self.farm_id} date={self.date}>'
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'date': self.date.isoformat() if self.date else None,
'task_to_do': self.task_to_do,
'task_to_avoid': self.task_to_avoid,
'reason_explanation': self.reason_explanation,
'crop_stage': self.crop_stage,
'weather_context': json.loads(self.weather_context) if self.weather_context else None,
'gemini_response': json.loads(self.gemini_response) if self.gemini_response else None,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class SMSLog(db.Model):
__tablename__ = 'sms_logs'
id = db.Column(db.Integer, primary_key=True)
# Adjusted to match the current DB schema in instance/farm_management.db
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=True)
phone_number = db.Column(db.String(15), nullable=True)
message_content = db.Column(db.Text, nullable=True)
twilio_sid = db.Column(db.String(100), nullable=True)
status = db.Column(db.String(20), nullable=True)
sent_at = db.Column(db.DateTime, default=datetime.utcnow)
delivered_at = db.Column(db.DateTime, nullable=True)
error_message = db.Column(db.Text, nullable=True)
def as_dict(self):
return {
'id': self.id,
'farmer_id': self.farmer_id,
'phone_number': self.phone_number,
'message_content': self.message_content,
'twilio_sid': self.twilio_sid,
'status': self.status,
'sent_at': self.sent_at.isoformat() if self.sent_at else None,
'delivered_at': self.delivered_at.isoformat() if self.delivered_at else None,
'error_message': self.error_message
}
def __repr__(self):
return f'<SMSLog {self.id} to={self.recipient} status={self.status}>'
class AdminUser(db.Model):
__tablename__ = 'admin_users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(150), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
last_login = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def set_password(self, password: str):
self.password_hash = generate_password_hash(password)
def check_password(self, password: str) -> bool:
if not self.password_hash:
return False
return check_password_hash(self.password_hash, password)
def __repr__(self):
return f'<AdminUser {self.username}>'
class YearlyPlan(db.Model):
__tablename__ = 'yearly_plans'
id = db.Column(db.Integer, primary_key=True)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
year = db.Column(db.Integer, nullable=False)
plan_json = db.Column(db.Text, nullable=True) # JSON blob of the full plan
summary_text = db.Column(db.Text, nullable=True) # short text summary for messages
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def as_dict(self):
try:
return {
'id': self.id,
'farmer_id': self.farmer_id,
'year': self.year,
'plan': json.loads(self.plan_json) if self.plan_json else {},
'summary': self.summary_text,
'created_at': self.created_at.isoformat() if self.created_at else None
}
except Exception:
return {
'id': self.id,
'farmer_id': self.farmer_id,
'year': self.year,
'plan': {},
'summary': self.summary_text,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class WeatherAlert(db.Model):
__tablename__ = 'weather_alerts'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
alert_type = db.Column(db.String(100), nullable=False) # 'rain', 'storm', 'drought', 'frost', 'heat_wave'
severity = db.Column(db.String(50), nullable=False) # 'low', 'medium', 'high', 'critical'
title = db.Column(db.String(200), nullable=False)
message = db.Column(db.Text, nullable=False)
recommended_action = db.Column(db.Text, nullable=True)
weather_data = db.Column(db.Text, nullable=True) # JSON string
is_active = db.Column(db.Boolean, default=True)
is_sent = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
expires_at = db.Column(db.DateTime, nullable=True)
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'alert_type': self.alert_type,
'severity': self.severity,
'title': self.title,
'message': self.message,
'recommended_action': self.recommended_action,
'is_active': self.is_active,
'is_sent': self.is_sent,
'created_at': self.created_at.isoformat() if self.created_at else None,
'expires_at': self.expires_at.isoformat() if self.expires_at else None
}
class MarketPrice(db.Model):
__tablename__ = 'market_prices'
id = db.Column(db.Integer, primary_key=True)
crop_name = db.Column(db.String(100), nullable=False)
market_name = db.Column(db.String(200), nullable=False)
state = db.Column(db.String(100), nullable=False)
district = db.Column(db.String(100), nullable=False)
min_price = db.Column(db.Float, nullable=True)
max_price = db.Column(db.Float, nullable=True)
avg_price = db.Column(db.Float, nullable=False)
price_unit = db.Column(db.String(50), default='per quintal')
price_date = db.Column(db.Date, nullable=False)
source = db.Column(db.String(100), nullable=True) # 'government', 'market_api', 'manual'
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def as_dict(self):
return {
'id': self.id,
'crop_name': self.crop_name,
'market_name': self.market_name,
'state': self.state,
'district': self.district,
'min_price': self.min_price,
'max_price': self.max_price,
'avg_price': self.avg_price,
'price_unit': self.price_unit,
'price_date': self.price_date.isoformat() if self.price_date else None,
'source': self.source,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class DiseaseDetection(db.Model):
__tablename__ = 'disease_detections'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
crop_name = db.Column(db.String(100), nullable=False)
disease_name = db.Column(db.String(200), nullable=True)
confidence_score = db.Column(db.Float, nullable=True)
symptoms = db.Column(db.Text, nullable=True)
treatment = db.Column(db.Text, nullable=True)
prevention = db.Column(db.Text, nullable=True)
image_path = db.Column(db.String(500), nullable=True)
ai_analysis = db.Column(db.Text, nullable=True) # JSON string
status = db.Column(db.String(50), default='detected') # 'detected', 'treating', 'resolved'
severity = db.Column(db.String(50), nullable=True) # 'mild', 'moderate', 'severe'
detected_at = db.Column(db.DateTime, default=datetime.utcnow)
resolved_at = db.Column(db.DateTime, nullable=True)
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'crop_name': self.crop_name,
'disease_name': self.disease_name,
'confidence_score': self.confidence_score,
'symptoms': self.symptoms,
'treatment': self.treatment,
'prevention': self.prevention,
'status': self.status,
'severity': self.severity,
'detected_at': self.detected_at.isoformat() if self.detected_at else None,
'resolved_at': self.resolved_at.isoformat() if self.resolved_at else None
}
class DailyTask(db.Model):
__tablename__ = 'daily_tasks'
id = db.Column(db.Integer, primary_key=True)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=True)
task_date = db.Column(db.Date, nullable=False, default=date.today)
task_type = db.Column(db.String(100), nullable=False) # 'irrigation', 'fertilizing', 'pest_control', etc.
task_title = db.Column(db.String(200), nullable=False)
task_description = db.Column(db.Text, nullable=False)
priority = db.Column(db.String(20), default='medium') # 'high', 'medium', 'low'
estimated_duration = db.Column(db.Integer, nullable=True) # in minutes
weather_dependent = db.Column(db.Boolean, default=False)
crop_specific = db.Column(db.String(100), nullable=True)
# Completion tracking
is_completed = db.Column(db.Boolean, default=False)
completed_at = db.Column(db.DateTime, nullable=True)
completion_notes = db.Column(db.Text, nullable=True)
completion_rating = db.Column(db.Integer, nullable=True) # 1-5 rating from farmer
# Notification tracking
sent_via_telegram = db.Column(db.Boolean, default=False)
telegram_sent_at = db.Column(db.DateTime, nullable=True)
reminder_sent = db.Column(db.Boolean, default=False)
reminder_sent_at = db.Column(db.DateTime, nullable=True)
# Meta
created_at = db.Column(db.DateTime, default=datetime.utcnow)
created_by = db.Column(db.String(50), default='system') # 'system', 'ai', 'manual'
def as_dict(self):
return {
'id': self.id,
'farmer_id': self.farmer_id,
'farm_id': self.farm_id,
'task_date': self.task_date.isoformat() if self.task_date else None,
'task_type': self.task_type,
'task_title': self.task_title,
'task_description': self.task_description,
'priority': self.priority,
'estimated_duration': self.estimated_duration,
'weather_dependent': self.weather_dependent,
'crop_specific': self.crop_specific,
'is_completed': self.is_completed,
'completed_at': self.completed_at.isoformat() if self.completed_at else None,
'completion_notes': self.completion_notes,
'completion_rating': self.completion_rating,
'sent_via_telegram': self.sent_via_telegram,
'telegram_sent_at': self.telegram_sent_at.isoformat() if self.telegram_sent_at else None,
'reminder_sent': self.reminder_sent,
'reminder_sent_at': self.reminder_sent_at.isoformat() if self.reminder_sent_at else None,
'created_at': self.created_at.isoformat() if self.created_at else None,
'created_by': self.created_by
}
def __repr__(self):
return f'<DailyTask {self.id} {self.task_title} for {self.task_date}>'
class TaskCompletion(db.Model):
__tablename__ = 'task_completions'
id = db.Column(db.Integer, primary_key=True)
task_id = db.Column(db.Integer, db.ForeignKey('daily_tasks.id'), nullable=False)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
completed_at = db.Column(db.DateTime, default=datetime.utcnow)
completion_method = db.Column(db.String(50), default='dashboard') # 'dashboard', 'telegram', 'mobile'
completion_status = db.Column(db.String(50), default='completed') # 'completed', 'partially_completed', 'skipped'
notes = db.Column(db.Text, nullable=True)
rating = db.Column(db.Integer, nullable=True) # 1-5 rating
issues_faced = db.Column(db.Text, nullable=True)
time_taken = db.Column(db.Integer, nullable=True) # actual time taken in minutes
def as_dict(self):
return {
'id': self.id,
'task_id': self.task_id,
'farmer_id': self.farmer_id,
'completed_at': self.completed_at.isoformat() if self.completed_at else None,
'completion_method': self.completion_method,
'completion_status': self.completion_status,
'notes': self.notes,
'rating': self.rating,
'issues_faced': self.issues_faced,
'time_taken': self.time_taken
}
def __repr__(self):
return f'<TaskCompletion {self.id} for task {self.task_id}>'
# ==================== LIVESTOCK MANAGEMENT MODELS ====================
class LivestockRecord(db.Model):
"""Model for individual animal records in livestock farming"""
__tablename__ = 'livestock_records'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
# Animal identification
animal_id = db.Column(db.String(50), nullable=False) # Tag number or custom ID
animal_type = db.Column(db.String(50), nullable=False) # cow, buffalo, goat, chicken, fish
breed = db.Column(db.String(100), nullable=True)
gender = db.Column(db.String(10), nullable=True) # male, female
# Basic info
date_of_birth = db.Column(db.Date, nullable=True)
date_acquired = db.Column(db.Date, nullable=False, default=datetime.utcnow)
acquisition_method = db.Column(db.String(50), nullable=True) # purchased, born, gifted
current_weight = db.Column(db.Float, nullable=True) # in kg
# Health and breeding
vaccination_status = db.Column(db.Text, nullable=True) # JSON of vaccination records
health_status = db.Column(db.String(50), default='healthy') # healthy, sick, treatment
breeding_status = db.Column(db.String(50), nullable=True) # pregnant, lactating, dry
# Production tracking (for dairy, poultry, etc.)
milk_production_avg = db.Column(db.Float, nullable=True) # liters per day
egg_production_avg = db.Column(db.Integer, nullable=True) # eggs per day
# Status
is_active = db.Column(db.Boolean, default=True)
status = db.Column(db.String(50), default='active') # active, sold, deceased, transferred
notes = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'animal_id': self.animal_id,
'animal_type': self.animal_type,
'breed': self.breed,
'gender': self.gender,
'date_of_birth': self.date_of_birth.isoformat() if self.date_of_birth else None,
'date_acquired': self.date_acquired.isoformat() if self.date_acquired else None,
'current_weight': self.current_weight,
'health_status': self.health_status,
'breeding_status': self.breeding_status,
'milk_production_avg': self.milk_production_avg,
'egg_production_avg': self.egg_production_avg,
'is_active': self.is_active,
'status': self.status,
'created_at': self.created_at.isoformat() if self.created_at else None
}
def __repr__(self):
return f'<LivestockRecord {self.animal_id} - {self.animal_type}>'
class ProductionRecord(db.Model):
"""Model for tracking daily production (milk, eggs, etc.)"""
__tablename__ = 'production_records'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
livestock_id = db.Column(db.Integer, db.ForeignKey('livestock_records.id'), nullable=True)
# Production details
production_date = db.Column(db.Date, nullable=False, default=datetime.utcnow)
production_type = db.Column(db.String(50), nullable=False) # milk, eggs, meat, fish
quantity = db.Column(db.Float, nullable=False) # liters, pieces, kg
unit = db.Column(db.String(20), nullable=False) # liters, pieces, kg
quality_grade = db.Column(db.String(20), nullable=True) # A, B, C or premium, standard
# Economic data
unit_price = db.Column(db.Float, nullable=True) # price per unit
total_value = db.Column(db.Float, nullable=True) # total value
buyer_info = db.Column(db.String(200), nullable=True) # where sold
# Additional info
notes = db.Column(db.Text, nullable=True)
weather_conditions = db.Column(db.String(100), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'livestock_id': self.livestock_id,
'production_date': self.production_date.isoformat() if self.production_date else None,
'production_type': self.production_type,
'quantity': self.quantity,
'unit': self.unit,
'quality_grade': self.quality_grade,
'unit_price': self.unit_price,
'total_value': self.total_value,
'buyer_info': self.buyer_info,
'created_at': self.created_at.isoformat() if self.created_at else None
}
def __repr__(self):
return f'<ProductionRecord {self.production_type} - {self.quantity} {self.unit}>'
class FeedRecord(db.Model):
"""Model for tracking feed consumption and costs"""
__tablename__ = 'feed_records'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
# Feed details
feed_date = db.Column(db.Date, nullable=False, default=datetime.utcnow)
feed_type = db.Column(db.String(100), nullable=False) # grass, concentrate, grain, etc.
feed_name = db.Column(db.String(200), nullable=False) # specific feed name/brand
quantity = db.Column(db.Float, nullable=False) # in kg or specified unit
unit = db.Column(db.String(20), nullable=False, default='kg')
# Cost tracking
cost_per_unit = db.Column(db.Float, nullable=True)
total_cost = db.Column(db.Float, nullable=True)
supplier = db.Column(db.String(200), nullable=True)
# Target animals
target_animals = db.Column(db.Text, nullable=True) # JSON list of animal IDs or types
animals_count = db.Column(db.Integer, nullable=True) # number of animals fed
# Nutritional info
protein_content = db.Column(db.Float, nullable=True) # percentage
energy_content = db.Column(db.Float, nullable=True) # kcal/kg
notes = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'feed_date': self.feed_date.isoformat() if self.feed_date else None,
'feed_type': self.feed_type,
'feed_name': self.feed_name,
'quantity': self.quantity,
'unit': self.unit,
'cost_per_unit': self.cost_per_unit,
'total_cost': self.total_cost,
'supplier': self.supplier,
'animals_count': self.animals_count,
'created_at': self.created_at.isoformat() if self.created_at else None
}
def __repr__(self):
return f'<FeedRecord {self.feed_name} - {self.quantity} {self.unit}>'
class HealthRecord(db.Model):
"""Model for tracking animal health, vaccinations, and treatments"""
__tablename__ = 'health_records'
id = db.Column(db.Integer, primary_key=True)
farm_id = db.Column(db.Integer, db.ForeignKey('farms.id'), nullable=False)
farmer_id = db.Column(db.Integer, db.ForeignKey('farmers.id'), nullable=False)
livestock_id = db.Column(db.Integer, db.ForeignKey('livestock_records.id'), nullable=False)
# Health event details
event_date = db.Column(db.Date, nullable=False, default=datetime.utcnow)
event_type = db.Column(db.String(50), nullable=False) # vaccination, treatment, checkup, illness
# Vaccination details
vaccine_name = db.Column(db.String(200), nullable=True)
vaccine_batch = db.Column(db.String(100), nullable=True)
next_due_date = db.Column(db.Date, nullable=True)
# Treatment details
symptoms = db.Column(db.Text, nullable=True)
diagnosis = db.Column(db.Text, nullable=True)
treatment_given = db.Column(db.Text, nullable=True)
medication = db.Column(db.Text, nullable=True)
dosage = db.Column(db.String(100), nullable=True)
# Veterinary info
vet_name = db.Column(db.String(200), nullable=True)
vet_contact = db.Column(db.String(100), nullable=True)
cost = db.Column(db.Float, nullable=True)
# Follow-up
follow_up_required = db.Column(db.Boolean, default=False)
follow_up_date = db.Column(db.Date, nullable=True)
recovery_status = db.Column(db.String(50), nullable=True) # recovered, recovering, chronic
notes = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def as_dict(self):
return {
'id': self.id,
'farm_id': self.farm_id,
'livestock_id': self.livestock_id,
'event_date': self.event_date.isoformat() if self.event_date else None,
'event_type': self.event_type,
'vaccine_name': self.vaccine_name,
'symptoms': self.symptoms,
'diagnosis': self.diagnosis,
'treatment_given': self.treatment_given,
'vet_name': self.vet_name,
'cost': self.cost,
'recovery_status': self.recovery_status,
'created_at': self.created_at.isoformat() if self.created_at else None
}
def __repr__(self):
return f'<HealthRecord {self.event_type} for livestock {self.livestock_id}>'