Source code for gantt_project_maker.gantt

# -*- coding: utf-8 -*-

"""
This is a python class to create gantt chart using SVG

Author: Alexandre Norman - norman at xael.org

Contributors:

* Sébastien NOBILI - pipoprods at free.fr

Modified by:

* Eelco van Vliet
"""

import codecs
import datetime
import io
import logging
import re
import sys
from datetime import date
from logging import Logger

import dateutil.relativedelta
import svgwrite
from svgwrite.container import Group as svg_Group
from svgwrite.shapes import Rect as svg_Rect
from svgwrite.text import Text as svg_Text

# original author: Alexandre Norman (norman at xael.org)
# modified by Eelco van Vliet

# we do conversion from mm/cm to pixel ourselves as firefox seems
# to have a bug for big numbers...
# 3.543307 is for conversion from mm to pt units!
mm = 3.543307
cm = 35.43307

# noinspection PyTypeChecker
_logger: Logger = logging.getLogger(__name__)

COLOR_OVERCHARGE_DEFAULT = "#AA0000"
COLOR_VACATION_DEFAULT = "#008000"
COLOR_RESOURCE_DEFAULT = "#c5f0eb"

DRAW_WITH_DAILY_SCALE = "d"
DRAW_WITH_WEEKLY_SCALE = "w"
DRAW_WITH_MONTHLY_SCALE = "m"
DRAW_WITH_QUARTERLY_SCALE = "q"

# Unworked days (0: Monday ... 6: Sunday)
NOT_WORKED_DAYS = [5, 6]

FONT_ATTR = {
    "fill": "black",
    "stroke": "black",
    "stroke_width": 0,
    "font_family": "Verdana",
    "font_size": 15,
    "font_weight": "normal",
}
VACATIONS = []


