external/SevReqCircle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial

Browse Source
This commit is contained in:
dresber
2024-03-10 21:26:57 +01:00
commit 0b1a507f89
6 changed files with 463 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
/requirements.txt
/release.env

19
Dockerfile Normal file
View File

@@ -0,0 +1,19 @@
# set base image (host OS)
FROM python:3.11
COPY src /app/src
COPY requirements.txt /app/src
# set the working directory in the container
WORKDIR /app/src
RUN apt-get update
# install dependencies
RUN pip install -r requirements.txt
RUN rm requirements.txt
# command to run on container start
ENTRYPOINT ["streamlit", "run"]
CMD ["start_page.py"]

5
requirements.in Normal file
View File

@@ -0,0 +1,5 @@
streamlit
bbutils
bbprojectutils
invoke
plotly

221
src/request_load_estimator.py Normal file
View File

@@ -0,0 +1,221 @@
"""
"""
# --------------------------------------- #
# imports #
# --------------------------------------- #
from pandas import DataFrame, concat
from BBUtils.logging_handle import logger
# --------------------------------------- #
# definitions #
# --------------------------------------- #
MODULE_LOGGER_HEAD = "request_load_estimator -> "
START_NR_OF_EMPLOYEES = 50
START_NR_OF_APPLICATIONS = 5
START_NR_OF_SERVICE_REQUESTS = 0
START_NR_OF_SERVICE_DESK_EMPLOYEES = 3
START_DOCUMENTATION_LEVEL = 1
START_KNOWLEDGE_LEVEL = 1
START_SYSTEMS_COMPLEXITY = 1
KNOWLEDGE_REDUCTION_FACTOR = 0.2
DOCUMENTATION_LEVEL_MAX = 4
DOCUMENTATION_LEVEL_MIN = 0.5
KNOWLEDGE_LEVEL_MAX = 4
KNOWLEDGE_LEVEL_MIN = 0.5
# --------------------------------------- #
# global vars #
# --------------------------------------- #
# --------------------------------------- #
# functions #
# --------------------------------------- #
def limit_value(value: float, min_value: float, max_value: float) -> float:
if value < min_value:
return min_value
elif value > max_value:
return max_value
else:
return value
# --------------------------------------- #
# classes #
# --------------------------------------- #
class RequestLoadEstimator:
AVERAGE_REQUEST_WORKING_TIME = 0.5
AVERAGE_DAY_WORKING_TIME = 6
AVERAGE_PROBLEM_LEVEL_IN_PERCENT = 0.1
def __init__(self):
self._weeks_cycle = 4
self.service_desk_employees = 0
self.documentation_level = 0
self._service_desk_knowledge_level = 0
self._systems_complexity = 0
self.nr_of_applications = 0
self.nr_of_employees = 0
self.new_application_problems = 0
self.nr_of_open_service_requests = 0
self.processed_service_requests = 0
self._desired_request_working_time = 0
self.personal_available_time = 0
self.actual_week = 0
self.data_set = DataFrame()
self.initialize_data()
pass
def set_weeks_cycle(self, days: int):
self._weeks_cycle = days
def add_new_applications(self, nr: int):
if nr > 0:
logger.info(MODULE_LOGGER_HEAD + "Adding new applications: " + str(nr))
documentation_level_reduction = (nr / self.nr_of_applications) * 0.2
self.documentation_level -= documentation_level_reduction
if self.documentation_level < DOCUMENTATION_LEVEL_MIN:
self.documentation_level = DOCUMENTATION_LEVEL_MIN
complexity_increase = (nr / self.nr_of_applications) * 0.2
self._systems_complexity += complexity_increase
self.nr_of_applications += nr
elif nr < 0:
logger.info(MODULE_LOGGER_HEAD + "Removing applications: " + str(nr))
complexity_reduction = (nr / self.nr_of_applications) * 0.2
self._systems_complexity -= complexity_reduction
self.nr_of_applications += nr
def add_new_employees(self, nr: int):
if nr > 0:
logger.info(MODULE_LOGGER_HEAD + "Adding new employees: " + str(nr))
self._systems_complexity += nr / self.nr_of_employees
self.nr_of_employees += nr
elif nr < 0:
logger.info(MODULE_LOGGER_HEAD + "Removing employees: " + str(nr))
self._systems_complexity -= nr / self.nr_of_employees
self.nr_of_employees += nr
def add_service_desk_employees(self, nr: int):
if nr > 0:
logger.info(MODULE_LOGGER_HEAD + "Adding new service desk employees: " + str(nr))
knowledge_level_reduction = (nr / self.service_desk_employees) * 0.2
self._service_desk_knowledge_level = limit_value(self._service_desk_knowledge_level - knowledge_level_reduction,
KNOWLEDGE_LEVEL_MIN,
KNOWLEDGE_LEVEL_MAX)
self.service_desk_employees += nr
elif nr < 0:
logger.info(MODULE_LOGGER_HEAD + "Removing service desk employees: " + str(nr))
self.service_desk_employees += nr
def process_weeks(self):
logger.info(MODULE_LOGGER_HEAD + "Processing weeks")
for week in range(self._weeks_cycle):
self._calc_nr_of_service_requests()
self._calc_request_working_time()
self._calc_personal_available_time()
self._calc_documentation_and_knowledge_level()
self._calc_application_problems()
self.actual_week += 1
self._process_data_set()
pass
def initialize_data(self):
logger.info(MODULE_LOGGER_HEAD + "Initializing data")
self._weeks_cycle = 7
self.service_desk_employees = START_NR_OF_SERVICE_DESK_EMPLOYEES
self.documentation_level = START_DOCUMENTATION_LEVEL
self._service_desk_knowledge_level = START_KNOWLEDGE_LEVEL
self._systems_complexity = START_SYSTEMS_COMPLEXITY
self.nr_of_applications = START_NR_OF_APPLICATIONS
self.nr_of_employees = START_NR_OF_EMPLOYEES
self.new_application_problems = 0
self.processed_service_requests = 0
self.nr_of_open_service_requests = START_NR_OF_SERVICE_REQUESTS
self._desired_request_working_time = 0
self.personal_available_time = 0
self.actual_week = 0
self.data_set = DataFrame()
def _calc_application_problems(self):
self.new_application_problems = ((self.nr_of_employees * self.nr_of_applications *
self._systems_complexity / self.documentation_level)
* self.AVERAGE_PROBLEM_LEVEL_IN_PERCENT)
if self.new_application_problems < 0:
self.new_application_problems = 0
pass
def _calc_nr_of_service_requests(self):
self.nr_of_open_service_requests += int(self.new_application_problems)
def _calc_request_working_time(self):
self._desired_request_working_time = (self.nr_of_open_service_requests *
(self.AVERAGE_REQUEST_WORKING_TIME / self._service_desk_knowledge_level))
if self._desired_request_working_time < 0:
self._desired_request_working_time = 0
if self._desired_request_working_time < (self.service_desk_employees * self.AVERAGE_DAY_WORKING_TIME):
self.processed_service_requests = self.nr_of_open_service_requests
self.nr_of_open_service_requests = 0
else:
self.processed_service_requests = int((self.service_desk_employees * self.AVERAGE_DAY_WORKING_TIME) /
(
self.AVERAGE_REQUEST_WORKING_TIME / self._service_desk_knowledge_level))
self.nr_of_open_service_requests -= self.processed_service_requests
if self.nr_of_open_service_requests < 0:
self.nr_of_open_service_requests = 0
pass
def _calc_personal_available_time(self):
self.personal_available_time = (self.service_desk_employees * self.AVERAGE_DAY_WORKING_TIME) - \
self._desired_request_working_time
pass
def _calc_documentation_and_knowledge_level(self):
self.documentation_level += (self.personal_available_time * KNOWLEDGE_REDUCTION_FACTOR /
(self.nr_of_employees * self.nr_of_applications))
self._service_desk_knowledge_level += self.personal_available_time * KNOWLEDGE_REDUCTION_FACTOR / (
self.nr_of_employees *
self.nr_of_applications *
self.service_desk_employees)
self.documentation_level = limit_value(self.documentation_level, DOCUMENTATION_LEVEL_MIN,
DOCUMENTATION_LEVEL_MAX)
self._service_desk_knowledge_level = limit_value(self._service_desk_knowledge_level, KNOWLEDGE_LEVEL_MIN,
KNOWLEDGE_LEVEL_MAX)
def _process_data_set(self):
row = {
"week": self.actual_week,
"weeks_cycle": self._weeks_cycle,
"service_desk_employees": self.service_desk_employees,
"documentation_level": self.documentation_level,
"service_desk_knowledge_level": self._service_desk_knowledge_level,
"systems_complexity": self._systems_complexity,
"nr_of_applications": self.nr_of_applications,
"new_application_problems": self.new_application_problems,
"nr_of_open_service_requests": self.nr_of_open_service_requests,
"request_working_time": self._desired_request_working_time,
"personal_available_time": self.personal_available_time,
"nr_of_processed_service_requests": self.processed_service_requests,
"nr_of_employees": self.nr_of_employees,
}
new_df = DataFrame([row])
if self.data_set.empty:
self.data_set = new_df
else:
self.data_set = concat([self.data_set, new_df], ignore_index=False)
pass
# --------------------------------------- #
# main #
# --------------------------------------- #

