Skip to content

Normalize hardware response keys across APIs #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jithinAB wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Normalize hardware response keys across APIs #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 47 additions & 37 deletions BudSimulator/apis/routers/hardware.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from src.hardware import BudHardware
from src.hardware_recommendation import HardwareRecommendation
from src.utils.hardware_formatting import normalize_hardware_keys


router = APIRouter(prefix="/api/hardware", tags=["hardware"])
Expand Down Expand Up @@ -170,8 +171,10 @@ async def create_hardware(hardware: HardwareCreate):
created = hardware_manager.get_hardware_by_name(hardware.name)
if not created:
raise HTTPException(status_code=500, detail="Failed to create hardware")

return HardwareResponse(**created)

created_normalized = normalize_hardware_keys(created)

return HardwareResponse(**created_normalized)

except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
Expand Down Expand Up @@ -203,22 +206,24 @@ async def list_hardware(
# Add price indicators to each hardware item
enhanced_results = []
for hw in results:
normalized_hw = normalize_hardware_keys(hw)

# Calculate price indicator if specs are available
flops = hw.get('flops', 0)
memory_size = hw.get('memory_size', 0)
memory_bw = hw.get('memory_bw', 0)
flops = normalized_hw.get('flops', 0)
memory_size = normalized_hw.get('memory_size', 0)
memory_bw = normalized_hw.get('memory_bw', 0)

if flops > 0 and memory_size > 0 and memory_bw > 0:
hw['price_approx'] = BudHardware.calculate_price_indicator(
normalized_hw['price_approx'] = BudHardware.calculate_price_indicator(
flops=flops,
memory_gb=memory_size,
bandwidth_gbs=memory_bw
)
else:
hw['price_approx'] = None
enhanced_results.append(hw)
normalized_hw['price_approx'] = None

enhanced_results.append(normalized_hw)
Comment on lines 207 to +225
Copy link

@gemini-code-assist gemini-code-assist bot Sep 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

There is significant code duplication between this loop in list_hardware and a similar loop in filter_hardware (lines 282-300). To improve maintainability and reduce redundancy, consider extracting this logic into a private helper function.

For example, you could create a function _enhance_hardware_results:

def _enhance_hardware_results(hardware_list: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    enhanced_results = []
    for hw in hardware_list:
        normalized_hw = normalize_hardware_keys(hw)

        # Calculate price indicator if specs are available
        flops = normalized_hw.get('flops', 0)
        memory_size = normalized_hw.get('memory_size', 0)
        memory_bw = normalized_hw.get('memory_bw', 0)

        if flops > 0 and memory_size > 0 and memory_bw > 0:
            normalized_hw['price_approx'] = BudHardware.calculate_price_indicator(
                flops=flops,
                memory_gb=memory_size,
                bandwidth_gbs=memory_bw
            )
        else:
            normalized_hw['price_approx'] = None

        enhanced_results.append(normalized_hw)
    return enhanced_results

Then, both list_hardware and filter_hardware can be simplified to:

        enhanced_results = _enhance_hardware_results(results)
        return [HardwareResponse(**hw) for hw in enhanced_results]


return [HardwareResponse(**hw) for hw in enhanced_results]

except Exception as e:
Expand Down Expand Up @@ -276,22 +281,24 @@ async def filter_hardware(
# Add price indicators to each hardware item
enhanced_results = []
for hw in results:
normalized_hw = normalize_hardware_keys(hw)

# Calculate price indicator if specs are available
flops = hw.get('flops', 0)
memory_size = hw.get('memory_size', 0)
memory_bw = hw.get('memory_bw', 0)
flops = normalized_hw.get('flops', 0)
memory_size = normalized_hw.get('memory_size', 0)
memory_bw = normalized_hw.get('memory_bw', 0)

if flops > 0 and memory_size > 0 and memory_bw > 0:
hw['price_approx'] = BudHardware.calculate_price_indicator(
normalized_hw['price_approx'] = BudHardware.calculate_price_indicator(
flops=flops,
memory_gb=memory_size,
bandwidth_gbs=memory_bw
)
else:
hw['price_approx'] = None
enhanced_results.append(hw)
normalized_hw['price_approx'] = None

enhanced_results.append(normalized_hw)

return [HardwareResponse(**hw) for hw in enhanced_results]

except Exception as e:
Expand All @@ -306,30 +313,32 @@ async def get_hardware(hardware_name: str):
hardware = hardware_manager.get_hardware_by_name(hardware_name)
if not hardware:
raise HTTPException(status_code=404, detail="Hardware not found")


normalized_hardware = normalize_hardware_keys(hardware)

# Get vendor details with pricing
vendors = hardware_manager.get_hardware_vendors(hardware_name)

# Get cloud details with instance pricing
clouds = hardware_manager.get_hardware_clouds(hardware_name)

# Build detailed response
response = HardwareDetailResponse(
name=hardware['name'],
type=hardware['type'],
manufacturer=hardware.get('manufacturer'),
flops=hardware['flops'],
memory_size=hardware['memory_size'],
memory_bw=hardware['memory_bw'],
icn=hardware.get('icn'),
icn_ll=hardware.get('icn_ll'),
power=hardware.get('power'),
real_values=hardware.get('real_values', True),
url=hardware.get('url'),
description=hardware.get('description'),
name=normalized_hardware['name'],
type=normalized_hardware['type'],
manufacturer=normalized_hardware.get('manufacturer'),
flops=normalized_hardware['flops'],
memory_size=normalized_hardware['memory_size'],
memory_bw=normalized_hardware['memory_bw'],
icn=normalized_hardware.get('icn'),
icn_ll=normalized_hardware.get('icn_ll'),
power=normalized_hardware.get('power'),
real_values=normalized_hardware.get('real_values', True),
url=normalized_hardware.get('url'),
description=normalized_hardware.get('description'),
vendors=vendors,
clouds=clouds,
source=hardware.get('source', 'manual')
source=normalized_hardware.get('source', 'manual')
)

return response
Expand All @@ -354,7 +363,8 @@ async def update_hardware(hardware_name: str, updates: HardwareUpdate):

# Get updated hardware
updated = hardware_manager.get_hardware_by_name(hardware_name)
return HardwareResponse(**updated)
normalized_updated = normalize_hardware_keys(updated)
return HardwareResponse(**normalized_updated)

except HTTPException:
raise
Expand Down
7 changes: 5 additions & 2 deletions BudSimulator/src/api/hardware_routes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from flask import request, jsonify
from BudSimulator.src.hardware import BudHardware
from BudSimulator.src.hardware_recommendation import HardwareRecommendation
from BudSimulator.src.utils.hardware_formatting import normalize_hardware_keys


def create_hardware_routes(app):
Expand Down Expand Up @@ -66,10 +67,12 @@ def list_hardware():

results = hardware.search_hardware(**params)

normalized_results = [normalize_hardware_keys(hw) for hw in results]

return jsonify({
'success': True,
'hardware': results,
'count': len(results)
'hardware': normalized_results,
'count': len(normalized_results)
}), 200

except Exception as e:
Expand Down
22 changes: 22 additions & 0 deletions BudSimulator/src/utils/hardware_formatting.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
"""Utility helpers for formatting hardware payloads."""
from typing import Any, Dict


def normalize_hardware_keys(hardware: Dict[str, Any]) -> Dict[str, Any]:
"""Ensure lowercase hardware keys exist for their uppercase counterparts.
Some hardware records may come from sources that use uppercase keys (e.g.,
``FLOPS`` instead of ``flops``). This helper preserves the original keys
while populating the expected lowercase versions when they are missing so
that API serializers can rely on a consistent shape.
"""
if hardware is None:
return {}

normalized = dict(hardware)
for key, value in hardware.items():
if isinstance(key, str) and key.isupper():
lower_key = key.lower()
if lower_key not in normalized:
normalized[lower_key] = value
return normalized
Comment on lines +5 to +22
Copy link

@gemini-code-assist gemini-code-assist bot Sep 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The current implementation of normalize_hardware_keys only handles fully uppercase keys (e.g., FLOPS) because of key.isupper(). However, other parts of the codebase suggest that keys can also be in PascalCase or mixed case (e.g., Memory_size, Flops). These keys would not be correctly normalized.

To make this utility more robust and handle all non-lowercase keys, I suggest modifying the logic to check if key.lower() != key. This will correctly handle uppercase, PascalCase, and mixed-case keys. I've also updated the docstring to reflect this broader scope.

This change will ensure that all expected keys are normalized, making the API responses more consistent.

Suggested change
def normalize_hardware_keys(hardware: Dict[str, Any]) -> Dict[str, Any]:
"""Ensure lowercase hardware keys exist for their uppercase counterparts.
Some hardware records may come from sources that use uppercase keys (e.g.,
``FLOPS`` instead of ``flops``). This helper preserves the original keys
while populating the expected lowercase versions when they are missing so
that API serializers can rely on a consistent shape.
"""
if hardware is None:
return {}
normalized = dict(hardware)
for key, value in hardware.items():
if isinstance(key, str) and key.isupper():
lower_key = key.lower()
if lower_key not in normalized:
normalized[lower_key] = value
return normalized
def normalize_hardware_keys(hardware: Dict[str, Any]) -> Dict[str, Any]:
"""Ensure lowercase hardware keys exist for their non-lowercase counterparts.
Some hardware records may come from sources that use uppercase or mixed-case
keys (e.g., ``FLOPS`` or ``Memory_size`` instead of ``flops`` or ``memory_size``).
This helper preserves the original keys while populating the expected lowercase
versions when they are missing so that API serializers can rely on a consistent shape.
"""
if hardware is None:
return {}
normalized = dict(hardware)
for key, value in hardware.items():
if isinstance(key, str):
lower_key = key.lower()
if key != lower_key and lower_key not in normalized:
normalized[lower_key] = value
return normalized