599 lines
20 KiB
Python
599 lines
20 KiB
Python
import frappe
|
|
from frappe import _
|
|
from frappe.utils import nowdate
|
|
import json
|
|
|
|
|
|
def _ok(payload, code=200):
|
|
frappe.response.status_code = code
|
|
frappe.response.message = payload
|
|
|
|
|
|
def _err(msg, code=500):
|
|
frappe.response.status_code = code
|
|
frappe.response.message = {"error": msg}
|
|
|
|
|
|
@frappe.whitelist(allow_guest = True)
|
|
def get_number_cards(company=None, site_name=None):
|
|
"""
|
|
Returns counts for Number Cards:
|
|
- total_assets
|
|
- work_orders_open
|
|
- work_orders_in_progress
|
|
- work_orders_completed
|
|
- work_orders_closed
|
|
|
|
Optional filters: company (Hospital), site_name (Mobile Team Site)
|
|
"""
|
|
try:
|
|
from asset_lite.api.dashboard_filters import expand_dashboard_filters
|
|
|
|
from asset_lite.api.userperm_api import get_permission_filters
|
|
|
|
asset_filters = expand_dashboard_filters({"company": company, "site_name": site_name}, "Asset")
|
|
wo_base = expand_dashboard_filters({"company": company, "site_name": site_name}, "Work_Order")
|
|
|
|
asset_perm = get_permission_filters("Asset").get("filters") or {}
|
|
wo_perm = get_permission_filters("Work_Order").get("filters") or {}
|
|
|
|
asset_filters = {**asset_perm, **(asset_filters or {})}
|
|
wo_base = {**wo_perm, **(wo_base or {})}
|
|
|
|
total_assets = frappe.db.count("Asset", asset_filters or {})
|
|
|
|
open_filters = {**wo_base, "repair_status": "Open"}
|
|
in_progress_filters = {**wo_base, "repair_status": "Work In Progress"}
|
|
completed_filters = {**wo_base, "repair_status": "Completed"}
|
|
closed_filters = {**wo_base, "repair_status": "Closed"}
|
|
|
|
work_orders_open = frappe.db.count("Work_Order", open_filters)
|
|
work_orders_in_progress = frappe.db.count("Work_Order", in_progress_filters)
|
|
work_orders_completed = frappe.db.count("Work_Order", completed_filters)
|
|
work_orders_closed = frappe.db.count("Work_Order", closed_filters)
|
|
|
|
_ok({
|
|
"total_assets": total_assets,
|
|
"work_orders_open": work_orders_open,
|
|
"work_orders_in_progress": work_orders_in_progress,
|
|
"work_orders_completed": work_orders_completed,
|
|
"work_orders_closed": work_orders_closed,
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_number_cards")
|
|
_err(str(e))
|
|
|
|
|
|
WO_STATUS_SORT_ORDER = [
|
|
"Open",
|
|
"Work In Progress",
|
|
"Pending Review",
|
|
"Completed",
|
|
"Closed",
|
|
"Rejected",
|
|
"Cancelled",
|
|
]
|
|
|
|
WO_STATUS_COLORS = {
|
|
"Open": "#F59E0B",
|
|
"Work In Progress": "#3B82F6",
|
|
"Pending Review": "#E11D48",
|
|
"Completed": "#10B981",
|
|
"Closed": "#A855F7",
|
|
"Rejected": "#EF4444",
|
|
"Cancelled": "#6B7280",
|
|
}
|
|
|
|
|
|
def _is_null_or_unknown_label(label):
|
|
key = (label or "").strip().lower()
|
|
return key in ("", "null", "unknown", "none", "undefined")
|
|
|
|
|
|
def _sort_status_labels(labels):
|
|
order = {name: idx for idx, name in enumerate(WO_STATUS_SORT_ORDER)}
|
|
|
|
def sort_key(label):
|
|
if label in order:
|
|
return (0, order[label])
|
|
return (1, label)
|
|
|
|
return sorted(labels, key=sort_key)
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_dashboard_work_order_metrics(
|
|
company=None,
|
|
site_name=None,
|
|
work_order_type=None,
|
|
from_date=None,
|
|
to_date=None,
|
|
):
|
|
"""
|
|
Aggregate Work Orders by repair_status for dashboard pie chart.
|
|
Optional filters: company, site_name, work_order_type, from_date, to_date (creation).
|
|
"""
|
|
try:
|
|
from asset_lite.api.dashboard_filters import expand_dashboard_filters
|
|
from asset_lite.api.userperm_api import get_permission_filters
|
|
|
|
wo_filters = expand_dashboard_filters(
|
|
{"company": company, "site_name": site_name}, "Work_Order"
|
|
)
|
|
wo_perm = get_permission_filters("Work_Order").get("filters") or {}
|
|
wo_filters = {**wo_perm, **(wo_filters or {})}
|
|
|
|
if work_order_type:
|
|
wo_filters["work_order_type"] = work_order_type
|
|
|
|
if from_date and to_date:
|
|
wo_filters["creation"] = ["between", [f"{from_date} 00:00:00", f"{to_date} 23:59:59"]]
|
|
elif from_date:
|
|
wo_filters["creation"] = [">=", f"{from_date} 00:00:00"]
|
|
elif to_date:
|
|
wo_filters["creation"] = ["<=", f"{to_date} 23:59:59"]
|
|
|
|
from asset_lite.api.work_order_api import get_work_order_role_query
|
|
|
|
role_filters, or_filters = get_work_order_role_query()
|
|
wo_filters.update(role_filters or {})
|
|
|
|
rows = frappe.get_all(
|
|
"Work_Order",
|
|
filters=wo_filters or None,
|
|
or_filters=or_filters,
|
|
fields=["repair_status", "work_order_type", "workflow_state"],
|
|
limit=0,
|
|
)
|
|
|
|
status_counts = {}
|
|
type_stats = {}
|
|
|
|
for row in rows:
|
|
status = (row.get("repair_status") or "Unknown").strip()
|
|
status_counts[status] = status_counts.get(status, 0) + 1
|
|
|
|
wo_type = (row.get("work_order_type") or "").strip() or "Unknown"
|
|
if _is_null_or_unknown_label(wo_type):
|
|
continue
|
|
|
|
bucket = type_stats.setdefault(
|
|
wo_type,
|
|
{
|
|
"type": wo_type,
|
|
"total": 0,
|
|
"open": 0,
|
|
"inProgress": 0,
|
|
"pendingReview": 0,
|
|
"completed": 0,
|
|
"closed": 0,
|
|
"rejected": 0,
|
|
"cancelled": 0,
|
|
"wfCompleted": 0,
|
|
"wfInProgress": 0,
|
|
"wfRejected": 0,
|
|
},
|
|
)
|
|
bucket["total"] += 1
|
|
|
|
if status == "Open":
|
|
bucket["open"] += 1
|
|
elif status == "Work In Progress":
|
|
bucket["inProgress"] += 1
|
|
elif status == "Pending Review":
|
|
bucket["pendingReview"] += 1
|
|
elif status == "Completed":
|
|
bucket["completed"] += 1
|
|
elif status == "Closed":
|
|
bucket["closed"] += 1
|
|
elif status == "Rejected":
|
|
bucket["rejected"] += 1
|
|
elif status == "Cancelled":
|
|
bucket["cancelled"] += 1
|
|
|
|
wf_state = (row.get("workflow_state") or "").strip()
|
|
if wf_state == "Completed":
|
|
bucket["wfCompleted"] += 1
|
|
elif wf_state in ("Work In Progress", "In Progress"):
|
|
bucket["wfInProgress"] += 1
|
|
elif wf_state == "Rejected":
|
|
bucket["wfRejected"] += 1
|
|
|
|
labels = _sort_status_labels(list(status_counts.keys()))
|
|
values = [status_counts[label] for label in labels]
|
|
colors = [WO_STATUS_COLORS.get(label, "#6366F1") for label in labels]
|
|
|
|
completed_raw = status_counts.get("Completed", 0)
|
|
closed_raw = status_counts.get("Closed", 0)
|
|
completed_combined = completed_raw + closed_raw
|
|
|
|
counts = {
|
|
"open": status_counts.get("Open", 0),
|
|
"inProgress": status_counts.get("Work In Progress", 0),
|
|
"completed": completed_raw,
|
|
"completedCombined": completed_combined,
|
|
"pendingReview": status_counts.get("Pending Review", 0),
|
|
"rejected": status_counts.get("Rejected", 0),
|
|
"closed": closed_raw,
|
|
"cancelled": status_counts.get("Cancelled", 0),
|
|
"total": len(rows),
|
|
}
|
|
|
|
type_rates = []
|
|
for wo_type, bucket in type_stats.items():
|
|
combined = bucket["completed"] + bucket["closed"]
|
|
bucket["completedCombined"] = combined
|
|
type_rates.append(bucket)
|
|
|
|
type_rates.sort(key=lambda item: item.get("total", 0), reverse=True)
|
|
|
|
type_status_matrix = {}
|
|
for row in rows:
|
|
status = (row.get("repair_status") or "Unknown").strip()
|
|
wo_type = (row.get("work_order_type") or "").strip() or "Unknown"
|
|
if _is_null_or_unknown_label(wo_type):
|
|
continue
|
|
bucket = type_status_matrix.setdefault(wo_type, {})
|
|
bucket[status] = bucket.get(status, 0) + 1
|
|
|
|
type_status_labels = sorted(
|
|
type_status_matrix.keys(),
|
|
key=lambda item: sum(type_status_matrix[item].values()),
|
|
reverse=True,
|
|
)
|
|
matrix_statuses = set()
|
|
for status_map in type_status_matrix.values():
|
|
matrix_statuses.update(status_map.keys())
|
|
type_status_columns = _sort_status_labels(list(matrix_statuses))
|
|
type_status_datasets = [
|
|
{
|
|
"name": status,
|
|
"values": [
|
|
type_status_matrix.get(wo_type, {}).get(status, 0)
|
|
for wo_type in type_status_labels
|
|
],
|
|
"color": WO_STATUS_COLORS.get(status, "#6366F1"),
|
|
}
|
|
for status in type_status_columns
|
|
]
|
|
|
|
_ok({
|
|
"counts": counts,
|
|
"completion_by_type": {
|
|
"labels": [item["type"] for item in type_rates],
|
|
"rates": type_rates,
|
|
},
|
|
"work_order_chart": {
|
|
"labels": labels,
|
|
"datasets": [{
|
|
"name": "Work Orders",
|
|
"values": values,
|
|
"colors": colors,
|
|
}],
|
|
"type": "Pie",
|
|
},
|
|
"type_status_chart": {
|
|
"labels": type_status_labels,
|
|
"datasets": type_status_datasets,
|
|
"type": "Bar",
|
|
},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_dashboard_work_order_metrics")
|
|
_err(str(e))
|
|
|
|
|
|
ASSET_DEVICE_STATUS_COLORS = {
|
|
"up": "#10B981",
|
|
"down": "#EF4444",
|
|
"under maintenance": "#F59E0B",
|
|
"decommissioned": "#6B7280",
|
|
}
|
|
|
|
ASSET_STATUS_FALLBACK_COLORS = [
|
|
"#6366F1",
|
|
"#8B5CF6",
|
|
"#06B6D4",
|
|
"#EC4899",
|
|
"#F59E0B",
|
|
"#10B981",
|
|
]
|
|
|
|
|
|
def _asset_device_status_color(label, fallback_index=0):
|
|
key = (label or "").strip().lower()
|
|
if key in ASSET_DEVICE_STATUS_COLORS:
|
|
return ASSET_DEVICE_STATUS_COLORS[key]
|
|
if "maintenance" in key:
|
|
return ASSET_DEVICE_STATUS_COLORS["under maintenance"]
|
|
if "decommission" in key:
|
|
return ASSET_DEVICE_STATUS_COLORS["decommissioned"]
|
|
return ASSET_STATUS_FALLBACK_COLORS[fallback_index % len(ASSET_STATUS_FALLBACK_COLORS)]
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_asset_device_status_summary(
|
|
company=None,
|
|
site_name=None,
|
|
from_date=None,
|
|
to_date=None,
|
|
):
|
|
"""
|
|
Aggregate Assets by custom_device_status for dashboard pie chart.
|
|
Optional filters: company, site_name (custom_site on Asset), from_date, to_date (creation).
|
|
"""
|
|
try:
|
|
from asset_lite.api.dashboard_filters import expand_dashboard_filters
|
|
from asset_lite.api.userperm_api import get_permission_filters
|
|
|
|
asset_filters = expand_dashboard_filters(
|
|
{"company": company, "site_name": site_name}, "Asset"
|
|
)
|
|
asset_perm = get_permission_filters("Asset").get("filters") or {}
|
|
asset_filters = {**asset_perm, **(asset_filters or {})}
|
|
|
|
if frappe.db.has_column("Asset", "custom_delete_status"):
|
|
asset_filters["custom_delete_status"] = ["!=", "Deleted"]
|
|
|
|
if from_date and to_date:
|
|
asset_filters["creation"] = [
|
|
"between",
|
|
[f"{from_date} 00:00:00", f"{to_date} 23:59:59"],
|
|
]
|
|
elif from_date:
|
|
asset_filters["creation"] = [">=", f"{from_date} 00:00:00"]
|
|
elif to_date:
|
|
asset_filters["creation"] = ["<=", f"{to_date} 23:59:59"]
|
|
|
|
rows = frappe.get_all(
|
|
"Asset",
|
|
filters=asset_filters or None,
|
|
fields=["custom_device_status"],
|
|
limit=0,
|
|
)
|
|
|
|
status_counts = {}
|
|
for row in rows:
|
|
raw = (row.get("custom_device_status") or "").strip()
|
|
label = raw or "Unknown"
|
|
status_counts[label] = status_counts.get(label, 0) + 1
|
|
|
|
sorted_items = sorted(status_counts.items(), key=lambda item: item[1], reverse=True)
|
|
labels = [label for label, _count in sorted_items]
|
|
values = [count for _label, count in sorted_items]
|
|
colors = [
|
|
_asset_device_status_color(label, idx) for idx, label in enumerate(labels)
|
|
]
|
|
|
|
assets_down = 0
|
|
for label, count in status_counts.items():
|
|
if label.strip().lower() == "down":
|
|
assets_down += count
|
|
|
|
total_assets = len(rows)
|
|
|
|
_ok({
|
|
"total_assets": total_assets,
|
|
"assets_down": assets_down,
|
|
"chart": {
|
|
"labels": labels,
|
|
"datasets": [{
|
|
"name": "Assets",
|
|
"values": values,
|
|
"colors": colors,
|
|
}],
|
|
"type": "Pie",
|
|
"total": total_assets,
|
|
},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_asset_device_status_summary")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest = True)
|
|
def list_dashboard_charts(search=None, public_only=True, limit=50):
|
|
"""
|
|
List available Dashboard Chart docs and their y-axis rows.
|
|
"""
|
|
try:
|
|
filters = {}
|
|
if str(public_only) in ("1", "true", "True"): # tolerate string flags
|
|
filters["is_public"] = 1
|
|
|
|
charts = frappe.get_all(
|
|
"Dashboard Chart",
|
|
filters=filters,
|
|
fields=[
|
|
"name",
|
|
"chart_name",
|
|
"type",
|
|
"is_public",
|
|
"chart_type",
|
|
"report_name",
|
|
"use_report_chart",
|
|
"x_field",
|
|
"time_interval",
|
|
"timespan",
|
|
"custom_options",
|
|
],
|
|
limit=int(limit or 50),
|
|
order_by="modified desc",
|
|
)
|
|
|
|
for c in charts:
|
|
y_rows = frappe.get_all(
|
|
"Dashboard Chart Field",
|
|
filters={"parenttype": "Dashboard Chart", "parent": c["name"]},
|
|
fields=["y_field", "color"],
|
|
order_by="idx asc",
|
|
)
|
|
c["y_axes"] = y_rows
|
|
|
|
_ok({"charts": charts})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "list_dashboard_charts")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest = True)
|
|
def get_dashboard_chart_data(chart_name, report_filters=None):
|
|
"""
|
|
Return chart-ready JSON for any Dashboard Chart (Report-based or Custom).
|
|
"""
|
|
try:
|
|
if isinstance(report_filters, str):
|
|
report_filters = json.loads(report_filters or "{}")
|
|
report_filters = report_filters or {}
|
|
|
|
chart = frappe.get_doc("Dashboard Chart", chart_name)
|
|
|
|
# Handle Custom charts (non-Report based)
|
|
if chart.chart_type != "Report" or not chart.use_report_chart:
|
|
# For Custom charts, query the source doctype directly
|
|
try:
|
|
# Get chart configuration
|
|
source = chart.document_type
|
|
based_on = chart.based_on
|
|
value_based_on = chart.value_based_on or "name"
|
|
|
|
# Build aggregation query
|
|
if chart.type == "Pie":
|
|
# Group by based_on field and count
|
|
data = frappe.db.sql(f"""
|
|
SELECT {based_on} as label, COUNT({value_based_on}) as value
|
|
FROM `tab{source}`
|
|
GROUP BY {based_on}
|
|
ORDER BY value DESC
|
|
""", as_dict=True)
|
|
|
|
labels = [str(d.get("label")) for d in data]
|
|
values = [float(d.get("value") or 0) for d in data]
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": "count", "values": values}],
|
|
"type": "Pie",
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"doctype": source},
|
|
})
|
|
else:
|
|
# Bar chart: group by based_on
|
|
data = frappe.db.sql(f"""
|
|
SELECT {based_on} as label, COUNT({value_based_on}) as value
|
|
FROM `tab{source}`
|
|
GROUP BY {based_on}
|
|
ORDER BY value DESC
|
|
LIMIT 20
|
|
""", as_dict=True)
|
|
|
|
labels = [str(d.get("label")) for d in data]
|
|
values = [float(d.get("value") or 0) for d in data]
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": "count", "values": values, "color": "#4F46E5"}],
|
|
"type": "Bar",
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"doctype": source},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), f"Custom Chart Error: {chart_name}")
|
|
_err(f"Error processing custom chart: {str(e)}")
|
|
return
|
|
|
|
y_axes = frappe.get_all(
|
|
"Dashboard Chart Field",
|
|
filters={"parenttype": "Dashboard Chart", "parent": chart.name},
|
|
fields=["y_field", "color"],
|
|
order_by="idx asc",
|
|
)
|
|
|
|
run = frappe.get_attr("frappe.desk.query_report.run")
|
|
report_result = run(chart.report_name, filters=report_filters)
|
|
rows = report_result.get("result", []) or []
|
|
data_rows = [r for r in rows if not r.get("is_total_row")]
|
|
|
|
x_key = chart.x_field
|
|
labels = [str(r.get(x_key)) for r in data_rows if r.get(x_key) is not None]
|
|
|
|
datasets = []
|
|
for y in y_axes:
|
|
series_name = y.get("y_field") # Use field name as series name
|
|
values = []
|
|
for r in data_rows:
|
|
val = r.get(y.get("y_field"))
|
|
try:
|
|
values.append(float(val) if val is not None else 0)
|
|
except Exception:
|
|
values.append(0)
|
|
datasets.append({"name": series_name, "values": values, "color": y.get("color")})
|
|
|
|
chart_type = (chart.type or "Bar").title()
|
|
if chart_type.lower() == "pie":
|
|
ds = datasets[0] if datasets else {"name": "value", "values": []}
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [ds],
|
|
"type": "Pie",
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"report": chart.report_name},
|
|
})
|
|
return
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": datasets,
|
|
"type": chart_type,
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"report": chart.report_name},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_dashboard_chart_data")
|
|
_err(str(e))
|
|
|
|
|
|
def _parse_custom_options(raw):
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
if isinstance(raw, dict):
|
|
return raw
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
@frappe.whitelist(allow_guest = True)
|
|
def get_repair_cost_by_item(year=None):
|
|
"""
|
|
Example specialized endpoint for 'Repair Cost' report style chart
|
|
(X: item_code, Y: amount, Filter: Year)
|
|
"""
|
|
try:
|
|
year = int(year or frappe.utils.getdate(nowdate()).year)
|
|
rows = frappe.db.sql(
|
|
"""
|
|
SELECT item_code, SUM(amount) as amount
|
|
FROM `tabWork Order` wo
|
|
WHERE YEAR(wo.posting_date) = %(year)s
|
|
GROUP BY item_code
|
|
ORDER BY amount DESC
|
|
""",
|
|
{"year": year},
|
|
as_dict=True,
|
|
)
|
|
labels = [r.item_code or "Unknown" for r in rows]
|
|
values = [float(r.amount or 0) for r in rows]
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": f"Repair Cost {year}", "values": values}],
|
|
"type": "Bar",
|
|
"options": {},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_repair_cost_by_item")
|
|
_err(str(e))
|
|
|
|
|