import os
import re
import ssl
import json
import secrets
import hashlib
import smtplib
import sqlite3
import calendar
import threading
import tkinter as tk
from tkinter import ttk, messagebox, filedialog
from datetime import datetime, date
from tkcalendar import DateEntry as _TkCalendarDateEntry
APP_TITLE = "ERT CONSORTIUM - Instrument Management System"
APP_BUILD = "v35-ConfigFullControl"
DB_FILENAME = "instrument_management.db"
LOGO_FILENAME = "ert_consortium_logo.png"
UI_FONT = "Bahnschrift"
DEFAULT_PAGE_SIZE = 150
SMTP_HOST = os.environ.get("ERT_SMTP_HOST", "")
SMTP_PORT = int(os.environ.get("ERT_SMTP_PORT", "465"))
SMTP_USER = os.environ.get("ERT_SMTP_USER", "")
SMTP_PASSWORD = os.environ.get("ERT_SMTP_PASSWORD", "")
SMTP_FROM = os.environ.get("ERT_SMTP_FROM", SMTP_USER)
VERIFICATION_EXPIRY_MINUTES = 10
MAX_LOGIN_ATTEMPTS = 3
ERT_BLUE = "#00AEEF"
ERT_DARK_BLUE = "#283593"
ERT_NAVY = "#1A237E"
ERT_LIGHT_BLUE = "#E8F7FC"
ERT_SELECTED = "#00AEEF"
class DateEntry(_TkCalendarDateEntry):
"""DateEntry with geometry clamp, single-instance popup safeguard, and empty-date support."""
_open_instances = []
def __init__(self, master=None, **kw):
super().__init__(master, **kw)
self.configure(validate='none')
def drop_down(self):
try:
self.winfo_toplevel().update_idletasks()
except Exception:
pass
for other in list(DateEntry._open_instances):
if other is not self:
try:
other._top_cal.withdraw()
other.state(["!pressed"])
except Exception:
pass
DateEntry._open_instances = [w for w in DateEntry._open_instances if w is self]
try:
already_open = self._calendar.winfo_ismapped()
except Exception:
already_open = False
if already_open:
self._top_cal.withdraw()
self.state(["!pressed"])
if self in DateEntry._open_instances:
DateEntry._open_instances.remove(self)
return
current_text = self.get()
try:
selected_date = self.parse_date(current_text) if current_text else date.today()
except Exception:
selected_date = date.today()
x = self.winfo_rootx()
y = self.winfo_rooty() + self.winfo_height()
self._top_cal.update_idletasks()
popup_w = self._top_cal.winfo_reqwidth() or self._top_cal.winfo_width()
popup_h = self._top_cal.winfo_reqheight() or self._top_cal.winfo_height()
screen_w = self.winfo_screenwidth()
screen_h = self.winfo_screenheight()
if popup_w and x + popup_w > screen_w:
x = max(0, screen_w - popup_w)
if popup_h and y + popup_h > screen_h:
y = max(0, self.winfo_rooty() - popup_h)
self._top_cal.attributes("-topmost", bool(self.winfo_toplevel().attributes("-topmost")))
self._top_cal.geometry(f"+{x}+{y}")
self._top_cal.deiconify()
self._top_cal.lift()
self._calendar.focus_force()
self._calendar.selection_set(selected_date)
DateEntry._open_instances.append(self)
class WrappedCategoryList(tk.Frame):
def __init__(self, master, **kwargs):
super().__init__(master, **kwargs)
self._rows = []
self._selected = None
self._active = None
self.canvas = tk.Canvas(self, bd=0, highlightthickness=0, bg="white", yscrollincrement=1)
self.scrollbar = tk.Scrollbar(self, orient="vertical", command=self.canvas.yview)
self.inner = tk.Frame(self.canvas, bg="white")
self._window_id = self.canvas.create_window(0, 0, anchor="nw", window=self.inner)
self.canvas.configure(yscrollcommand=self.scrollbar.set)
self.canvas.pack(side="left", fill="both", expand=True)
self.scrollbar.pack(side="right", fill="y")
self.inner.bind("<Configure>", self._on_inner_configure)
self.canvas.bind("<Configure>", self._on_canvas_configure)
def _on_inner_configure(self, event=None):
self.canvas.configure(scrollregion=self.canvas.bbox("all"))
self._update_wrap_lengths()
def _on_canvas_configure(self, event=None):
self.canvas.itemconfigure(self._window_id, width=self.canvas.winfo_width())
self._update_wrap_lengths()
def _update_wrap_lengths(self):
width = max(100, self.canvas.winfo_width() - 12)
for row in self._rows:
row["label"].configure(wraplength=width)
self.inner.update_idletasks()
self.canvas.configure(scrollregion=self.canvas.bbox("all"))
def _row_clicked(self, index):
self._selected = index
self._active = index
self._apply_selection()
self.event_generate("<<ListboxSelect>>")
def _apply_selection(self):
for i, row in enumerate(self._rows):
selected = (i == self._selected)
row["label"].configure(
bg=ERT_SELECTED if selected else "white",
fg="white" if selected else "black"
)
def insert(self, index, text):
row_frame = tk.Frame(self.inner, bg="white")
label = tk.Label(
row_frame, text=text, anchor="w", justify="left",
font=(UI_FONT, 11), bg="white", fg="black", padx=5, pady=3
)
label.pack(fill="x", expand=True)
row_frame.pack(fill="x", expand=True)
row = {"frame": row_frame, "label": label, "text": text}
row_index = len(self._rows)
self._rows.append(row)
label.bind("<Button-1>", lambda event, i=row_index: self._row_clicked(i))
row_frame.bind("<Button-1>", lambda event, i=row_index: self._row_clicked(i))
self._update_wrap_lengths()
def delete(self, first, last=None):
for row in self._rows:
row["frame"].destroy()
self._rows.clear()
self._selected = None
self._active = None
self.canvas.yview_moveto(0)
self._update_wrap_lengths()
def curselection(self):
return () if self._selected is None else (self._selected,)
def selection_clear(self, first, last=None):
self._selected = None
self._active = None
self._apply_selection()
def selection_set(self, index):
try:
index = int(index)
except (TypeError, ValueError):
return
if 0 <= index < len(self._rows):
self._selected = index
self._apply_selection()
def activate(self, index):
try:
index = int(index)
except (TypeError, ValueError):
return
if 0 <= index < len(self._rows):
self._active = index
def see(self, index):
try:
index = int(index)
except (TypeError, ValueError):
return
if not (0 <= index < len(self._rows)):
return
self.update_idletasks()
row = self._rows[index]["frame"]
y = row.winfo_y()
h = max(1, self.inner.winfo_height())
self.canvas.yview_moveto(max(0.0, min(1.0, y / h)))
MANUFACTURER_INSTRUMENTS = {
"Labconco": {
"Fume Hood": ("Size", ["4ft", "5ft", "6ft"]),
"Biosafety Cabinet": ("Size", ["4ft", "5ft", "6ft"]),
"Balance Enclosure": ("Size", ["4ft", "5ft", "6ft"]),
"Water Pro PS": (None, []),
"Freeze Dryer": ("Capacity", []),
},
"Distek": {
"Dissolution System": ("Station", ["6 Station", "7 Station", "8 Station"]),
"Autosampler": (None, []),
"Disintegration Tester": ("Station", ["2 Station", "4 Station", "6 Station"]),
"Media Preparation": (None, []),
},
"Sherwood Scientific": {
"Flame Photometer": (None, []),
},
"Heidolph": {
"Rotary Evaporator": (None, []),
"Magnetic Stirrer with Hotplate": (None, []),
"Vortex Mixer": (None, []),
"Shaker": (None, []),
},
"BEL": {},
"Other": {},
}
CONFIG_MAIN_INSTRUMENTS = [
"Fume Hoods", "Biosafety Cabinet", "Laminar Airflow", "Balance Enclosure",
"Double Filtered Balance Enclosure", "Bulk Powder Enclosure", "Water Pro PS",
"Freeze Dryer", "Dissolution Tester", "Disintegration Tester",
"Dissolution Media Preparation", "Glassware Washer", "Rotary Evaporator",
"Flame Photometer", "Shaker", "Mixer",
]
class Database:
def __init__(self, db_path):
self.db_path = db_path
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
self.conn.row_factory = sqlite3.Row
self._configure_pragmas()
self.create_tables()
self.ensure_performance_indexes()
def _configure_pragmas(self):
try:
self.conn.execute("PRAGMA journal_mode=WAL")
self.conn.execute("PRAGMA synchronous=NORMAL")
self.conn.execute("PRAGMA temp_store=MEMORY")
self.conn.execute("PRAGMA cache_size=-64000")
self.conn.execute("PRAGMA foreign_keys=ON")
except sqlite3.DatabaseError:
pass
def ensure_performance_indexes(self):
indexes = [
("idx_cs_cust_id", "customer_sites", "customer_id"),
("idx_inst_site_id", "instruments", "site_id"),
("idx_inst_name", "instruments", "instrument_name"),
("idx_inst_manufacturer", "instruments", "manufacturer"),
("idx_inst_model", "instruments", "model"),
("idx_inst_serial", "instruments", "serial_no"),
("idx_inst_eq_id", "instruments", "equipment_id"),
("idx_inst_cal_date", "instruments", "calibration_date"),
("idx_inst_inst_date", "instruments", "installation_date"),
("idx_sr_inst_id", "service_records", "instrument_id"),
("idx_sr_date", "service_records", "service_date"),
("idx_cr_inst_id", "calibration_records", "instrument_id"),
("idx_cr_date", "calibration_records", "calibration_date"),
("idx_amc_inst_id", "amc_records", "instrument_id"),
("idx_amc_year_no", "amc_records", "amc_year, amc_no"),
("idx_pkg_main", "instrument_packages", "main_instrument"),
("idx_pkg_items_pkg_id", "instrument_package_items", "package_id")
]
with self.conn:
for name, table, cols in indexes:
try:
self.conn.execute(f"CREATE INDEX IF NOT EXISTS {name} ON {table}({cols})")
except sqlite3.OperationalError:
pass
def create_tables(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE COLLATE NOCASE
);
CREATE TABLE IF NOT EXISTS customer_sites (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id INTEGER NOT NULL,
location TEXT NOT NULL,
department TEXT NOT NULL,
UNIQUE(customer_id, location, department),
FOREIGN KEY(customer_id) REFERENCES customers(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS instruments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL,
manufacturer TEXT NOT NULL,
instrument_name TEXT NOT NULL,
model TEXT,
serial_no TEXT,
equipment_id TEXT,
size_station TEXT,
installation_date TEXT,
calibration_date TEXT,
calibration_due_date TEXT,
configuration TEXT,
FOREIGN KEY(site_id) REFERENCES customer_sites(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS service_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument_id INTEGER NOT NULL,
service_date TEXT NOT NULL,
service_text TEXT NOT NULL,
FOREIGN KEY(instrument_id) REFERENCES instruments(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS amc_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument_id INTEGER NOT NULL,
amc_year INTEGER NOT NULL,
amc_no INTEGER NOT NULL,
amc_date TEXT NOT NULL,
notes TEXT,
UNIQUE(instrument_id, amc_year, amc_no),
FOREIGN KEY(instrument_id) REFERENCES instruments(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS calibration_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument_id INTEGER NOT NULL,
calibration_date TEXT NOT NULL,
completed_at TEXT NOT NULL,
notes TEXT,
FOREIGN KEY(instrument_id) REFERENCES instruments(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS instrument_packages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
main_instrument TEXT NOT NULL,
package_name TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(main_instrument, package_name)
);
CREATE TABLE IF NOT EXISTS instrument_package_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
package_id INTEGER NOT NULL,
line_no INTEGER NOT NULL,
quantity INTEGER NOT NULL DEFAULT 1,
part_no TEXT,
description TEXT NOT NULL,
is_main INTEGER NOT NULL DEFAULT 0,
FOREIGN KEY(package_id) REFERENCES instrument_packages(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS instrument_config_categories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE COLLATE NOCASE,
order_index INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS datasheets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument_id INTEGER NOT NULL UNIQUE,
file_path TEXT NOT NULL,
file_name TEXT NOT NULL,
original_name TEXT,
added_at TEXT NOT NULL,
FOREIGN KEY(instrument_id) REFERENCES instruments(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
email TEXT NOT NULL UNIQUE COLLATE NOCASE,
password_hash TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'Administrator',
verified INTEGER NOT NULL DEFAULT 0,
failed_attempts INTEGER NOT NULL DEFAULT 0,
locked INTEGER NOT NULL DEFAULT 0,
verification_code_hash TEXT,
verification_expires TEXT,
created_at TEXT NOT NULL,
last_login TEXT
);
""")
self.conn.commit()
# Seed categories if absent
existing = self.conn.execute("SELECT COUNT(*) FROM instrument_config_categories").fetchone()[0]
if existing == 0:
now = datetime.now().isoformat(timespec="seconds")
self.conn.executemany(
"INSERT INTO instrument_config_categories (name, order_index, created_at) VALUES (?, ?, ?)",
[(name, idx, now) for idx, name in enumerate(CONFIG_MAIN_INSTRUMENTS)]
)
self.conn.commit()
def config_categories(self):
return self.conn.execute(
"SELECT name FROM instrument_config_categories ORDER BY order_index, id"
).fetchall()
def add_config_category(self, name):
name = str(name).strip()
if not name:
return False
if self.conn.execute(
"SELECT 1 FROM instrument_config_categories WHERE name = ?", (name,)
).fetchone():
return False
next_order = self.conn.execute(
"SELECT COALESCE(MAX(order_index), -1) + 1 FROM instrument_config_categories"
).fetchone()[0]
self.conn.execute(
"INSERT INTO instrument_config_categories (name, order_index, created_at) VALUES (?, ?, ?)",
(name, int(next_order), datetime.now().isoformat(timespec="seconds"))
)
self.conn.commit()
return True
def reorder_config_categories(self, categories):
with self.conn:
for idx, name in enumerate(categories):
self.conn.execute(
"UPDATE instrument_config_categories SET order_index=? WHERE name=?",
(idx, name)
)
def delete_config_category(self, name):
name = str(name).strip()
if not name:
return False
with self.conn:
self.conn.execute("DELETE FROM instrument_packages WHERE main_instrument=?", (name,))
cur = self.conn.execute("DELETE FROM instrument_config_categories WHERE name=?", (name,))
return cur.rowcount > 0
def _find_customer_id(self, customer_name):
customer_name = (customer_name or "").strip()
if not customer_name:
return None
row = self.conn.execute(
"SELECT id FROM customers WHERE name = ?", (customer_name,)
).fetchone()
return row["id"] if row else None
def _get_or_create_customer_id(self, customer_name):
customer_id = self._find_customer_id(customer_name)
if customer_id is not None:
return customer_id
cur = self.conn.execute(
"INSERT INTO customers(name) VALUES (?)", (customer_name.strip(),)
)
self.conn.commit()
return cur.lastrowid
def add_customer_site(self, customer_name, location, department):
customer_name = customer_name.strip()
location = location.strip()
department = department.strip()
if not customer_name or not location:
raise ValueError("Customer Name and Location are required.")
customer_id = self._get_or_create_customer_id(customer_name)
try:
self.conn.execute(
"INSERT INTO customer_sites(customer_id, location, department) VALUES (?, ?, ?)",
(customer_id, location, department)
)
self.conn.commit()
except sqlite3.IntegrityError:
raise ValueError("This exact Customer + Location + Department combination already exists.")
def update_customer_site(self, site_id, customer_name, location, department):
customer_name = customer_name.strip()
location = location.strip()
department = department.strip()
if not customer_name or not location:
raise ValueError("Customer Name and Location are required.")
customer_id = self._get_or_create_customer_id(customer_name)
try:
with self.conn:
self.conn.execute(
"UPDATE customer_sites SET customer_id=?, location=?, department=? WHERE id=?",
(customer_id, location, department, site_id)
)
self.conn.execute("""
DELETE FROM customers
WHERE id NOT IN (SELECT DISTINCT customer_id FROM customer_sites)
""")
except sqlite3.IntegrityError:
raise ValueError("Another record already has this Customer + Location + Department combination.")
def delete_customer_site(self, site_id):
count = self.conn.execute(
"SELECT COUNT(*) AS n FROM instruments WHERE site_id=?", (site_id,)
).fetchone()["n"]
if count:
raise ValueError("This customer site has assigned instruments. Remove/reassign them first.")
with self.conn:
self.conn.execute("DELETE FROM customer_sites WHERE id=?", (site_id,))
self.conn.execute("""
DELETE FROM customers
WHERE id NOT IN (SELECT DISTINCT customer_id FROM customer_sites)
""")
def customer_names(self):
return [r["name"] for r in self.conn.execute(
"SELECT name FROM customers ORDER BY name"
).fetchall()]
def locations_for_customer(self, customer_name):
customer_name = (customer_name or "").strip()
if not customer_name:
return []
return [r["location"] for r in self.conn.execute("""
SELECT DISTINCT cs.location
FROM customer_sites cs
JOIN customers c ON c.id=cs.customer_id
WHERE c.name = ?
ORDER BY cs.location
""", (customer_name,)).fetchall()]
def departments_for_customer_location(self, customer_name, location):
customer_name = (customer_name or "").strip()
location = (location or "").strip()
if not customer_name or not location:
return []
return [r["department"] for r in self.conn.execute("""
SELECT DISTINCT cs.department
FROM customer_sites cs
JOIN customers c ON c.id=cs.customer_id
WHERE c.name = ? AND cs.location = ?
ORDER BY cs.department
""", (customer_name, location)).fetchall()]
def departments_for_customer(self, customer_name):
customer_name = (customer_name or "").strip()
if not customer_name:
return []
return [r["department"] for r in self.conn.execute("""
SELECT DISTINCT cs.department FROM customer_sites cs
JOIN customers c ON c.id=cs.customer_id
WHERE c.name = ? ORDER BY cs.department
""", (customer_name,)).fetchall()]
def search_customer_names(self, prefix=""):
prefix = (prefix or "").strip()
query = "SELECT name FROM customers WHERE name LIKE ? ORDER BY name LIMIT 50" if prefix else "SELECT name FROM customers ORDER BY name LIMIT 50"
params = (prefix + "%",) if prefix else ()
return [r["name"] for r in self.conn.execute(query, params).fetchall()]
def all_locations(self, prefix=""):
prefix = (prefix or "").strip()
query = "SELECT DISTINCT location FROM customer_sites WHERE location LIKE ? ORDER BY location LIMIT 50" if prefix else "SELECT DISTINCT location FROM customer_sites ORDER BY location LIMIT 50"
params = (prefix + "%",) if prefix else ()
return [r["location"] for r in self.conn.execute(query, params).fetchall()]
def all_departments(self, prefix=""):
prefix = (prefix or "").strip()
query = "SELECT DISTINCT department FROM customer_sites WHERE department LIKE ? ORDER BY department LIMIT 50" if prefix else "SELECT DISTINCT department FROM customer_sites ORDER BY department LIMIT 50"
params = (prefix + "%",) if prefix else ()
return [r["department"] for r in self.conn.execute(query, params).fetchall()]
def site_id(self, customer_name, location, department):
row = self.conn.execute("""
SELECT cs.id
FROM customer_sites cs
JOIN customers c ON c.id=cs.customer_id
WHERE c.name = ? AND cs.location = ? AND COALESCE(cs.department, '') = ?
""", (customer_name.strip(), location.strip(), department.strip())).fetchone()
return row["id"] if row else None
def get_or_create_site_id(self, customer_name, location, department):
customer_name = (customer_name or "").strip()
location = (location or "").strip()
department = (department or "").strip()
if not customer_name or not location:
raise ValueError("Customer and Location are required.")
existing = self.site_id(customer_name, location, department)
if existing:
return existing
customer_id = self._get_or_create_customer_id(customer_name)
try:
cur = self.conn.execute(
"INSERT INTO customer_sites(customer_id, location, department) VALUES (?, ?, ?)",
(customer_id, location, department)
)
self.conn.commit()
return cur.lastrowid
except sqlite3.IntegrityError:
return self.site_id(customer_name, location, department)
def add_instrument(self, data):
site_id = self.get_or_create_site_id(data["customer"], data["location"], data["department"])
self.conn.execute("""
INSERT INTO instruments
(site_id, manufacturer, instrument_name, model, serial_no, equipment_id,
size_station, installation_date, calibration_date, calibration_due_date, configuration)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
site_id, data["manufacturer"], data["instrument"], data["model"],
data["serial_no"], data["equipment_id"], data["size_station"],
data["installation_date"], data["calibration_date"],
data.get("calibration_due_date", ""), data.get("configuration", "")
))
self.conn.commit()
def update_instrument(self, instrument_id, data):
site_id = self.get_or_create_site_id(data["customer"], data["location"], data["department"])
self.conn.execute("""
UPDATE instruments
SET site_id=?, manufacturer=?, instrument_name=?, model=?,
serial_no=?, equipment_id=?, size_station=?, installation_date=?,
calibration_date=?, calibration_due_date=?, configuration=?
WHERE id=?
""", (
site_id, data["manufacturer"], data["instrument"], data["model"],
data["serial_no"], data["equipment_id"], data["size_station"],
data["installation_date"], data["calibration_date"],
data.get("calibration_due_date", ""), data.get("configuration", ""),
instrument_id
))
self.conn.commit()
def delete_instrument(self, instrument_id):
self.conn.execute("DELETE FROM instruments WHERE id=?", (instrument_id,))
self.conn.commit()
def instrument(self, instrument_id):
return self.conn.execute("""
SELECT i.*, c.name AS customer_name, cs.location, cs.department
FROM instruments i
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
WHERE i.id=?
""", (instrument_id,)).fetchone()
def instruments(self, limit=DEFAULT_PAGE_SIZE, offset=0):
return self.conn.execute("""
SELECT i.*, c.name AS customer_name, cs.location, cs.department
FROM instruments i
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
ORDER BY c.name, cs.location, i.instrument_name
LIMIT ? OFFSET ?
""", (limit, offset)).fetchall()
def search_instruments(self, term="", limit=DEFAULT_PAGE_SIZE, offset=0):
term = (term or "").strip()
if not term:
return self.instruments(limit, offset)
pattern = f"%{term}%"
return self.conn.execute("""
SELECT i.*, c.name AS customer_name, cs.location, cs.department
FROM instruments i
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
WHERE i.instrument_name LIKE ?
OR i.manufacturer LIKE ?
OR i.model LIKE ?
OR i.serial_no LIKE ?
OR i.equipment_id LIKE ?
OR c.name LIKE ?
OR cs.location LIKE ?
OR cs.department LIKE ?
ORDER BY c.name, cs.location, i.instrument_name
LIMIT ? OFFSET ?
""", (pattern, pattern, pattern, pattern, pattern, pattern, pattern, pattern, limit, offset)).fetchall()
def search_customer_sites(self, term="", limit=DEFAULT_PAGE_SIZE, offset=0):
term = (term or "").strip()
if not term:
return self.conn.execute("""
SELECT cs.id, c.name AS customer_name, cs.location, cs.department
FROM customer_sites cs
JOIN customers c ON c.id=cs.customer_id
ORDER BY c.name, cs.location, cs.department
LIMIT ? OFFSET ?
""", (limit, offset)).fetchall()
pattern = f"%{term}%"
return self.conn.execute("""
SELECT cs.id, c.name AS customer_name, cs.location, cs.department
FROM customer_sites cs
JOIN customers c ON c.id=cs.customer_id
WHERE c.name LIKE ? OR cs.location LIKE ? OR cs.department LIKE ?
ORDER BY c.name, cs.location, cs.department
LIMIT ? OFFSET ?
""", (pattern, pattern, pattern, limit, offset)).fetchall()
def search_service_records(self, term="", limit=DEFAULT_PAGE_SIZE, offset=0):
term = (term or "").strip()
if not term:
return self.conn.execute("""
SELECT sr.*, i.instrument_name, i.model, i.serial_no,
i.manufacturer, c.name AS customer_name, cs.location, cs.department
FROM service_records sr
JOIN instruments i ON i.id=sr.instrument_id
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
ORDER BY sr.id DESC
LIMIT ? OFFSET ?
""", (limit, offset)).fetchall()
pattern = f"%{term}%"
return self.conn.execute("""
SELECT sr.*, i.instrument_name, i.model, i.serial_no,
i.manufacturer, c.name AS customer_name, cs.location, cs.department
FROM service_records sr
JOIN instruments i ON i.id=sr.instrument_id
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
WHERE i.instrument_name LIKE ?
OR i.manufacturer LIKE ?
OR i.model LIKE ?
OR i.serial_no LIKE ?
OR c.name LIKE ?
OR cs.location LIKE ?
OR cs.department LIKE ?
OR sr.service_date LIKE ?
OR sr.service_text LIKE ?
ORDER BY sr.id DESC
LIMIT ? OFFSET ?
""", (pattern, pattern, pattern, pattern, pattern, pattern, pattern, pattern, pattern, limit, offset)).fetchall()
def add_service(self, instrument_id, service_date, text):
self.conn.execute("""
INSERT INTO service_records(instrument_id, service_date, service_text)
VALUES (?, ?, ?)
""", (instrument_id, service_date, text.strip()))
self.conn.commit()
def update_service(self, service_id, instrument_id, service_date, text):
self.conn.execute("""
UPDATE service_records
SET instrument_id=?, service_date=?, service_text=?
WHERE id=?
""", (instrument_id, service_date, text.strip(), service_id))
self.conn.commit()
def delete_service(self, service_id):
self.conn.execute("DELETE FROM service_records WHERE id=?", (service_id,))
self.conn.commit()
def search_amc_records(self, term="", limit=DEFAULT_PAGE_SIZE, offset=0):
term = (term or "").strip()
if not term:
return self.conn.execute("""
SELECT a.*, i.instrument_name, i.model, i.manufacturer,
i.serial_no, i.size_station, i.equipment_id,
c.name AS customer_name, cs.location, cs.department
FROM amc_records a
JOIN instruments i ON i.id=a.instrument_id
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
ORDER BY a.amc_year DESC, a.amc_no ASC
LIMIT ? OFFSET ?
""", (limit, offset)).fetchall()
pattern = f"%{term}%"
return self.conn.execute("""
SELECT a.*, i.instrument_name, i.model, i.manufacturer,
i.serial_no, i.size_station, i.equipment_id,
c.name AS customer_name, cs.location, cs.department
FROM amc_records a
JOIN instruments i ON i.id=a.instrument_id
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
WHERE i.instrument_name LIKE ?
OR i.manufacturer LIKE ?
OR i.model LIKE ?
OR i.serial_no LIKE ?
OR c.name LIKE ?
OR cs.location LIKE ?
OR cs.department LIKE ?
OR CAST(a.amc_year AS TEXT) LIKE ?
ORDER BY a.amc_year DESC, a.amc_no ASC
LIMIT ? OFFSET ?
""", (pattern, pattern, pattern, pattern, pattern, pattern, pattern, pattern, limit, offset)).fetchall()
def add_amc(self, instrument_id, year, amc_no, amc_date, notes):
if self.conn.execute("""
SELECT 1 FROM amc_records WHERE instrument_id=? AND amc_year=? AND amc_no=?
""", (instrument_id, year, amc_no)).fetchone():
raise ValueError(f"AMC {amc_no} for {year} already exists for this instrument.")
self.conn.execute("""
INSERT INTO amc_records (instrument_id, amc_year, amc_no, amc_date, notes)
VALUES (?, ?, ?, ?, ?)
""", (instrument_id, year, amc_no, amc_date, notes.strip()))
self.conn.commit()
def delete_amc(self, amc_id):
self.conn.execute("DELETE FROM amc_records WHERE id=?", (amc_id,))
self.conn.commit()
def upcoming_calibrations(self):
rows = self.conn.execute("""
SELECT i.*, c.name AS customer_name, cs.location, cs.department
FROM instruments i
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
WHERE i.calibration_date IS NOT NULL AND TRIM(i.calibration_date) <> ''
""").fetchall()
today = date.today()
result = []
for row in rows:
raw = (row["calibration_date"] or "").strip()
cal_date = None
for fmt in ("%d-%m-%Y", "%Y-%m-%d", "%d/%m/%Y", "%Y/%m/%d"):
try:
cal_date = datetime.strptime(raw, fmt).date()
break
except ValueError:
continue
if cal_date is not None and cal_date >= today:
result.append((cal_date, row))
result.sort(key=lambda item: item[0])
return [row for _, row in result]
def mark_calibration_completed(self, instrument_id, notes=""):
row = self.instrument(instrument_id)
if not row or not row["calibration_date"]:
raise ValueError("Instrument does not have an active calibration date.")
with self.conn:
self.conn.execute("""
INSERT INTO calibration_records (instrument_id, calibration_date, completed_at, notes)
VALUES (?, ?, ?, ?)
""", (instrument_id, row["calibration_date"], datetime.now().strftime("%Y-%m-%d %H:%M:%S"), notes.strip()))
self.conn.execute("UPDATE instruments SET calibration_date='' WHERE id=?", (instrument_id,))
def search_upcoming_calibrations(self, term=""):
rows = self.upcoming_calibrations()
term = (term or "").strip().lower()
if not term:
return rows
return [
r for r in rows if any(term in str(r[k] or "").lower() for k in [
"instrument_name", "manufacturer", "model", "serial_no",
"equipment_id", "customer_name", "location", "department", "calibration_date"
])
]
def search_calibration_history(self, term="", limit=DEFAULT_PAGE_SIZE, offset=0):
term = (term or "").strip()
if not term:
return self.conn.execute("""
SELECT cr.id, cr.calibration_date, cr.completed_at, cr.notes,
i.instrument_name, i.manufacturer, i.model, i.serial_no,
c.name AS customer_name, cs.location, cs.department
FROM calibration_records cr
JOIN instruments i ON i.id=cr.instrument_id
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
ORDER BY cr.completed_at DESC
LIMIT ? OFFSET ?
""", (limit, offset)).fetchall()
pattern = f"%{term}%"
return self.conn.execute("""
SELECT cr.id, cr.calibration_date, cr.completed_at, cr.notes,
i.instrument_name, i.manufacturer, i.model, i.serial_no,
c.name AS customer_name, cs.location, cs.department
FROM calibration_records cr
JOIN instruments i ON i.id=cr.instrument_id
JOIN customer_sites cs ON cs.id=i.site_id
JOIN customers c ON c.id=cs.customer_id
WHERE i.instrument_name LIKE ?
OR i.manufacturer LIKE ?
OR i.model LIKE ?
OR i.serial_no LIKE ?
OR c.name LIKE ?
OR cs.location LIKE ?
OR cs.department LIKE ?
OR cr.calibration_date LIKE ?
ORDER BY cr.completed_at DESC
LIMIT ? OFFSET ?
""", (pattern, pattern, pattern, pattern, pattern, pattern, pattern, pattern, limit, offset)).fetchall()
def dashboard_counts(self):
return {
"customers": self.conn.execute("SELECT COUNT(*) n FROM customer_sites").fetchone()["n"],
"instruments": self.conn.execute("SELECT COUNT(*) n FROM instruments").fetchone()["n"],
"service": self.conn.execute("SELECT COUNT(*) n FROM service_records").fetchone()["n"],
"amc": self.conn.execute("SELECT COUNT(*) n FROM amc_records").fetchone()["n"],
"calibration": self.conn.execute("SELECT COUNT(*) n FROM calibration_records").fetchone()["n"],
}
def monthly_activity(self, year, month):
num_days = calendar.monthrange(year, month)[1]
service = {d: 0 for d in range(1, num_days + 1)}
installation = {d: 0 for d in range(1, num_days + 1)}
calibration_counts = {d: 0 for d in range(1, num_days + 1)}
def process_dates(query, target_dict):
for row in self.conn.execute(query).fetchall():
raw = (row[0] or "").strip()
for fmt in ("%d-%m-%Y", "%Y-%m-%d", "%d/%m/%Y", "%Y/%m/%d"):
try:
p = datetime.strptime(raw, fmt).date()
if p.year == year and p.month == month:
target_dict[p.day] += 1
break
except ValueError:
pass
process_dates("SELECT service_date FROM service_records", service)
process_dates("SELECT installation_date FROM instruments WHERE installation_date IS NOT NULL AND TRIM(installation_date) <> ''", installation)
process_dates("SELECT calibration_date FROM calibration_records", calibration_counts)
return num_days, service, installation, calibration_counts
def instrument_packages(self, main_instrument):
main_instrument = (main_instrument or "").strip()
if not main_instrument:
return []
return self.conn.execute("""
SELECT * FROM instrument_packages
WHERE main_instrument=?
ORDER BY package_name, id
""", (main_instrument,)).fetchall()
def instrument_package(self, main_instrument, package_name=None, package_id=None):
if package_id is not None:
return self.conn.execute("SELECT * FROM instrument_packages WHERE id=?", (package_id,)).fetchone()
main_instrument = (main_instrument or "").strip()
if package_name is None:
return self.conn.execute("SELECT * FROM instrument_packages WHERE main_instrument=? ORDER BY id LIMIT 1", (main_instrument,)).fetchone()
return self.conn.execute("SELECT * FROM instrument_packages WHERE main_instrument=? AND package_name=?", (main_instrument, package_name.strip())).fetchone()
def create_instrument_package(self, main_instrument, package_name):
main_instrument = (main_instrument or "").strip()
package_name = (package_name or "").strip()
if not main_instrument or not package_name:
raise ValueError("Category and Package Name are required.")
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cur = self.conn.execute("""
INSERT INTO instrument_packages (main_instrument, package_name, created_at, updated_at)
VALUES (?, ?, ?, ?)
""", (main_instrument, package_name, now, now))
self.conn.commit()
return self.conn.execute("SELECT * FROM instrument_packages WHERE id=?", (cur.lastrowid,)).fetchone()
def update_instrument_package(self, package_id, package_name):
package_name = (package_name or "").strip()
if not package_name:
raise ValueError("Package Name is required.")
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.conn.execute("UPDATE instrument_packages SET package_name=?, updated_at=? WHERE id=?", (package_name, now, package_id))
self.conn.commit()
def instrument_package_items_by_id(self, package_id):
if not package_id:
return []
return self.conn.execute("SELECT * FROM instrument_package_items WHERE package_id=? ORDER BY line_no, id", (package_id,)).fetchall()
def delete_instrument_package(self, package_id):
if package_id:
self.conn.execute("DELETE FROM instrument_packages WHERE id=?", (package_id,))
self.conn.commit()
def datasheet_for_instrument(self, instrument_id):
return self.conn.execute("SELECT * FROM datasheets WHERE instrument_id=?", (instrument_id,)).fetchone()
def save_datasheet(self, instrument_id, file_path, file_name):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
original_name = os.path.basename(str(file_name))
self.conn.execute("""
INSERT INTO datasheets (instrument_id, file_path, file_name, original_name, added_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(instrument_id) DO UPDATE SET
file_path=excluded.file_path,
file_name=excluded.file_name,
original_name=excluded.original_name,
added_at=excluded.added_at
""", (instrument_id, str(file_path), original_name, original_name, now))
self.conn.commit()
def delete_datasheet(self, instrument_id):
self.conn.execute("DELETE FROM datasheets WHERE instrument_id=?", (instrument_id,))
self.conn.commit()
@staticmethod
def _hash_password(password):
salt = secrets.token_bytes(16)
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 100000)
return salt.hex() + ":" + digest.hex()
@staticmethod
def _verify_password(password, stored_hash):
try:
salt_hex, digest_hex = stored_hash.split(":", 1)
salt = bytes.fromhex(salt_hex)
expected = bytes.fromhex(digest_hex)
actual = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 100000)
return secrets.compare_digest(actual, expected)
except Exception:
return False
@staticmethod
def _hash_code(code):
return hashlib.sha256(code.encode("utf-8")).hexdigest()
def get_user(self, username):
return self.conn.execute("SELECT * FROM users WHERE username=?", (username.strip(),)).fetchone()
def create_user_pending(self, username, email, password, code, expires_at):
username = username.strip()
email = email.strip().lower()
if not re.fullmatch(r"[A-Za-z0-9_.-]{3,40}", username):
raise ValueError("User ID must be 3–40 characters (alphanumeric, dot, dash, underscore).")
if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", email):
raise ValueError("Invalid email address format.")
if len(password) < 8:
raise ValueError("Password must be at least 8 characters.")
if self.get_user(username):
raise ValueError("This User ID already exists.")
if self.conn.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone():
raise ValueError("This email is already registered.")
self.conn.execute("""
INSERT INTO users
(username, email, password_hash, role, verified, failed_attempts, locked,
verification_code_hash, verification_expires, created_at)
VALUES (?, ?, ?, 'Administrator', 0, 0, 0, ?, ?, ?)
""", (
username, email, self._hash_password(password),
self._hash_code(code), expires_at, datetime.now().isoformat(timespec="seconds")
))
self.conn.commit()
def verify_new_user(self, username, code):
user = self.get_user(username)
if not user:
return False, "Account not found."
expires = user["verification_expires"]
if not expires or datetime.fromisoformat(expires) < datetime.now():
return False, "Verification code expired."
if not secrets.compare_digest(self._hash_code(code.strip()), user["verification_code_hash"] or ""):
return False, "Incorrect verification code."
self.conn.execute("""
UPDATE users SET verified=1, verification_code_hash=NULL, verification_expires=NULL WHERE id=?
""", (user["id"],))
self.conn.commit()
return True, "Account verified successfully."
def authenticate(self, username, password):
user = self.get_user(username)
if not user:
return False, "Invalid credentials.", None
if not user["verified"]:
return False, "Account pending email verification.", user
if user["locked"]:
return False, "Account locked. Please reset password.", user
if not self._verify_password(password, user["password_hash"]):
attempts = int(user["failed_attempts"]) + 1
if attempts >= MAX_LOGIN_ATTEMPTS:
self.conn.execute("UPDATE users SET failed_attempts=?, locked=1 WHERE id=?", (attempts, user["id"]))
self.conn.commit()
return False, "Account locked after 3 failed attempts.", user
self.conn.execute("UPDATE users SET failed_attempts=? WHERE id=?", (attempts, user["id"]))
self.conn.commit()
return False, f"Incorrect password ({attempts}/{MAX_LOGIN_ATTEMPTS}).", user
self.conn.execute("UPDATE users SET failed_attempts=0, locked=0, last_login=? WHERE id=?",
(datetime.now().isoformat(timespec="seconds"), user["id"]))
self.conn.commit()
return True, "Login successful.", user
def start_password_reset(self, username, code, expires_at):
user = self.get_user(username)
if not user:
return False, "User ID not found."
self.conn.execute("""
UPDATE users SET verification_code_hash=?, verification_expires=? WHERE id=?
""", (self._hash_code(code), expires_at, user["id"]))
self.conn.commit()
return True, user["email"]
def complete_password_reset(self, username, code, new_password):
user = self.get_user(username)
if not user:
return False, "User ID not found."
if len(new_password) < 8:
return False, "Password must be at least 8 characters."
expires = user["verification_expires"]
if not expires or datetime.fromisoformat(expires) < datetime.now():
return False, "Verification code expired."
if not secrets.compare_digest(self._hash_code(code.strip()), user["verification_code_hash"] or ""):
return False, "Incorrect verification code."
self.conn.execute("""
UPDATE users SET password_hash=?, failed_attempts=0, locked=0,
verification_code_hash=NULL, verification_expires=NULL
WHERE id=?
""", (self._hash_password(new_password), user["id"]))
self.conn.commit()
return True, "Password reset successfully."
def close(self):
try:
self.conn.close()
except Exception:
pass
class AutoCompleteCombobox(ttk.Combobox):
def __init__(self, master, provider, select_callback=None, **kwargs):
super().__init__(master, state="normal", **kwargs)
self.provider = provider
self.select_callback = select_callback
self._updating = False
self._query_after_id = None
self.bind("<KeyRelease>", self._filter_values, add="+")
self.bind("<<ComboboxSelected>>", self._selected, add="+")
self.bind("<FocusIn>", self._focus_in, add="+")
def _focus_in(self, event=None):
self._set_values(self.get(), open_dropdown=False)
def _filter_values(self, event=None):
if self._updating:
return
if event and event.keysym in ("Up", "Down", "Left", "Right", "Return", "Escape", "Tab", "Shift_L", "Control_L"):
return
query = self.get()
if self._query_after_id:
try:
self.after_cancel(self._query_after_id)
except Exception:
pass
self._query_after_id = self.after(150, lambda: self._set_values(query, open_dropdown=True))
def _set_values(self, query="", open_dropdown=False):
if self._updating or not self.winfo_exists():
return
self._updating = True
try:
vals = list(self.provider(query))
self.configure(values=vals)
finally:
self._updating = False
def _selected(self, event=None):
if self.select_callback:
self.select_callback(event)
class DataTable(tk.Frame):
def __init__(self, master, columns, widths, height=12, selectmode="browse", fit_width=False):
super().__init__(master, bd=1, relief="solid", bg="#9b9b9b")
self.columns = columns
self.widths = widths
self.height = height
self.fit_width = fit_width
self.rows = []
self.selected_iid = None
self.on_select_callback = None
self._sort_column = None
self._sort_reverse = False
self.row_tooltip_provider = None
self._tooltip_window = None
self._tooltip_after_id = None
self._tooltip_row = None
self.tree = ttk.Treeview(
self, columns=[c[0] for c in columns], show="headings",
height=height, selectmode=selectmode
)
for (key, title), width in zip(columns, widths):
self.tree.heading(key, text=title, anchor="center", command=lambda k=key: self.sort_by_column(k))
self.tree.column(key, width=width, minwidth=55, anchor="center", stretch=not fit_width)
vsb = ttk.Scrollbar(self, orient="vertical", command=self.tree.yview)
hsb = ttk.Scrollbar(self, orient="horizontal", command=self.tree.xview)
self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
self.tree.grid(row=0, column=0, sticky="nsew")
vsb.grid(row=0, column=1, sticky="ns")
hsb.grid(row=1, column=0, sticky="ew")
self.grid_rowconfigure(0, weight=1)
self.grid_columnconfigure(0, weight=1)
self.tree.bind("<<TreeviewSelect>>", self._select)
self.tree.bind("<Button-3>", self._right_click)
self.tree.bind("<Motion>", self._on_motion, add="+")
self.tree.bind("<Leave>", self._hide_tooltip, add="+")
if self.fit_width:
self.bind("<Configure>", self._fit_columns)
self._fit_after_id = None
def _on_motion(self, event):
if not self.row_tooltip_provider:
return
iid = self.tree.identify_row(event.y)
if iid == self._tooltip_row:
return
self._tooltip_row = iid
self._hide_tooltip()
if not iid:
return
if self._tooltip_after_id:
try:
self.after_cancel(self._tooltip_after_id)
except Exception:
pass
self._tooltip_after_id = self.after(350, lambda: self._show_tooltip(iid))
def _show_tooltip(self, iid):
self._tooltip_after_id = None
if not self.winfo_exists() or self._tooltip_row != iid:
return
try:
text = self.row_tooltip_provider(iid)
except Exception:
text = None
if not text:
return
self._hide_tooltip()
tip = tk.Toplevel(self)
tip.wm_overrideredirect(True)
try:
tip.wm_attributes("-topmost", True)
except Exception:
pass
label = tk.Label(
tip, text=text, justify="left", wraplength=320,
bg="#FFFFE0", fg="#000000", relief="solid", bd=1,
font=("Segoe UI", 9), padx=6, pady=4
)
label.pack()
tip.wm_geometry(f"+{self.tree.winfo_pointerx() + 14}+{self.tree.winfo_pointery() + 14}")
self._tooltip_window = tip
def _hide_tooltip(self, event=None):
if self._tooltip_after_id:
try:
self.after_cancel(self._tooltip_after_id)
except Exception:
pass
self._tooltip_after_id = None
if self._tooltip_window is not None:
try:
self._tooltip_window.destroy()
except Exception:
pass
self._tooltip_window = None
def _fit_columns(self, event=None):
if not self.fit_width or not self.winfo_exists():
return
if getattr(self, "_fit_after_id", None):
try:
self.after_cancel(self._fit_after_id)
except Exception:
pass
self._fit_after_id = self.after(30, self._do_fit_columns)
def _do_fit_columns(self):
self._fit_after_id = None
if not self.winfo_exists():
return
available = max(100, self.winfo_width() - 18)
total_base = sum(self.widths) or 1
for (key, _), base in zip(self.columns, self.widths):
width = max(55, int(available * base / total_base))
self.tree.column(key, width=width, minwidth=55, stretch=False)
def _select(self, event=None):
selection = self.tree.selection()
if selection:
self.selected_iid = selection[0]
if self.on_select_callback:
self.on_select_callback(self.selected_iid)
def _right_click(self, event):
iid = self.tree.identify_row(event.y)
if iid:
self.tree.selection_set(iid)
self.tree.focus(iid)
self.selected_iid = iid
self._select()
return "break"
def bind_right_click(self, callback):
self.tree.bind("<Button-3>", lambda e: self._right_click_custom(e, callback))
def _right_click_custom(self, event, callback):
iid = self.tree.identify_row(event.y)
if iid:
self.tree.selection_set(iid)
self.tree.focus(iid)
self.selected_iid = iid
if self.on_select_callback:
self.on_select_callback(iid)
callback(event, iid)
return "break"
def sort_by_column(self, column):
if self._sort_column == column:
self._sort_reverse = not self._sort_reverse
else:
self._sort_column = column
self._sort_reverse = False
items = []
for iid in self.tree.get_children(""):
values = self.tree.item(iid, "values")
col_index = [c[0] for c in self.columns].index(column)
items.append((str(values[col_index] if col_index < len(values) else "").lower(), iid))
items.sort(key=lambda item: item[0], reverse=self._sort_reverse)
for index, (_, iid) in enumerate(items):
self.tree.move(iid, "", index)
for key, title in self.columns:
arrow = (" ▼" if self._sort_reverse else " ▲") if key == self._sort_column else ""
self.tree.heading(key, text=title + arrow)
def clear(self):
children = self.tree.get_children()
if children:
self.tree.delete(*children)
self.rows.clear()
self.selected_iid = None
def insert(self, values, iid=None):
iid = str(iid if iid is not None else len(self.rows) + 1)
self.tree.insert("", "end", iid=iid, values=values)
self.rows.append((iid, values))
def clear_selection(self):
try:
self.tree.selection_remove(self.tree.selection())
except Exception:
pass
def send_verification_email_async(root, recipient, code, purpose, username, on_done):
def worker():
err = None
try:
if not all([SMTP_HOST, SMTP_USER, SMTP_PASSWORD, SMTP_FROM]):
raise RuntimeError("SMTP environment variables are missing.")
subj = "ERT CONSORTIUM - Account Verification" if purpose == "create" else "ERT CONSORTIUM - Password Reset"
body = f"User ID: {username}\nVerification Code: {code}\nExpires in {VERIFICATION_EXPIRY_MINUTES} mins."
msg = f"From: {SMTP_FROM}\r\nTo: {recipient}\r\nSubject: {subj}\r\n\r\n{body}\r\n"
ctx = ssl.create_default_context()
with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=ctx, timeout=12) as s:
s.login(SMTP_USER, SMTP_PASSWORD)
s.sendmail(SMTP_FROM, [recipient], msg)
except Exception as e:
err = e
try:
root.after(0, lambda: on_done(err))
except Exception:
pass
threading.Thread(target=worker, daemon=True).start()
class AuthWindow(tk.Toplevel):
def __init__(self, master, db, on_success):
super().__init__(master)
self.master = master
self.db = db
self.on_success = on_success
self.title("ERT CONSORTIUM - Login")
self.geometry("500x420")
self.resizable(False, False)
self.protocol("WM_DELETE_WINDOW", self._cancel)
self.grab_set()
self.configure(bg="#F7FAFC")
top = tk.Frame(self, bg="#F7FAFC")
top.pack(fill="x", padx=30, pady=(20, 5))
tk.Label(top, text="ERT CONSORTIUM", font=(UI_FONT, 20, "bold"), bg="#F7FAFC", fg=ERT_NAVY).pack()
tk.Label(self, text="Management Authentication", font=(UI_FONT, 12, "bold"), bg="#F7FAFC", fg="#607D8B").pack(pady=(2, 12))
form = tk.Frame(self, bg="#F7FAFC")
form.pack(fill="x", padx=60)
tk.Label(form, text="User ID", bg="#F7FAFC", font=(UI_FONT, 10, "bold")).pack(anchor="w", pady=(4, 2))
self.username_var = tk.StringVar()
tk.Entry(form, textvariable=self.username_var, font=(UI_FONT, 11)).pack(fill="x", pady=(0, 8))
tk.Label(form, text="Password", bg="#F7FAFC", font=(UI_FONT, 10, "bold")).pack(anchor="w", pady=(4, 2))
self.password_var = tk.StringVar()
self.password_entry = tk.Entry(form, textvariable=self.password_var, show="•", font=(UI_FONT, 11))
self.password_entry.pack(fill="x", pady=(0, 14))
tk.Button(
form, text="LOGIN", command=self.login, bg=ERT_BLUE, fg="white",
relief="flat", font=(UI_FONT, 10, "bold"), cursor="hand2", pady=6
).pack(fill="x")
links = tk.Frame(self, bg="#F7FAFC")
links.pack(pady=12)
tk.Button(links, text="Create Account", command=lambda: CreateAccountWindow(self, self.db), relief="flat", bg="#F7FAFC", fg=ERT_NAVY, font=(UI_FONT, 9, "bold"), cursor="hand2").pack(side="left", padx=8)
tk.Button(links, text="Forgot Password?", command=lambda: ResetPasswordWindow(self, self.db, self.username_var.get().strip()), relief="flat", bg="#F7FAFC", fg=ERT_NAVY, font=(UI_FONT, 9), cursor="hand2").pack(side="left", padx=8)
self.bind("<Return>", lambda e: self.login())
def login(self):
u = self.username_var.get().strip()
p = self.password_var.get()
if not u or not p:
messagebox.showwarning("Login", "Enter User ID and Password.", parent=self)
return
ok, msg, user = self.db.authenticate(u, p)
if ok:
self.grab_release()
self.destroy()
self.on_success(user)
else:
messagebox.showerror("Login", msg, parent=self)
def _cancel(self):
try:
self.grab_release()
except Exception:
pass
self.master.destroy()
class CreateAccountWindow(tk.Toplevel):
def __init__(self, parent, db):
super().__init__(parent)
self.db = db
self.title("Create Administrator Account")
self.geometry("520x480")
self.resizable(False, False)
self.configure(bg="#F7FAFC")
self.transient(parent)
self.grab_set()
self.username_var = tk.StringVar()
self.email_var = tk.StringVar()
self.password_var = tk.StringVar()
self.confirm_var = tk.StringVar()
tk.Label(self, text="Create Administrator Account", font=(UI_FONT, 16, "bold"), bg="#F7FAFC", fg=ERT_NAVY).pack(pady=(18, 12))
form = tk.Frame(self, bg="#F7FAFC")
form.pack(fill="x", padx=50)
for label, var, hide in [
("User ID", self.username_var, False),
("Email Address", self.email_var, False),
("Password", self.password_var, True),
("Confirm Password", self.confirm_var, True)
]:
tk.Label(form, text=label, bg="#F7FAFC", font=(UI_FONT, 9, "bold")).pack(anchor="w", pady=(3, 1))
tk.Entry(form, textvariable=var, show="•" if hide else "", font=(UI_FONT, 10)).pack(fill="x", pady=(0, 6))
self.create_btn = tk.Button(
self, text="CREATE ACCOUNT", command=self.create_account,
bg=ERT_BLUE, fg="white", relief="flat", font=(UI_FONT, 10, "bold"), cursor="hand2", pady=6
)
self.create_btn.pack(fill="x", padx=50, pady=14)
def create_account(self):
u, e, p, c = self.username_var.get().strip(), self.email_var.get().strip(), self.password_var.get(), self.confirm_var.get()
if p != c:
messagebox.showerror("Error", "Passwords do not match.", parent=self)
return
code = f"{secrets.randbelow(1000000):06d}"
expires = datetime.fromtimestamp(datetime.now().timestamp() + VERIFICATION_EXPIRY_MINUTES * 60).isoformat(timespec="seconds")
try:
self.db.create_user_pending(u, e, p, code, expires)
except Exception as ex:
messagebox.showerror("Registration Error", str(ex), parent=self)
return
self.create_btn.config(state="disabled", text="SENDING CODE...")
def on_done(err):
self.create_btn.config(state="normal", text="CREATE ACCOUNT")
if err:
self.db.conn.execute("DELETE FROM users WHERE username=?", (u,))
self.db.conn.commit()
messagebox.showerror("Email Error", f"Failed to send email:\n{err}", parent=self)
return
messagebox.showinfo("Verification Sent", "A verification code was dispatched to your email.", parent=self)
VerifyAccountWindow(self, self.db, u, on_verified=self._verified)
send_verification_email_async(self, e, code, "create", u, on_done)
def _verified(self, u):
self.destroy()
class VerifyAccountWindow(tk.Toplevel):
def __init__(self, parent, db, username, on_verified=None):
super().__init__(parent)
self.db, self.username, self.on_verified = db, username, on_verified
self.title("Verify Account")
self.geometry("400x220")
self.resizable(False, False)
self.configure(bg="#F7FAFC")
self.transient(parent)
self.grab_set()
self.code_var = tk.StringVar()
tk.Label(self, text="Email Verification", font=(UI_FONT, 15, "bold"), bg="#F7FAFC", fg=ERT_NAVY).pack(pady=(20, 6))
tk.Label(self, text=f"Enter code sent to '{username}'", bg="#F7FAFC", font=(UI_FONT, 9)).pack()
tk.Entry(self, textvariable=self.code_var, justify="center", font=(UI_FONT, 14)).pack(padx=60, fill="x", pady=10)
tk.Button(self, text="VERIFY", command=self.verify, bg=ERT_BLUE, fg="white", relief="flat", font=(UI_FONT, 10, "bold"), cursor="hand2", pady=6).pack(fill="x", padx=60)
def verify(self):
code = self.code_var.get().strip()
ok, msg = self.db.verify_new_user(self.username, code)
if ok:
messagebox.showinfo("Success", msg, parent=self)
if self.on_verified:
self.on_verified(self.username)
self.destroy()
else:
messagebox.showerror("Error", msg, parent=self)
class ResetPasswordWindow(tk.Toplevel):
def __init__(self, parent, db, prefilled_username=""):
super().__init__(parent)
self.db = db
self.title("Reset Password")
self.geometry("480x420")
self.resizable(False, False)
self.configure(bg="#F7FAFC")
self.transient(parent)
self.grab_set()
self.username_var = tk.StringVar(value=prefilled_username)
self.code_var = tk.StringVar()
self.pass_var = tk.StringVar()
self.confirm_var = tk.StringVar()
tk.Label(self, text="Password Recovery", font=(UI_FONT, 15, "bold"), bg="#F7FAFC", fg=ERT_NAVY).pack(pady=(16, 8))
form = tk.Frame(self, bg="#F7FAFC")
form.pack(fill="x", padx=50)
for label, var, hide in [
("User ID", self.username_var, False),
("Verification Code", self.code_var, False),
("New Password", self.pass_var, True),
("Confirm Password", self.confirm_var, True)
]:
tk.Label(form, text=label, bg="#F7FAFC", font=(UI_FONT, 9, "bold")).pack(anchor="w", pady=(2, 1))
tk.Entry(form, textvariable=var, show="•" if hide else "", font=(UI_FONT, 10)).pack(fill="x", pady=(0, 5))
self.send_btn = tk.Button(self, text="SEND RESET CODE", command=self.send_code, bg="#607D8B", fg="white", relief="flat", font=(UI_FONT, 9, "bold"), pady=5)
self.send_btn.pack(fill="x", padx=50, pady=(6, 4))
tk.Button(self, text="RESET PASSWORD", command=self.reset, bg=ERT_BLUE, fg="white", relief="flat", font=(UI_FONT, 10, "bold"), pady=6).pack(fill="x", padx=50, pady=4)
def send_code(self):
u = self.username_var.get().strip()
code = f"{secrets.randbelow(1000000):06d}"
expires = datetime.fromtimestamp(datetime.now().timestamp() + VERIFICATION_EXPIRY_MINUTES * 60).isoformat(timespec="seconds")
ok, res = self.db.start_password_reset(u, code, expires)
if not ok:
messagebox.showerror("Error", res, parent=self)
return
self.send_btn.config(state="disabled", text="SENDING CODE...")
def on_done(err):
self.send_btn.config(state="normal", text="SEND RESET CODE")
if err:
messagebox.showerror("Error", str(err), parent=self)
return
messagebox.showinfo("Sent", "Reset verification code dispatched to email.", parent=self)
send_verification_email_async(self, res, code, "reset", u, on_done)
def reset(self):
u, c, p, conf = self.username_var.get().strip(), self.code_var.get().strip(), self.pass_var.get(), self.confirm_var.get()
if p != conf:
messagebox.showerror("Error", "Passwords do not match.", parent=self)
return
ok, msg = self.db.complete_password_reset(u, c, p)
if ok:
messagebox.showinfo("Success", msg, parent=self)
self.destroy()
else:
messagebox.showerror("Error", msg, parent=self)
class App(tk.Tk):
def __init__(self):
super().__init__()
self.title(APP_TITLE)
self.geometry("1400x850")
try:
self.state("zoomed")
except tk.TclError:
self.attributes("-zoomed", True)
# --- ADD THIS BLOCK TO REMOVE THE FEATHER ICON ---
try:
base_dir = os.path.dirname(os.path.abspath(__file__))
ico_path = os.path.join(base_dir, "ERT_ico.ico")
if os.path.exists(ico_path):
self.iconbitmap(default=ico_path) # 'default' applies it to the login screen too!
except Exception:
pass
# -------------------------------------------------
self.minsize(1120, 700)
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), DB_FILENAME)
self.db = Database(self.db_path)
self.current_user = None
self.selected_instrument_id = None
self.selected_site_id = None
self.nav_buttons = {}
self.theme = "White"
self.settings_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "app_settings.json")
self.clock_visible = self.load_app_setting("clock_visible", True)
self._debounce_timers = {}
self.setup_styles()
self.setup_menu()
self.build_layout()
self.update_clock()
self.withdraw()
self.after(100, self.start_authentication)
self.protocol("WM_DELETE_WINDOW", self.exit_app)
self.bind("<Map>", self._ensure_main_window_maximized, add="+")
def _debounce(self, key, func, delay=250):
if key in self._debounce_timers:
try:
self.after_cancel(self._debounce_timers[key])
except Exception:
pass
self._debounce_timers[key] = self.after(delay, func)
def load_app_settings(self):
try:
if os.path.isfile(self.settings_file):
with open(self.settings_file, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def load_app_setting(self, key, default=None):
return self.load_app_settings().get(key, default)
def save_app_setting(self, key, value):
data = self.load_app_settings()
data[key] = value
try:
with open(self.settings_file, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
except Exception as ex:
messagebox.showerror("Settings Error", str(ex), parent=self)
def start_authentication(self):
AuthWindow(self, self.db, self.authentication_success)
def authentication_success(self, user):
self.current_user = user
self.deiconify()
self.update_user_display()
self.show_dashboard()
def update_user_display(self):
if hasattr(self, "user_label"):
self.user_label.config(text=f"Username: {self.current_user['username']}" if self.current_user else "")
def setup_styles(self):
self.style = ttk.Style(self)
try:
self.style.theme_use("clam")
except tk.TclError:
pass
self.apply_theme("White")
def apply_theme(self, theme):
self.theme = theme
bg, fg, field, accent, selected = ("#17202A", "#F5F7FA", "#25313C", "#34495E", "#0077B6") if theme == "Dark" else ("#F7FAFC", "#111827", "#FFFFFF", "#D9F3FB", "#00AEEF")
self.configure(bg=bg)
self.option_add("*Font", (UI_FONT, 9))
self.style.configure(".", font=(UI_FONT, 9))
self.style.configure("TFrame", background=bg)
self.style.configure("TLabel", background=bg, foreground=fg, font=(UI_FONT, 9))
self.style.configure("Title.TLabel", font=(UI_FONT, 20, "bold"), background=bg, foreground=ERT_NAVY if theme != "Dark" else "#8EDCFF")
self.style.configure("Clock.TLabel", font=(UI_FONT, 10, "bold"), background=bg, foreground=fg)
self.style.configure("Treeview", background=field, foreground=fg, fieldbackground=field, rowheight=25, borderwidth=1, relief="solid")
self.style.configure("Treeview.Heading", background=accent, foreground=fg, relief="raised", borderwidth=1, font=(UI_FONT, 9, "bold"))
self.style.map("Treeview", background=[("selected", selected)], foreground=[("selected", "white")])
def setup_menu(self):
menu = tk.Menu(self)
file_menu = tk.Menu(menu, tearoff=0)
file_menu.add_command(label="Signout", command=self.sign_out)
file_menu.add_separator()
file_menu.add_command(label="Exit", command=self.exit_app)
menu.add_cascade(label="File", menu=file_menu)
menu.add_command(label="Datasheets", command=self.show_datasheets)
menu.add_command(label="Instrument Configuration", command=self.show_instrument_configuration)
menu.add_command(label="About", command=lambda: messagebox.showinfo("About", "ERT CONSORTIUM Management System\nBuild: " + APP_BUILD))
self.config(menu=menu)
def build_layout(self):
self.main = tk.Frame(self, bg=self.cget("bg"))
self.main.pack(fill="both", expand=True)
self.sidebar = tk.Frame(self.main, width=205, bg="#FFFFFF")
self.sidebar.pack(side="left", fill="y")
self.sidebar.pack_propagate(False)
self.content = tk.Frame(self.main, bg=self.cget("bg"))
self.content.pack(side="left", fill="both", expand=True)
self.build_sidebar()
self.header = tk.Frame(self.content, height=52, bg=self.cget("bg"))
self.header.pack(fill="x")
self.header.pack_propagate(False)
self.page_title = ttk.Label(self.header, text="", style="Title.TLabel")
self.page_title.pack(side="left", padx=14, pady=8)
right_header = tk.Frame(self.header, bg=self.cget("bg"))
right_header.pack(side="right", padx=14)
self.user_label = tk.Label(right_header, text="", bg=self.cget("bg"), fg=ERT_NAVY, font=(UI_FONT, 9, "bold"), width=20, anchor="e")
self.user_label.pack(side="left", padx=(0, 18))
self.clock_label = ttk.Label(right_header, text="", style="Clock.TLabel", width=23, anchor="e")
if self.clock_visible:
self.clock_label.pack(side="left")
self.page = tk.Frame(self.content, bg=self.cget("bg"))
self.page.pack(fill="both", expand=True, padx=10, pady=(0, 10))
def build_sidebar(self):
tk.Label(self.sidebar, text="MANAGEMENT", bg="#FFFFFF", fg=ERT_NAVY, font=(UI_FONT, 11, "bold")).pack(fill="x", padx=12, pady=(18, 10))
buttons = [
("Dashboard", self.show_dashboard),
("Customers", self.show_customers),
("Instruments", self.show_instruments),
("Service Records", self.show_service),
("AMC", self.show_amc),
("Calibration", self.show_calibration),
("Monthly Records", self.show_monthly_records),
("Settings", self.show_settings),
]
self.nav_buttons = {}
for text, command in buttons:
btn = tk.Button(
self.sidebar, text=text, command=lambda c=command, t=text: self.navigate(t, c),
anchor="center", bg="#FFFFFF", fg=ERT_NAVY, activebackground=ERT_BLUE,
activeforeground="white", relief="solid", bd=1, font=(UI_FONT, 11, "bold"), cursor="hand2"
)
btn.pack(fill="x", padx=10, pady=4, ipady=8)
self.nav_buttons[text] = btn
def navigate(self, title, command):
self.set_active_nav(title)
command()
def set_active_nav(self, title):
self.active_nav = title
if not hasattr(self, "nav_buttons"):
return
for name, button in self.nav_buttons.items():
is_active = (name == title)
button.configure(
bg=ERT_BLUE if is_active else "#FFFFFF",
fg="white" if is_active else ERT_NAVY,
activebackground=ERT_BLUE,
activeforeground="white"
)
def clear_page(self, title):
if hasattr(self, "sidebar") and self.sidebar.winfo_exists():
if not self.sidebar.winfo_manager():
self.sidebar.pack(side="left", fill="y", before=self.content)
self.sidebar.pack_propagate(False)
if hasattr(self, "content") and self.content.winfo_exists():
self.content.pack_configure(side="left", fill="both", expand=True)
for w in self.page.winfo_children():
w.destroy()
self.page_title.config(text=title)
self.set_active_nav(title)
def section(self, parent, title):
frame = tk.LabelFrame(parent, text=title, font=(UI_FONT, 9, "bold"), bd=1, relief="solid", padx=6, pady=5)
frame.pack(fill="x", pady=(0, 7))
return frame
def show_dashboard(self):
self.clear_page("Dashboard")
counts = self.db.dashboard_counts()
cards = tk.Frame(self.page, bg=self.cget("bg"))
cards.pack(fill="x", pady=5)
card_data = [
("Customer Sites", counts["customers"]),
("Instruments", counts["instruments"]),
("Service Records", counts["service"]),
("AMC Records", counts["amc"]),
("Completed Calibrations", counts["calibration"]),
]
for i, (lbl, val) in enumerate(card_data):
c = tk.Frame(cards, bd=1, relief="solid", padx=18, pady=12)
c.grid(row=0, column=i, padx=5, sticky="ew")
cards.grid_columnconfigure(i, weight=1)
tk.Label(c, text=str(val), font=(UI_FONT, 22, "bold")).pack()
tk.Label(c, text=lbl, font=(UI_FONT, 9)).pack()
upcoming = self.section(self.page, "Upcoming Calibration")
upcoming.pack_forget()
upcoming.pack(fill="both", expand=True, pady=(5, 0))
upcoming.configure(fg="#C62828", font=(UI_FONT, 10, "bold"))
table_frame = tk.Frame(upcoming)
table_frame.pack(fill="both", expand=True)
table = DataTable(
table_frame,
[("date", "Calibration"), ("days", "Days Left"), ("instrument", "Instrument"),
("manufacturer", "Manufacturer"), ("model", "Model"), ("customer", "Customer"),
("location", "Location"), ("department", "Department")],
[105, 85, 165, 125, 105, 180, 160, 135], height=12, fit_width=True
)
table.pack(fill="both", expand=True)
for r in self.db.upcoming_calibrations():
raw = (r["calibration_date"] or "").strip()
cal_date = None
for fmt in ("%d-%m-%Y", "%Y-%m-%d", "%d/%m/%Y", "%Y/%m/%d"):
try:
cal_date = datetime.strptime(raw, fmt).date()
break
except ValueError:
continue
days_left = (cal_date - date.today()).days if cal_date else ""
table.insert([
r["calibration_date"] or "", str(days_left) if days_left != "" else "",
r["instrument_name"], r["manufacturer"], r["model"] or "",
r["customer_name"], r["location"], r["department"]
])
def show_customers(self):
self.clear_page("Customers")
form = self.section(self.page, "Customer / Location / Department")
self.customer_name_var = tk.StringVar()
self.location_var = tk.StringVar()
self.department_var = tk.StringVar()
for i, label in enumerate(["Customer Name", "Location", "Department"]):
tk.Label(form, text=label).grid(row=0, column=i, sticky="w", padx=5)
self.customer_name_cb = AutoCompleteCombobox(form, provider=self.db.search_customer_names, textvariable=self.customer_name_var)
self.location_cb = AutoCompleteCombobox(form, provider=self.db.all_locations, textvariable=self.location_var)
self.department_cb = AutoCompleteCombobox(form, provider=self.db.all_departments, textvariable=self.department_var)
self.customer_name_cb.grid(row=1, column=0, sticky="ew", padx=5, pady=(3, 4))
self.location_cb.grid(row=1, column=1, sticky="ew", padx=5, pady=(3, 4))
self.department_cb.grid(row=1, column=2, sticky="ew", padx=5, pady=(3, 4))
tk.Button(form, text="Add Customer", command=self.add_customer_site).grid(row=1, column=3, padx=8)
for i in range(3):
form.grid_columnconfigure(i, weight=1)
search_bar = tk.LabelFrame(self.page, text="Search Customer Database", font=(UI_FONT, 9, "bold"), padx=6, pady=5)
search_bar.pack(fill="x", pady=(4, 5))
self.customer_search_var = tk.StringVar()
entry = tk.Entry(search_bar, textvariable=self.customer_search_var, width=55)
entry.pack(side="left", padx=4, fill="x", expand=True)
entry.bind("<KeyRelease>", lambda e: self._debounce("customer_search", self.refresh_customer_table))
tk.Button(search_bar, text="Search", command=self.refresh_customer_table, font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
tk.Button(search_bar, text="Clear Search", command=lambda: [self.customer_search_var.set(""), self.refresh_customer_table()]).pack(side="left", padx=4)
db_frame = self.section(self.page, "Customer Database")
self.customer_table = DataTable(db_frame, [("customer", "Customer Name"), ("location", "Location"), ("department", "Department")], [320, 300, 280], height=18, fit_width=True)
self.customer_table.pack(fill="both", expand=True)
self.customer_table.on_select_callback = self.customer_row_selected
self.customer_table.bind_right_click(self.customer_context_menu)
self.refresh_customer_table()
def add_customer_site(self):
try:
self.db.add_customer_site(self.customer_name_var.get(), self.location_var.get(), self.department_var.get())
self.refresh_customer_table()
messagebox.showinfo("Saved", "Customer site saved successfully.")
except Exception as e:
messagebox.showwarning("Customer Error", str(e))
def customer_row_selected(self, iid):
try:
site_id = int(iid)
self.selected_site_id = site_id
row = self.db.conn.execute("SELECT c.name, cs.location, cs.department FROM customer_sites cs JOIN customers c ON c.id=cs.customer_id WHERE cs.id=?", (site_id,)).fetchone()
if row:
self.customer_name_var.set(row["name"])
self.location_var.set(row["location"])
self.department_var.set(row["department"])
except Exception:
pass
def customer_context_menu(self, event, iid):
menu = tk.Menu(self, tearoff=0)
menu.add_command(label="Update Record", command=self.update_customer_site)
menu.add_command(label="Delete Record", command=self.delete_customer_site)
menu.tk_popup(event.x_root, event.y_root)
def update_customer_site(self):
if not self.selected_site_id:
return
try:
self.db.update_customer_site(self.selected_site_id, self.customer_name_var.get(), self.location_var.get(), self.department_var.get())
self.refresh_customer_table()
messagebox.showinfo("Updated", "Customer site updated.")
except Exception as e:
messagebox.showwarning("Error", str(e))
def delete_customer_site(self):
if not self.selected_site_id or not messagebox.askyesno("Confirm Delete", "Remove this customer entry?"):
return
try:
self.db.delete_customer_site(self.selected_site_id)
self.selected_site_id = None
self.refresh_customer_table()
except Exception as e:
messagebox.showwarning("Delete Error", str(e))
def refresh_customer_table(self):
if not hasattr(self, "customer_table"):
return
self.customer_table.clear()
for r in self.db.search_customer_sites(self.customer_search_var.get()):
self.customer_table.insert([r["customer_name"], r["location"], r["department"]], iid=r["id"])
def make_date_entry(self, parent, variable):
entry = DateEntry(parent, textvariable=variable, date_pattern="dd-mm-yyyy", width=13, showweeknumbers=False, font=(UI_FONT, 10))
try:
entry.delete(0, tk.END)
except Exception:
pass
variable.set("")
return entry
def show_instruments(self):
self.clear_page("Instruments")
self.selected_instrument_id = None
self._instrument_configurations = {}
if hasattr(self, "sidebar") and self.sidebar.winfo_exists():
self.sidebar.pack_forget()
if hasattr(self, "content") and self.content.winfo_exists():
self.content.pack_configure(side="left", fill="both", expand=True)
back_bar = tk.Frame(self.page)
back_bar.pack(fill="x", pady=(0, 6))
tk.Button(back_bar, text="← Back", command=self.show_dashboard, font=(UI_FONT, 10, "bold"), padx=10, pady=5).pack(side="left")
form = self.section(self.page, "Add / Edit Instrument")
for c in range(10):
form.grid_columnconfigure(c, weight=1)
self.manufacturer_var = tk.StringVar()
self.instrument_var = tk.StringVar()
self.other_manufacturer_var = tk.StringVar()
self.model_var = tk.StringVar()
self.serial_no_var = tk.StringVar()
self.equipment_id_var = tk.StringVar()
self.installation_var = tk.StringVar()
self.calibration_var = tk.StringVar()
self.calibration_due_var = tk.StringVar()
self.configuration_var = tk.StringVar()
self.dynamic_var = tk.StringVar()
tk.Label(form, text="Manufacturer").grid(row=0, column=0, sticky="w", padx=4)
self.manufacturer_cb = ttk.Combobox(form, textvariable=self.manufacturer_var, values=list(MANUFACTURER_INSTRUMENTS.keys()), state="readonly")
self.manufacturer_cb.grid(row=1, column=0, sticky="ew", padx=4, pady=2)
self.manufacturer_cb.bind("<<ComboboxSelected>>", self.manufacturer_changed)
self.instrument_label = tk.Label(form, text="Instrument")
self.instrument_label.grid(row=0, column=1, sticky="w", padx=4)
self.instrument_cb = ttk.Combobox(form, textvariable=self.instrument_var, state="readonly")
self.instrument_cb.grid(row=1, column=1, sticky="ew", padx=4, pady=2)
self.instrument_cb.bind("<<ComboboxSelected>>", self.instrument_changed)
self.other_manufacturer_label = tk.Label(form, text="Manufacturer Name")
self.other_manufacturer_entry = tk.Entry(form, textvariable=self.other_manufacturer_var)
self.other_instrument_label = tk.Label(form, text="Instrument Name")
self.other_instrument_entry = tk.Entry(form, textvariable=self.instrument_var)
self.dynamic_label = tk.Label(form, text="")
self.dynamic_cb = ttk.Combobox(form, textvariable=self.dynamic_var, state="readonly")
self.dynamic_entry = tk.Entry(form, textvariable=self.dynamic_var)
for label, var, col in [("Model", self.model_var, 3), ("Serial No.", self.serial_no_var, 4), ("Equipment ID", self.equipment_id_var, 5)]:
tk.Label(form, text=label).grid(row=0, column=col, sticky="w", padx=4)
tk.Entry(form, textvariable=var).grid(row=1, column=col, sticky="ew", padx=4, pady=2)
tk.Label(form, text="Installation Date").grid(row=0, column=6, sticky="w", padx=4)
self.installation_entry = self.make_date_entry(form, self.installation_var)
self.installation_entry.grid(row=1, column=6, sticky="ew", padx=4, pady=2)
tk.Label(form, text="Calibration Date").grid(row=0, column=7, sticky="w", padx=4)
self.calibration_entry = self.make_date_entry(form, self.calibration_var)
self.calibration_entry.grid(row=1, column=7, sticky="ew", padx=4, pady=2)
tk.Label(form, text="Calib. Due Date").grid(row=0, column=8, sticky="w", padx=4)
self.calibration_due_entry = self.make_date_entry(form, self.calibration_due_var)
self.calibration_due_entry.grid(row=1, column=8, sticky="ew", padx=4, pady=2)
self.instrument_customer_var = tk.StringVar()
self.instrument_location_var = tk.StringVar()
self.instrument_department_var = tk.StringVar()
tk.Label(form, text="Customer").grid(row=2, column=0, sticky="w", padx=4, pady=(6, 0))
tk.Label(form, text="Location").grid(row=2, column=1, sticky="w", padx=4, pady=(6, 0))
tk.Label(form, text="Department").grid(row=2, column=2, sticky="w", padx=4, pady=(6, 0))
self.instrument_customer_cb = AutoCompleteCombobox(form, provider=self.db.search_customer_names, textvariable=self.instrument_customer_var)
self.instrument_location_cb = AutoCompleteCombobox(form, provider=self.db.all_locations, textvariable=self.instrument_location_var)
self.instrument_department_cb = AutoCompleteCombobox(form, provider=self.db.all_departments, textvariable=self.instrument_department_var)
self.instrument_customer_cb.grid(row=3, column=0, sticky="ew", padx=4, pady=2)
self.instrument_location_cb.grid(row=3, column=1, sticky="ew", padx=4, pady=2)
self.instrument_department_cb.grid(row=3, column=2, sticky="ew", padx=4, pady=2)
tk.Label(form, text="Configuration (optional)").grid(row=2, column=3, columnspan=4, sticky="w", padx=4, pady=(6, 0))
tk.Entry(form, textvariable=self.configuration_var).grid(row=3, column=3, columnspan=4, sticky="ew", padx=4, pady=2)
self.save_button = tk.Button(form, text="SAVE", command=self.save_instrument, bg="#287b2f", fg="white", font=(UI_FONT, 10, "bold"), width=11, relief="flat")
self.save_button.grid(row=1, column=9, rowspan=2, padx=8, pady=2, sticky="e")
tk.Button(form, text="Clear Filters", command=self.clear_instrument_filters, font=(UI_FONT, 9, "bold")).grid(row=3, column=9, padx=8, pady=3, sticky="e")
search_bar = tk.LabelFrame(self.page, text="Search Instrument Database", font=(UI_FONT, 9, "bold"), padx=6, pady=5)
search_bar.pack(fill="x", pady=(4, 5))
self.instrument_search_var = tk.StringVar()
entry = tk.Entry(search_bar, textvariable=self.instrument_search_var, width=55)
entry.pack(side="left", padx=4, fill="x", expand=True)
entry.bind("<KeyRelease>", lambda e: self._debounce("instrument_search", self.refresh_instrument_table))
tk.Button(search_bar, text="Clear", command=lambda: [self.instrument_search_var.set(""), self.refresh_instrument_table()]).pack(side="left", padx=4)
db_frame = self.section(self.page, "Instrument Database")
self.instrument_table = DataTable(
db_frame,
[("instrument", "Instrument"), ("manufacturer", "Manufacturer"), ("model", "Model"),
("serial", "Serial No."), ("size", "Size / Station"), ("equipment", "Equipment ID"),
("installation", "Installation"), ("calibration", "Calibration"), ("calib_due", "Calib. Due"),
("customer", "Customer"), ("location", "Location"), ("department", "Department")],
[120, 105, 95, 105, 90, 95, 90, 90, 90, 135, 120, 105], height=18, fit_width=True
)
self.instrument_table.pack(fill="both", expand=True)
self.instrument_table.on_select_callback = self.instrument_row_selected
self.instrument_table.bind_right_click(self.instrument_context_menu)
self.instrument_table.row_tooltip_provider = lambda iid: self._instrument_configurations.get(str(iid))
self.hide_dynamic_control()
self.hide_other_manufacturer()
self.hide_other_instrument()
self.refresh_instrument_table()
def hide_dynamic_control(self):
self.dynamic_cb.grid_remove()
self.dynamic_entry.grid_remove()
self.dynamic_label.grid_remove()
self.dynamic_var.set("")
def hide_other_manufacturer(self):
self.other_manufacturer_label.grid_remove()
self.other_manufacturer_entry.grid_remove()
self.other_manufacturer_var.set("")
def hide_other_instrument(self):
self.other_instrument_label.grid_remove()
self.other_instrument_entry.grid_remove()
def manufacturer_changed(self, event=None):
m = self.manufacturer_var.get().strip()
self.instrument_var.set("")
self.hide_dynamic_control()
self.hide_other_manufacturer()
self.hide_other_instrument()
if m == "BEL":
self.instrument_label.grid_remove()
self.instrument_cb.grid_remove()
self.other_instrument_label.grid(row=0, column=1, sticky="w", padx=4)
self.other_instrument_entry.grid(row=1, column=1, sticky="ew", padx=4, pady=2)
elif m == "Other":
self.instrument_label.grid_remove()
self.instrument_cb.grid_remove()
self.other_manufacturer_label.grid(row=0, column=2, sticky="w", padx=4)
self.other_manufacturer_entry.grid(row=1, column=2, sticky="ew", padx=4, pady=2)
self.other_instrument_label.grid(row=0, column=1, sticky="w", padx=4)
self.other_instrument_entry.grid(row=1, column=1, sticky="ew", padx=4, pady=2)
else:
self.instrument_label.grid(row=0, column=1, sticky="w", padx=4)
self.instrument_cb.grid(row=1, column=1, sticky="ew", padx=4, pady=2)
self.instrument_cb["values"] = list(MANUFACTURER_INSTRUMENTS.get(m, {}).keys())
def instrument_changed(self, event=None):
m, ins = self.manufacturer_var.get(), self.instrument_var.get()
spec = MANUFACTURER_INSTRUMENTS.get(m, {}).get(ins)
self.hide_dynamic_control()
if not spec:
return
lbl, vals = spec
if not lbl:
return
self.dynamic_label.config(text=lbl)
self.dynamic_label.grid(row=0, column=2, sticky="w", padx=4)
if lbl in ("Size", "Station"):
self.dynamic_cb["values"] = vals
self.dynamic_cb.grid(row=1, column=2, sticky="ew", padx=4, pady=2)
elif lbl == "Capacity":
self.dynamic_entry.grid(row=1, column=2, sticky="ew", padx=4, pady=2)
def clear_instrument_filters(self):
self.selected_instrument_id = None
for v in (self.manufacturer_var, self.instrument_var, self.other_manufacturer_var, self.model_var,
self.serial_no_var, self.equipment_id_var, self.installation_var, self.calibration_var,
self.calibration_due_var, self.configuration_var, self.dynamic_var,
self.instrument_customer_var, self.instrument_location_var, self.instrument_department_var):
v.set("")
self.hide_dynamic_control()
self.hide_other_manufacturer()
self.hide_other_instrument()
self.save_button.config(text="SAVE", bg="#287b2f")
self.instrument_table.clear_selection()
def refresh_instrument_table(self):
if not hasattr(self, "instrument_table"):
return
self.instrument_table.clear()
self._instrument_configurations = {}
term = self.instrument_search_var.get()
for r in self.db.search_instruments(term):
self.instrument_table.insert([
r["instrument_name"] or "", r["manufacturer"], r["model"] or "",
r["serial_no"] or "", r["size_station"] or "", r["equipment_id"] or "",
r["installation_date"] or "", r["calibration_date"] or "",
r["calibration_due_date"] or "", r["customer_name"], r["location"], r["department"]
], iid=r["id"])
if r["configuration"]:
self._instrument_configurations[str(r["id"])] = r["configuration"]
def instrument_row_selected(self, iid):
try:
row = self.db.instrument(int(iid))
if not row:
return
self.selected_instrument_id = int(iid)
m = row["manufacturer"]
if m in MANUFACTURER_INSTRUMENTS:
self.manufacturer_var.set(m)
self.other_manufacturer_var.set("")
else:
self.manufacturer_var.set("Other")
self.other_manufacturer_var.set(m or "")
self.manufacturer_changed()
self.instrument_var.set(row["instrument_name"] or "")
self.instrument_changed()
self.dynamic_var.set(row["size_station"] or "")
self.model_var.set(row["model"] or "")
self.serial_no_var.set(row["serial_no"] or "")
self.equipment_id_var.set(row["equipment_id"] or "")
self.installation_var.set(row["installation_date"] or "")
self.calibration_var.set(row["calibration_date"] or "")
self.calibration_due_var.set(row["calibration_due_date"] or "")
self.configuration_var.set(row["configuration"] or "")
self.instrument_customer_var.set(row["customer_name"])
self.instrument_location_var.set(row["location"])
self.instrument_department_var.set(row["department"])
self.save_button.config(text="UPDATE", bg="#1769aa")
except Exception:
pass
def save_instrument(self):
sel_m = self.manufacturer_var.get().strip()
actual_m = self.other_manufacturer_var.get().strip() if sel_m == "Other" else sel_m
data = {
"manufacturer": actual_m,
"instrument": self.instrument_var.get().strip(),
"size_station": self.dynamic_var.get().strip(),
"model": self.model_var.get().strip(),
"serial_no": self.serial_no_var.get().strip(),
"equipment_id": self.equipment_id_var.get().strip(),
"installation_date": self.installation_var.get().strip(),
"calibration_date": self.calibration_var.get().strip(),
"calibration_due_date": self.calibration_due_var.get().strip(),
"configuration": self.configuration_var.get().strip(),
"customer": self.instrument_customer_var.get().strip(),
"location": self.instrument_location_var.get().strip(),
"department": self.instrument_department_var.get().strip(),
}
if not data["manufacturer"] or not data["instrument"] or not data["customer"] or not data["location"]:
messagebox.showwarning("Validation Error", "Manufacturer, Instrument Name, Customer, and Location are required.")
return
try:
if self.selected_instrument_id:
self.db.update_instrument(self.selected_instrument_id, data)
messagebox.showinfo("Success", "Instrument updated.")
else:
self.db.add_instrument(data)
messagebox.showinfo("Success", "Instrument saved.")
self.refresh_instrument_table()
self.clear_instrument_filters()
except Exception as e:
messagebox.showerror("Error", str(e))
def instrument_context_menu(self, event, iid):
menu = tk.Menu(self, tearoff=0)
menu.add_command(label="Datasheet", command=lambda: self._open_datasheet_for_instrument(int(iid)))
menu.add_separator()
menu.add_command(label="Edit Selected", command=self.clear_instrument_filters)
menu.add_separator()
menu.add_command(label="Delete Instrument", command=self.remove_selected_instrument)
menu.tk_popup(event.x_root, event.y_root)
def remove_selected_instrument(self):
if not self.selected_instrument_id or not messagebox.askyesno("Confirm", "Delete this instrument from system?"):
return
self.db.delete_instrument(self.selected_instrument_id)
self.selected_instrument_id = None
self.refresh_instrument_table()
self.clear_instrument_filters()
def show_service(self):
self.clear_page("Service Records")
self.selected_service_id = None
top = self.section(self.page, "Add / Edit Service Record")
for c in range(4):
top.grid_columnconfigure(c, weight=1)
tk.Label(top, text="Instrument").grid(row=0, column=0, sticky="w", padx=5)
self.service_instrument_var = tk.StringVar()
self.service_instrument_cb = ttk.Combobox(top, textvariable=self.service_instrument_var, state="readonly")
self.service_instrument_cb.grid(row=1, column=0, columnspan=2, padx=5, pady=3, sticky="ew")
self.service_map = {}
vals = []
for r in self.db.instruments(limit=300):
lbl = f"{r['id']} | {r['instrument_name']} | {r['manufacturer']} | {r['model'] or '-'} | {r['customer_name']}"
vals.append(lbl)
self.service_map[lbl] = r["id"]
self.service_instrument_cb["values"] = vals
tk.Label(top, text="Service Date").grid(row=0, column=2, sticky="w", padx=5)
self.service_date_var = tk.StringVar(value=date.today().strftime("%d-%m-%Y"))
DateEntry(top, textvariable=self.service_date_var, date_pattern="dd-mm-yyyy", width=13).grid(row=1, column=2, padx=5, sticky="w")
self.service_text = tk.Text(top, height=4)
self.service_text.grid(row=2, column=0, columnspan=3, padx=5, pady=5, sticky="nsew")
self.service_save_btn = tk.Button(top, text="Save Service", command=self.save_service_record, bg=ERT_BLUE, fg="white", font=(UI_FONT, 9, "bold"), relief="flat")
self.service_save_btn.grid(row=2, column=3, padx=7, pady=5, sticky="e")
search_bar = tk.LabelFrame(self.page, text="Search Service Records", font=(UI_FONT, 9, "bold"), padx=6, pady=5)
search_bar.pack(fill="x", pady=(4, 5))
self.service_search_var = tk.StringVar()
entry = tk.Entry(search_bar, textvariable=self.service_search_var, width=55)
entry.pack(side="left", padx=4, fill="x", expand=True)
entry.bind("<KeyRelease>", lambda e: self._debounce("service_search", self.refresh_service_table))
tk.Button(search_bar, text="Search", command=self.refresh_service_table, font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
db_frame = self.section(self.page, "Service Database")
self.service_table = DataTable(
db_frame,
[("date", "Date"), ("instrument", "Instrument"), ("model", "Model"),
("serial", "Serial No."), ("customer", "Customer"), ("location", "Location"),
("department", "Department"), ("work", "Service Performed")],
[85, 125, 90, 105, 135, 120, 100, 360], height=15, fit_width=True
)
self.service_table.pack(fill="both", expand=True)
self.service_table.on_select_callback = self.service_row_selected
self.service_table.bind_right_click(self.service_context_menu)
self.refresh_service_table()
def save_service_record(self):
lbl = self.service_instrument_var.get()
inst_id = self.service_map.get(lbl)
txt = self.service_text.get("1.0", "end").strip()
if not inst_id or not txt:
messagebox.showwarning("Validation", "Select an instrument and provide description of service.")
return
if self.selected_service_id:
self.db.update_service(self.selected_service_id, inst_id, self.service_date_var.get(), txt)
messagebox.showinfo("Updated", "Service record updated.")
else:
self.db.add_service(inst_id, self.service_date_var.get(), txt)
messagebox.showinfo("Saved", "Service record added.")
self.selected_service_id = None
self.service_text.delete("1.0", "end")
self.service_save_btn.config(text="Save Service")
self.refresh_service_table()
def service_row_selected(self, iid):
try:
row = self.db.conn.execute("SELECT * FROM service_records WHERE id=?", (int(iid),)).fetchone()
if not row:
return
self.selected_service_id = int(iid)
for k, v in self.service_map.items():
if v == row["instrument_id"]:
self.service_instrument_var.set(k)
break
self.service_date_var.set(row["service_date"] or "")
self.service_text.delete("1.0", "end")
self.service_text.insert("1.0", row["service_text"] or "")
self.service_save_btn.config(text="Update Service")
except Exception:
pass
def service_context_menu(self, event, iid):
menu = tk.Menu(self, tearoff=0)
menu.add_command(label="Delete Record", command=self.delete_service_record)
menu.tk_popup(event.x_root, event.y_root)
def delete_service_record(self):
if not self.selected_service_id or not messagebox.askyesno("Confirm", "Delete service entry?"):
return
self.db.delete_service(self.selected_service_id)
self.selected_service_id = None
self.service_text.delete("1.0", "end")
self.service_save_btn.config(text="Save Service")
self.refresh_service_table()
def refresh_service_table(self):
if not hasattr(self, "service_table"):
return
self.service_table.clear()
for r in self.db.search_service_records(self.service_search_var.get()):
self.service_table.insert([
r["service_date"], r["instrument_name"] or "", r["model"] or "",
r["serial_no"] or "", r["customer_name"], r["location"],
r["department"], r["service_text"]
], iid=r["id"])
def show_amc(self):
self.clear_page("AMC")
top = self.section(self.page, "Assign AMC Record")
self.amc_instrument_var = tk.StringVar()
self.amc_instrument_cb = ttk.Combobox(top, textvariable=self.amc_instrument_var, state="readonly", width=90)
self.amc_instrument_cb.grid(row=0, column=0, columnspan=2, padx=5, pady=3, sticky="ew")
top.grid_columnconfigure(0, weight=1)
self.amc_map = {}
vals = []
for r in self.db.instruments(limit=300):
lbl = f"{r['id']} | {r['instrument_name']} | {r['manufacturer']} | {r['customer_name']}"
vals.append(lbl)
self.amc_map[lbl] = r["id"]
self.amc_instrument_cb["values"] = vals
self.amc_year_var = tk.StringVar(value=str(date.today().year))
self.amc_no_var = tk.StringVar(value="1")
self.amc_date_var = tk.StringVar(value=date.today().strftime("%d-%m-%Y"))
self.amc_notes_var = tk.StringVar()
tk.Label(top, text="Year").grid(row=1, column=0, sticky="w", padx=5)
tk.Entry(top, textvariable=self.amc_year_var, width=10).grid(row=2, column=0, sticky="w", padx=5)
tk.Label(top, text="AMC No.").grid(row=1, column=1, sticky="w", padx=5)
ttk.Combobox(top, textvariable=self.amc_no_var, values=["1", "2", "3", "4"], state="readonly", width=8).grid(row=2, column=1, sticky="w", padx=5)
tk.Label(top, text="AMC Date").grid(row=1, column=2, sticky="w", padx=5)
DateEntry(top, textvariable=self.amc_date_var, date_pattern="dd-mm-yyyy", width=13).grid(row=2, column=2, sticky="w", padx=5)
tk.Label(top, text="Notes").grid(row=1, column=3, sticky="w", padx=5)
tk.Entry(top, textvariable=self.amc_notes_var, width=35).grid(row=2, column=3, sticky="ew", padx=5)
tk.Button(top, text="Add AMC", command=self.add_amc_record, bg=ERT_BLUE, fg="white", font=(UI_FONT, 9, "bold")).grid(row=2, column=4, padx=8)
search_bar = tk.LabelFrame(self.page, text="Search AMC Records", font=(UI_FONT, 9, "bold"), padx=6, pady=5)
search_bar.pack(fill="x", pady=(4, 5))
self.amc_search_var = tk.StringVar()
entry = tk.Entry(search_bar, textvariable=self.amc_search_var, width=55)
entry.pack(side="left", padx=4, fill="x", expand=True)
entry.bind("<KeyRelease>", lambda e: self._debounce("amc_search", self.refresh_amc_table))
tk.Button(search_bar, text="Clear", command=lambda: [self.amc_search_var.set(""), self.refresh_amc_table()]).pack(side="left", padx=4)
db_frame = self.section(self.page, "AMC Database")
self.amc_table = DataTable(
db_frame,
[("year", "Year"), ("no", "AMC"), ("date", "AMC Date"), ("instrument", "Instrument"),
("manufacturer", "Manufacturer"), ("size", "Size / Station"), ("customer", "Customer"),
("location", "Location"), ("department", "Department")],
[60, 50, 90, 140, 105, 95, 145, 120, 100], height=14, fit_width=True
)
self.amc_table.pack(fill="both", expand=True)
self.amc_table.bind_right_click(lambda e, iid: self._show_amc_context(e, iid))
self.refresh_amc_table()
def add_amc_record(self):
inst_id = self.amc_map.get(self.amc_instrument_var.get())
if not inst_id:
messagebox.showwarning("Warning", "Select an instrument.")
return
try:
self.db.add_amc(inst_id, int(self.amc_year_var.get()), int(self.amc_no_var.get()), self.amc_date_var.get(), self.amc_notes_var.get())
self.refresh_amc_table()
messagebox.showinfo("Saved", "AMC entry added.")
except Exception as e:
messagebox.showwarning("Error", str(e))
def _show_amc_context(self, event, iid):
menu = tk.Menu(self, tearoff=0)
menu.add_command(label="Delete AMC", command=lambda: self.delete_amc_record(int(iid)))
menu.tk_popup(event.x_root, event.y_root)
def delete_amc_record(self, aid):
if not messagebox.askyesno("Confirm", "Delete AMC record?"):
return
self.db.delete_amc(aid)
self.refresh_amc_table()
def refresh_amc_table(self):
if not hasattr(self, "amc_table"):
return
self.amc_table.clear()
for r in self.db.search_amc_records(self.amc_search_var.get()):
self.amc_table.insert([
r["amc_year"], f"AMC-{r['amc_no']}", r["amc_date"],
r["instrument_name"], r["manufacturer"], r["size_station"] or "",
r["customer_name"], r["location"], r["department"]
], iid=r["id"])
def show_calibration(self):
self.clear_page("Calibration")
top = self.section(self.page, "Upcoming Calibrations")
search_bar = tk.LabelFrame(top, text="Search Upcoming / History", font=(UI_FONT, 9, "bold"), padx=6, pady=5)
search_bar.pack(fill="x", pady=(2, 5))
self.cal_search_var = tk.StringVar()
entry = tk.Entry(search_bar, textvariable=self.cal_search_var, width=55)
entry.pack(side="left", padx=4, fill="x", expand=True)
entry.bind("<KeyRelease>", lambda e: self._debounce("cal_search", self.refresh_calibration_tables))
tk.Button(search_bar, text="Clear", command=lambda: [self.cal_search_var.set(""), self.refresh_calibration_tables()]).pack(side="left", padx=4)
self.cal_table = DataTable(
top,
[("date", "Calibration Date"), ("instrument", "Instrument"), ("manufacturer", "Manufacturer"),
("model", "Model"), ("size", "Size / Station"), ("equipment", "Equipment ID"),
("customer", "Customer"), ("location", "Location"), ("department", "Department")],
[100, 130, 105, 85, 90, 90, 145, 120, 100], height=9, fit_width=True
)
self.cal_table.pack(fill="both", expand=True)
self.cal_table.on_select_callback = lambda iid: self.cal_complete_btn.config(state="normal")
self.cal_complete_btn = tk.Button(top, text="Mark Selected as Calibrated", command=self.complete_calibration, bg="#1769aa", fg="white", font=(UI_FONT, 9, "bold"), state="disabled")
self.cal_complete_btn.pack(anchor="e", padx=5, pady=6)
history = self.section(self.page, "Calibration Log History")
self.cal_history_table = DataTable(
history,
[("date", "Calibration Date"), ("instrument", "Instrument"), ("manufacturer", "Manufacturer"),
("model", "Model"), ("customer", "Customer"), ("location", "Location"),
("department", "Department"), ("completed", "Completed Timestamp")],
[105, 130, 105, 90, 145, 120, 100, 155], height=7, fit_width=True
)
self.cal_history_table.pack(fill="both", expand=True)
self.refresh_calibration_tables()
def complete_calibration(self):
iid = self.cal_table.selected_iid
if not iid or not messagebox.askyesno("Confirm", "Record calibration as complete?"):
return
try:
self.db.mark_calibration_completed(int(iid))
self.cal_complete_btn.config(state="disabled")
self.refresh_calibration_tables()
messagebox.showinfo("Success", "Calibration logged.")
except Exception as e:
messagebox.showwarning("Error", str(e))
def refresh_calibration_tables(self):
if hasattr(self, "cal_table"):
self.cal_table.clear()
for r in self.db.search_upcoming_calibrations(self.cal_search_var.get()):
self.cal_table.insert([
r["calibration_date"], r["instrument_name"], r["manufacturer"],
r["model"] or "", r["size_station"] or "", r["equipment_id"] or "",
r["customer_name"], r["location"], r["department"]
], iid=r["id"])
if hasattr(self, "cal_history_table"):
self.cal_history_table.clear()
for r in self.db.search_calibration_history(self.cal_search_var.get()):
self.cal_history_table.insert([
r["calibration_date"], r["instrument_name"], r["manufacturer"],
r["model"] or "", r["customer_name"], r["location"],
r["department"], r["completed_at"]
], iid=r["id"])
def show_monthly_records(self):
self.clear_page("Monthly Records")
controls = self.section(self.page, "Filter Period")
now = date.today()
self.month_var = tk.StringVar(value=str(now.month))
self.year_var = tk.StringVar(value=str(now.year))
month_names = ["January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December"]
self.month_cb = ttk.Combobox(controls, textvariable=self.month_var, values=[f"{i+1:02d} - {name}" for i, name in enumerate(month_names)], state="readonly", width=18)
self.month_cb.grid(row=0, column=0, padx=6, pady=4)
self.month_cb.current(now.month - 1)
self.month_cb.bind("<<ComboboxSelected>>", lambda e: self.draw_monthly_graph())
self.year_cb = ttk.Combobox(controls, textvariable=self.year_var, values=[str(y) for y in range(now.year - 5, now.year + 6)], state="readonly", width=10)
self.year_cb.grid(row=0, column=1, padx=6, pady=4)
self.year_cb.bind("<<ComboboxSelected>>", lambda e: self.draw_monthly_graph())
self.month_summary_var = tk.StringVar()
tk.Label(controls, textvariable=self.month_summary_var, font=(UI_FONT, 10, "bold")).grid(row=0, column=2, padx=20)
graph_sec = self.section(self.page, "Services & Calibrations Timeline")
graph_sec.pack(fill="both", expand=True)
self.month_graph = tk.Canvas(graph_sec, bg="white", highlightthickness=1, highlightbackground="#B0BEC5")
self.month_graph.pack(fill="both", expand=True, padx=6, pady=6)
self.month_graph.bind("<Configure>", lambda e: self.draw_monthly_graph())
self.draw_monthly_graph()
def draw_monthly_graph(self):
if not hasattr(self, "month_graph") or not self.month_graph.winfo_exists():
return
try:
m = int(self.month_var.get().split("-")[0].strip())
y = int(self.year_var.get())
except Exception:
return
days, srv, inst, cal = self.db.monthly_activity(y, m)
act = {d: srv[d] + inst[d] for d in range(1, days + 1)}
self.month_summary_var.set(f"Services: {sum(srv.values())} | Installations: {sum(inst.values())} | Calibrations: {sum(cal.values())}")
canvas = self.month_graph
canvas.delete("all")
w, h = max(600, canvas.winfo_width()), max(300, canvas.winfo_height())
l, r, t, b = 60, 30, 40, 50
pw, ph = w - l - r, h - t - b
max_val = max([max(act[d], cal[d]) for d in range(1, days + 1)] + [5])
y_top = ((max_val + 4) // 5) * 5
for v in range(0, y_top + 1, max(1, y_top // 5)):
y_pos = t + ph - (v / y_top) * ph
canvas.create_line(l, y_pos, w - r, y_pos, fill="#EEEEEE")
canvas.create_text(l - 8, y_pos, text=str(v), anchor="e", font=(UI_FONT, 8))
canvas.create_line(l, t, l, t + ph, width=1)
canvas.create_line(l, t + ph, w - r, t + ph, width=1)
pts_srv, pts_cal = [], []
for d in range(1, days + 1):
x = l + ((d - 1) / max(1, days - 1)) * pw
if d == 1 or d == days or d % 2 == 0:
canvas.create_text(x, t + ph + 15, text=str(d), font=(UI_FONT, 8))
sy = t + ph - (act[d] / y_top) * ph
cy = t + ph - (cal[d] / y_top) * ph
pts_srv.append((x, sy))
pts_cal.append((x, cy))
if len(pts_srv) > 1:
canvas.create_line(pts_srv, fill="#00AEEF", width=2)
canvas.create_line(pts_cal, fill="#C62828", width=2)
for x, y_coord in pts_srv:
canvas.create_oval(x - 2, y_coord - 2, x + 2, y_coord + 2, fill="#00AEEF", outline="#00AEEF")
for x, y_coord in pts_cal:
canvas.create_oval(x - 2, y_coord - 2, x + 2, y_coord + 2, fill="#C62828", outline="#C62828")
def show_datasheets(self):
self.clear_page("Datasheets")
self.selected_datasheet_instrument_id = None
search = tk.Frame(self.page)
search.pack(fill="x", pady=(0, 6))
self.datasheet_search_var = tk.StringVar()
entry = tk.Entry(search, textvariable=self.datasheet_search_var)
entry.pack(side="left", fill="x", expand=True, padx=4)
entry.bind("<KeyRelease>", lambda e: self._debounce("ds_search", self.refresh_datasheet_table))
tk.Button(search, text="Clear", command=lambda: [self.datasheet_search_var.set(""), self.refresh_datasheet_table()]).pack(side="left", padx=4)
db_frame = self.section(self.page, "Instrument Catalog")
self.datasheet_table = DataTable(
db_frame,
[("instrument", "Instrument"), ("manufacturer", "Manufacturer"), ("model", "Model"),
("serial", "Serial No."), ("customer", "Customer"), ("location", "Location"), ("datasheet", "Datasheet")],
[175, 130, 115, 120, 185, 155, 185], height=16, fit_width=True
)
self.datasheet_table.pack(fill="both", expand=True)
self.datasheet_table.on_select_callback = lambda iid: setattr(self, "selected_datasheet_instrument_id", int(iid))
self.datasheet_table.tree.bind("<Double-1>", lambda e: self.open_selected_datasheet())
controls = tk.Frame(self.page, bd=1, relief="solid", padx=8, pady=8)
controls.pack(fill="x", pady=(7, 0))
tk.Button(controls, text="OPEN PDF", command=self.open_selected_datasheet, bg=ERT_NAVY, fg="white", font=(UI_FONT, 9, "bold"), width=12).pack(side="left", padx=4)
tk.Button(controls, text="ATTACH PDF", command=self.add_datasheet_file, bg=ERT_BLUE, fg="white", font=(UI_FONT, 9, "bold"), width=12).pack(side="left", padx=4)
tk.Button(controls, text="REMOVE PDF", command=self.remove_datasheet_file, bg="#B71C1C", fg="white", font=(UI_FONT, 9, "bold"), width=12).pack(side="left", padx=4)
self.refresh_datasheet_table()
def refresh_datasheet_table(self):
if not hasattr(self, "datasheet_table"):
return
self.datasheet_table.clear()
for r in self.db.search_instruments(self.datasheet_search_var.get()):
ds = self.db.datasheet_for_instrument(r["id"])
self.datasheet_table.insert([
r["instrument_name"] or "", r["manufacturer"] or "", r["model"] or "",
r["serial_no"] or "", r["customer_name"] or "", r["location"] or "",
"Attached" if ds else "Not Attached"
], iid=r["id"])
def open_selected_datasheet(self):
iid = getattr(self, "selected_datasheet_instrument_id", None)
if not iid:
messagebox.showwarning("Datasheet", "Select an instrument first.")
return
self._open_datasheet_for_instrument(iid)
def _open_datasheet_for_instrument(self, instrument_id):
ds = self.db.datasheet_for_instrument(instrument_id)
if not ds or not os.path.exists(ds["file_path"]):
messagebox.showinfo("Datasheet", "PDF attachment is missing or does not exist.")
return
try:
os.startfile(ds["file_path"])
except Exception as e:
messagebox.showerror("Error", str(e))
def add_datasheet_file(self):
iid = getattr(self, "selected_datasheet_instrument_id", None)
if not iid:
messagebox.showwarning("Datasheet", "Select an instrument first.")
return
src = filedialog.askopenfilename(title="Select PDF Datasheet", filetypes=[("PDF files", "*.pdf")])
if not src:
return
target_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Datasheets")
os.makedirs(target_dir, exist_ok=True)
dest = os.path.join(target_dir, f"inst_{iid}_{os.path.basename(src)}")
try:
import shutil
shutil.copy2(src, dest)
self.db.save_datasheet(iid, dest, os.path.basename(dest))
self.refresh_datasheet_table()
messagebox.showinfo("Saved", "Datasheet attached successfully.")
except Exception as e:
messagebox.showerror("Error", str(e))
def remove_datasheet_file(self):
iid = getattr(self, "selected_datasheet_instrument_id", None)
if not iid or not messagebox.askyesno("Confirm", "Remove attached datasheet?"):
return
self.db.delete_datasheet(iid)
self.refresh_datasheet_table()
def show_instrument_configuration(self):
self.clear_page("Instrument Configuration")
if hasattr(self, "sidebar") and self.sidebar.winfo_exists():
self.sidebar.pack_forget()
if hasattr(self, "content") and self.content.winfo_exists():
self.content.pack_configure(side="left", fill="both", expand=True)
self.config_main_categories = [r["name"] for r in self.db.config_categories()]
self.configuration_main = None
self.configuration_package_id = None
self.configuration_work_items = []
self.configuration_expanded = set()
self.configuration_row_map = []
self.selected_config_item_index = None
header = tk.Frame(self.page)
header.pack(fill="x", pady=(0, 6))
tk.Button(header, text="← Back", command=self.show_dashboard, font=(UI_FONT, 10, "bold"), padx=10, pady=5).pack(side="left", padx=(0, 10))
tk.Label(header, text="Instrument Configuration Builder", font=(UI_FONT, 18, "bold"), fg=ERT_NAVY).pack(side="left")
body = tk.Frame(self.page)
body.pack(fill="both", expand=True)
body.grid_columnconfigure(0, weight=0, minsize=300)
body.grid_columnconfigure(1, weight=1)
body.grid_rowconfigure(0, weight=1)
left = tk.LabelFrame(body, text="Categories", font=(UI_FONT, 10, "bold"), bd=1, relief="solid", padx=6, pady=6)
left.grid(row=0, column=0, sticky="nsew", padx=(0, 8))
# NEW: Add and Remove Package Buttons at the top of the categories list
cat_controls = tk.Frame(left)
cat_controls.pack(fill="x", pady=(0, 6))
tk.Button(
cat_controls, text="+ Add Package", command=self.new_config_package,
bg=ERT_BLUE, fg="white", font=(UI_FONT, 9, "bold"), relief="flat"
).pack(side="left", fill="x", expand=True, padx=(0, 2))
tk.Button(
cat_controls, text="- Remove Package", command=self.delete_config_package,
bg="#B71C1C", fg="white", font=(UI_FONT, 9, "bold"), relief="flat"
).pack(side="left", fill="x", expand=True, padx=(2, 0))
self.config_main_list = WrappedCategoryList(left, bd=0, relief="flat")
self.config_main_list.pack(fill="both", expand=True)
self.config_main_list.bind("<<ListboxSelect>>", self.configuration_main_selected)
right = tk.Frame(body)
right.grid(row=0, column=1, sticky="nsew")
right.grid_rowconfigure(1, weight=1)
right.grid_columnconfigure(0, weight=1)
pkg_head = tk.LabelFrame(right, text="Package Details", font=(UI_FONT, 10, "bold"), bd=1, relief="solid", padx=8, pady=7)
pkg_head.grid(row=0, column=0, sticky="ew", pady=(0, 7))
pkg_head.grid_columnconfigure(1, weight=1)
self.config_main_label = tk.Label(pkg_head, text="Select a main category on left.", fg=ERT_NAVY, font=(UI_FONT, 11, "bold"))
self.config_main_label.grid(row=0, column=0, columnspan=3, sticky="w", pady=(0, 4))
tk.Label(pkg_head, text="Package Name:").grid(row=1, column=0, sticky="w", padx=4)
self.config_package_name_var = tk.StringVar()
tk.Entry(pkg_head, textvariable=self.config_package_name_var, font=(UI_FONT, 10)).grid(row=1, column=1, sticky="ew", padx=4)
tk.Button(
pkg_head, text="Copy Table", command=self.copy_config_to_clipboard,
bg=ERT_NAVY, fg="white", font=(UI_FONT, 9, "bold")
).grid(row=1, column=2, padx=4, sticky="e")
table_frame = tk.Frame(right, bd=1, relief="solid")
table_frame.grid(row=1, column=0, sticky="nsew")
self.config_table = DataTable(table_frame, [("sl", "Sl No."), ("qty", "Qty."), ("part", "Part No."), ("description", "Description")], [70, 70, 200, 500], height=15, fit_width=True)
self.config_table.pack(fill="both", expand=True)
self.config_table.on_select_callback = self.config_item_selected
editor = tk.LabelFrame(right, text="Add / Edit Item", font=(UI_FONT, 10, "bold"), bd=1, relief="solid", padx=8, pady=6)
editor.grid(row=2, column=0, sticky="ew", pady=(5, 0))
editor.grid_columnconfigure(0, weight=6)
editor.grid_columnconfigure(1, weight=3)
editor.grid_columnconfigure(2, weight=1)
self.config_description_var = tk.StringVar()
self.config_part_var = tk.StringVar()
self.config_qty_var = tk.StringVar(value="1")
tk.Label(editor, text="Description").grid(row=0, column=0, sticky="w")
tk.Label(editor, text="Part / Model No.").grid(row=0, column=1, sticky="w")
tk.Label(editor, text="Qty").grid(row=0, column=2, sticky="w")
tk.Entry(editor, textvariable=self.config_description_var).grid(row=1, column=0, sticky="ew", padx=3, pady=3)
tk.Entry(editor, textvariable=self.config_part_var).grid(row=1, column=1, sticky="ew", padx=3, pady=3)
# NEW: Mini-frame for Qty with native +/- buttons
qty_frame = tk.Frame(editor)
qty_frame.grid(row=1, column=2, sticky="w", padx=3, pady=3)
tk.Button(
qty_frame, text=" − ", command=self.decrease_config_qty,
font=(UI_FONT, 10, "bold"), bg=ERT_BLUE, fg="white", relief="flat", padx=2, pady=0
).pack(side="left")
tk.Entry(
qty_frame, textvariable=self.config_qty_var, width=5,
justify="center", font=(UI_FONT, 10)
).pack(side="left", padx=3)
tk.Button(
qty_frame, text=" + ", command=self.increase_config_qty,
font=(UI_FONT, 10, "bold"), bg=ERT_BLUE, fg="white", relief="flat", padx=2, pady=0
).pack(side="left")
btn_frame = tk.Frame(editor)
btn_frame.grid(row=2, column=0, columnspan=3, sticky="e", pady=(8, 0))
tk.Button(btn_frame, text="Add Item", command=self.add_config_item, bg=ERT_BLUE, fg="white", font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
tk.Button(btn_frame, text="Update Selected", command=self.update_config_item, font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
tk.Button(btn_frame, text="Remove Selected", command=self.remove_config_item, bg="#B71C1C", fg="white", font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
tk.Button(btn_frame, text="↑ Up", command=self.move_config_item_up, font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
tk.Button(btn_frame, text="↓ Down", command=self.move_config_item_down, font=(UI_FONT, 9, "bold")).pack(side="left", padx=4)
tk.Button(btn_frame, text="Save Full Configuration", command=self.save_full_configuration, bg="#287b2f", fg="white", font=(UI_FONT, 9, "bold")).pack(side="left", padx=(15, 4))
self.refresh_configuration_tree()
# NEW: Logic for Quantity +/- Buttons
def increase_config_qty(self):
try:
qty = int(self.config_qty_var.get())
except ValueError:
qty = 1
self.config_qty_var.set(str(max(1, qty + 1)))
def decrease_config_qty(self):
try:
qty = int(self.config_qty_var.get())
except ValueError:
qty = 1
self.config_qty_var.set(str(max(1, qty - 1)))
# NEW: Logic for Adding a Brand New Package
def new_config_package(self):
if not self.configuration_main:
messagebox.showwarning("Validation", "Please select a Main Instrument Category from the list first.")
return
self.configuration_package_id = None
self.config_package_name_var.set("")
self.configuration_work_items = []
self.selected_config_item_index = None
self.config_description_var.set("")
self.config_part_var.set("")
self.config_qty_var.set("1")
self.config_main_label.config(text=f"New Package for: {self.configuration_main}")
self.refresh_configuration_table()
# NEW: Logic for Deleting a Package
def delete_config_package(self):
if not self.configuration_package_id:
messagebox.showwarning("Validation", "Please select an existing Package to remove.")
return
pkg_name = self.config_package_name_var.get()
if not messagebox.askyesno("Confirm Delete", f"Are you sure you want to delete this package:\n\n'{pkg_name}'?\n\nThis action cannot be undone."):
return
self.db.delete_instrument_package(self.configuration_package_id)
self.configuration_package_id = None
self.config_package_name_var.set("")
self.configuration_work_items = []
self.refresh_configuration_table()
self.refresh_configuration_tree()
self.config_main_label.config(text=f"Category: {self.configuration_main}")
messagebox.showinfo("Deleted", "Package deleted successfully.")
def copy_config_to_clipboard(self):
if not getattr(self, "configuration_work_items", None):
messagebox.showwarning("Copy", "There are no items to copy.")
return
lines = ["Sl No.\tQty.\tPart No.\tDescription"]
for idx, itm in enumerate(self.configuration_work_items, 1):
lines.append(f"{idx:02d}\t{itm['qty']}\t{itm['part']}\t{itm['description']}")
text = "\n".join(lines)
self.clipboard_clear()
self.clipboard_append(text)
self.update()
messagebox.showinfo("Copied", "Table data copied to clipboard.")
def config_item_selected(self, iid):
try:
self.selected_config_item_index = int(iid) - 1
item = self.configuration_work_items[self.selected_config_item_index]
self.config_description_var.set(item["description"])
self.config_part_var.set(item["part"])
self.config_qty_var.set(str(item["qty"]))
except Exception:
self.selected_config_item_index = None
def refresh_configuration_tree(self):
if not hasattr(self, "config_main_list"):
return
target_rows = []
for cat in self.config_main_categories:
exp = cat in self.configuration_expanded
target_rows.append(("category", cat, f"{'▼' if exp else '▷'} {cat}"))
if exp:
for p in self.db.instrument_packages(cat):
target_rows.append(("package", p["id"], f" └─ {p['package_name']}"))
self.configuration_row_map = [(k, v) for k, v, _ in target_rows]
self.config_main_list.delete(0, "end")
for _, _, text in target_rows:
self.config_main_list.insert("end", text)
def configuration_main_selected(self, event=None):
sel = self.config_main_list.curselection()
if not sel or sel[0] >= len(self.configuration_row_map):
return
kind, val = self.configuration_row_map[sel[0]]
self.selected_config_item_index = None
self.config_description_var.set("")
self.config_part_var.set("")
self.config_qty_var.set("1")
if kind == "category":
if val in self.configuration_expanded:
self.configuration_expanded.remove(val)
else:
self.configuration_expanded.add(val)
self.configuration_main = val
self.configuration_package_id = None
self.configuration_work_items = []
self.config_package_name_var.set("")
self.config_main_label.config(text=f"Category: {val}")
self.refresh_configuration_table()
self.refresh_configuration_tree()
else:
pkg = self.db.instrument_package(None, package_id=int(val))
if pkg:
self.configuration_main = pkg["main_instrument"]
self.configuration_package_id = pkg["id"]
self.config_package_name_var.set(pkg["package_name"])
self.config_main_label.config(text=f"Category: {pkg['main_instrument']}")
self.configuration_work_items = [
{"qty": r["quantity"], "part": r["part_no"] or "", "description": r["description"] or ""}
for r in self.db.instrument_package_items_by_id(pkg["id"])
]
self.refresh_configuration_table()
def refresh_configuration_table(self):
if not hasattr(self, "config_table"):
return
self.config_table.clear()
for idx, itm in enumerate(self.configuration_work_items, 1):
self.config_table.insert([f"{idx:02d}", str(itm["qty"]), itm["part"], itm["description"]], iid=str(idx))
if getattr(self, "selected_config_item_index", None) is not None:
iid = str(self.selected_config_item_index + 1)
try:
self.config_table.tree.selection_set(iid)
self.config_table.tree.focus(iid)
except Exception:
pass
def add_config_item(self):
desc = self.config_description_var.get().strip()
part = self.config_part_var.get().strip()
try:
qty = int(self.config_qty_var.get())
except ValueError:
messagebox.showwarning("Validation", "Quantity must be an integer.")
return
if not desc:
messagebox.showwarning("Validation", "Description is required.")
return
self.configuration_work_items.append({"qty": qty, "part": part, "description": desc})
self.selected_config_item_index = None
self.refresh_configuration_table()
self.config_description_var.set("")
self.config_part_var.set("")
self.config_qty_var.set("1")
def update_config_item(self):
idx = getattr(self, "selected_config_item_index", None)
if idx is None:
messagebox.showwarning("Selection", "Select an item from the list to update.")
return
try:
qty = int(self.config_qty_var.get())
except ValueError:
messagebox.showwarning("Validation", "Quantity must be an integer.")
return
desc = self.config_description_var.get().strip()
if not desc:
messagebox.showwarning("Validation", "Description is required.")
return
self.configuration_work_items[idx] = {
"qty": qty,
"part": self.config_part_var.get().strip(),
"description": desc
}
self.refresh_configuration_table()
def remove_config_item(self):
idx = getattr(self, "selected_config_item_index", None)
if idx is None or idx >= len(self.configuration_work_items):
messagebox.showwarning("Selection", "Select an item from the list to remove.")
return
del self.configuration_work_items[idx]
self.selected_config_item_index = None
self.config_description_var.set("")
self.config_part_var.set("")
self.config_qty_var.set("1")
self.refresh_configuration_table()
def move_config_item_up(self):
idx = getattr(self, "selected_config_item_index", None)
if idx is None or idx <= 0:
return
self.configuration_work_items[idx], self.configuration_work_items[idx-1] = \
self.configuration_work_items[idx-1], self.configuration_work_items[idx]
self.selected_config_item_index = idx - 1
self.refresh_configuration_table()
def move_config_item_down(self):
idx = getattr(self, "selected_config_item_index", None)
if idx is None or idx >= len(self.configuration_work_items) - 1:
return
self.configuration_work_items[idx], self.configuration_work_items[idx+1] = \
self.configuration_work_items[idx+1], self.configuration_work_items[idx]
self.selected_config_item_index = idx + 1
self.refresh_configuration_table()
def save_full_configuration(self):
if not self.configuration_main:
messagebox.showwarning("Validation", "Select a category first.")
return
name = self.config_package_name_var.get().strip()
if not name:
messagebox.showwarning("Validation", "Package Name is required.")
return
try:
if self.configuration_package_id:
self.db.update_instrument_package(self.configuration_package_id, name)
pkg_id = self.configuration_package_id
self.db.conn.execute("DELETE FROM instrument_package_items WHERE package_id=?", (pkg_id,))
else:
pkg = self.db.create_instrument_package(self.configuration_main, name)
pkg_id = pkg["id"]
self.configuration_package_id = pkg_id
for idx, itm in enumerate(self.configuration_work_items, 1):
self.db.conn.execute("""
INSERT INTO instrument_package_items (package_id, line_no, quantity, part_no, description, is_main)
VALUES (?, ?, ?, ?, ?, 0)
""", (pkg_id, idx, itm["qty"], itm["part"], itm["description"]))
self.db.conn.commit()
messagebox.showinfo("Saved", "Configuration package saved successfully.")
self.refresh_configuration_tree()
except Exception as e:
messagebox.showerror("Error", str(e))
def show_settings(self):
self.clear_page("Settings")
db_sec = self.section(self.page, "Database Management")
self.db_path_var = tk.StringVar(value=self.db_path)
tk.Entry(db_sec, textvariable=self.db_path_var, state="readonly", width=80).pack(side="left", padx=5, fill="x", expand=True)
tk.Button(db_sec, text="Backup DB", command=self.backup_database_to_pc, bg=ERT_NAVY, fg="white", font=(UI_FONT, 9, "bold")).pack(side="left", padx=5)
disp_sec = self.section(self.page, "Display & Theme")
self.clock_var = tk.BooleanVar(value=bool(self.clock_visible))
tk.Checkbutton(disp_sec, text="Display Clock", variable=self.clock_var, command=self.toggle_clock).pack(side="left", padx=8)
def toggle_clock(self):
self.clock_visible = bool(self.clock_var.get())
self.save_app_setting("clock_visible", self.clock_visible)
if self.clock_visible:
self.clock_label.pack(side="left")
else:
self.clock_label.pack_forget()
def backup_database_to_pc(self):
dest = filedialog.asksaveasfilename(defaultextension=".db", initialfile=f"backup_{datetime.now().strftime('%Y%m%d')}.db")
if dest:
import shutil
self.db.conn.commit()
shutil.copy2(self.db_path, dest)
messagebox.showinfo("Backup", "Database backup saved.")
def _ensure_main_window_maximized(self, event=None):
try:
if self.state() != "zoomed": self.state("zoomed")
except Exception: pass
def update_clock(self):
if getattr(self, "clock_visible", True) and hasattr(self, "clock_label"):
self.clock_label.config(text=datetime.now().strftime("%d-%m-%Y %H:%M:%S"))
self.after(1000, self.update_clock)
def sign_out(self):
if messagebox.askyesno("Signout", "Sign out?"):
self.current_user = None
self.withdraw()
self.after(100, self.start_authentication)
def exit_app(self):
self.db.close()
self.destroy()
if __name__ == "__main__":
app = App()
app.mainloop()