gestioCOF/kfet/statistic.py
Aurélien Delobelle 1bb83ccdd7 simplify StatScale
2017-04-03 15:10:53 +02:00

187 lines
5 KiB
Python

# -*- coding: utf-8 -*-
import ast
from dateutil.relativedelta import relativedelta
from django.utils import timezone
from django.db.models import Sum
KFET_WAKES_UP_AT = 7
def kfet_day(year, month, day, start_at=KFET_WAKES_UP_AT):
return timezone.datetime(year, month, day, hour=start_at)
def to_kfet_day(dt, start_at=KFET_WAKES_UP_AT):
kfet_dt = kfet_day(year=dt.year, month=dt.month, day=dt.day)
if dt.hour < start_at:
kfet_dt -= timezone.timedelta(days=1)
return kfet_dt
class StatScale(object):
name = None
step = None
def __init__(self, n_steps=0, begin=None, end=None,
last=False, std_chunk=True):
self.std_chunk = std_chunk
if last:
end = timezone.now()
if begin is not None and n_steps != 0:
self.begin = self.get_from(begin)
self.end = self.do_step(self.begin, n_steps=n_steps)
elif end is not None and n_steps != 0:
self.end = self.get_from(end)
self.begin = self.do_step(self.end, n_steps=-n_steps)
elif begin is not None and end is not None:
self.begin = self.get_from(begin)
self.end = self.get_from(end)
else:
raise Exception('Two of these args must be specified: '
'n_steps, begin, end; '
'or use last and n_steps')
self.datetimes = self.get_datetimes()
@staticmethod
def by_name(name):
for cls in StatScale.__subclasses__():
if cls.name == name:
return cls
return None
def get_from(self, dt):
return self.std_chunk and self.get_chunk_start(dt) or dt
def __getitem__(self, i):
return self.datetimes[i], self.datetimes[i+1]
def __len__(self):
return len(self.datetimes) - 1
def do_step(self, dt, n_steps=1):
return dt + self.step * n_steps
def get_datetimes(self):
datetimes = [self.begin]
tmp = self.begin
while tmp <= self.end:
tmp = self.do_step(tmp)
datetimes.append(tmp)
return datetimes
def get_labels(self, label_fmt=None):
if label_fmt is None:
label_fmt = self.label_fmt
return [begin.strftime(label_fmt) for begin, end in self]
class DayStatScale(StatScale):
name = 'day'
step = timezone.timedelta(days=1)
label_fmt = '%A'
@classmethod
def get_chunk_start(cls, dt):
return to_kfet_day(dt)
class WeekStatScale(StatScale):
name = 'week'
step = timezone.timedelta(days=7)
label_fmt = 'Semaine %W'
@classmethod
def get_chunk_start(cls, dt):
dt_kfet = to_kfet_day(dt)
offset = timezone.timedelta(days=dt_kfet.weekday())
return dt_kfet - offset
class MonthStatScale(StatScale):
name = 'month'
step = relativedelta(months=1)
label_fmt = '%B'
@classmethod
def get_chunk_start(cls, dt):
return to_kfet_day(dt).replace(day=1)
def stat_manifest(scales_def=None, scale_args=None, **url_params):
if scales_def is None:
scales_def = []
if scale_args is None:
scale_args = {}
return [
dict(
label=label,
url_params=dict(
scale=cls.name,
scale_args=scale_args,
**url_params,
),
)
for label, cls in scales_def
]
def last_stats_manifest(scales_def=None, scale_args=None, **url_params):
scales_def = [
('Derniers mois', MonthStatScale, ),
('Dernières semaines', WeekStatScale, ),
('Derniers jours', DayStatScale, ),
]
if scale_args is None:
scale_args = {}
scale_args.update(dict(
last=True,
n_steps=7,
))
return stat_manifest(scales_def=scales_def, scale_args=scale_args,
**url_params)
# Étant donné un queryset d'operations
# rend la somme des article_nb
def tot_ventes(queryset):
res = queryset.aggregate(Sum('article_nb'))['article_nb__sum']
return res and res or 0
class ScaleMixin(object):
def get_context_data(self, *args, **kwargs):
context = super().get_context_data(*args, **kwargs)
scale_name = self.request.GET.get('scale', None)
cls = StatScale.by_name(scale_name)
if cls is None:
scale = self.get_default_scale()
else:
scale_args = self.request.GET.get('scale_args', {})
if isinstance(scale_args, str):
scale_args = ast.literal_eval(scale_args)
scale = cls(**scale_args)
self.scale = scale
context['labels'] = scale.get_labels()
return context
def get_default_scale(self):
return DayStatScale(n_steps=7, last=True)
def chunkify_qs(self, qs, scale, field=None):
if field is None:
field = 'at'
begin_f = '{}__gte'.format(field)
end_f = '{}__lte'.format(field)
return [
qs.filter(**{begin_f: begin, end_f: end})
for begin, end in scale
]