Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[#128] Implemented ScheduleModel Enum class to handle schedule mode… #131

Merged
merged 3 commits into from
Dec 11, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/stalker/models/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ class attribute to control auto naming behavior.
String(64), nullable=True, default=""
)

stalker_version: Mapped[str] = mapped_column(String(256))
stalker_version: Mapped[Optional[str]] = mapped_column(String(256))

def __init__(
self,
Expand Down
179 changes: 125 additions & 54 deletions src/stalker/models/mixins.py
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,7 @@ class TimeUnitDecorator(TypeDecorator):

impl = Enum(*[u.value for u in TimeUnit], name="TimeUnit")

def process_bind_param(self, value, dialect) -> str:
def process_bind_param(self, value: TimeUnit, dialect: str) -> str:
"""Return the str value of the TimeUnit.

Args:
Expand All @@ -357,7 +357,7 @@ def process_bind_param(self, value, dialect) -> str:
# just return the value
return value.value

def process_result_value(self, value, dialect):
def process_result_value(self, value: str, dialect: str) -> TimeUnit:
"""Return a TimeUnit.

Args:
Expand All @@ -367,6 +367,90 @@ def process_result_value(self, value, dialect):
return TimeUnit.to_unit(value)


class ScheduleModel(PythonEnum):
"""The schedule model enum."""

Effort = "effort"
Duration = "duration"
Length = "length"

def __str__(self) -> str:
"""Return the string representation.

Returns:
str: The string representation.
"""
return str(self.value)

@classmethod
def to_model(cls, model: Union[str, "ScheduleModel"]) -> "ScheduleModel":
"""Convert the given model value to a ScheduleModel enum.

Args:
model (Union[str, ScheduleModel]): The value to convert to a
ScheduleModel.

Raises:
TypeError: Input value type is invalid.
ValueError: Input value is invalid.

Returns:
ScheduleModel: The enum.
"""
if not isinstance(model, (str, ScheduleModel)):
raise TypeError(
"model should be a ScheduleModel enum value or one of {}, "
"not {}: '{}'".format(
[u.name.title() for u in cls] + [u.value for u in cls],
model.__class__.__name__,
model,
)
)
if isinstance(model, str):
model_name_lut = dict([(m.name.lower(), m.name) for m in cls])
model_name_lut.update(dict([(m.value.lower(), m.name) for m in cls]))
model_lower_case = model.lower()
if model_lower_case not in model_name_lut:
raise ValueError(
"model should be a ScheduleModel enum value or one of {}, "
"not '{}'".format(
[m.name.title() for m in cls] + [m.value for m in cls], model
)
)

return cls.__members__[model_name_lut[model_lower_case]]

return model


class ScheduleModelDecorator(TypeDecorator):
"""Store ScheduleModel as a str and restore as ScheduleModel."""

impl = Enum(*[m.value for m in ScheduleModel], name=f"ScheduleModel")

def process_bind_param(self, value, dialect) -> str:
"""Return the str value of the ScheduleModel.

Args:
value (ScheduleModel): The ScheduleModel value.
dialect (str): The name of the dialect.

Returns:
str: The value of the ScheduleModel.
"""
# just return the value
return value.value

def process_result_value(self, value: str, dialect: str) -> ScheduleModel:
"""Return a ScheduleModel.

Args:
value (str): The string value to convert to ScheduleModel.
dialect (str): The name of the dialect.
"""
return ScheduleModel.to_model(value)


class TargetEntityTypeMixin(object):
"""Adds target_entity_type attribute to mixed in class.

Expand Down Expand Up @@ -1459,7 +1543,7 @@ def __init__(
self.working_hours = working_hours

@declared_attr
def working_hours_id(cls) -> Mapped[int]:
def working_hours_id(cls) -> Mapped[Optional[int]]:
"""Create the working_hours_id attribute as a declared attribute.

Returns:
Expand All @@ -1468,7 +1552,7 @@ def working_hours_id(cls) -> Mapped[int]:
return mapped_column("working_hours_id", Integer, ForeignKey("WorkingHours.id"))

@declared_attr
def working_hours(cls) -> Mapped["WorkingHours"]:
def working_hours(cls) -> Mapped[Optional["WorkingHours"]]:
"""Create the working_hours attribute as a declared attribute.

Returns:
Expand Down Expand Up @@ -1525,13 +1609,13 @@ class ScheduleMixin(object):
__default_schedule_attr_name__ = "schedule"
__default_schedule_timing__ = defaults.timing_resolution.seconds / 60
__default_schedule_unit__ = TimeUnit.Hour
__default_schedule_models__ = defaults.task_schedule_models
__default_schedule_model__ = ScheduleModel.Effort

def __init__(
self,
schedule_timing: Optional[float] = None,
schedule_unit: TimeUnit = TimeUnit.Hour,
schedule_model: Optional[str] = None,
schedule_model: Optional[ScheduleModel] = ScheduleModel.Effort,
schedule_constraint: ScheduleConstraint = ScheduleConstraint.NONE,
**kwargs: Dict[str, Any],
) -> None:
Expand Down Expand Up @@ -1581,23 +1665,21 @@ def schedule_unit(cls) -> Mapped[Optional[TimeUnit]]:
)

@declared_attr
def schedule_model(cls) -> Mapped[str]:
def schedule_model(cls) -> Mapped[ScheduleModel]:
"""Create the schedule_model attribute as a declared attribute.

Returns:
Column: The Column related to the schedule_model attribute.
"""
return mapped_column(
f"{cls.__default_schedule_attr_name__}_model",
Enum(
*cls.__default_schedule_models__,
name=f"{cls.__name__}{cls.__default_schedule_attr_name__.title()}Model",
),
default=cls.__default_schedule_models__[0],
ScheduleModelDecorator(),
default=ScheduleModel.Effort,
nullable=False,
doc="""Defines the schedule model which is going to be used by
**TaskJuggler** while scheduling this Task. It has three possible
values; **effort**, **duration**, **length**. ``effort`` is the
doc="""Defines the schedule model which is used by **TaskJuggler**
while scheduling this Projects. It is handled as a ScheduleModel
enum value which has three possible values; **effort**,
**duration**, **length**. :attr:`.ScheduleModel.Effort` is the
default value. Each value causes this task to be scheduled in
different ways:

Expand All @@ -1611,8 +1693,9 @@ def schedule_model(cls) -> Mapped[str]:
doesn't mean that the task duration will be 4 days. If the
resource works overtime then the task will be finished
before 4 days or if the resource will not be available
(due to a vacation) then the task duration can be much
more.
(due to a vacation or task coinciding to a weekend day)
then the task duration can be much more bigger than
required effort.

duration The duration of the task will exactly be equal to
:attr:`.schedule_timing` regardless of the resource
Expand Down Expand Up @@ -1695,42 +1778,22 @@ def _validate_schedule_constraint(

@validates("schedule_model")
def _validate_schedule_model(
self, key: str, schedule_model: Union[None, str]
) -> str:
self, key: str, schedule_model: Union[None, str, ScheduleModel]
) -> ScheduleModel:
"""Validate the given schedule_model value.

Args:
key (str): The name of the validated column.
schedule_model (Union[None, str]): The schedule_model value to be
validated.

Raises:
TypeError: If the schedule_model is not a str.
ValueError: If the schedule_model value is not one of the values in
the self.__default_schedule_models__ list.

Returns:
str: The validated schedule_model value.
ScheduleModel: The validated schedule_model value.
"""
if not schedule_model:
schedule_model = self.__default_schedule_models__[0]

error_message = (
"{cls}.{attr}_model should be one of {defaults}, not "
"{model_class}: '{model}'".format(
cls=self.__class__.__name__,
attr=self.__default_schedule_attr_name__,
defaults=self.__default_schedule_models__,
model_class=schedule_model.__class__.__name__,
model=schedule_model,
)
)

if not isinstance(schedule_model, str):
raise TypeError(error_message)

if schedule_model not in self.__default_schedule_models__:
raise ValueError(error_message)
if schedule_model is None:
schedule_model = self.__default_schedule_model__
else:
schedule_model = ScheduleModel.to_model(schedule_model)

return schedule_model

Expand Down Expand Up @@ -1873,7 +1936,7 @@ def to_seconds(
cls,
timing: float,
unit: Union[None, str, TimeUnit],
model: str,
model: Union[str, ScheduleModel],
) -> Union[None, float]:
"""Convert the schedule values to seconds.

Expand All @@ -1887,8 +1950,8 @@ def to_seconds(
unit (Union[None, str, TimeUnit]): The unit value, a TimeUnit enum
value or one of ['min', 'h', 'd', 'w', 'm', 'y', 'Minute',
'Hour', 'Day', 'Week', 'Month', 'Year'].
model (str): The schedule model, one of 'effort', 'length' or
'duration'.
model (str): The schedule model, preferably a ScheduleModel enum
value or one of 'effort', 'length' or 'duration'.

Returns:
Union[None, float]: The converted seconds value.
Expand All @@ -1907,7 +1970,7 @@ def to_seconds(
TimeUnit.Year: 31536000,
}

if model in ["effort", "length"]:
if model in [ScheduleModel.Effort, ScheduleModel.Length]:
day_wt = defaults.daily_working_hours * 3600
week_wt = defaults.weekly_working_days * day_wt
month_wt = 4 * week_wt
Expand All @@ -1926,30 +1989,38 @@ def to_seconds(

@classmethod
def to_unit(
cls, seconds: int, unit: Union[None, str, TimeUnit], model: str
cls,
seconds: int,
unit: Union[None, str, TimeUnit],
model: Union[str, ScheduleModel],
) -> float:
"""Convert the ``seconds`` value to the given ``unit``.

Depending on to the ``schedule_model`` the value will differ. So if the
``schedule_model`` is 'effort' or 'length' then the ``seconds`` and
``schedule_unit`` values are interpreted as work time, if the
``schedule_model`` is 'duration' then the ``seconds`` and
``schedule_unit`` values are considered as calendar time.
``schedule_model`` is :attr:`ScheduleModel.Duration` then the
``seconds`` and ``schedule_unit`` values are considered as calendar
time.

Args:
seconds (int): The seconds to convert.
unit (Union[None, str, TimeUnit]): The unit value, a TimeUnit enum
value one of ['min', 'h', 'd', 'w', 'm', 'y', 'Minute', 'Hour',
'Day', 'Week', 'Month', 'Year'] or a TimeUnit enum value.
model (str): The schedule model, one of 'effort', 'length' or 'duration'.
model (Union[str, ScheduleModel]): The schedule model, either a
ScheduleModel enum value or one of 'effort', 'length' or
'duration'.

Returns:
float: The seconds converted to the given unit considering the given model.
float: The seconds converted to the given unit considering the
given model.
"""
if unit is None:
return None

unit = TimeUnit.to_unit(unit)
model = ScheduleModel.to_model(model)

lut = {
TimeUnit.Minute: 60,
Expand All @@ -1960,7 +2031,7 @@ def to_unit(
TimeUnit.Year: 31536000,
}

if model in ["effort", "length"]:
if model in [ScheduleModel.Effort, ScheduleModel.Length]:
day_wt = defaults.daily_working_hours * 3600
week_wt = defaults.weekly_working_days * day_wt
month_wt = 4 * week_wt
Expand Down
Loading
Loading