Add CSV input support for license calculation

README.md

@@ -120,19 +120,36 @@ lmci collect -p <provider> [options]
 
 ### `calculate` Command
 
-Calculate licenses from an existing inventory JSON file.
+Calculate licenses from an existing inventory JSON file or a simple CSV upload.
 
 ```bash
 lmci calculate -i <inventory.json> [options]
+lmci calculate --csv-input <inventory.csv> [options]
 ```
 
 | Option | Description |
 |--------|-------------|
-| `-i, --input` | Input inventory JSON file **(required)** |
+| `-i, --input` | Input inventory JSON file |
+| `--csv-input` | Input inventory CSV file with `provider,resource_type,count` headers |
+| `-p, --provider` | Optional provider filter for CSV input |
 | `-o, --output` | Output CSV file (default: `license_summary.csv`) |
 | `-d, --detailed` | Generate detailed CSV |
 | `--show-unmapped` | List unmapped resource types |
 
+Provide exactly one of `--input` or `--csv-input`.
+
+**CSV format**
+
+```csv
+provider,resource_type,count
+aws,AWS::EC2::Instance,42
+aws,lambda function,85
+azure,Virtual machine,18
+azure,App Service plan,6
+```
+
+CSV imports are case-insensitive and support common aliases. For example, AWS values do not need to exactly match the internal `ec2:instance` mapping format, and Azure display names such as `Virtual machine` or `App Service plan` are accepted.
+
 ### `permissions` Command
 
 Show required permissions for each cloud provider.

src/calculator/license_calculator.py

@@ -3,7 +3,9 @@
 import json
 import logging
 import math
-from typing import Dict, List, Set, Tuple
+import re
+from functools import lru_cache
+from typing import Dict, List, Optional, Set, Tuple
 from collections import defaultdict
 
 logger = logging.getLogger(__name__)
@@ -85,13 +87,15 @@ def get_category(self, provider: str, resource_type: str) -> str:
         # Look up in mappings
         provider_mappings = self.resource_mappings.get(provider, {})
 
+        canonical_type = self.normalize_resource_type(provider, resource_type)
+
         # Try exact match first
-        resource_info = provider_mappings.get(resource_type)
+        resource_info = provider_mappings.get(canonical_type)
         if resource_info:
             return resource_info.get('category', 'Unsupported')
 
         # Try case-insensitive match
-        resource_type_lower = resource_type.lower()
+        resource_type_lower = canonical_type.lower()
         for mapped_type, info in provider_mappings.items():
             if mapped_type.lower() == resource_type_lower:
                 return info.get('category', 'Unsupported')
@@ -100,6 +104,86 @@ def get_category(self, provider: str, resource_type: str) -> str:
         self._unsupported_types.add((provider, resource_type))
         return "Unsupported"
 
