434 lines
17 KiB
Python
Executable File
434 lines
17 KiB
Python
Executable File
from django.db import models
|
|
from django.contrib.auth.models import User
|
|
from phonenumber_field.modelfields import PhoneNumberField
|
|
from django.utils import timezone
|
|
from django_enum import EnumField
|
|
from django.utils.text import slugify
|
|
from django.db.models import Sum
|
|
import datetime
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
def user_str(self):
|
|
if self.first_name and self.last_name:
|
|
return f"{self.first_name} {self.last_name}"
|
|
if self.username:
|
|
return self.username
|
|
if self.email:
|
|
return self.email
|
|
return str(self.id)
|
|
|
|
User.__str__ = user_str
|
|
|
|
# Abstract Model Classes
|
|
class TimeMixin(models.Model):
|
|
|
|
created = models.DateTimeField(default=timezone.now)
|
|
last_modified = models.DateTimeField(default=timezone.now)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='+')
|
|
last_modified_BY = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='+')
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
class IdMixin(models.Model):
|
|
slug = models.SlugField(blank=True, unique=True, null=True)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def save(self, *args, **kwargs):
|
|
if not self.slug:
|
|
self.slug = slugify(datetime.datetime.now().time())
|
|
super(IdMixin, self).save(*args, **kwargs)
|
|
|
|
# Concrete Classes
|
|
class Contract(IdMixin, TimeMixin):
|
|
class ContractTypeEnum(models.TextChoices):
|
|
FIRM_FIX_PRICED = "FFP", "FIRM_FIX_PRICED"
|
|
COST_PLUS_FIXED_FEE = "CPFF", "COST_PLUS_FIXED_FEE"
|
|
MAX_NUM_CONRACT_TYPES = "MAX", "MAX_NUM_CONTRACT_TYPES"
|
|
|
|
contract_type = EnumField(ContractTypeEnum)
|
|
|
|
name = models.CharField(max_length=100)
|
|
proposed_amount = models.FloatField(default=0.0)
|
|
baseline_amount = models.FloatField(default=0.0)
|
|
funded_amount = models.FloatField(default=0.0)
|
|
award_amount = models.FloatField(default=0.0)
|
|
budget_hours = models.FloatField(default=0.0)
|
|
|
|
baseline_start = models.DateField(null=True, blank=True, default=None)
|
|
baseline_end = models.DateField(null=True, blank=True, default=None)
|
|
linkedContracts = models.ForeignKey("self", on_delete=models.CASCADE, null=True, blank=True)
|
|
|
|
def __str__(self):
|
|
return f"{self.name} - {self.contract_type}"
|
|
|
|
@property
|
|
def ticket_percent_complete(self):
|
|
charge_numbers = self.chargenumber_set.all()
|
|
total_hours = sum(cn.budget_hours for cn in charge_numbers)
|
|
if total_hours == 0:
|
|
return 0.0
|
|
|
|
weighted_sum = sum(cn.budget_hours * cn.get_percent_complete for cn in charge_numbers)
|
|
return weighted_sum / total_hours
|
|
|
|
def get_evm_data(self):
|
|
"""Compute all Earned Value Management metrics for this contract."""
|
|
today = timezone.now().date()
|
|
|
|
# --- Budget at Completion (BAC) ---
|
|
bac_hours = self.budget_hours or 0.0
|
|
bac_dollars = self.baseline_amount if self.baseline_amount > 0 else (self.funded_amount or 0.0)
|
|
|
|
# --- Time fraction elapsed ---
|
|
if self.baseline_start and self.baseline_end and self.baseline_end > self.baseline_start:
|
|
total_days = (self.baseline_end - self.baseline_start).days
|
|
elapsed_days = max(0, (min(today, self.baseline_end) - self.baseline_start).days)
|
|
time_fraction = min(1.0, elapsed_days / total_days) if total_days > 0 else 0.0
|
|
else:
|
|
total_days = 0
|
|
elapsed_days = 0
|
|
time_fraction = 0.0
|
|
|
|
# --- Planned Value (PV) ---
|
|
pv_hours = bac_hours * time_fraction
|
|
pv_dollars = bac_dollars * time_fraction
|
|
|
|
# --- Earned Value (EV) ---
|
|
percent_complete = self.ticket_percent_complete # 0-100
|
|
ev_hours = bac_hours * (percent_complete / 100.0)
|
|
ev_dollars = bac_dollars * (percent_complete / 100.0)
|
|
|
|
# --- Actual Cost (AC) ---
|
|
from financial.models import TimeCardCell # avoid circular at module level
|
|
cells = TimeCardCell.objects.filter(
|
|
charge_number__contract=self
|
|
).select_related('timeCard__employee')
|
|
ac_hours = cells.aggregate(total=Sum('hour'))['total'] or 0.0
|
|
ac_dollars = sum(
|
|
(cell.hour or 0.0) * (cell.timeCard.employee.hourly_salary if cell.timeCard and cell.timeCard.employee else 0.0)
|
|
for cell in cells
|
|
)
|
|
|
|
# --- Variances ---
|
|
sv_hours = ev_hours - pv_hours
|
|
sv_dollars = ev_dollars - pv_dollars
|
|
cv_hours = ev_hours - ac_hours
|
|
cv_dollars = ev_dollars - ac_dollars
|
|
|
|
# --- Performance Indices ---
|
|
spi = (ev_dollars / pv_dollars) if pv_dollars > 0 else 0.0
|
|
cpi = (ev_dollars / ac_dollars) if ac_dollars > 0 else 0.0
|
|
spi_hours = (ev_hours / pv_hours) if pv_hours > 0 else 0.0
|
|
cpi_hours = (ev_hours / ac_hours) if ac_hours > 0 else 0.0
|
|
|
|
# --- Monthly Time-Series for S-Curve ---
|
|
time_series = []
|
|
if self.baseline_start and self.baseline_end and total_days > 0:
|
|
# Build cumulative AC by month
|
|
ac_by_month = {}
|
|
for cell in cells:
|
|
if cell.date:
|
|
key = cell.date.strftime('%Y-%m')
|
|
cost = (cell.hour or 0.0) * (
|
|
cell.timeCard.employee.hourly_salary if cell.timeCard and cell.timeCard.employee else 0.0
|
|
)
|
|
ac_by_month[key] = ac_by_month.get(key, 0.0) + cost
|
|
|
|
cursor = self.baseline_start.replace(day=1)
|
|
end_limit = max(self.baseline_end, today)
|
|
cumulative_ac = 0.0
|
|
|
|
while cursor <= end_limit:
|
|
month_key = cursor.strftime('%Y-%m')
|
|
# PV at this point in time
|
|
days_into = max(0, (cursor - self.baseline_start).days)
|
|
pv_at = bac_dollars * min(1.0, days_into / total_days)
|
|
# Cumulative AC
|
|
cumulative_ac += ac_by_month.get(month_key, 0.0)
|
|
# EV: proportional to current % complete, scaled by time
|
|
if cursor <= today:
|
|
ev_at = ev_dollars * min(1.0, days_into / max(1, (today - self.baseline_start).days)) if (today - self.baseline_start).days > 0 else 0.0
|
|
else:
|
|
ev_at = ev_dollars # flat after today
|
|
|
|
time_series.append({
|
|
'month': month_key,
|
|
'pv': round(pv_at, 2),
|
|
'ev': round(ev_at, 2),
|
|
'ac': round(cumulative_ac, 2),
|
|
})
|
|
cursor += relativedelta(months=1)
|
|
|
|
return {
|
|
'bac_hours': round(bac_hours, 2),
|
|
'bac_dollars': round(bac_dollars, 2),
|
|
'pv_hours': round(pv_hours, 2),
|
|
'pv_dollars': round(pv_dollars, 2),
|
|
'ev_hours': round(ev_hours, 2),
|
|
'ev_dollars': round(ev_dollars, 2),
|
|
'ac_hours': round(ac_hours, 2),
|
|
'ac_dollars': round(ac_dollars, 2),
|
|
'sv_hours': round(sv_hours, 2),
|
|
'sv_dollars': round(sv_dollars, 2),
|
|
'cv_hours': round(cv_hours, 2),
|
|
'cv_dollars': round(cv_dollars, 2),
|
|
'spi': round(spi, 2),
|
|
'cpi': round(cpi, 2),
|
|
'spi_hours': round(spi_hours, 2),
|
|
'cpi_hours': round(cpi_hours, 2),
|
|
'percent_complete': round(percent_complete, 2),
|
|
'time_series': time_series,
|
|
}
|
|
|
|
class ChargeNumber(IdMixin, TimeMixin):
|
|
class ChargeNumberTypeEnum(models.TextChoices):
|
|
LEVEL_OF_EFFORT = "LOE", "LEVEL_OF_EFFORT"
|
|
QBD = "QBD", "QBD"
|
|
ZERO_ONE_HUNDRED = "0_100", "ZERO_ONE_HUNDRED"
|
|
FIFTY_FIFTY = "50_50", "FIFTY_FIFTY"
|
|
MAX_NUM_TASK_TYPES = "MAX", "MAX_NUM_TASK_TYPES"
|
|
|
|
charge_number_type = EnumField(ChargeNumberTypeEnum)
|
|
name = models.CharField(max_length=100, blank=True, null=True)
|
|
|
|
contract = models.ForeignKey(Contract, on_delete=models.CASCADE)
|
|
amount = models.FloatField(default=0.0)
|
|
budget_hours = models.FloatField(default=0.0)
|
|
percent_complete = models.FloatField(default=0.0)
|
|
# TODO: add validator to make sure the range is 0.0 - 100
|
|
start_date = models.DateField(null=True, blank=True, default = None)
|
|
end_date = models.DateField(null=True, blank=True, default = None)
|
|
|
|
def __str__(self):
|
|
return self.name or self.slug or f"CN-{self.id}"
|
|
|
|
@property
|
|
def get_percent_complete(self):
|
|
if self.charge_number_type == self.ChargeNumberTypeEnum.QBD:
|
|
total = self.tickets.count()
|
|
if total == 0:
|
|
return 0.0
|
|
done = self.tickets.filter(status='DONE').count()
|
|
return (done / total) * 100.0
|
|
return self.percent_complete
|
|
|
|
def clean(self):
|
|
from django.core.exceptions import ValidationError
|
|
super().clean()
|
|
|
|
if self.charge_number_type == "0_100":
|
|
if self.percent_complete not in [0.0, 100.0]:
|
|
raise ValidationError({"percent_complete": "For ZERO_ONE_HUNDRED, percent complete must be 0 or 100."})
|
|
if self.start_date and self.end_date:
|
|
delta = self.end_date - self.start_date
|
|
if delta.days > 31:
|
|
raise ValidationError({"end_date": "Duration cannot be more than a month for ZERO_ONE_HUNDRED."})
|
|
|
|
if self.charge_number_type == "50_50":
|
|
if self.percent_complete not in [0.0, 50.0, 100.0]:
|
|
raise ValidationError({"percent_complete": "For FIFTY_FIFTY, percent complete must be 0, 50, or 100."})
|
|
if self.start_date and self.end_date:
|
|
delta = self.end_date - self.start_date
|
|
if delta.days > 62:
|
|
raise ValidationError({"end_date": "Duration cannot be more than two months for FIFTY_FIFTY."})
|
|
|
|
class AddressModel(models.Model):
|
|
address_1 = models.CharField(max_length=128)
|
|
address_2 = models.CharField(max_length=128, blank=True)
|
|
|
|
city = models.CharField(max_length=64)
|
|
state = models.CharField(max_length=2)
|
|
zip_code = models.CharField(max_length=5)
|
|
|
|
|
|
class UserProfile(models.Model):
|
|
class UserType(models.TextChoices):
|
|
EMPLOYEE = "employee", "Employee"
|
|
CLIENT = "client", "Client"
|
|
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="profile")
|
|
user_type = models.CharField(
|
|
max_length=10,
|
|
choices=UserType.choices,
|
|
default=UserType.CLIENT,
|
|
)
|
|
|
|
def __str__(self):
|
|
return f"{self.user} ({self.get_user_type_display()})"
|
|
|
|
def is_employee(self):
|
|
return self.user_type == self.UserType.EMPLOYEE
|
|
|
|
def is_client(self):
|
|
return self.user_type == self.UserType.CLIENT
|
|
|
|
|
|
class Employee(IdMixin, TimeMixin):
|
|
manager = models.ForeignKey("self", on_delete=models.CASCADE, related_name="manager_employee", null=True, blank=True)
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE)
|
|
primaryAddress = models.ForeignKey(AddressModel, on_delete=models.CASCADE, related_name="primary_address_employee", null=True, blank=True)
|
|
workAddress = models.ForeignKey(AddressModel, on_delete=models.CASCADE, related_name="work_address_employee", null=True, blank=True)
|
|
phoneNumber = PhoneNumberField(null=True, blank=True, unique=True)
|
|
slary= models.FloatField(default=0.0)
|
|
|
|
def __str__(self):
|
|
return str(self.user)
|
|
|
|
@property
|
|
def hourly_salary(self):
|
|
return (self.slary or 0.0) / 2040.0
|
|
# TODO: roles, jpbTitles
|
|
|
|
|
|
|
|
class TimeCard(IdMixin, TimeMixin):
|
|
startDate = models.DateField(null=True, default = None)
|
|
endDate = models.DateField(null=True, default = None)
|
|
employee = models.ForeignKey(Employee, on_delete=models.CASCADE, related_name="employee")
|
|
approvedBy = models.ForeignKey(Employee, on_delete=models.CASCADE, null=True, related_name="approved_by")
|
|
approvedDate = models.DateField(null=True, default=None)
|
|
approvalComment = models.TextField(max_length=256, default="")
|
|
# TODO: do the is approved comment
|
|
|
|
class TimeCardCell(IdMixin, TimeMixin):
|
|
timeCard = models.ForeignKey(TimeCard, on_delete=models.CASCADE)
|
|
date = models.DateField(null=True, default = None)
|
|
start_time = models.TimeField(null=True, blank=True)
|
|
end_time = models.TimeField(null=True, blank=True)
|
|
hour = models.FloatField(default = 0.0)
|
|
charge_number = models.ForeignKey(ChargeNumber, on_delete=models.CASCADE, null=True, blank=True)
|
|
|
|
|
|
class BillingCustomer(IdMixin, TimeMixin):
|
|
"""Customer who pays us (one-off invoices or subscriptions)."""
|
|
|
|
name = models.CharField(max_length=200)
|
|
email = models.EmailField()
|
|
company = models.CharField(max_length=200, blank=True)
|
|
stripe_customer_id = models.CharField(max_length=255, blank=True, default="")
|
|
notes = models.TextField(blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["name"]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} <{self.email}>"
|
|
|
|
|
|
class Invoice(IdMixin, TimeMixin):
|
|
"""One-off invoice — customer pays us via Stripe hosted invoice URL."""
|
|
|
|
class Status(models.TextChoices):
|
|
DRAFT = "draft", "Draft"
|
|
OPEN = "open", "Open"
|
|
PAID = "paid", "Paid"
|
|
VOID = "void", "Void"
|
|
UNCOLLECTIBLE = "uncollectible", "Uncollectible"
|
|
FAILED = "failed", "Failed"
|
|
|
|
customer = models.ForeignKey(
|
|
BillingCustomer, on_delete=models.PROTECT, related_name="invoices"
|
|
)
|
|
description = models.CharField(max_length=500)
|
|
amount_cents = models.PositiveIntegerField(help_text="Amount in cents (USD)")
|
|
currency = models.CharField(max_length=3, default="usd")
|
|
status = models.CharField(
|
|
max_length=20, choices=Status.choices, default=Status.DRAFT
|
|
)
|
|
due_date = models.DateField(null=True, blank=True)
|
|
notes = models.TextField(blank=True)
|
|
stripe_invoice_id = models.CharField(max_length=255, blank=True, default="")
|
|
hosted_invoice_url = models.URLField(max_length=500, blank=True, default="")
|
|
invoice_pdf_url = models.URLField(max_length=500, blank=True, default="")
|
|
pay_link_emailed_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-created"]
|
|
|
|
def __str__(self):
|
|
return f"Invoice {self.pk} — {self.customer} — {self.status}"
|
|
|
|
@property
|
|
def amount_dollars(self):
|
|
return self.amount_cents / 100.0
|
|
|
|
|
|
class RecurringSubscription(IdMixin, TimeMixin):
|
|
"""Monthly (or custom) recurring charge — customer pays us via Stripe Subscription."""
|
|
|
|
class Status(models.TextChoices):
|
|
INCOMPLETE = "incomplete", "Incomplete"
|
|
ACTIVE = "active", "Active"
|
|
PAST_DUE = "past_due", "Past due"
|
|
CANCELED = "canceled", "Canceled"
|
|
UNPAID = "unpaid", "Unpaid"
|
|
TRIALING = "trialing", "Trialing"
|
|
PAUSED = "paused", "Paused"
|
|
|
|
class Interval(models.TextChoices):
|
|
MONTH = "month", "Monthly"
|
|
YEAR = "year", "Yearly"
|
|
WEEK = "week", "Weekly"
|
|
|
|
customer = models.ForeignKey(
|
|
BillingCustomer, on_delete=models.PROTECT, related_name="subscriptions"
|
|
)
|
|
description = models.CharField(max_length=500)
|
|
amount_cents = models.PositiveIntegerField(help_text="Amount per period in cents (USD)")
|
|
currency = models.CharField(max_length=3, default="usd")
|
|
interval = models.CharField(
|
|
max_length=10, choices=Interval.choices, default=Interval.MONTH
|
|
)
|
|
status = models.CharField(
|
|
max_length=20, choices=Status.choices, default=Status.INCOMPLETE
|
|
)
|
|
notes = models.TextField(blank=True)
|
|
stripe_product_id = models.CharField(max_length=255, blank=True, default="")
|
|
stripe_price_id = models.CharField(max_length=255, blank=True, default="")
|
|
stripe_subscription_id = models.CharField(max_length=255, blank=True, default="")
|
|
checkout_session_id = models.CharField(max_length=255, blank=True, default="")
|
|
checkout_url = models.URLField(max_length=500, blank=True, default="")
|
|
pay_link_emailed_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-created"]
|
|
|
|
def __str__(self):
|
|
return f"Subscription {self.pk} — {self.customer} — {self.status}"
|
|
|
|
@property
|
|
def amount_dollars(self):
|
|
return self.amount_cents / 100.0
|
|
|
|
|
|
def set_user_type(user, user_type):
|
|
"""Set user type and sync the Employee record (mutually exclusive types)."""
|
|
user.__dict__.pop("profile", None)
|
|
profile, _ = UserProfile.objects.get_or_create(
|
|
user=user,
|
|
defaults={"user_type": user_type},
|
|
)
|
|
if profile.user_type == user_type:
|
|
if user_type == UserProfile.UserType.EMPLOYEE:
|
|
Employee.objects.get_or_create(user=user)
|
|
return profile
|
|
|
|
if user_type == UserProfile.UserType.CLIENT:
|
|
employee = Employee.objects.filter(user=user).first()
|
|
if employee and TimeCardCell.objects.filter(timeCard__employee=employee).exists():
|
|
raise ValueError("Cannot set Client: user has time log entries.")
|
|
|
|
profile.user_type = user_type
|
|
profile.save()
|
|
user.__dict__.pop("profile", None)
|
|
|
|
if user_type == UserProfile.UserType.EMPLOYEE:
|
|
Employee.objects.get_or_create(user=user)
|
|
else:
|
|
Employee.objects.filter(user=user).delete()
|
|
|
|
return profile
|