Spaces:
Sleeping
Sleeping
| import json | |
| import gradio as gr | |
| import pandas as pd | |
| import datetime | |
| import os | |
| import uuid | |
| from typing import Dict, Any | |
| from azure.cosmos import CosmosClient, PartitionKey, exceptions | |
| from azure.cosmos.exceptions import CosmosResourceNotFoundError | |
| from dotenv import load_dotenv | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| # File paths | |
| DATA_PATH = "./converters_with_links_and_pricelist.json" | |
| META_PATH = "./converters_metadata.json" | |
| # Cosmos DB configuration from environment variables | |
| COSMOS_ENDPOINT = os.getenv("AZURE_COSMOS_DB_ENDPOINT") | |
| COSMOS_KEY = os.getenv("AZURE_COSMOS_DB_KEY") | |
| DATABASE_NAME = os.getenv("AZURE_COSMOS_DB_DATABASE", "TAL_DB") # Default to TAL_DB | |
| print(DATABASE_NAME) | |
| CONTAINER_NAME = os.getenv("AZURE_COSMOS_DB_CONTAINER", "Converters") # Default to Converters_with_embeddings | |
| # Indexing policy | |
| INDEXING_POLICY = { | |
| "indexingMode": "consistent", | |
| "includedPaths": [{"path": "/*"}], | |
| "excludedPaths": [ | |
| {"path": "/\"_etag\"/?", | |
| "path": "/embedding/*"} | |
| ] | |
| } | |
| # Validate environment variables | |
| if not all([COSMOS_ENDPOINT, COSMOS_KEY, DATABASE_NAME]): | |
| raise ValueError("Missing required Cosmos DB environment variables (AZURE_COSMOS_DB_ENDPOINT, AZURE_COSMOS_DB_KEY, AZURE_COSMOS_DB_DATABASE). Check .env file.") | |
| # Initialize Cosmos DB client | |
| try: | |
| client = CosmosClient(COSMOS_ENDPOINT, COSMOS_KEY) | |
| database = client.create_database_if_not_exists(id=DATABASE_NAME) | |
| container = database.create_container_if_not_exists( | |
| id=CONTAINER_NAME, | |
| partition_key=PartitionKey(path="/artnr"), | |
| indexing_policy=INDEXING_POLICY, | |
| default_ttl=-1 | |
| ) | |
| except exceptions.CosmosHttpResponseError as e: | |
| raise ValueError(f"Failed to initialize Cosmos DB client or create database/container: {str(e)}") | |
| def load_json(path) -> Dict[str, Any]: | |
| try: | |
| with open(path, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| except FileNotFoundError: | |
| return {} | |
| def save_json(data: Dict[str, Any], path): | |
| with open(path, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=4, ensure_ascii=False) | |
| def get_current_time(): | |
| return datetime.datetime.now().isoformat() | |
| def transform_to_cosmos_format(converter_id: str, converter_data: Dict[str, Any]) -> Dict[str, Any]: | |
| """Transform converter data to match Cosmos DB document structure.""" | |
| return { | |
| "id": str(uuid.uuid4()), # Generate UUID for Cosmos DB | |
| "type": converter_data.get("TYPE", ""), | |
| "artnr": int(converter_data.get("ARTNR", 0)), | |
| "converter_description": converter_data.get("CONVERTER DESCRIPTION:", ""), | |
| "dimlist_type": converter_data.get("dimlist_type", ""), | |
| "strain_relief": converter_data.get("STRAIN RELIEF", ""), | |
| "location": converter_data.get("LOCATION", ""), | |
| "dimmability": converter_data.get("DIMMABILITY", ""), | |
| "ccr_(amplitude)": converter_data.get("CCR (AMPLITUDE)", ""), | |
| "size_l*b*h_(mm)": converter_data.get("SIZE: L*B*H (mm)", ""), | |
| "ip": float(converter_data.get("IP", 0)), | |
| "class": float(converter_data.get("CLASS", 0)), | |
| "barcode": converter_data.get("Barcode", ""), | |
| "name": converter_data.get("Name", ""), | |
| "listprice": float(converter_data.get("Listprice", 0)), | |
| "unit": converter_data.get("Unit", ""), | |
| "gross_weight": float(converter_data.get("gross_weight", 0)), | |
| "lifecycle": converter_data.get("LifeCycle", ""), | |
| "pdf_link": converter_data.get("pdf_link", ""), | |
| "lamps": converter_data.get("lamps", {}), | |
| "output_voltage": converter_data.get("OUTPUT VOLTAGE (V)", ""), | |
| "nom_input_voltage": converter_data.get("NOM. INPUT VOLTAGE (V)", ""), | |
| "efficiency_full_load": float(converter_data.get("EFFICIENCY @full load", 0)) | |
| } | |
| def sync_to_cosmos_db(converter_id: str, converter_data: Dict[str, Any], meta_data: Dict[str, Any], operation="upsert"): | |
| """Sync converter data to Cosmos DB with UUID and transformed format.""" | |
| if operation == "delete": | |
| cosmos_id = meta_data.get("cosmos_id") | |
| if cosmos_id: | |
| try: | |
| # Use artnr as partition key value | |
| artnr = int(converter_data.get("ARTNR", 0)) if converter_data else 0 | |
| container.delete_item(item=cosmos_id, partition_key=artnr) | |
| except CosmosResourceNotFoundError: | |
| print(f"Document with ID {cosmos_id} not found in Cosmos DB. Continuing deletion.") | |
| except exceptions.CosmosHttpResponseError as e: | |
| print(f"Error deleting document {cosmos_id}: {str(e)}") | |
| return False | |
| return True | |
| # Transform data to Cosmos DB format | |
| document = transform_to_cosmos_format(converter_id, converter_data) | |
| # Store the UUID in metadata | |
| meta_data["cosmos_id"] = document["id"] | |
| try: | |
| # Use artnr as partition key value | |
| doc = container.create_item(document) | |
| if doc: | |
| print(doc) | |
| print(container.read_item(item=doc["id"], partition_key=doc["artnr"])) | |
| return True | |
| except exceptions.CosmosHttpResponseError as e: | |
| print(f"Error syncing to Cosmos DB: {str(e)}") | |
| return False | |
| def add_converter( | |
| converter_id, type_, artnr, converter_description, dimlist_type, strain_relief, location, dimmability, | |
| ccr_amplitude, size, efficiency, ip, class_, input_voltage, output_voltage, barcode, name, | |
| price, unit, gross_weight, lifecycle, pdf_link | |
| ): | |
| converter_id = converter_id.strip() | |
| data = load_json(DATA_PATH) | |
| meta = load_json(META_PATH) | |
| if converter_id in data: | |
| return f"Converter '{converter_id}' already exists." | |
| info = { | |
| "TYPE": type_ or "", | |
| "ARTNR": float(artnr) if artnr else 0, | |
| "CONVERTER DESCRIPTION:": converter_description or "", | |
| "dimlist_type": dimlist_type or "", | |
| "STRAIN RELIEF": strain_relief or "", | |
| "LOCATION": location or "", | |
| "DIMMABILITY": dimmability or "", | |
| "CCR (AMPLITUDE)": ccr_amplitude or "", | |
| "SIZE: L*B*H (mm)": size or "", | |
| "EFFICIENCY @full load": float(efficiency) if efficiency else 0, | |
| "IP": float(ip) if ip else 0, | |
| "CLASS": float(class_) if class_ else 0, | |
| "NOM. INPUT VOLTAGE (V)": input_voltage or "", | |
| "OUTPUT VOLTAGE (V)": output_voltage or "", | |
| "Barcode": barcode or "", | |
| "Name": name or "", | |
| "Listprice": float(price) if price else 0, | |
| "Unit": unit or "", | |
| "gross_weight": float(gross_weight) if gross_weight else 0, | |
| "LifeCycle": lifecycle or "", | |
| "pdf_link": pdf_link or "", | |
| "lamps": {} | |
| } | |
| data[converter_id] = info | |
| now = get_current_time() | |
| meta[converter_id] = { | |
| "created_at": now, | |
| "updated_at": now, | |
| "deleted_at": None, | |
| "price_history": [{"timestamp": now, "price": float(price)}] if price else [], | |
| "cosmos_id": None # Will be updated after syncing | |
| } | |
| # Sync to Cosmos DB and update metadata with cosmos_id | |
| if sync_to_cosmos_db(converter_id, info, meta[converter_id]): | |
| # Save to JSON files | |
| save_json(data, DATA_PATH) | |
| save_json(meta, META_PATH) | |
| return f"Added converter '{converter_id}' and synced to Cosmos DB." | |
| else: | |
| return f"Failed to add converter '{converter_id}' to Cosmos DB." | |
| def update_converter( | |
| converter_id, type_, artnr, converter_description, dimlist_type, strain_relief, location, dimmability, | |
| ccr_amplitude, size, efficiency, ip, class_, input_voltage, output_voltage, barcode, name, | |
| price, unit, gross_weight, lifecycle, pdf_link | |
| ): | |
| converter_id = converter_id.strip() | |
| data = load_json(DATA_PATH) | |
| meta = load_json(META_PATH) | |
| if converter_id not in data: | |
| return f"Converter '{converter_id}' does not exist." | |
| info = data[converter_id] | |
| now = get_current_time() | |
| # Update fields if provided | |
| if type_: info["TYPE"] = type_ | |
| if artnr: info["ARTNR"] = float(artnr) | |
| if converter_description: info["CONVERTER DESCRIPTION:"] = converter_description | |
| if dimlist_type: info["dimlist_type"] = dimlist_type | |
| if strain_relief: info["STRAIN RELIEF"] = strain_relief | |
| if location: info["LOCATION"] = location | |
| if dimmability: info["DIMMABILITY"] = dimmability | |
| if ccr_amplitude: info["CCR (AMPLITUDE)"] = ccr_amplitude | |
| if size: info["SIZE: L*B*H (mm)"] = size | |
| if efficiency: info["EFFICIENCY @full load"] = float(efficiency) | |
| if ip: info["IP"] = float(ip) | |
| if class_: info["CLASS"] = float(class_) | |
| if input_voltage: info["NOM. INPUT VOLTAGE (V)"] = input_voltage | |
| if output_voltage: info["OUTPUT VOLTAGE (V)"] = output_voltage | |
| if barcode: info["Barcode"] = barcode | |
| if name: info["Name"] = name | |
| if price: | |
| price = float(price) | |
| if "Listprice" not in info or price != info.get("Listprice"): | |
| if converter_id not in meta: | |
| meta[converter_id] = { | |
| "created_at": now, | |
| "updated_at": now, | |
| "deleted_at": None, | |
| "price_history": [], | |
| "cosmos_id": None | |
| } | |
| if "price_history" not in meta[converter_id]: | |
| meta[converter_id]["price_history"] = [] | |
| meta[converter_id]["price_history"].append({"timestamp": now, "price": price}) | |
| info["Listprice"] = price | |
| if unit: info["Unit"] = unit | |
| if gross_weight: info["gross_weight"] = float(gross_weight) | |
| if lifecycle: info["LifeCycle"] = lifecycle | |
| if pdf_link: info["pdf_link"] = pdf_link | |
| # Update metadata | |
| if converter_id not in meta: | |
| meta[converter_id] = { | |
| "created_at": now, | |
| "updated_at": now, | |
| "deleted_at": None, | |
| "price_history": [], | |
| "cosmos_id": None | |
| } | |
| meta[converter_id]["updated_at"] = now | |
| # Compute new id | |
| new_type = info["TYPE"] | |
| new_artnr = int(info["ARTNR"]) | |
| new_id = f"{new_type}mA - {new_artnr}" | |
| if new_id != converter_id: | |
| # Delete old Cosmos DB document | |
| sync_to_cosmos_db(converter_id, data[converter_id], meta[converter_id], operation="delete") | |
| data[new_id] = info | |
| meta[new_id] = meta.pop(converter_id) | |
| del data[converter_id] | |
| # Sync new document | |
| if sync_to_cosmos_db(new_id, info, meta[new_id]): | |
| save_json(data, DATA_PATH) | |
| save_json(meta, META_PATH) | |
| return f"Updated converter. ID changed to '{new_id}' and synced to Cosmos DB." | |
| else: | |
| return f"Failed to update converter to '{new_id}' in Cosmos DB." | |
| else: | |
| if sync_to_cosmos_db(converter_id, info, meta[converter_id]): | |
| data[converter_id] = info | |
| save_json(data, DATA_PATH) | |
| save_json(meta, META_PATH) | |
| return f"Updated converter '{converter_id}' and synced to Cosmos DB." | |
| else: | |
| return f"Failed to update converter '{converter_id}' in Cosmos DB." | |
| def delete_converter(converter_id): | |
| converter_id = converter_id.strip() | |
| data = load_json(DATA_PATH) | |
| meta = load_json(META_PATH) | |
| if converter_id not in data: | |
| return f"Converter '{converter_id}' does not exist." | |
| now = get_current_time() | |
| if converter_id not in meta: | |
| meta[converter_id] = { | |
| "created_at": None, | |
| "updated_at": now, | |
| "deleted_at": now, | |
| "price_history": [], | |
| "cosmos_id": None | |
| } | |
| else: | |
| meta[converter_id]["deleted_at"] = now | |
| # Delete from Cosmos DB | |
| sync_to_cosmos_db(converter_id, data[converter_id], meta[converter_id], operation="delete") | |
| del data[converter_id] | |
| save_json(data, DATA_PATH) | |
| save_json(meta, META_PATH) | |
| return f"Deleted converter '{converter_id}' and removed from Cosmos DB." | |
| def add_or_update_lamp(converter_id, lamp_name, min_val, max_val): | |
| converter_id = converter_id.strip() | |
| lamp_name = lamp_name.strip() | |
| data = load_json(DATA_PATH) | |
| meta = load_json(META_PATH) | |
| if converter_id not in data: | |
| return f"Converter '{converter_id}' does not exist." | |
| if "lamps" not in data[converter_id]: | |
| data[converter_id]["lamps"] = {} | |
| data[converter_id]["lamps"][lamp_name] = {"min": min_val, "max": max_val} | |
| # Update metadata | |
| now = get_current_time() | |
| if converter_id not in meta: | |
| meta[converter_id] = { | |
| "created_at": now, | |
| "updated_at": now, | |
| "deleted_at": None, | |
| "price_history": [], | |
| "cosmos_id": None | |
| } | |
| meta[converter_id]["updated_at"] = now | |
| # Sync to Cosmos DB | |
| if sync_to_cosmos_db(converter_id, data[converter_id], meta[converter_id]): | |
| save_json(data, DATA_PATH) | |
| save_json(meta, META_PATH) | |
| return f"Added/updated lamp '{lamp_name}' in converter '{converter_id}' and synced to Cosmos DB." | |
| else: | |
| return f"Failed to add/update lamp '{lamp_name}' in Cosmos DB." | |
| def delete_lamp(converter_id, lamp_name): | |
| converter_id = converter_id.strip() | |
| lamp_name = lamp_name.strip() | |
| data = load_json(DATA_PATH) | |
| meta = load_json(META_PATH) | |
| if converter_id not in data: | |
| return f"Converter '{converter_id}' does not exist." | |
| lamps = data[converter_id].get("lamps", {}) | |
| if lamp_name not in lamps: | |
| return f"Lamp '{lamp_name}' does not exist in converter '{converter_id}'." | |
| del lamps[lamp_name] | |
| # Update metadata | |
| now = get_current_time() | |
| if converter_id not in meta: | |
| meta[converter_id] = { | |
| "created_at": now, | |
| "updated_at": now, | |
| "deleted_at": None, | |
| "price_history": [], | |
| "cosmos_id": None | |
| } | |
| meta[converter_id]["updated_at"] = now | |
| # Sync to Cosmos DB | |
| if sync_to_cosmos_db(converter_id, data[converter_id], meta[converter_id]): | |
| save_json(data, DATA_PATH) | |
| save_json(meta, META_PATH) | |
| return f"Deleted lamp '{lamp_name}' from converter '{converter_id}' and synced to Cosmos DB." | |
| else: | |
| return f"Failed to delete lamp '{lamp_name}' in Cosmos DB." | |
| def get_converter(converter_id): | |
| converter_id = converter_id.strip() | |
| meta = load_json(META_PATH) | |
| cosmos_id = meta.get(converter_id, {}).get("cosmos_id") | |
| if cosmos_id: | |
| try: | |
| # Fetch from Cosmos DB using artnr as partition key | |
| data = load_json(DATA_PATH) | |
| artnr = int(data.get(converter_id, {}).get("ARTNR", 0)) | |
| item = container.read_item(item=cosmos_id, partition_key=artnr) | |
| # Add metadata from local file | |
| item["metadata"] = meta.get(converter_id, {}) | |
| return json.dumps(item, indent=2, ensure_ascii=False) | |
| except CosmosResourceNotFoundError: | |
| print(f"Document with ID {cosmos_id} not found in Cosmos DB. Falling back to local JSON.") | |
| except exceptions.CosmosHttpResponseError as e: | |
| print(f"Error reading document {cosmos_id}: {str(e)}") | |
| # Fallback to local JSON | |
| data = load_json(DATA_PATH) | |
| item = data.get(converter_id, {}) | |
| if item: | |
| item["metadata"] = meta.get(converter_id, {}) | |
| return json.dumps(item, indent=2, ensure_ascii=False) | |
| def filter_lamps(filter_type, n_latest): | |
| # Load local JSON files | |
| data = load_json(DATA_PATH) | |
| meta = load_json(META_PATH) | |
| records = [] | |
| for cid, cinfo in data.items(): | |
| m = meta.get(cid, {}) | |
| created_at = m.get("created_at", "") | |
| updated_at = m.get("updated_at", "") | |
| deleted_at = m.get("deleted_at", None) | |
| price_history = m.get("price_history", []) | |
| price = cinfo.get("Listprice", "") | |
| lamps = cinfo.get("lamps", {}) | |
| record = { | |
| "Converter ID": cid, | |
| "Created At": created_at, | |
| "Updated At": updated_at, | |
| "Deleted At": deleted_at, | |
| "Price": price, | |
| "Lamps": ", ".join(lamps.keys()) | |
| } | |
| if filter_type == "Show All": | |
| records.append(record) | |
| elif filter_type == "Latest Added": | |
| if not deleted_at: | |
| records.append(record) | |
| elif filter_type == "Latest Updated": | |
| if not deleted_at: | |
| records.append(record) | |
| elif filter_type == "Deleted": | |
| if deleted_at: | |
| records.append(record) | |
| elif filter_type == "Price Change": | |
| if len(price_history) > 1: | |
| record["Price History"] = str(price_history) | |
| records.append(record) | |
| # Apply sorting and slicing | |
| if filter_type == "Latest Added": | |
| records = sorted(records, key=lambda x: x.get("Created At", ""), reverse=True)[:n_latest] | |
| elif filter_type == "Latest Updated": | |
| records = sorted(records, key=lambda x: x.get("Updated At", ""), reverse=True)[:n_latest] | |
| return pd.DataFrame(records) | |
| # Gradio interface | |
| with gr.Blocks(title="TAL Converter JSON Editor") as demo: | |
| gr.Markdown("# TAL Converter JSON Editor") | |
| with gr.Tab("Add Converter"): | |
| gr.HTML("<label style='font-weight:bold; color:red;'>Converter ID (e.g. 350mA - 930537) *</label>") | |
| converter_id = gr.Textbox(show_label=False) | |
| gr.HTML("<label style='font-weight:bold; color:red;'>Article Number *</label>") | |
| artnr = gr.Textbox(show_label=False) | |
| type_ = gr.Textbox(label="Type") | |
| # artnr = gr.Textbox(label="Article Number !*required field") | |
| converter_description = gr.Textbox(label="Converter Description") | |
| dimlist_type = gr.Textbox(label="Dimlist Type") | |
| strain_relief = gr.Textbox(label="Strain Relief") | |
| location = gr.Textbox(label="Location") | |
| dimmability = gr.Textbox(label="Dimmability") | |
| ccr_amplitude = gr.Textbox(label="CCR (Amplitude)") | |
| size = gr.Textbox(label="Size: L*B*H (mm)") | |
| efficiency = gr.Textbox(label="Efficiency @ Full Load") | |
| ip = gr.Textbox(label="IP") | |
| class_ = gr.Textbox(label="Class") | |
| input_voltage = gr.Textbox(label="Nominal Input Voltage (V)") | |
| output_voltage = gr.Textbox(label="Output Voltage (V)") | |
| barcode = gr.Textbox(label="Barcode") | |
| name = gr.Textbox(label="Name") | |
| price = gr.Textbox(label="List Price") | |
| unit = gr.Textbox(label="Unit") | |
| gross_weight = gr.Textbox(label="Gross Weight") | |
| lifecycle = gr.Textbox(label="Lifecycle") | |
| pdf_link = gr.Textbox(label="PDF Link") | |
| add_btn = gr.Button("Add Converter") | |
| add_output = gr.Textbox(label="Result") | |
| add_btn.click( | |
| add_converter, | |
| inputs=[ | |
| converter_id, type_, artnr, converter_description, dimlist_type, strain_relief, location, dimmability, | |
| ccr_amplitude, size, efficiency, ip, class_, input_voltage, output_voltage, barcode, name, | |
| price, unit, gross_weight, lifecycle, pdf_link | |
| ], | |
| outputs=add_output | |
| ).then( | |
| lambda *args: [gr.update(value="") for _ in args], | |
| inputs=[ | |
| converter_id, type_, artnr, converter_description, dimlist_type, strain_relief, location, dimmability, | |
| ccr_amplitude, size, efficiency, ip, class_, input_voltage, output_voltage, barcode, name, | |
| price, unit, gross_weight, lifecycle, pdf_link | |
| ], | |
| outputs=[ | |
| converter_id, type_, artnr, converter_description, dimlist_type, strain_relief, location, dimmability, | |
| ccr_amplitude, size, efficiency, ip, class_, input_voltage, output_voltage, barcode, name, | |
| price, unit, gross_weight, lifecycle, pdf_link | |
| ] | |
| ) | |
| with gr.Tab("Update Converter"): | |
| gr.HTML("<label style='font-weight:bold; color:red;'>Converter ID (e.g. 350mA - 930537) *</label>") | |
| converter_id_u = gr.Textbox(show_label=False) | |
| type_u = gr.Textbox(label="Type") | |
| artnr_u = gr.Textbox(label="Article Number") | |
| converter_description_u = gr.Textbox(label="Converter Description") | |
| dimlist_type_u = gr.Textbox(label="Dimlist Type") | |
| strain_relief_u = gr.Textbox(label="Strain Relief") | |
| location_u = gr.Textbox(label="Location") | |
| dimmability_u = gr.Textbox(label="Dimmability") | |
| ccr_amplitude_u = gr.Textbox(label="CCR (Amplitude)") | |
| size_u = gr.Textbox(label="Size: L*B*H (mm)") | |
| efficiency_u = gr.Textbox(label="Efficiency @ Full Load") | |
| ip_u = gr.Textbox(label="IP") | |
| class_u = gr.Textbox(label="Class") | |
| input_voltage_u = gr.Textbox(label="Nominal Input Voltage (V)") | |
| output_voltage_u = gr.Textbox(label="Output Voltage (V)") | |
| barcode_u = gr.Textbox(label="Barcode") | |
| name_u = gr.Textbox(label="Name") | |
| price_u = gr.Textbox(label="List Price") | |
| unit_u = gr.Textbox(label="Unit") | |
| gross_weight_u = gr.Textbox(label="Gross Weight") | |
| lifecycle_u = gr.Textbox(label="Lifecycle") | |
| pdf_link_u = gr.Textbox(label="PDF Link") | |
| update_btn = gr.Button("Update Converter") | |
| update_output = gr.Textbox(label="Result") | |
| update_btn.click( | |
| update_converter, | |
| inputs=[ | |
| converter_id_u, type_u, artnr_u, converter_description_u, dimlist_type_u, strain_relief_u, location_u, dimmability_u, | |
| ccr_amplitude_u, size_u, efficiency_u, ip_u, class_u, input_voltage_u, output_voltage_u, barcode_u, name_u, | |
| price_u, unit_u, gross_weight_u, lifecycle_u, pdf_link_u | |
| ], | |
| outputs=update_output | |
| ).then( | |
| lambda *args: [gr.update(value="") for _ in args], | |
| inputs=[ | |
| converter_id_u, type_u, artnr_u, converter_description_u, dimlist_type_u, strain_relief_u, location_u, dimmability_u, | |
| ccr_amplitude_u, size_u, efficiency_u, ip_u, class_u, input_voltage_u, output_voltage_u, barcode_u, name_u, | |
| price_u, unit_u, gross_weight_u, lifecycle_u, pdf_link_u | |
| ], | |
| outputs=[ | |
| converter_id_u, type_u, artnr_u, converter_description_u, dimlist_type_u, strain_relief_u, location_u, dimmability_u, | |
| ccr_amplitude_u, size_u, efficiency_u, ip_u, class_u, input_voltage_u, output_voltage_u, barcode_u, name_u, | |
| price_u, unit_u, gross_weight_u, lifecycle_u, pdf_link_u | |
| ] | |
| ) | |
| with gr.Tab("Delete Converter"): | |
| gr.HTML("<label style='font-weight:bold; color:red;'>Converter ID (e.g. 350mA - 930537) *</label>") | |
| converter_id_d = gr.Textbox(show_label=False) | |
| delete_btn = gr.Button("Delete Converter") | |
| delete_output = gr.Textbox(label="Result") | |
| delete_btn.click( | |
| delete_converter, | |
| inputs=converter_id_d, | |
| outputs=delete_output | |
| ).then( | |
| lambda _: gr.update(value=""), | |
| inputs=converter_id_d, | |
| outputs=converter_id_d | |
| ) | |
| with gr.Tab("Add/Update Lamp"): | |
| gr.HTML("<label style='font-weight:bold; color:red;'>Converter ID (e.g. 350mA - 930537) *</label>") | |
| converter_id_l = gr.Textbox(show_label=False) | |
| lamp_name = gr.Textbox(label="Lamp Name") | |
| min_val = gr.Textbox(label="Min") | |
| max_val = gr.Textbox(label="Max") | |
| lamp_btn = gr.Button("Add/Update Lamp") | |
| lamp_output = gr.Textbox(label="Result") | |
| lamp_btn.click( | |
| add_or_update_lamp, | |
| inputs=[converter_id_l, lamp_name, min_val, max_val], | |
| outputs=lamp_output | |
| ).then( | |
| lambda *args: [gr.update(value="") for _ in args], | |
| inputs=[converter_id_l, lamp_name, min_val, max_val], | |
| outputs=[converter_id_l, lamp_name, min_val, max_val] | |
| ) | |
| with gr.Tab("Delete Lamp"): | |
| converter_id_ld = gr.Textbox(label="Converter ID (e.g. 350mA - 930537)") | |
| lamp_name_d = gr.Textbox(label="Lamp Name") | |
| lamp_delete_btn = gr.Button("Delete Lamp") | |
| lamp_delete_output = gr.Textbox(label="Result") | |
| lamp_delete_btn.click( | |
| delete_lamp, | |
| inputs=[converter_id_ld, lamp_name_d], | |
| outputs=lamp_delete_output | |
| ).then( | |
| lambda *args: [gr.update(value="") for _ in args], | |
| inputs=[converter_id_ld, lamp_name_d], | |
| outputs=[converter_id_ld, lamp_name_d] | |
| ) | |
| with gr.Tab("View Converter"): | |
| converter_id_v = gr.Textbox(label="Converter ID (e.g. 350mA - 930537)") | |
| view_btn = gr.Button("View Converter") | |
| view_output = gr.Textbox(label="Converter Data", lines=10) | |
| view_btn.click(get_converter, inputs=converter_id_v, outputs=view_output) | |
| with gr.Tab("Converters Table & Filters"): | |
| filter_type = gr.Dropdown( | |
| choices=["Show All", "Latest Added", "Latest Updated", "Price Change"], | |
| value="Show All", | |
| label="Filter Type" | |
| ) | |
| n_latest = gr.Slider(1, 20, value=5, step=1, label="How many (for latest)") | |
| lamp_table = gr.DataFrame(label="Converters Table", interactive=False, show_search="filter") | |
| filter_btn = gr.Button("Apply Filter") | |
| filter_btn.click( | |
| filter_lamps, | |
| inputs=[filter_type, n_latest], | |
| outputs=lamp_table | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() # Run locally without sharing | |