Spaces:
Sleeping
Sleeping
File size: 7,159 Bytes
3d4cf60 71180ec 3d4cf60 71180ec 3d4cf60 38ee18f 3d4cf60 864b8b2 3d4cf60 38ee18f 3d4cf60 71180ec 3d4cf60 71180ec 3d4cf60 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 |
#converterPlugin.py
from typing import Annotated, Optional
from CosmosDBHandlers.cosmosConnector import CosmosLampHandler
from semantic_kernel.functions import kernel_function
class ConverterPlugin:
def __init__(self, logger):
self.logger = logger
self.db = CosmosLampHandler(logger=logger)
@kernel_function(
name="query_converters",
description="Execute SQL query against Cosmos DB converters collection"
)
async def query_converters(
self,
user_input:Annotated[str,"Natural Language question the user asked"],
query: Annotated[str,"SQL generated from NL2SQL plugin"]) -> str:
try:
items = await self.db.query_converters(query, user_input)
self.logger.info(f"Executed query: {query}")
if not items:
return "No items found for the given query."
return str(items)
except Exception as e:
return f"Query failed: {str(e)}"
@kernel_function(
name = "get_converter_info",
description="Get information about a converter using its artnr (partition key)"
)
async def get_converter_info(self, artnr:int) -> str:
try:
converter = await self.db.get_converter_info(artnr)
self.logger.info(f"Used get_converter_info with artrn: {artnr}")
return f"{converter.model_dump()})"
except Exception as e:
f"Failed to retrieve converter {artnr} - {e}"
@kernel_function
async def get_compatible_lamps(
self,
artnr: Annotated[int, "Converter artnr (partition key)"]
) -> str:
"""Get compatible lamps for a converter by artnr"""
try:
lamps = await self.db.get_compatible_lamps(artnr)
self.logger.info(f"Used get_compatible_lamps with artnr: {artnr}")
return f"Compatible lamps: {', '.join(lamps)}" if lamps else "No lamps found"
except Exception as e:
return f"Error retrieving compatible lamps: {str(e)}"
@kernel_function(
name="get_converters_by_lamp_type",
description="Find converters compatible with a specific lamp type"
)
async def get_converters_by_lamp_type(
self,
lamp_type: Annotated[str, "Lamp model (e.g., Haloled, B4)"]
) -> str:
"""Find converters compatible with specific lamp type"""
try:
converters = await self.db.get_converters_by_lamp_type(lamp_type)
self.logger.info(f"Used get_converters_by_lamp_type with lamp_type: {lamp_type}")
if not converters:
return "No compatible converters found"
return "\n".join([f"{c.model_dump()})" for c in converters]) if converters else "No converters found"
except Exception as e:
return f"Error retrieving converters: {str(e)}"
@kernel_function(
name="get_converters_by_dimming",
description="Find converters of a specified dimming type"
)
async def get_converters_by_dimming(
self,
dimming_type: Annotated[str, "Dimming type mentioned like dali, mains, 1-10v"],
voltage_current: Annotated[str | None,"Voltage or current specification like 350mA, 24V DC"] = None,
lamp_type: Annotated[str | None, "Lamp model (e.g., Haloled, B4)"] = None,
threshold: int = 75) -> str:
"""Search converters by dimming type with technical specifications"""
try:
converters = await self.db.get_converters_by_dimming(
dimming_type=dimming_type,
voltage_current=voltage_current,
lamp_type=lamp_type,
threshold=threshold)
self.logger.info(f"""Used get_converters_by_dimming with dimming type: {dimming_type}
voltage_current: {voltage_current}
lamp_type: {lamp_type}""")
if not converters:
return "No relavent converters found"
return "\n".join([f"{c.model_dump()})" for c in converters]) if converters else "No converters found"
except Exception as e:
return f"Error returning converters: {str(e)}"
@kernel_function(
name="get_lamp_limits",
description="Get min/max lamps for a converter by artnr and lamp type"
)
async def get_lamp_limits(
self,
artnr: Annotated[int, "Converter artnr"],
lamp_type: Annotated[str, "Lamp model (e.g., Haloled)"]
) -> str:
"""Get min/max lamps for a converter"""
try:
limits = await self.db.get_lamp_limits(artnr, lamp_type)
self.logger.info(f"Used get_lamp_limits with ARTNR: {artnr} and lamp_type: {lamp_type}")
return f"{lamp_type}: Min {limits['min']} - Max {limits['max']} lamps"
except Exception as e:
return f"Error retrieving lamp limits: {str(e)}"
@kernel_function(
name="get_converters_by_voltage_current",
description="Get converters that have the mentioned input/output voltage range or current"
)
async def get_converters_by_voltage_current(
self,
artnr: Annotated[int | None, ""] = None,
current: Annotated[str | None, "Current like 350mA, 700mA"]=None,
input_voltage: Annotated[str | None, "Input voltage range like '198-464' NEVER ip, null if no voltage"] = None,
output_voltage: Annotated[str | None, "Output voltage range like '24', '2-25' null if no voltage"] = None,
lamp_type: Annotated[str | None, "Lamp model (e.g., Haloled, B4)"] = None,
) -> str:
try:
converters = await self.db.get_converters_by_voltage_current(artnr=artnr,
current=current,
input_voltage=input_voltage,
output_voltage=output_voltage,
lamp_type=lamp_type)
self.logger.info(f"""Used get_converters_by_voltage_current with input_voltage: {input_voltage}
output_voltage: {output_voltage}
current: {current}
lamp_type: {lamp_type}
artnr: {artnr}""")
if not converters:
return "No relavent converters found"
return "\n".join([f"{c.model_dump()})" for c in converters]) if converters else "No converters found"
except Exception as e:
return f"Error retrieving converters" |