Spaces:
Sleeping
Sleeping
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}>' | |