124
src/start_app.py Normal file
View File

@@ -0,0 +1,124 @@
"""
"""
# --------------------------------------- #
# imports #
# --------------------------------------- #
import streamlit as st
import plotly.express as px
from BBUtils.logging_handle import logger
from request_load_estimator import RequestLoadEstimator
# --------------------------------------- #
# definitions #
# --------------------------------------- #
MODULE_LOGGER_HEAD = "start_app -> "
APP_VERSION = "99.99.99"
APP_NAME = "Service Request Complexity Estimator"
st.set_page_config(page_title="Service Request Complexity Estimator", page_icon="📊", layout="wide")
FOOTER = f"""
<style>
.footer {{
position: fixed;
left: 0;
bottom: 0;
width: 100%;
background-color: white;
color: lightgrey;
text-align: center;
}}
</style>
<div class="footer">
{APP_NAME} - v{APP_VERSION} - by <a href="https://www.bit-buddy.at" target="_blank">BitBuddySolutions</a>
</div>
"""
# --------------------------------------- #
# global vars #
# --------------------------------------- #
# --------------------------------------- #
# functions #
# --------------------------------------- #
@st.cache_resource
def setup_logging():
logger.set_logging_level("debug")
logger.set_cmd_line_logging_output()
pass
@st.cache_resource
def get_estimator():
return RequestLoadEstimator()
# --------------------------------------- #
# classes #
# --------------------------------------- #
# --------------------------------------- #
# main #
# --------------------------------------- #
if __name__ == "__main__":
setup_logging()
st.header("Service Request Complexity Estimator")
st.write("This app is designed to help estimate the complexity of a service request system to avoid problems "
"within your IT department.")
form = st.form(key='complex_form')
col1, col2, col3, col4 = form.columns(4)
form.divider()
dis_col1, dis_col2 = form.columns([5, 1])
weeks_cycle = col1.number_input("Weeks cycle", min_value=1, max_value=20, value=3)
add_new_employee = col2.number_input("Add new employees", min_value=0, max_value=50, value=0)
add_new_applications = col3.number_input("Change number of applications", min_value=-10, max_value=10, value=0)
add_new_service_desk_employees = col4.number_input("Change number of service desk employees", min_value=-10,
max_value=10,
value=0)
submit_button = col1.form_submit_button(label='Process')
request_estimator = get_estimator()
if submit_button:
request_estimator.set_weeks_cycle(weeks_cycle)
request_estimator.add_new_applications(add_new_applications)
request_estimator.add_service_desk_employees(add_new_service_desk_employees)
request_estimator.add_new_employees(add_new_employee)
request_estimator.process_weeks()
if not request_estimator.data_set.empty:
fig_requests = px.area(request_estimator.data_set, x="week", y=["nr_of_processed_service_requests"], line_shape="spline")
dis_col1.plotly_chart(fig_requests)
detail_expander = dis_col1.expander("Details")
detail_expander.dataframe(request_estimator.data_set)
dis_col2.metric("Company Employees", request_estimator.nr_of_employees, delta=add_new_employee)
dis_col2.metric("Service Desk Employees", request_estimator.service_desk_employees, delta=add_new_service_desk_employees)
dis_col2.metric("Nr of Applications", request_estimator.nr_of_applications, delta=add_new_applications)
dis_col2.metric("Nr of open Service Requests", request_estimator.nr_of_open_service_requests)
dis_col2.metric("Nr of processed Service Requests", request_estimator.processed_service_requests)
if request_estimator.personal_available_time > 0:
st.info(f"Your team has available time of {request_estimator.personal_available_time:.0f} hours.")
else:
st.error("Your team has no available time. You are in the cycle of death! Hire new Guys!!!!")
else:
form.write(request_estimator.data_set)
st.divider()
reset_button = st.button("Reset")
if reset_button:
request_estimator.initialize_data()
st.rerun()
st.markdown(FOOTER, unsafe_allow_html=True)