[docs] class MySVGWriteDrawingWrapper(svgwrite.Drawing): """ Hack to allow using a file descriptor as filename """
[docs] def save(self, width="100%", height="100%"): """Write the XML string to **filename**.""" # Fix height and width self["height"] = height self["width"] = width this_file_type = type(self.filename) test = this_file_type == io.TextIOWrapper if test: self.write(self.filename) else: with io.open(str(self.filename), mode="w", encoding="utf-8") as stream: self.write(stream)
[docs] def define_not_worked_days(list_of_days): """ Define specific days off Keyword arguments: list_of_days -- list of integer (0: Monday ... 6: Sunday) - default [5, 6] """ global NOT_WORKED_DAYS NOT_WORKED_DAYS = list_of_days return
def _not_worked_days(): """ Returns list of days off (0: Monday ... 6: Sunday) """ global NOT_WORKED_DAYS return NOT_WORKED_DAYS
[docs] def define_font_attributes( fill: str = "black", stroke: str = "black", stroke_width: float = 0, font_family: str = "Verdana", font_weight: str = "normal", font_size: int = 15, ): """ Define font attributes Args: fill (str): Fill color. Defaults to 'black' stroke (str): Stroke color. Defaults to 'black' stroke_width (float): stroke width. Defaults to 0 font_family (str): Font family. Defaults to 'Verdana' font_weight (str): Font weight. Defaults to 'normal' font_size (int): Font size. Defaults to '15' """ global FONT_ATTR FONT_ATTR = { "fill": fill, "stroke": stroke, "stroke_width": stroke_width, "font_family": font_family, "font_weight": font_weight, "font_size": font_size, }
def _font_attributes(): """ Return dictionary of font attributes Returns: dict with font attributes Example: FONT_ATTR = { 'fill': 'black', 'stroke': 'black', 'stroke_width': 0, 'font_family': 'Verdana', } """ global FONT_ATTR return FONT_ATTR
[docs] def get_font_attributes( fill=None, stroke=None, stroke_width=None, font_family=None, font_weight=None, font_size=None, ): """ Return dictionary of font attributes """ global FONT_ATTR font_attributes = FONT_ATTR.copy() if fill is not None: font_attributes["fill"] = fill if stroke is not None: font_attributes["stroke"] = stroke if stroke_width is not None: font_attributes["stroke_width"] = stroke_width if font_family is not None: font_attributes["font_family"] = font_family if font_weight is not None: font_attributes["font_weight"] = font_weight if font_size is not None: font_attributes["font_size"] = font_size return font_attributes
[docs] def add_vacations(start_date: date, end_date: date = None): """ Add vacations to a resource beginning at *start_date* to *end_date* (included). If *end_date* is not defined, vacation will be for *start_date* day only Args: start_date (date): Beginning of a vacation end_date: (date): End of a vacation """ _logger.debug( "** add_vacations {0}".format({"start_date": start_date, "end_date": end_date}) ) global VACATIONS if end_date is None: if start_date not in VACATIONS: VACATIONS.append(start_date) else: while start_date <= end_date: if start_date not in VACATIONS: VACATIONS.append(start_date) start_date += datetime.timedelta(days=1) _logger.debug( "** add_vacations {0}".format( {"start_date": start_date, "end_date": end_date, "vac": VACATIONS} ) )
[docs] def init_log_to_sysout(level=logging.INFO): """ Init global variable __LOG__ used for logging purpose Keyword arguments: level -- logging level (from logging.debug to logging.critical) """ global _logger logger = logging.getLogger("Gantt") logger.setLevel(level) fh = logging.StreamHandler() formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") fh.setFormatter(formatter) logger.addHandler(fh) _logger = logging.getLogger("Gantt") return
def _flatten(nested_list, list_types=(list, tuple)): """ Return a flattened list from a list like [1,2,[4,5,1]] """ list_type = type(nested_list) nested_list = list(nested_list) i = 0 while i < len(nested_list): while isinstance(nested_list[i], list_types): if not nested_list[i]: nested_list.pop(i) i -= 1 break else: nested_list[i : i + 1] = nested_list[i] i += 1 return list_type(nested_list) ############################################################################
[docs] class GroupOfResources: """ Class for grouping resources """ def __init__(self, name, fullname=None): """ Init a group of resources Keyword arguments: name -- name given to the resource (id) fullname -- long name given to the resource """ _logger.debug("** GroupOfResources::__init__ {0}".format({"name": name})) self.name = name self.vacations = [] if fullname is not None: self.fullname = fullname else: self.fullname = name self.resources = [] self.tasks = [] return
[docs] def add_resource(self, resource): """ Add a resource to the group of resources Keyword arguments: resource -- Resource object """ if resource not in self.resources: self.resources.append(resource) resource.add_group(self) return
[docs] def add_vacations(self, dfrom, dto=None): """ Add vacations to a resource beginning at [dfrom] to [dto] (included). If [dto] is not defined, vacation will be for [dfrom] day only Keyword arguments: dfrom -- datetime.date beginning of vacation dto -- datetime.date end of vacation """ _logger.debug( "** Resource::add_vacations {0}".format( {"name": self.name, "dfrom": dfrom, "dto": dto} ) ) if dto is None: self.vacations.append((dfrom, dfrom)) else: self.vacations.append((dfrom, dto)) return
[docs] def nb_elements(self): """ Returns the number of resources """ _logger.debug( "** GroupOfResources::nb_elements ({0})".format({"name": self.name}) ) return len(self.resources)
[docs] def is_available(self, date_to_check): """ Returns True if any resource is available at a given date, False if not. Availability is tasks from the global VACATIONS and resource's ones. Args: date_to_check (object): date day to look for """ # Global VACATIONS if date_to_check in VACATIONS: _logger.debug( "** GroupOfResources::is_available {0} : False (global vacation)".format( {"name": self.name, "date": date_to_check} ) ) return False # Group vacations for h in self.vacations: data_from, data_to = h if data_from <= date_to_check <= data_to: _logger.debug( "** GroupOfResources::is_available {0} : False (group vacation)".format( {"name": self.name, "date": date_to_check} ) ) return False # Test if at least one resource is available for r in self.resources: if r.is_available(date_to_check): _logger.debug( "** GroupOfResources::is_available {0} : True {1}".format( {"name": self.name, "date": date_to_check}, r.name ) ) return True _logger.debug( "** GroupOfResources::is_available {0} : False".format( {"name": self.name, "date": date_to_check} ) ) return False
[docs] def add_task(self, task): """ Tell the resource that we have assigned a task Keyword arguments: task -- Task object """ if task not in self.tasks: self.tasks.append(task)
[docs] def search_for_task_conflicts(self, all_tasks=False): """ Returns a dictionary of all days (datetime.date) containing for each overcharged day the list of task for this day. It examines all resources' member and group tasks. Keyword arguments: all_tasks -- if True return all tasks for all days, not just overcharged days """ # Get for each resource affected_days = {} for r in self.resources: ad = r.search_for_task_conflicts(all_tasks=True) for d in ad: try: affected_days[d].append(ad[d]) except KeyError: affected_days[d] = [ad[d]] # inspect the project for t in self.tasks: current_day = t.start_date while current_day <= t.end_date: if current_day.weekday() not in _not_worked_days(): try: affected_days[current_day].append(t.fullname) except KeyError: affected_days[current_day] = [t.fullname] current_day += datetime.timedelta(days=1) # compile everything overcharged_days = {} ke = list(affected_days.keys()) ke.sort() for d in ke: affected_days[d] = _flatten(affected_days[d]) if all_tasks: overcharged_days[d] = affected_days[d] elif len(affected_days[d]) > self.nb_elements(): overcharged_days[d] = affected_days[d] _logger.warning( '** GroupOfResources "{2}" has more than {3} tasks on day {0} / {1}'.format( d, affected_days[d], self.name, self.nb_elements() ) ) return overcharged_days
[docs] def is_vacant(self, from_date: date, to_date: date): """ Check if any resource from the group is unallocated between for a given timeframe. Returns a list of available ressource names. Args: from_date(date): First day to_date (date): Last day """ available = [] for r in self.resources: if len(r.is_vacant(from_date, to_date)) > 0: available.append(r.name) return available
[docs] class Resource: """ Class for handling resources assigned to tasks Args: name (str): Name given to the resource (id) fullname (str): Long name given to the resource color (str): Color used to represent the resource in the resources' overview """ def __init__(self, name, fullname=None, color=None): """ Init a resource """ _logger.debug("** Resource::__init__ {0}".format({"name": name})) self.name = name self.color = color if fullname is not None: self.fullname = fullname else: self.fullname = name self.vacations = [] self.member_of_groups = [] self.tasks = [] self.task_hours = []
[docs] def add_vacations(self, from_date: date, to_date: date = None): """ Add vacations to a resource beginning at *from_date* to *to_date* (included). If *to_date* is not defined, vacation will be for *from_data* day only Args: from_date (date): Beginning of vacation to_date (date): End of vacation """ _logger.debug( "** Resource::add_vacations {0}".format( {"name": self.name, "from_date": from_date, "to_date": to_date} ) ) if to_date is None: self.vacations.append((from_date, from_date)) else: self.vacations.append((from_date, to_date))
[docs] def nb_elements(self): """ Returns the number of resources, 1 here """ _logger.debug("** Resource::nb_elements ({0})".format({"name": self.name})) return 1
[docs] def is_available(self, date_of_this_day): """ Returns True if the resource is available at given date, False if not. Availability is tasks from the global VACATIONS and resource's ones. Args: date_of_this_day (date): Day to look for """ # global VACATIONS if date_of_this_day in VACATIONS: _logger.debug( "** Resource::is_available {0} : False (global vacation)".format( {"name": self.name, "date": date_of_this_day} ) ) return False # GroupOfResources vacation for g in self.member_of_groups: for h in g.vacations: date_from, date_to = h if date_from <= date_of_this_day <= date_to: _logger.debug( "** Resource::is_available {0} : False (Group {1})".format( {"name": self.name, "date": date}, g.name ) ) return False # Resource vacation for h in self.vacations: date_from, date_to = h if date_from <= date_of_this_day <= date_to: _logger.debug( "** Resource::is_available {0} : False".format( {"name": self.name, "date": date} ) ) return False _logger.debug( "** Resource::is_available {0} : True".format( {"name": self.name, "date": date} ) ) return True
[docs] def add_group(self, group_of_resources): """ Tell the resource it belongs to a GroupOfResources Args: group_of_resources (GroupOfResources): The GroupOfResources to which a resource belongs to """ if group_of_resources not in self.member_of_groups: self.member_of_groups.append(group_of_resources) return
[docs] def add_task(self, task, hours_for_resource=None): """ Tell the resource that we have assigned a task Args: task(Task): The task object to add hours_for_resource(int): The number of hours to assign to the task for this resource """ if task not in self.tasks: self.tasks.append(task) self.task_hours.append(hours_for_resource)
[docs] def search_for_task_conflicts(self, all_tasks=False): """ Returns a dictionary of all days (datetime.date) containing for each overcharged day the list of task for this day. Keyword arguments: all_tasks -- if True return all tasks for all days, not just overcharged days """ affected_days = {} for t in self.tasks: try: task_start_date = t.start_date except TypeError as err: _logger.warning(err) _logger.warning( f"Could not get initial start date for task {t.fullname}. Is is properly defined?" ) raise try: task_end_date = t.end_date except TypeError as err: _logger.warning(err) _logger.warning( f"Could not get end date for task {t.fullname}. Is is properly defined?" ) raise try: while task_start_date <= task_end_date: if task_start_date.weekday() not in _not_worked_days(): try: affected_days[task_start_date].append(t.fullname) except KeyError: affected_days[task_start_date] = [t.fullname] task_start_date += datetime.timedelta(days=1) except TypeError as err: _logger.warning(err) _logger.warning( f"Failing for task {t.fullname} with {task_start_date} (init {t.start_date} and {t.end_date}" ) raise # return all if all_tasks: return affected_days # compile only overcharge overcharged_days = {} ke = list(affected_days.keys()) ke.sort() for d in ke: if len(affected_days[d]) > 1: overcharged_days[d] = affected_days[d] _logger.warning( '** Resource "{2}" has more than one task on day {0} / {1}'.format( d, affected_days[d], self.name ) ) return overcharged_days
[docs] def is_vacant(self, from_date, to_date): """ Check if the resource is unallocated between for a given timeframe. Returns True if the resource is free, False otherwise Args: from_date (date): First day to_date (date): Last day """ non_vacant_days = self.search_for_task_conflicts(all_tasks=True) current_day = from_date while current_day <= to_date: if current_day.weekday() not in _not_worked_days(): if not self.is_available(current_day): _logger.debug( '** Ressource "{0}" is not available on day {1} (vacation)'.format( self.name, current_day ) ) return [] if current_day in non_vacant_days: _logger.debug( '** Ressource "{0}" is not available on day {1} (other task : {2})'.format( self.name, current_day, non_vacant_days[current_day] ) ) return [] current_day += datetime.timedelta(days=1) return [self.name]
############################################################################
[docs] class Task: """ Class for manipulating Tasks Notes: Initialize task object. Two of start, stop or duration may be given. This task can rely on other task and will be completed with resources. If percent done is given, a progress bar will be included on the task. If color is specified, it will be used for the task. Args: name (str): name of the task (id) fullname (str): Long name given to the resource start (date): First day of the task, default None stop (date): Last day of the task, default None duration (int): Duration of the task, default None depends_of (list|None): Tasks which are parents of this one, default None resources (list|None): Resources assigned to the task, default None percent_done (int): Percent of achievement, default 0 color (str, html color): default None display (bool): Display this task, default True state (str): State of the task owner (str): Owner of the task parent (str): Parent of the task """ depends_of: list def __init__( self, name: str, start: date = None, stop: date = None, duration: int = None, depends_of: list | None = None, resources: list = None, percent_done: int = 0, color: str = None, fullname: str = None, display: bool = True, state: str = "", owner: str = "", parent: str = "", ): """ Constructor for the class Task """ _logger.debug( "** Task::__init__ {0}".format( { "name": name, "start": start, "stop": stop, "duration": duration, "depends_of": depends_of, "resources": resources, "percent_done": percent_done, "owner": owner, "parent": parent, } ) ) self.name = name if fullname is not None: self.fullname = fullname else: self.fullname = name self._start = start self._stop = stop self.duration: int = duration self.color = color self.display = display self.state = state self.owner = owner self.parent = parent self._end = None ends = (self._start, self._stop, self.duration) none_count = 0 for e in ends: if e is None: none_count += 1 # check limits (2 must-be set on 4) or scheduling is defined by duration and dependencies if none_count != 1 and (self.duration is None or depends_of is None): _logger.error( '** Task "{1}" must be defined by two of three limits ({0})'.format( { "start": self._start, "stop": self._stop, "duration": self.duration, }, fullname, ) ) self.depends_of: list | None = None if depends_of is not None: if isinstance(depends_of, list): self.depends_of = depends_of elif isinstance(depends_of, str): self.depends_of = [depends_of] if resources is not None: if not isinstance(resources, list): _logger.debug( "** Task::__init__ {0} - resources is not a list".format(name) ) self.resources = resources else: self.resources = [] self.percent_done = percent_done self.drawn_x_begin_coord = None self.drawn_x_end_coord = None self.drawn_y_coord = None self._cache_start_date = None self._cache_end_date = None # tell each resource we have # assigned a new task if resources is not None: for r in resources: r.add_task(self)
[docs] def add_depends(self, depends_of): """ Adds dependency to a task Args: depends_of (list): Task which are parents of this one """ if self.depends_of is None: self.depends_of = [] if isinstance(depends_of, list): if self.depends_of is None: self.depends_of = depends_of else: for d in depends_of: self.depends_of.append(d) else: if self.depends_of is None: self.depends_of = depends_of else: self.depends_of.append(depends_of)
@property def start(self): return self._start @start.setter def start(self, value): self._start = value @property def stop(self): return self._stop @stop.setter def stop(self, value): self._stop = value @property def start_date(self): """ Returns the first day of the task, either the one which was given at task creation or the one calculated after checking dependencies """ if self._cache_start_date is not None: return self._cache_start_date _logger.debug("** Task::start_date ({0})".format(self.name)) if self._start is not None: # start date set, calculate beginning if self.depends_of is None: # depends on nothing... start date is start # __LOG__.debug('*** Do not depend of another task') start = self._start # avoid weekends and vacations start = self._start while start.weekday() in _not_worked_days() or start in VACATIONS: start = start + datetime.timedelta(days=1) if start > self._start: # if the start date is changed, warn _logger.warning( '** Due to vacations, Task "{0}", will not start on date {1} but {2}'.format( self.fullname, self._start, start ) ) self._cache_start_date = start return self._cache_start_date else: # depends on another task, start date could vary # __LOG__.debug('*** Do depend of other tasks') start = self._start # avoid weekends and vacations while start.weekday() in _not_worked_days() or start in VACATIONS: start = start + datetime.timedelta(days=1) # get the latest end date of the dependencies prev_task_end = self._start prev_task_end = start for t in self.depends_of: if isinstance(t, Milestone): if t.end_date >= prev_task_end: prev_task_end = t.end_date elif isinstance(t, Task): if t.end_date >= prev_task_end: prev_task_end = t.end_date + datetime.timedelta(days=1) # avoid weekends and vacations while ( prev_task_end.weekday() in _not_worked_days() or prev_task_end in VACATIONS ): prev_task_end = prev_task_end + datetime.timedelta(days=1) if prev_task_end > self._start: # if the start date is changed, warn _logger.warning( '** Due to dependencies, Task "{0}", will not start on date {1} but {2}'.format( self.fullname, self._start, prev_task_end ) ) self._cache_start_date = prev_task_end return self._cache_start_date elif self.duration is None: # start and stop fixed current_day = self._start # check depends if self.depends_of is not None: prev_task_end = self.depends_of[0].end_date for t in self.depends_of: if isinstance(t, Milestone): if t.end_date > prev_task_end: prev_task_end = t.end_date - datetime.timedelta(days=1) elif isinstance(t, Task): if t.end_date > prev_task_end: prev_task_end = t.end_date # if t.end_date > prev_task_end: # #__LOG__.debug('*** latest one {0} which end on {1}'.format(t.name, t.end_date)) # prev_task_end = t.end_date if prev_task_end > current_day: depend_start_date = prev_task_end else: start = self._start while start.weekday() in _not_worked_days() or start in VACATIONS: start = start + datetime.timedelta(days=1) depend_start_date = start if depend_start_date > current_day: _logger.error( '** Due to dependencies, Task "{0}", could not be finished on time (should start as last ' "on {1} but will start on {2})".format( self.fullname, current_day, depend_start_date ) ) self._cache_start_date = depend_start_date else: # should be first day of start... self._cache_start_date = current_day return self._cache_start_date elif ( self.duration is not None and self.depends_of is not None and self._stop is None ): # duration and dependencies fixed prev_task_end = self.depends_of[0].end_date for t in self.depends_of: if isinstance(t, Milestone): if t.end_date > prev_task_end: prev_task_end = t.end_date - datetime.timedelta(days=1) elif isinstance(t, Task): if t.end_date > prev_task_end: prev_task_end = t.end_date # if t.end_date > prev_task_end: # __LOG__.debug('*** latest one {0} which end on {1}'.format(t.name, t.end_date)) # prev_task_end = t.end_date start = prev_task_end + datetime.timedelta(days=1) while start.weekday() in _not_worked_days() or start in VACATIONS: start = start + datetime.timedelta(days=1) # should be first day of start... self._cache_start_date = start elif self._start is None and self._stop is not None: # stop and duration fixed # start date not setted, calculate from end_date + depends current_day = self._stop real_duration = 0 duration = self.duration while duration > 0: if not ( current_day.weekday() in _not_worked_days() or current_day in VACATIONS ): real_duration = real_duration + 1 duration -= 1 else: real_duration = real_duration + 1 current_day = self._stop - datetime.timedelta(days=real_duration) current_day = self._stop - datetime.timedelta(days=real_duration - 1) # check depends if self.depends_of is not None: prev_task_end = self.depends_of[0].end_date for t in self.depends_of: if isinstance(t, Milestone): if t.end_date > prev_task_end: prev_task_end = t.end_date elif isinstance(t, Task): if t.end_date > prev_task_end: prev_task_end = t.end_date # if t.end_date > prev_task_end: # __LOG__.debug('*** latest one {0} which end on {1}'.format(t.name, t.end_date)) # prev_task_end = t.end_date if prev_task_end > current_day: start = prev_task_end + datetime.timedelta(days=1) # return prev_task_end else: start = current_day while start.weekday() in _not_worked_days() or start in VACATIONS: start = start + datetime.timedelta(days=1) depend_start_date = start if depend_start_date > current_day: _logger.error( f"** Due to dependencies, Task '{self.full_name}', could not be finished on time (should start " f"as last on {current_day} but will start on {depend_start_date})" ) self._cache_start_date = depend_start_date else: # should be first day of start... self._cache_start_date = depend_start_date else: # should be first day of start... self._cache_start_date = current_day if self._cache_start_date != self._start: _logger.warning( '** starting date for task "{0}" is changed from {1} to {2}'.format( self.fullname, self._start, self._cache_start_date ) ) return self._cache_start_date
[docs] def set_end_date(self, end_date): """ Set a end date, overriding the previous end date Parameters ---------- end_date: datetime End date """ self._cache_end_date = end_date self._end = end_date
@property def end_date(self): """ Returns the last day of the task, either the one which was given at task creation or the one calculated after checking dependencies """ # Should take care of resources vacations ? if self._cache_end_date is not None: return self._cache_end_date _logger.debug("** Task::end_date ({0})".format(self.name)) if self.duration is None or self._start is None and self._stop is not None: real_end = self._stop # Take care of vacations while real_end.weekday() in _not_worked_days() or real_end in VACATIONS: real_end -= datetime.timedelta(days=1) if real_end <= self.start_date: current_day = self.start_date real_duration = 0 if self.duration is None: msg = ( f"End time {real_end} is before start time {self.start_date} and no duration is given for\n" f"project '{self.name}'. Please fix" ) raise AssertionError(msg) duration: int = self.duration while duration > 1 or ( current_day.weekday() in _not_worked_days() or current_day in VACATIONS ): if not ( current_day.weekday() in _not_worked_days() or current_day in VACATIONS ): real_duration = real_duration + 1 duration -= 1 else: real_duration = real_duration + 1 current_day = self.start_date + datetime.timedelta( days=real_duration ) self._cache_end_date = self.start_date + datetime.timedelta( days=real_duration ) _logger.warning( '** task "{0}" will not be finished on time : end_date is changed from {1} to {2}'.format( self.fullname, self._stop, self._cache_end_date ) ) return self._cache_end_date self._cache_end_date = real_end if real_end != self._stop: _logger.warning( '** task "{0}" will not be finished on time : end_date is changed from {1} to {2}'.format( self.fullname, self._stop, self._cache_end_date ) ) return self._cache_end_date if self._stop is None: current_day = self.start_date real_duration = 0 duration = self.duration while duration > 1 or ( current_day.weekday() in _not_worked_days() or current_day in VACATIONS ): if not ( current_day.weekday() in _not_worked_days() or current_day in VACATIONS ): real_duration = real_duration + 1 duration -= 1 else: real_duration = real_duration + 1 current_day = self.start_date + datetime.timedelta(days=real_duration) self._cache_end_date = self.start_date + datetime.timedelta( days=real_duration ) return self._cache_end_date raise AssertionError("Something happend that should not")
[docs] def svg( self, prev_y: int = 0, start: date = None, end: date = None, planning_start: date = None, planning_end: date = None, color: str = None, level: int = None, scale: str = DRAW_WITH_DAILY_SCALE, title_align_on_left: bool = False, offset: float = 0, ): """ Get the SVG for drawing this task. Args: prev_y (int): line to start to draw start (date): First day to draw end (date): Last day to draw planning_start (date): Not used here planning_end (date): Not used here color (str): color for drawing the project level (int): Indentation level of the project, not used here scale (str): Drawing scale (d: days, w: weeks, m: months, q: quarterly) title_align_on_left (bool): align task title on left offset (float): X offset from image border to start of drawing zone Returns: Container, number of lines: the svg containter with the start line """ _logger.debug( "** Task::svg ({0})".format( { "name": self.name, "prev_y": prev_y, "start": start, "end": end, "color": color, "level": level, } ) ) if not self.display: _logger.debug("** Task::svg ({0}) display off".format({"name": self.name})) return None, 0 add_modified_begin_mark = False add_modified_end_mark = False if start is None: start = self.start_date if self._start is not None and self.start_date != self._start: add_modified_begin_mark = True if end is None: end = self.end_date if self._stop is not None and self.end_date != self._stop: add_modified_end_mark = True # override project color if defined if self.color is not None: color = self.color add_begin_mark = False add_end_mark = False y = prev_y * 10 if scale == DRAW_WITH_DAILY_SCALE: def _time_diff(e, s): return (e - s).days def _time_diff_d(e, s): return _time_diff(e, s) + 1 elif scale == DRAW_WITH_WEEKLY_SCALE: def _time_diff(end_date, start_date): td = 0 guess = start_date while guess.weekday() != 0: guess = guess + dateutil.relativedelta.relativedelta(days=-1) while end_date.weekday() != 6: end_date = end_date + dateutil.relativedelta.relativedelta(days=+1) while guess + dateutil.relativedelta.relativedelta(days=+6) < end_date: td += 1 guess = guess + dateutil.relativedelta.relativedelta(weeks=+1) return td def _time_diff_d(e, s): return _time_diff(e, s) + 1 elif scale == DRAW_WITH_MONTHLY_SCALE: def _time_diff(end_date, start_date): return ( dateutil.relativedelta.relativedelta(end_date, start_date).months + dateutil.relativedelta.relativedelta(end_date, start_date).years * 12 ) def _time_diff_d(e, s): return _time_diff(e, s) + 1 elif scale == DRAW_WITH_QUARTERLY_SCALE: raise ValueError("DRAW_WITH_QUARTERLY_SCALE not implemented yet") else: raise AssertionError(f"scale {scale} not recognised") # cas 1 -s--S==E--e- if self.start_date >= start and self.end_date <= end: x = _time_diff(self.start_date, start) * 10 d = _time_diff_d(self.end_date, self.start_date) * 10 self.drawn_x_begin_coord = x self.drawn_x_end_coord = x + d # cas 5 -s--e--S==E- elif self.start_date > end: return None, 0 # cas 6 -S==E-s--e- elif self.end_date < start: return None, 0 # cas 2 -S==s==E--e- elif self.start_date < start and self.end_date <= end: x = 0 d = _time_diff_d(self.end_date, start) * 10 self.drawn_x_begin_coord = x self.drawn_x_end_coord = x + d add_begin_mark = True # cas 3 -s--S==e==E- elif self.start_date >= start and self.end_date > end: x = _time_diff(self.start_date, start) * 10 d = _time_diff_d(end, self.start_date) * 10 self.drawn_x_begin_coord = x self.drawn_x_end_coord = x + d add_end_mark = True # cas 4 -S==s==e==E- elif self.start_date < start and self.end_date > end: x = 0 d = _time_diff_d(end, start) * 10 self.drawn_x_begin_coord = x self.drawn_x_end_coord = x + d add_end_mark = True add_begin_mark = True else: return None, 0 self.drawn_y_coord = y svg = svg_Group(id=re.sub(r"[ ,'/()]", "_", self.name)) svg.add( svg_Rect( insert=((x + 1 + offset) * mm, (y + 1) * mm), size=((d - 2) * mm, 8 * mm), fill=color, stroke=color, stroke_width=2, opacity=0.85, ) ) svg.add( svg_Rect( insert=((x + 1 + offset) * mm, (y + 6) * mm), size=((d - 2) * mm, 3 * mm), fill="#909090", stroke=color, stroke_width=1, opacity=0.2, ) ) if add_modified_begin_mark: svg.add( svg_Rect( insert=((x + 1) * mm, (y + 1) * mm), size=(5 * mm, 4 * mm), fill="#0000FF", stroke=color, stroke_width=1, opacity=0.35, ) ) if add_modified_end_mark: svg.add( svg_Rect( insert=((x + d - 7 + 1) * mm, (y + 1) * mm), size=(5 * mm, 4 * mm), fill="#0000FF", stroke=color, stroke_width=1, opacity=0.35, ) ) if add_begin_mark: svg.add( svg_Rect( insert=((x + 1) * mm, (y + 1) * mm), size=(5 * mm, 8 * mm), fill="#000000", stroke=color, stroke_width=1, opacity=0.2, ) ) if add_end_mark: svg.add( svg_Rect( insert=((x + d - 7 + 1) * mm, (y + 1) * mm), size=(5 * mm, 8 * mm), fill="#000000", stroke=color, stroke_width=1, opacity=0.2, ) ) if self.percent_done is not None and self.percent_done > 0: # Bar shade svg.add( svg_Rect( insert=((x + 1 + offset) * mm, (y + 6) * mm), size=(((d - 2) * self.percent_done / 100) * mm, 3 * mm), fill="#F08000", stroke=color, stroke_width=1, opacity=0.35, ) ) if not title_align_on_left: tx = x + 2 else: tx = 5 svg.add( svg_Text( self.fullname, insert=(tx * mm, (y + 5) * mm), fill=_font_attributes()["fill"], stroke=_font_attributes()["stroke"], stroke_width=_font_attributes()["stroke_width"], font_family=_font_attributes()["font_family"], font_size=15, ) ) if self.resources is not None: t = " / ".join(["{0}".format(r.name) for r in self.resources]) svg.add( svg_Text( "{0}".format(t), insert=(tx * mm, (y + 8.5) * mm), fill="purple", stroke=_font_attributes()["stroke"], stroke_width=_font_attributes()["stroke_width"], font_family=_font_attributes()["font_family"], font_size=15 - 5, ) ) return svg, 1
[docs] def svg_dependencies(self, prj): """ Draws svg dependencies between task and project according to coordinates cached when drawing tasks Keyword arguments: prj -- Project object to check against """ _logger.debug( "** Task::svg_dependencies ({0})".format({"name": self.name, "prj": prj}) ) if self.depends_of is None: return None else: svg = svg_Group() for t in self.depends_of: if isinstance(t, Milestone): if not ( t.drawn_x_end_coord is None or t.drawn_y_coord is None or self.drawn_x_begin_coord is None ) and prj.is_in_project(t): if t.drawn_x_end_coord < self.drawn_x_begin_coord: # horizontal line svg.add( svgwrite.shapes.Line( start=( (t.drawn_x_end_coord + 9) * mm, (t.drawn_y_coord + 5) * mm, ), end=( self.drawn_x_begin_coord * mm, (t.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) marker = svgwrite.container.Marker( insert=(5, 5), size=(10, 10) ) marker.add( svgwrite.shapes.Circle( (5, 5), r=5, fill="#000000", opacity=0.5, stroke_width=0, ) ) svg.add(marker) # vertical line eline = svgwrite.shapes.Line( start=( self.drawn_x_begin_coord * mm, (t.drawn_y_coord + 5) * mm, ), end=( self.drawn_x_begin_coord * mm, (self.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) eline["marker-end"] = marker.get_funciri() svg.add(eline) else: # horizontal line svg.add( svgwrite.shapes.Line( start=( (t.drawn_x_end_coord + 9) * mm, (t.drawn_y_coord + 5) * mm, ), end=( (self.drawn_x_begin_coord + 10) * mm, (t.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) # vertical svg.add( svgwrite.shapes.Line( start=( (self.drawn_x_begin_coord + 10) * mm, (t.drawn_y_coord + 5) * mm, ), end=( (self.drawn_x_begin_coord + 10) * mm, (t.drawn_y_coord + 15) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) # horizontal line svg.add( svgwrite.shapes.Line( start=( self.drawn_x_begin_coord * mm, (t.drawn_y_coord + 15) * mm, ), end=( (self.drawn_x_begin_coord + 10) * mm, (t.drawn_y_coord + 15) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) marker = svgwrite.container.Marker( insert=(5, 5), size=(10, 10) ) marker.add( svgwrite.shapes.Circle( (5, 5), r=5, fill="#000000", opacity=0.5, stroke_width=0, ) ) svg.add(marker) # vertical line eline = svgwrite.shapes.Line( start=( self.drawn_x_begin_coord * mm, (t.drawn_y_coord + 15) * mm, ), end=( self.drawn_x_begin_coord * mm, (self.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) eline["marker-end"] = marker.get_funciri() svg.add(eline) elif isinstance(t, Task): if not ( t.drawn_x_end_coord is None or t.drawn_y_coord is None or self.drawn_x_begin_coord is None ) and prj.is_in_project(t): # horizontal line svg.add( svgwrite.shapes.Line( start=( (t.drawn_x_end_coord - 2) * mm, (t.drawn_y_coord + 5) * mm, ), end=( self.drawn_x_begin_coord * mm, (t.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) marker = svgwrite.container.Marker(insert=(5, 5), size=(10, 10)) marker.add( svgwrite.shapes.Circle( (5, 5), r=5, fill="#000000", opacity=0.5, stroke_width=0 ) ) svg.add(marker) # vertical line eline = svgwrite.shapes.Line( start=( self.drawn_x_begin_coord * mm, (t.drawn_y_coord + 5) * mm, ), end=( self.drawn_x_begin_coord * mm, (self.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) eline["marker-end"] = marker.get_funciri() svg.add(eline) return svg
[docs] def nb_elements(self): """ Returns the number of task, 1 here """ _logger.debug("** Task::nb_elements ({0})".format({"name": self.name})) return 1
def _reset_coord(self): """ Reset cached elements of task """ _logger.debug("** Task::reset_coord ({0})".format({"name": self.name})) self.drawn_x_begin_coord = None self.drawn_x_end_coord = None self.drawn_y_coord = None self._cache_start_date = None self._cache_end_date = None return
[docs] def is_in_project(self, task): """ Return True if the given Task is itself... (lazy coding ;) Keyword arguments: task -- Task object """ _logger.debug( "** Task::is_in_project ({0})".format({"name": self.name, "task": task}) ) if task is self: return True return False
[docs] def get_resources(self): """ Returns Resources used in the task """ return self.resources
[docs] def check_conflicts_between_task_and_resources_vacations(self): """ Displays a warning for each conflict between tasks and vacation of resources affected to the task And returns a dictionary for resource vacation conflicts """ conflicts = [] if self.get_resources() is None: return conflicts for r in self.get_resources(): current_day = self.start_date while current_day <= self.end_date: if ( current_day.weekday() not in _not_worked_days() and not r.is_available(current_day) ): conflicts.append( {"resource": r.name, "date": current_day, "task": self.name} ) _logger.warning( '** Caution resource "{0}" is affected on task "{2}" during vacations on day {1}'.format( r.name, current_day, self.fullname ) ) current_day += datetime.timedelta(days=1) return conflicts
[docs] def csv(self, csv=None): """ Create CSV output from tasks Keyword arguments: csv -- None, dymmy object """ if self.resources is not None: resources = ", ".join([x.fullname for x in self.resources]) else: resources = "" csv_text = '"{0}";"{1}";{2};{3};{4};"{5}";\r\n'.format( self.state.replace('"', '\\"'), self.fullname.replace('"', '\\"'), self.start_date, self.end_date, self.duration, resources.replace('"', '\\"'), ) return csv_text
############################################################################
[docs] class Milestone(Task): """ Class for manipulating Milestones """ def __init__( self, name, start=None, depends_of=None, color=None, fullname=None, display=True, parent=None, ): """ Initialize a milestone object. Two of start, stop or duration may be given. This milestone can rely on another milestone and will be completed with resources. If percent done is given, a progress bar will be included on the milestone. If color is specified, it will be used for the milestone. Keyword arguments: name -- name of the milestone (id) fullname -- long name given to the resource start -- datetime.date, first day of the milestone, default None depends_of -- list of Milestone which are parents of this one, default None color -- string, html color, default None display -- boolean, display this milestone, default True """ super().__init__( name=name, start=start, depends_of=depends_of, color=color, fullname=fullname, display=display, parent=parent, ) _logger.debug( "** Milestone::__init__ {0}".format( {"name": name, "start": start, "depends_of": depends_of} ) ) self._stop = start self.duration = 0 if color is not None: self.color = color else: self.color = "#FF3030" self.state = "Milestone" if type(depends_of) is type([]): self.depends_of = depends_of elif depends_of is not None: self.depends_of = [depends_of] else: self.depends_of = None self.drawn_x_begin_coord = None self.drawn_x_end_coord = None self.drawn_y_coord = None self._cache_start_date = None self._cache_end_date = None return @property def end_date(self): """ Returns the last day of the milestone, either the one which was given at milestone creation or the one calculated after checking dependencies """ _logger.debug("** Milestone::end_date ({0})".format(self.name)) # return self.start_date - datetime.timedelta(days=1) return self.start_date
[docs] def svg( self, prev_y=0, start=None, end=None, planning_start=None, planning_end=None, color=None, level=None, scale=DRAW_WITH_DAILY_SCALE, title_align_on_left=False, offset=0, ): """ Return SVG for drawing this milestone. Keyword arguments: prev_y -- int, line to start to draw start -- datetime.date of first day to draw end -- datetime.date of last day to draw planning_start -- datetime.date start date of planning planning_end -- datetime.date end date of planning color -- string of color for drawing the project level -- int, indentation level of the project, not used here scale -- drawing scale (d: days, w: weeks, m: months, q: quarterly) title_align_on_left -- boolean, align milestone title on left offset -- X offset from image border to start of drawing zone """ _logger.debug( "** Milestone::svg ({0})".format( { "name": self.name, "prev_y": prev_y, "start": start, "end": end, "color": color, "level": level, } ) ) if not self.display: _logger.debug( "** Milestone::svg ({0}) display off".format({"name": self.name}) ) return None, 0 # add_modified_begin_mark = False # add_modified_end_mark = False if start is None: start = self.start_date # if self.start_date != self.start and self.start is not None: # add_modified_begin_mark = True if end is None: end = self.end_date # if self.end_date != self.stop and self.stop is not None: # add_modified_end_mark = True # override project color if defined if self.color is not None: color = self.color # add_begin_mark = False # add_end_mark = False y = prev_y * 10 if scale == DRAW_WITH_DAILY_SCALE: def _time_diff(e, s): return (e - s).days def _time_diff_d(e, s): return _time_diff(e, s) + 1 elif scale == DRAW_WITH_WEEKLY_SCALE: def _time_diff(end_date, start_date): td = 0 guess = start_date # find first day of the week while guess.weekday() != 0: guess = guess + dateutil.relativedelta.relativedelta(days=-1) # find last day of the week while end_date.weekday() != 6: end_date = end_date + dateutil.relativedelta.relativedelta(days=+1) while guess <= end_date: td += 1 guess = guess + dateutil.relativedelta.relativedelta(weeks=+1) return td - 1 def _time_diff_d(e, s): return _time_diff(e, s) + 1 elif scale == DRAW_WITH_MONTHLY_SCALE: def _time_diff(end_date, start_date): return ( dateutil.relativedelta.relativedelta(end_date, start_date).months + dateutil.relativedelta.relativedelta(end_date, start_date).years * 12 ) def _time_diff_d(e, s): return _time_diff(e, s) + 1 elif scale == DRAW_WITH_QUARTERLY_SCALE: raise ValueError("DRAW_WITH_QUARTERLY_SCALE not implemented yet") else: raise AssertionError(f"scale {scale} not recognised") # cas 1 -s--X--e- if self.start_date >= start and self.end_date <= end: x = _time_diff(self.start_date, start) * 10 self.drawn_x_begin_coord = x self.drawn_x_end_coord = x else: return None, 0 self.drawn_y_coord = y # insert=((x+1)*mm, (y+1)*mm), # size=((d-2)*mm, 8*mm), svg = svg_Group(id=re.sub(r"[ ,'/()]", "_", self.name)) # 3.543307 is for conversion from mm to pt units ! svg.add( svgwrite.shapes.Polygon( points=[ ((x + 5 + offset) * mm, (y + 2) * mm), ((x + 8 + offset) * mm, (y + 5) * mm), ((x + 5 + offset) * mm, (y + 8) * mm), ((x + 2 + offset) * mm, (y + 5) * mm), ], fill=color, stroke=color, stroke_width=2, opacity=0.85, ) ) if not title_align_on_left: tx = x + 2 else: tx = 5 svg.add( svg_Text( self.fullname, insert=(tx * mm, (y + 5) * mm), fill=_font_attributes()["fill"], stroke=_font_attributes()["stroke"], stroke_width=_font_attributes()["stroke_width"], font_family=_font_attributes()["font_family"], font_size=15, ) ) return svg, 2
[docs] def svg_dependencies(self, prj): """ Draws svg dependencies between milestone and project according to coordinates cached when drawing milestones Keyword arguments: prj -- Project object to check against """ _logger.debug( "** Milestone::svg_dependencies ({0})".format( {"name": self.name, "prj": prj} ) ) if self.depends_of is None: return None else: svg = svg_Group() for t in self.depends_of: if isinstance(t, Milestone): if not ( t.drawn_x_end_coord is None or t.drawn_y_coord is None or self.drawn_x_begin_coord is None ) and prj.is_in_project(t): # horizontal line svg.add( svgwrite.shapes.Line( start=( (t.drawn_x_end_coord + 9) * mm, (t.drawn_y_coord + 5) * mm, ), end=( (self.drawn_x_begin_coord + 5) * mm, (t.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) marker = svgwrite.container.Marker(insert=(5, 5), size=(10, 10)) marker.add( svgwrite.shapes.Circle( (5, 5), r=5, fill="#000000", opacity=0.5, stroke_width=0 ) ) svg.add(marker) # vertical line eline = svgwrite.shapes.Line( start=( (self.drawn_x_begin_coord + 5) * mm, (t.drawn_y_coord + 5) * mm, ), end=( (self.drawn_x_begin_coord + 5) * mm, self.drawn_y_coord * mm, ), stroke="black", stroke_dasharray="5,3", ) eline["marker-end"] = marker.get_funciri() svg.add(eline) elif isinstance(t, Task): if not ( t.drawn_x_end_coord is None or t.drawn_y_coord is None or self.drawn_x_begin_coord is None ) and prj.is_in_project(t): # horizontal line svg.add( svgwrite.shapes.Line( start=( (t.drawn_x_end_coord - 2) * mm, (t.drawn_y_coord + 5) * mm, ), end=( (self.drawn_x_begin_coord + 5) * mm, (t.drawn_y_coord + 5) * mm, ), stroke="black", stroke_dasharray="5,3", ) ) marker = svgwrite.container.Marker(insert=(5, 5), size=(10, 10)) marker.add( svgwrite.shapes.Circle( (5, 5), r=5, fill="#000000", opacity=0.5, stroke_width=0 ) ) svg.add(marker) # vertical line eline = svgwrite.shapes.Line( start=( (self.drawn_x_begin_coord + 5) * mm, (t.drawn_y_coord + 5) * mm, ), end=( (self.drawn_x_begin_coord + 5) * mm, (self.drawn_y_coord + 0) * mm, ), stroke="black", stroke_dasharray="5,3", ) eline["marker-end"] = marker.get_funciri() svg.add(eline) return svg
[docs] def get_resources(self): """ Returns Resources used in the milestone """ return []
[docs] def check_conflicts_between_task_and_resources_vacations(self): """ Displays a warning for each conflict between milestones and vacation of resources affected to the milestone And returns a dictionary for resource vacation conflicts """ return []
[docs] def csv(self, csv=None): """ Create CSV output from milestones Keyword arguments: csv -- None, dymmy object """ if self.resources is not None: resources = ", ".join([x.fullname for x in self.resources]) else: resources = "" csv_text = '"{0}";"{1}";{2};{3};{4};"{5}";\r\n'.format( self.state.replace('"', '\\"'), self.fullname.replace('"', '\\"'), self.start_date, self.end_date, self.duration, resources.replace('"', '\\"'), ) return csv_text
[docs] class Project: """ Class for handling projects """ def __init__( self, name="", color=None, side_bar_color=None, project_start=None, project_end=None, font=None, ): """ Initialize project with a given name and color for all tasks Keyword arguments: name -- string, name of the project color -- color for all tasks of the project """ self.tasks = [] self.name = name if color is None: self.color = "#FFFF90" else: self.color = color self.font = font self.project_start = project_start self.project_end = project_end if side_bar_color is None: self.side_bar_color = self.color else: self.side_bar_color = side_bar_color self.cache_nb_elements = None return
[docs] def add_task(self, task): """ Add a Task to the Project. Task can also be a subproject Keyword arguments: task -- Task or Project object """ self.tasks.append(task) self.cache_nb_elements = None return
@staticmethod def _svg_calendar( maxx: int, maxy: int, start_date: date, today: date = None, scale: str = DRAW_WITH_DAILY_SCALE, offset: float = 0, ): """ Draw calendar in svg, beginning at start_date for maxx days, containing maxy lines. If today is given, draw a blue line at date Args: maxx (int): Number of days, weeks, months or quarters (depending on scale) to draw maxy (int): Number of lines to draw start_date (date): The first day to draw today (date): Day as today reference scale (str): Drawing scale (d: days, w: weeks, m: months, q: quarterly) offset (float): X offset from image border to start of drawing zone """ dwg = svg_Group() cal = {0: "Mo", 1: "Tu", 2: "We", 3: "Th", 4: "Fr", 5: "Sa", 6: "Su"} maxx += 1 vlines = dwg.add(svg_Group(id="vlines", stroke="lightgray")) for x in range(maxx): vlines.add( svgwrite.shapes.Line( start=((x + offset / 10) * cm, 2 * cm), end=((x + offset / 10) * cm, (maxy + 2) * cm), ) ) if scale == DRAW_WITH_DAILY_SCALE: jour = start_date + datetime.timedelta(days=x) elif scale == DRAW_WITH_WEEKLY_SCALE: jour = start_date + dateutil.relativedelta.relativedelta(weeks=+x) elif scale == DRAW_WITH_MONTHLY_SCALE: jour = start_date + dateutil.relativedelta.relativedelta(months=+x) elif scale == DRAW_WITH_QUARTERLY_SCALE: raise ValueError("DRAW_WITH_QUARTERLY_SCALE not implemented yet") else: raise AssertionError(f"scale {scale} not recognised") if today is not None and today == jour: vlines.add( svg_Rect( insert=((x + 0.4 + offset) * cm, 2 * cm), size=(0.2 * cm, maxy * cm), fill="#76e9ff", stroke="lightgray", stroke_width=0, opacity=0.8, ) ) if scale == DRAW_WITH_DAILY_SCALE: # draw vacations if ( start_date + datetime.timedelta(days=x) ).weekday() in _not_worked_days() or ( start_date + datetime.timedelta(days=x) ) in VACATIONS: vlines.add( svg_Rect( insert=((x + offset / 10) * cm, 2 * cm), size=(1 * cm, maxy * cm), fill="gray", stroke="lightgray", stroke_width=1, opacity=0.7, ) ) # Current day vlines.add( svg_Text( "{1} {0:02}".format(jour.day, cal[jour.weekday()][0]), insert=((x * 10 + 1 + offset) * mm, 19 * mm), fill="black", stroke="black", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 - 3, ) ) # Year if jour.day == 1 and jour.month == 1: vlines.add( svg_Text( "{0}".format(jour.year), insert=((x * 10 + 1 + offset) * mm, 5 * mm), fill="#400000", stroke="#400000", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 5, font_weight="bold", ) ) # Month name if jour.day == 1: vlines.add( svg_Text( "{0}".format(jour.strftime("%B")), insert=((x * 10 + 1 + offset) * mm, 10 * mm), fill="#800000", stroke="#800000", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 3, font_weight="bold", ) ) # Week number if jour.weekday() == 0: vlines.add( svg_Text( "{0:02}".format(jour.isocalendar()[1]), insert=((x * 10 + 1 + offset) * mm, 15 * mm), fill="black", stroke="black", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 1, font_weight="bold", ) ) elif scale == DRAW_WITH_WEEKLY_SCALE: # Year if jour.isocalendar()[1] == 1 and jour.month == 1: vlines.add( svg_Text( "{0}".format(jour.year), insert=((x * 10 + 1 + offset) * mm, 5 * mm), fill="#400000", stroke="#400000", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 5, font_weight="bold", ) ) # Month name if jour.day <= 7: vlines.add( svg_Text( "{0}".format(jour.strftime("%B")), insert=((x * 10 + 1 + offset) * mm, 10 * mm), fill="#800000", stroke="#800000", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 3, font_weight="bold", ) ) vlines.add( svg_Text( "{0:02}".format(jour.isocalendar()[1]), insert=((x * 10 + 1 + offset) * mm, 15 * mm), fill="black", stroke="black", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 1, font_weight="bold", ) ) elif scale == DRAW_WITH_MONTHLY_SCALE: # Month number vlines.add( svg_Text( "{0}".format(jour.strftime("%m")), insert=((x * 10 + 1 + offset) * mm, 19 * mm), fill="black", stroke="black", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 - 3, ) ) # Year if jour.month == 1: vlines.add( svg_Text( "{0}".format(jour.year), insert=((x * 10 + 1 + offset) * mm, 5 * mm), fill="#400000", stroke="#400000", stroke_width=0, font_family=_font_attributes()["font_family"], font_size=15 + 5, font_weight="bold", ) ) elif scale == DRAW_WITH_QUARTERLY_SCALE: raise ValueError("DRAW_WITH_QUARTERLY_SCALE not implemented yet") else: raise AssertionError(f"scale {scale} not recognised") vlines.add( svgwrite.shapes.Line( start=((maxx + offset / 10) * cm, 2 * cm), end=((maxx + offset / 10) * cm, (maxy + 2) * cm), ) ) hlines = dwg.add(svg_Group(id="hlines", stroke="lightgray")) dwg.add( svgwrite.shapes.Line( start=((0 + offset / 10) * cm, (2) * cm), end=((maxx + offset / 10) * cm, (2) * cm), stroke="black", ) ) dwg.add( svgwrite.shapes.Line( start=((0 + offset / 10) * cm, (maxy + 2) * cm), end=((maxx + offset / 10) * cm, (maxy + 2) * cm), stroke="black", ) ) for y in range(2, maxy + 3): hlines.add( svgwrite.shapes.Line( start=((0 + offset / 10) * cm, y * cm), end=((maxx + offset / 10) * cm, y * cm), ) ) return dwg
[docs] def make_svg_for_tasks( self, filename, today=None, start=None, end=None, margin_left=None, margin_right=None, scale=DRAW_WITH_DAILY_SCALE, title_align_on_left=False, offset=0, ): """ Draw gantt of tasks and output it to filename. Args: filename (str): Filename to save to OR file object today (date): Date of day marked as a reference start (date): The first day to draw end (date): The last day to draw margin_left (int): Number of week to add to the grid before the project start margin_right (int): Number of week to add to the grid after the project end scale (str): drawing scale (d: days, w: weeks, m: months, q: quarterly) title_align_on_left (bool): Align task title on left offset (float): X offset from image border to start of drawing zone Notes: * If start or end are given, use them as reference, otherwise use project first and last day """ if len(self.tasks) == 0: _logger.warning("** Empty project : {0}".format(self.name)) return self._reset_coord() if start is None: start_date = self.start_date else: start_date = start if end is None: end_date = self.end_date else: end_date = end # add a margin to the left and right of the right in order to make some space for the project labels if margin_left is not None: start_date -= datetime.timedelta(weeks=margin_left) if margin_right is not None: end_date += datetime.timedelta(weeks=margin_right) if start_date > end_date: _logger.critical( "start date {0} > end_date {1}".format(start_date, end_date) ) sys.exit(1) svg_container_group = svg_Group() psvg, pheight = self.svg( prev_y=2, start=start_date, end=end_date, planning_start=start, planning_end=end, color=self.side_bar_color, scale=scale, title_align_on_left=title_align_on_left, offset=offset, ) if psvg is not None: svg_container_group.add(psvg) dep = self.svg_dependencies(self) if dep is not None: svg_container_group.add(dep) if scale == DRAW_WITH_DAILY_SCALE: # how many days do we need to draw ? max_x_grid_lines = (end_date - start_date).days elif scale == DRAW_WITH_WEEKLY_SCALE: # how many weeks do we need to draw ? max_x_grid_lines = 0 guess = start_date while guess.weekday() != 0: guess = guess + dateutil.relativedelta.relativedelta(days=-1) while end_date.weekday() != 6: end_date = end_date + dateutil.relativedelta.relativedelta(days=+1) while guess <= end_date: max_x_grid_lines += 1 guess = guess + dateutil.relativedelta.relativedelta(weeks=+1) elif scale == DRAW_WITH_MONTHLY_SCALE: # how many months do we need to draw ? if dateutil.relativedelta.relativedelta(end_date, start_date).days == 0: max_x_grid_lines = ( dateutil.relativedelta.relativedelta(end_date, start_date).months + dateutil.relativedelta.relativedelta(end_date, start_date).years * 12 ) else: max_x_grid_lines = ( dateutil.relativedelta.relativedelta(end_date, start_date).months + dateutil.relativedelta.relativedelta(end_date, start_date).years * 12 + 1 ) elif scale == DRAW_WITH_QUARTERLY_SCALE: raise ValueError("DRAW_WITH_QUARTERLY_SCALE not implemented yet") else: raise AssertionError(f"scale {scale} not recognised") dwg = MySVGWriteDrawingWrapper(filename, debug=True) dwg.add( svg_Rect( insert=(0 * cm, 0 * cm), size=((max_x_grid_lines + 1 + offset / 10) * cm, (pheight + 3) * cm), fill="white", stroke_width=0, opacity=1, ) ) dwg.add( self._svg_calendar( max_x_grid_lines, pheight, start_date, today, scale, offset=offset ) ) dwg.add(svg_container_group) dwg.save( width=(max_x_grid_lines + 1 + offset / 10) * cm, height=(pheight + 3) * cm )
[docs] def make_svg_for_resources( self, filename: str, today: date = None, start: date = None, end: date = None, resources: list = None, one_line_for_tasks: bool = False, tag_filter: str = "", scale: str = DRAW_WITH_DAILY_SCALE, title_align_on_left: bool = False, offset: float = 0, color_per_taks: bool = False, ): """ Draw resources affectation and output it to filename. If start or end are given, use them as reference, otherwise use project first and last day And returns to a dictionary of dictionaries for vacation and task conflicts for resources Args: color_per_taks (bool): Use color per task filename (str): filename to save to OR file object today (date): Day marked as a reference start (date): First day to draw end (date): Last day to draw resources (list) Resources to check, default all one_line_for_tasks (bool): Use only one line to display all tasks ? tag_filter (bool): Display only those tags scale (str): Drawing scale (d: days, w: weeks, m: months, q: quarterly) title_align_on_left (bool): Align task title on left offset (float): X offset from image border to start of drawing zone """ if scale != DRAW_WITH_DAILY_SCALE: _logger.warning( "** Will draw ressource graph at day scale, not {0} as requested".format( scale ) ) scale = DRAW_WITH_DAILY_SCALE if len(self.tasks) == 0: _logger.warning("** Empty project : {0}".format(self.name)) return self._reset_coord() if start is None: start_date = self.start_date else: start_date = start if end is None: end_date = self.end_date else: end_date = end if start_date > end_date: _logger.critical( "start date {0} > end_date {1}".format(start_date, end_date) ) sys.exit(1) if resources is None: resources = self.get_resources() number_of_x_grid_lines = (end_date - start_date).days number_of_y_grid_lines = len(resources) * 2 if number_of_y_grid_lines == 0: # No resources return {} # detect conflicts between resources and holidays conflicts_vacations = [] for task in self.get_tasks(): conflicts_vacations.append( task.check_conflicts_between_task_and_resources_vacations() ) conflicts_vacations = _flatten(conflicts_vacations) ldwg = svg_Group() if not one_line_for_tasks: ldwg.add( svgwrite.shapes.Line( start=(0 * cm, 2 * cm), end=((number_of_x_grid_lines + 1 + offset / 10) * cm, 2 * cm), stroke="black", ) ) line_number = 2 conflicts_tasks = [] conflict_display_line = 1 for r in resources: # do stuff for each resource if tag_filter != "" and r.name not in tag_filter: continue ress = svg_Group() try: res_text = svg_Text( "{0}".format(r.fullname), insert=(3 * mm, (line_number * 10 + 7) * mm), fill=_font_attributes()["fill"], stroke=_font_attributes()["stroke"], stroke_width=_font_attributes()["stroke_width"], font_family=_font_attributes()["font_family"], font_size=15 + 3, ) except ValueError as err: _logger.warning(f"Failed making text object for {r.fullname}") else: ress.add(res_text) overcharged_days = r.search_for_task_conflicts() conflict_display_line = line_number line_number += 1 vac = svg_Group() conflicts = svg_Group() current_day = start_date while current_day <= end_date: # Vacations if ( current_day.weekday() not in _not_worked_days() and current_day not in VACATIONS and not r.is_available(current_day) ): vac.add( svg_Rect( insert=( ((current_day - start_date).days * 10 + 1 + offset) * mm, (conflict_display_line * 10 + 1) * mm, ), size=(4 * mm, 8 * mm), fill=COLOR_VACATION_DEFAULT, stroke=COLOR_VACATION_DEFAULT, stroke_width=1, opacity=0.65, ) ) # Overcharge if ( current_day.weekday() not in _not_worked_days() and current_day not in VACATIONS and current_day in overcharged_days ): conflicts.add( svg_Rect( insert=( ((current_day - start_date).days * 10 + 1 + 4 + offset) * mm, (conflict_display_line * 10 + 1) * mm, ), size=(4 * mm, 8 * mm), fill=COLOR_OVERCHARGE_DEFAULT, stroke=COLOR_OVERCHARGE_DEFAULT, stroke_width=1, opacity=0.65, ) ) current_day += datetime.timedelta(days=1) nb_tasks = 0 for task in self.get_tasks(): if task.get_resources() is not None and r in task.get_resources(): if color_per_taks: color = task.color else: color = r.color if color is None: color = COLOR_RESOURCE_DEFAULT psvg, void = task.svg( prev_y=line_number, start=start_date, end=end_date, color=color, scale=scale, title_align_on_left=title_align_on_left, offset=offset, ) if psvg is not None: ldwg.add(psvg) nb_tasks += 1 if not one_line_for_tasks: line_number += 1 if nb_tasks == 0: line_number -= 1 elif nb_tasks > 0: _logger.info(r.fullname, nb_tasks) ldwg.add(ress) ldwg.add(vac) ldwg.add(conflicts) if not one_line_for_tasks: ldwg.add( svgwrite.shapes.Line( start=(0 * cm, line_number * cm), end=( (number_of_x_grid_lines + 1 + offset / 10) * cm, line_number * cm, ), stroke="black", ) ) # nline += 1 if one_line_for_tasks: line_number += 1 ldwg.add( svgwrite.shapes.Line( start=(0 * cm, line_number * cm), end=((number_of_x_grid_lines + 1) * cm, line_number * cm), stroke="black", ) ) dwg = MySVGWriteDrawingWrapper(filename, debug=True) dwg.add( svg_Rect( insert=(0 * cm, 0 * cm), size=( (number_of_x_grid_lines + 1 + offset / 10) * cm, (line_number + 1) * cm, ), fill="white", stroke_width=0, opacity=1, ) ) dwg.add( self._svg_calendar( number_of_x_grid_lines, line_number - 2, start_date, today, scale, offset=offset, ) ) dwg.add(ldwg) dwg.save( width=(number_of_x_grid_lines + 1 + offset / 10) * cm, height=(line_number + 1) * cm, ) return { "conflicts_vacations": conflicts_vacations, "conflicts_tasks": conflicts_tasks, }
@property def start_date(self): """ Returns first day of the project """ if len(self.tasks) == 0: _logger.warning("** Empty project : {0}".format(self.name)) return datetime.date(9999, 1, 1) first = self.tasks[0].start_date for t in self.tasks: if t.start_date < first: first = t.start_date return first @property def end_date(self): """ Returns last day of the project """ if len(self.tasks) == 0: _logger.warning("** Empty project : {0}".format(self.name)) return datetime.date(1970, 1, 1) last = self.tasks[0].end_date for t in self.tasks: if t.end_date > last: last = t.end_date return last
[docs] def svg( self, prev_y=0, start=None, end=None, planning_start=None, planning_end=None, color=None, level=0, scale=DRAW_WITH_DAILY_SCALE, title_align_on_left=False, offset=0, ): """ Draw all tasks and add project name with a purple bar on the left side. Args: prev_y (int): Line to start to draw start (date): First day to draw end (date): Last day to draw planning_start (date): not used here planning_end (date): not used here color (str): Color for drawing the project level (int): Indentation level of the project scale (str): Drawing scale (d: days, w: weeks, m: months, q: quarterly) title_align_on_left (bool): Align task title on left offset (float): X offset from image border to start of drawing zone Returns: svg, int: SVG code and number of lines drawn for the project. """ if start is None: start = self.start_date if end is None: end = self.end_date if color is None or self.color is not None: color = self.color cy = prev_y + 1 * (self.name != "") container_project = svg_Group() for task in self.tasks: if isinstance(task, Task): if not task_within_range(task, planning_start, planning_end): continue trepr, theight = task.svg( cy, start=start, end=end, planning_start=planning_start, planning_end=planning_end, color=color, level=level + 1, scale=scale, title_align_on_left=title_align_on_left, offset=offset, ) if trepr is not None: container_project.add(trepr) cy += theight fprj = svg_Group() prj_bar = False # if margin_left is not None: # start += datetime.timedelta(days=margin_left) # if margin_right is not None: # end -= datetime.timedelta(days=margin_right) if self.name != "": if ( (self.start_date >= start and self.end_date <= end) or (self.end_date >= start and self.start_date <= end) ) or level == 1: # Adjust font level for level 0 (main project) and 1 (projects per project leader) # todo: allow modification in settings file if level == 0: font_weight = "bold" font_size = 18 elif level == 1: font_weight = "bold" font_size = 16 else: font_weight = "normal" font_size = 13 fprj.add( svg_Text( "{0}".format(self.name), insert=( (6 * level + 3 + offset) * mm, (prev_y * 10 + 7) * mm, ), fill=_font_attributes()["fill"], stroke=_font_attributes()["stroke"], stroke_width=_font_attributes()["stroke_width"], font_family=_font_attributes()["font_family"], font_weight=font_weight, font_size=font_size, ) ) fprj.add( svg_Rect( insert=((6 * level + 0.8 + offset) * mm, (prev_y + 0.5) * cm), size=(0.2 * cm, ((cy - prev_y - 1) + 0.4) * cm), fill=self.color, stroke="lightgray", stroke_width=0, opacity=0.5, ) ) prj_bar = True else: cy -= 1 # Do not display empty tasks if (cy - prev_y) == 0 or ((cy - prev_y) == 1 and prj_bar): return None, 0 fprj.add(container_project) return fprj, cy - prev_y
[docs] def svg_dependencies(self, prj): """ Draws svg dependencies between tasks according to coordinates cached when drawing tasks Keyword arguments: prj -- Project object to check against """ svg = svg_Group() for t in self.tasks: trepr = t.svg_dependencies(prj) if trepr is not None: svg.add(trepr) return svg
[docs] def nb_elements(self): """ Returns the number of tasks included in the project or subproject """ if self.cache_nb_elements is not None: return self.cache_nb_elements nb = 0 for t in self.tasks: nb += t.nb_elements() self.cache_nb_elements = nb return nb
def _reset_coord(self): """ Reset cached elements of all tasks and project """ self.cache_nb_elements = None for t in self.tasks: t._reset_coord() return
[docs] def is_in_project(self, task): """ Return True if the given Task is in the project, False if not Keyword arguments: task -- Task object """ for t in self.tasks: if t.is_in_project(task): return True return False
[docs] def get_resources(self): """ Returns Resources used in the project """ rlist = [] for t in self.tasks: r = t.get_resources() if r is not None: rlist.append(r) flist = [] for r in _flatten(rlist): if r not in flist: flist.append(r) return flist
[docs] def get_tasks(self): """ Returns flat list of Tasks used in the Project and subproject """ tlist = [] for t in self.tasks: # if it is a subproject, recurse if type(t) is type(self): st = t.get_tasks() tlist.append(st) else: # get task tlist.append(t) flist = [] for r in _flatten(tlist): if r not in flist: flist.append(r) return flist
[docs] def csv(self, csv=None): """ Create CSV output from projects Keyword arguments: csv -- string, filename to save to OR file object OR None """ if len(self.tasks) == 0: _logger.warning("** Empty project : {0}".format(self.name)) return if csv is not None: csv_text = bytes.decode(codecs.BOM_UTF8, "utf-8") csv_text += '"State";"Task Name";"Start date";"End date";"Duration";"Resources";\r\n' else: csv_text = "" for t in self.tasks: c = t.csv() if c is not None: csv_text += c if csv is not None: test = type(csv) == io.TextIOWrapper if test: csv.write(csv_text) else: with io.open(csv, mode="w", encoding="utf-8") as stream: stream.write(csv_text) return csv_text
[docs] def task_within_range(task: Task, planning_start: datetime, planning_end: datetime): """ Check if the task is in the range of the planning Parameters ---------- task: object The task planning_start: datetime Start of the planning planning_end: datetime End of the planning Returns ------- bool: True if the task is in range """ is_in_range = True if planning_start: try: task_start = task._start except AttributeError: pass else: if task_start < planning_start: is_in_range = False if planning_end: try: task_end = task.end_date except AttributeError: pass else: if task_end > planning_end: is_in_range = False return is_in_range
# MAIN ------------------- if __name__ == "__main__": import doctest # non regression test doctest.testmod() else: init_log_to_sysout(level=logging.CRITICAL) # <EOF>######################################################################