+    def normalize_resource_type(self, provider: str, resource_type: str) -> str:
+        """
+        Normalize imported resource types to canonical mapping keys.
+
+        This supports case-insensitive matching, friendly aliases, and
+        provider-specific variants such as AWS CloudFormation-style names.
+        """
+        provider_mappings = self.resource_mappings.get(provider, {})
+        if not provider_mappings or not resource_type:
+            return resource_type
+
+        if resource_type in provider_mappings:
+            return resource_type
+
+        alias_index = self._build_alias_index(provider)
+        normalized_key = self._normalize_alias(resource_type)
+        return alias_index.get(normalized_key, resource_type)
+
+    @staticmethod
+    def _normalize_alias(value: str) -> str:
+        """Normalize a resource label for case-insensitive alias matching."""
+        normalized = re.sub(r'([a-z0-9])([A-Z])', r'\1 \2', value or '')
+        normalized = normalized.replace('::', ' ')
+        normalized = normalized.replace('/', ' ')
+        normalized = normalized.replace(':', ' ')
+        normalized = normalized.replace('-', ' ')
+        normalized = normalized.replace('_', ' ')
+        normalized = normalized.replace('.', ' ')
+        normalized = re.sub(r'\s+', ' ', normalized.lower()).strip()
+        return normalized
+
+    @staticmethod
+    def _humanize_provider_leaf(provider: str, resource_type: str) -> str:
+        """Convert a provider resource key into a readable alias candidate."""
+        if provider == 'aws' and ':' in resource_type:
+            service, raw_type = resource_type.split(':', 1)
+            value = f"{service} {raw_type}"
+        elif provider == 'azure' and '/' in resource_type:
+            segments = resource_type.split('/')
+            value = ' '.join(segment for segment in segments[1:] if segment)
+        else:
+            value = resource_type.split('/')[-1].split(':')[-1]
+
+        value = re.sub(r'([a-z0-9])([A-Z])', r'\1 \2', value)
+        value = re.sub(r'(?<=[a-z])(?=(instances|machines|clusters|groups|accounts|servers|vaults|services|jobs|rules|plans|apps|sets|pools|zones|profiles|balances|interfaces|addresses|gateways|databases|workspaces|registries))', ' ', value, flags=re.IGNORECASE)
+        return re.sub(r'\s+', ' ', value).strip()
+
+    @classmethod
+    def _generate_aliases(cls, provider: str, resource_type: str, resource_info: Dict) -> Set[str]:
+        """Generate alias candidates for a canonical resource mapping."""
+        aliases = {resource_type}
+        aliases.add(cls._humanize_provider_leaf(provider, resource_type))
+
+        unit = resource_info.get('unit')
+        if unit:
+            aliases.add(unit)
+
+        if provider == 'aws' and ':' in resource_type:
+            service, raw_type = resource_type.split(':', 1)
+            aliases.add(f"AWS::{service.upper()}::{raw_type.replace('-', '').replace('/', '').upper()}")
+            aliases.add(f"AWS::{service}::{raw_type}")
+            aliases.add(f"{service} {raw_type}")
+
+        for alias in resource_info.get('aliases', []):
+            aliases.add(alias)
+
+        return {alias for alias in aliases if alias}
+
+    @lru_cache(maxsize=None)
+    def _build_alias_index(self, provider: str) -> Dict[str, str]:
+        """Build a normalized alias index for one provider."""
+        alias_index = {}
+        provider_mappings = self.resource_mappings.get(provider, {})
+
+        for resource_type, resource_info in provider_mappings.items():
+            for alias in self._generate_aliases(provider, resource_type, resource_info):
+                alias_index[self._normalize_alias(alias)] = resource_type
+
+        return alias_index
+
     def _matches_pattern(self, resource_type: str, pattern: str) -> bool:
         """
         Check if resource type matches a pattern.
@@ -177,7 +261,8 @@ def calculate(self, inventory: List[Dict]) -> Dict:
             provider = record.get('provider', 'unknown')
             account_id = record.get('account_id', '')
             region = record.get('region', '')
-            resource_type = record.get('resource_type', '')
+            original_resource_type = record.get('resource_type', '')
+            resource_type = self.normalize_resource_type(provider, original_resource_type)
             count = record.get('count', 0)
 
             category = self.get_category(provider, resource_type)
@@ -359,9 +444,10 @@ def print_summary(self, results: Dict) -> None:
         hybrid_units = results['summary'].get('hybrid_units', 0)
         iaas_count = results['summary']['totals'].get('IaaS', 0)
         paas_count = results['summary']['totals'].get('PaaS', 0)
+        paas_hru = math.ceil(paas_count / HYBRID_UNIT_RATIOS["PaaS"]) if paas_count > 0 else 0
         print("-" * 40)
         print(f"  {'HYBRID RESOURCE UNITS':20} {hybrid_units:>10}")
-        print(f"    (IaaS: {iaas_count} + PaaS: ceil({paas_count}/7))")
+        print(f"    (IaaS: {iaas_count} + PaaS: {paas_hru})")
 
         # Unmapped types notice
         unsupported = self.get_unsupported_types()

src/cli.py

@@ -8,6 +8,7 @@
 import json
 import logging
 import sys
+import csv
 from pathlib import Path
 from typing import Optional
 
@@ -24,6 +25,57 @@
 console = Console()
 
 
+def load_inventory_csv(csv_path: str, calculator, provider_filter: Optional[str] = None):
+    """Load inventory rows from a CSV file into the standard inventory schema."""
+    inventory = []
+
+    with open(csv_path, 'r', encoding='utf-8-sig', newline='') as f:
+        reader = csv.DictReader(f)
+        if not reader.fieldnames:
+            raise ValueError("CSV input is missing a header row.")
+
+        headers = {header.strip().lower(): header for header in reader.fieldnames if header}
+        required_headers = {'provider', 'resource_type', 'count'}
+        missing_headers = required_headers - set(headers)
+        if missing_headers:
+            missing_list = ', '.join(sorted(missing_headers))
+            raise ValueError(f"CSV input is missing required columns: {missing_list}")
+
+        for line_number, row in enumerate(reader, start=2):
+            provider = (row.get(headers['provider']) or '').strip().lower()
+            resource_type = (row.get(headers['resource_type']) or '').strip()
+            raw_count = (row.get(headers['count']) or '').strip()
+
+            if not provider or not resource_type or not raw_count:
+                raise ValueError(
+                    f"CSV row {line_number} must include provider, resource_type, and count."
+                )
+
+            if provider_filter and provider != provider_filter:
+                continue
+
+            try:
+                count = int(raw_count)
+            except ValueError as exc:
+                raise ValueError(
+                    f"CSV row {line_number} has a non-integer count: {raw_count}"
+                ) from exc
+
+            if count < 0:
+                raise ValueError(f"CSV row {line_number} has a negative count: {count}")
+
+            canonical_type = calculator.normalize_resource_type(provider, resource_type)
+            inventory.append({
+                'provider': provider,
+                'account_id': '',
+                'region': '',
+                'resource_type': canonical_type,
+                'count': count
+            })
+
+    return inventory
+
+
 def setup_logging(verbose: bool = False):
     """Configure logging with rich output."""
     # Configure our application logger only (not root logger)
@@ -166,8 +218,13 @@ def collect(
 
 
 @cli.command()
-@click.option('--input', '-i', 'input_path', required=True,
+@click.option('--input', '-i', 'input_path',
               help='Input inventory JSON file')
+@click.option('--csv-input', 'csv_input_path',
+              help='Input inventory CSV file with provider,resource_type,count headers')
+@click.option('--provider', '-p',
+              type=click.Choice(['aws', 'azure', 'gcp', 'oci']),
+              help='Optional provider filter for CSV input')
 @click.option('--output', '-o', default='license_summary.csv',
               help='Output summary CSV file')
 @click.option('--detailed', '-d', is_flag=True,
@@ -176,6 +233,8 @@ def collect(
               help='List resource types not mapped to license categories')
 def calculate(
     input_path: str,
+    csv_input_path: Optional[str],
+    provider: Optional[str],
     output: str,
     detailed: bool,
     show_unmapped: bool
@@ -188,19 +247,27 @@ def calculate(
       lm-cloud-inventory calculate -i inventory.json -o summary.csv
 
       lm-cloud-inventory calculate -i inventory.json -d --show-unmapped
+
+      lm-cloud-inventory calculate --csv-input inventory.csv
     """
     console.print("\n[bold blue]Calculating license requirements...[/bold blue]\n")
 
     try:
         from .calculator import LicenseCalculator
 
-        # Load inventory
-        with open(input_path, 'r', encoding='utf-8') as f:
-            inventory = json.load(f)
+        if bool(input_path) == bool(csv_input_path):
+            raise ValueError("Provide exactly one of --input or --csv-input.")
 
         # Create calculator with default config
         calculator = LicenseCalculator.from_config_files()
 
+        # Load inventory
+        if input_path:
+            with open(input_path, 'r', encoding='utf-8') as f:
+                inventory = json.load(f)
+        else:
+            inventory = load_inventory_csv(csv_input_path, calculator, provider_filter=provider)
+
         # Calculate
         results = calculator.calculate(inventory)
 
@@ -226,7 +293,8 @@ def calculate(
         console.print(f"\n[green]✓ Summary saved to {output}[/green]\n")
 
     except FileNotFoundError:
-        console.print(f"[red]Error: Input file not found: {input_path}[/red]")
+        missing_path = input_path or csv_input_path
+        console.print(f"[red]Error: Input file not found: {missing_path}[/red]")
         sys.exit(1)
 
     except Exception as e: