astrosat / django-astrosat-tasks Goto Github PK

    def get_timing_info(self, obj):

        schedule = obj.schedule
        last_run_at = obj.last_run_at
        
        if obj.interval is not None:
            schedule_type = "interval"
        elif obj.crontab is not None:
            schedule_type = "crontab"
        elif obj.solar is not None:
            schedule_type = "solar"
        elif obj.clocked is not None:
            schedule_type = "clocked"
        else:
            schedule_type = "UNKNOWN"
        
        (is_due, time_in_seconds_for_next_execution) = schedule.is_due(last_run_at)
        next_execution_time = timezone.now() + datetime.timedelta(seconds=time_in_seconds_for_next_execution)

        timing_info = {
            "type": f"{schedule_type}: {schedule}",
            "last_execution": last_run_at.replace(microsecond=0),
            "next_execution": next_execution_time.replace(microsecond=0),
            "is_due": is_due,
        }

        return timing_info