92
tasks.py Normal file
View File

@@ -0,0 +1,92 @@
"""
"""
# ------------------------------------------------------- #
# imports
# ------------------------------------------------------- #
import os
import subprocess
from invoke import task
from time import time
from BBProjectUtils.docker_utils import docker_build as lib_docker_build, docker_deployment as lib_docker_deployment, \
docker_login_registry
from BBProjectUtils.file_utils import set_version_nr_in_file
from BBProjectUtils.git_utils import check_ready_to_release, release_commit
from BBProjectUtils.standard_tasks import set_venv_name
# ------------------------------------------------------- #
# definitions
# ------------------------------------------------------- #
ARCHIVE_NAME = "SevReqCircle"
VIRTUALENV_NAME = f"py311_{ARCHIVE_NAME}"
DEVELOP_VERSION = "99.99.99"
DOCKER_REGISTRY = "gitea.tech-buddy.at/bitbuddydev"
DOCKER_IMAGE_NAME = "sev_req_circle"
set_venv_name(VIRTUALENV_NAME)
# ------------------------------------------------------- #
# global variables
# ------------------------------------------------------- #
# ------------------------------------------------------- #
# functions
# ------------------------------------------------------- #
# ------------------------------------------------------- #
# classes
# ------------------------------------------------------- #
# ------------------------------------------------------- #
# tasks
# ------------------------------------------------------- #
@task
def docker_build(c, tag=""):
lib_docker_build(c, DOCKER_IMAGE_NAME, tag)
@task
def docker_deployment(c, tag):
docker_login_registry(c, registry=DOCKER_REGISTRY, user=os.getenv('REP_USER'),
pwd=os.getenv('REP_PWD'))
lib_docker_deployment(c, DOCKER_IMAGE_NAME, DOCKER_REGISTRY, tag + "-amd64")
@task
def docker_deploy_raspi(c, tag):
start_time = time()
print("START deploying docker image")
docker_login_registry(c, DOCKER_REGISTRY, os.getenv("REP_USER"), os.getenv("REP_PWD"))
c.run("docker buildx use bbbuilder")
c.run("docker buildx build --platform linux/arm64 "
"-t {registry}/{img_name}{tag} .\\ --load".format(registry=DOCKER_REGISTRY,
img_name=DOCKER_IMAGE_NAME,
tag=":" + tag + "-arm64" if tag != "" else ":latest-arm64"))
c.run("docker push {registry}/{img_name}:{tag}".format(registry=DOCKER_REGISTRY,
img_name=DOCKER_IMAGE_NAME,
tag=tag + "-arm64"))
print(f"FINISHED deploying docker image in {int((time() - start_time)/60)} minutes")
return True
@task
def release(c, version):
if check_ready_to_release():
set_version_nr_in_file("src/start_page.py", 'APP_VERSION = "\d+.\d+.\d+"', f'APP_VERSION = "{version}"')
docker_deployment(c, version)
docker_deploy_raspi(c, version)
if release_commit(version):
set_version_nr_in_file("src/start_page.py", 'APP_VERSION = "\d+.\d+.\d+"', f'APP_VERSION = "{DEVELOP_VERSION}"')
subprocess.run('git commit -am "minor set develop version after release"')