TaskWizard / app.py
vincentiusyoshuac's picture
Update app.py
dd699be verified
Raw History Blame Contribute Delete
10.2 kB
import streamlit as st
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import networkx as nx
import plotly.graph_objs as go
class TaskDependencyManager:
def __init__(self):
self.dependency_graph = nx.DiGraph()
def add_task(self, task_data):
"""
Tambahkan tugas baru ke dalam graph dependency
"""
task_name = task_data['Task Name']
# Tambahkan node tugas ke graph
if task_name not in self.dependency_graph.nodes():
self.dependency_graph.add_node(task_name)
# Tambahkan dependency predecessor
predecessor = task_data.get('Predecessor', '-')
if predecessor and predecessor != '-':
if predecessor not in self.dependency_graph.nodes():
self.dependency_graph.add_node(predecessor)
if not self._is_cyclic_dependency(predecessor, task_name):
self.dependency_graph.add_edge(predecessor, task_name)
# Tambahkan dependency successor
successor = task_data.get('Successor', '-')
if successor and successor != '-':
if successor not in self.dependency_graph.nodes():
self.dependency_graph.add_node(successor)
if not self._is_cyclic_dependency(task_name, successor):
self.dependency_graph.add_edge(task_name, successor)
def _is_cyclic_dependency(self, source, target):
"""
Cek apakah menambahkan edge akan menciptakan siklus dalam graph
"""
try:
return nx.has_path(self.dependency_graph, target, source)
except nx.NetworkXError:
return False
def get_task_dependencies(self, task_name):
"""
Dapatkan predecessor dan successor suatu tugas
"""
try:
predecessors = list(self.dependency_graph.predecessors(task_name))
successors = list(self.dependency_graph.successors(task_name))
return {
'predecessors': ', '.join(predecessors) if predecessors else '-',
'successors': ', '.join(successors) if successors else '-'
}
except nx.NetworkXError:
return {
'predecessors': '-',
'successors': '-'
}
def create_dependency_visualization(self):
"""
Buat visualisasi graph dependency dengan posisi node yang optimal
"""
if not self.dependency_graph.nodes():
return None
# Layout untuk positioning node
try:
pos = nx.kamada_kawai_layout(self.dependency_graph)
except:
pos = nx.spring_layout(self.dependency_graph, k=0.5, seed=42)
# Trace untuk edges
edge_x, edge_y = [], []
for edge in self.dependency_graph.edges():
x0, y0 = pos[edge[0]]
x1, y1 = pos[edge[1]]
edge_x.extend([x0, x1, None])
edge_y.extend([y0, y1, None])
edge_trace = go.Scatter(
x=edge_x, y=edge_y,
line=dict(width=1.5, color='#888', dash='dot'),
hoverinfo='none',
mode='lines')
# Trace untuk nodes
node_x, node_y, node_text = [], [], []
node_degrees = dict(self.dependency_graph.degree())
for node in self.dependency_graph.nodes():
x, y = pos[node]
node_x.append(x)
node_y.append(y)
node_text.append(f"{node} (Koneksi: {node_degrees[node]})")
node_trace = go.Scatter(
x=node_x, y=node_y,
text=node_text,
mode='markers+text',
textposition='top center',
hoverinfo='text',
marker=dict(
showscale=True,
colorscale='Viridis',
size=[20 + 10 * node_degrees[node] for node in self.dependency_graph.nodes()],
color=[node_degrees[node] for node in self.dependency_graph.nodes()],
colorbar=dict(
thickness=15,
title='Tingkat Koneksi',
xanchor='left'
),
line_width=2
)
)
layout = go.Layout(
title='Graph Dependency Proyek',
titlefont_size=16,
showlegend=False,
hovermode='closest',
margin=dict(b=20, l=5, r=5, t=40),
xaxis=dict(showgrid=False, zeroline=False, showticklabels=False),
yaxis=dict(showgrid=False, zeroline=False, showticklabels=False)
)
return go.Figure(data=[edge_trace, node_trace], layout=layout)
class StreamlitProjectApp:
def __init__(self):
# Inisialisasi session state untuk task management
if 'tasks' not in st.session_state:
st.session_state.tasks = pd.DataFrame(columns=[
'Task Name', 'Start Date', 'End Date',
'Duration', 'Complexity', 'Predecessor', 'Successor',
'Progress', 'Cost'
])
# Inisialisasi dependency manager
self.dependency_manager = TaskDependencyManager()
# Load existing tasks from session state
for _, task in st.session_state.tasks.iterrows():
self.dependency_manager.add_task(task.to_dict())
def run(self):
# Konfigurasi halaman Streamlit
st.set_page_config(page_title="Project Dependency Optimizer", page_icon="๐Ÿš€", layout="wide")
st.title("๐Ÿš€ Advanced Project Dependency Optimizer")
# Tab untuk berbagai fungsionalitas
tab1, tab2, tab3 = st.tabs(["Manajemen Tugas", "Analisis Proyek", "Visualisasi Dependency"])
with tab1:
self._render_task_management()
with tab2:
self._render_project_metrics()
with tab3:
self._render_dependency_visualization()
def _render_task_management(self):
st.subheader("Daftar dan Manajemen Tugas")
# Buat kolom untuk editing
edited_tasks = st.data_editor(
st.session_state.tasks,
num_rows="dynamic",
column_config={
"Task Name": st.column_config.TextColumn("Nama Tugas"),
"Start Date": st.column_config.DateColumn("Tanggal Mulai"),
"End Date": st.column_config.DateColumn("Tanggal Selesai"),
"Duration": st.column_config.NumberColumn("Durasi (Hari)", min_value=1),
"Complexity": st.column_config.SelectboxColumn(
"Kompleksitas",
options=['Low', 'Medium', 'High']
),
"Predecessor": st.column_config.TextColumn("Predecessor"),
"Successor": st.column_config.TextColumn("Successor"),
"Progress": st.column_config.ProgressColumn(
"Progress", format="%d%%", min_value=0, max_value=100
),
"Cost": st.column_config.NumberColumn("Biaya", min_value=0)
},
use_container_width=True
)
# Tombol update
if st.button("Perbarui Data Tugas"):
try:
# Validasi duplikasi nama tugas
if len(edited_tasks['Task Name'].unique()) != len(edited_tasks):
st.error("Terdapat duplikasi nama tugas. Setiap tugas harus memiliki nama unik.")
return
# Update session state
st.session_state.tasks = edited_tasks
# Reset dependency manager
self.dependency_manager = TaskDependencyManager()
for _, task in st.session_state.tasks.iterrows():
self.dependency_manager.add_task(task.to_dict())
# Re-run dengan metode kompatibel multi-versi
if hasattr(st, 'experimental_rerun'):
st.experimental_rerun()
elif hasattr(st, 'rerun'):
st.rerun()
else:
st.experimental_rerun()
except Exception as e:
st.error(f"Gagal memperbarui data: {e}")
def _render_project_metrics(self):
st.subheader("Ringkasan Metrik Proyek")
metrics = self._calculate_project_metrics()
if metrics:
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Total Tugas", metrics['Total Tasks'])
st.metric("Total Durasi (Hari)", metrics['Total Duration'])
with col2:
st.metric("Total Biaya", f"Rp {metrics['Total Cost']:,}")
st.metric("Rata-rata Kompleksitas",
{1: 'Low', 2: 'Medium', 3: 'High'}.get(round(metrics['Avg Complexity']), 'N/A')
)
with col3:
st.metric("Rata-rata Progress", f"{metrics['Avg Progress']:.2f}%")
else:
st.info("Belum ada tugas yang ditambahkan")
def _calculate_project_metrics(self):
"""
Hitung metrik proyek secara keseluruhan
"""
if st.session_state.tasks.empty:
return None
tasks = st.session_state.tasks
return {
'Total Tasks': len(tasks),
'Total Duration': tasks['Duration'].sum(),
'Total Cost': tasks['Cost'].sum(),
'Avg Complexity': tasks['Complexity'].map({'Low': 1, 'Medium': 2, 'High': 3}).mean(),
'Avg Progress': tasks['Progress'].mean()
}
def _render_dependency_visualization(self):
st.subheader("Visualisasi Graph Dependency")
try:
dependency_graph_fig = self.dependency_manager.create_dependency_visualization()
if dependency_graph_fig:
st.plotly_chart(dependency_graph_fig, use_container_width=True)
else:
st.info("Belum ada dependency yang dibuat")
except Exception as e:
st.error(f"Gagal membuat visualisasi: {e}")
def main():
app = StreamlitProjectApp()
app.run()
if __name__ == "__main__":
main()