Europe Best Coworking Spaces: What No One Tells You

# coworking_netbench.py – Benchmark internet performance for coworking space validation
# Requires: pip install speedtest-cli requests
# Usage: python coworking_netbench.py --city "Berlin" --space "Factory Berlin" --iterations 5

import argparse
import json
import sys
import time
from datetime import datetime
from typing import Dict, List, Optional

try:
    import speedtest
except ImportError:
    print("ERROR: speedtest-cli not installed. Run: pip install speedtest-cli", file=sys.stderr)
    sys.exit(1)

try:
    import requests
except ImportError:
    print("ERROR: requests not installed. Run: pip install requests", file=sys.stderr)
    sys.exit(1)

# SLA thresholds for senior dev workloads (pair programming, CI/CD, video calls)
SLA_DOWNLOAD_MBPS = 100
SLA_UPLOAD_MBPS = 50
SLA_LATENCY_MS = 30
SLA_JITTER_MS = 10
SLA_PACKET_LOSS_PCT = 0.1

def run_speedtest() -> Optional[Dict]:
    """Run speedtest and return structured results, handle errors gracefully."""
    try:
        s = speedtest.Speedtest(secure=True)
        s.get_servers()
        s.get_best_server()
        s.download()
        s.upload()
        return s.results.dict()
    except speedtest.ConfigRetrievalError:
        print("ERROR: Failed to retrieve speedtest config. Check network connectivity.", file=sys.stderr)
        return None
    except speedtest.ServersRetrievalError:
        print("ERROR: No speedtest servers available. Try again later.", file=sys.stderr)
        return None
    except Exception as e:
        print(f"ERROR: Unexpected speedtest failure: {str(e)}", file=sys.stderr)
        return None

def check_packet_loss(target: str = "8.8.8.8", count: int = 10) -> float:
    """Check packet loss to target using ICMP pings (Unix-only, fallback to HTTP for Windows)."""
    import subprocess
    import platform
    packet_loss = 0.0
    try:
        if platform.system() != "Windows":
            result = subprocess.run(
                ["ping", "-c", str(count), target],
                capture_output=True,
                text=True,
                timeout=30
            )
            # Parse packet loss from ping output
            for line in result.stdout.split("\n"):
                if "packet loss" in line:
                    loss_str = line.split("%")[0].split(" ")[-1]
                    packet_loss = float(loss_str)
                    break
        else:
            # Fallback to HTTP HEAD requests for Windows (no native ping parsing)
            failures = 0
            for _ in range(count):
                try:
                    requests.head(f"http://{target}", timeout=2)
                except:
                    failures += 1
                time.sleep(0.1)
            packet_loss = (failures / count) * 100
    except Exception as e:
        print(f"WARNING: Packet loss check failed: {str(e)}", file=sys.stderr)
        packet_loss = -1.0  # Indicate check failure
    return packet_loss

def main():
    parser = argparse.ArgumentParser(description="Benchmark coworking space network performance")
    parser.add_argument("--city", required=True, help="City of the coworking space")
    parser.add_argument("--space", required=True, help="Name of the coworking space")
    parser.add_argument("--iterations", type=int, default=3, help="Number of speedtest iterations")
    parser.add_argument("--output", help="Output JSON file path")
    args = parser.parse_args()

    results = []
    print(f"Starting benchmark for {args.space} in {args.city} ({args.iterations} iterations)")

    for i in range(args.iterations):
        print(f"Iteration {i+1}/{args.iterations}...")
        speedtest_result = run_speedtest()
        if not speedtest_result:
            continue

        packet_loss = check_packet_loss()
        timestamp = datetime.utcnow().isoformat()

        result = {
            "timestamp": timestamp,
            "city": args.city,
            "space": args.space,
            "download_mbps": speedtest_result["download"] / 1_000_000,
            "upload_mbps": speedtest_result["upload"] / 1_000_000,
            "latency_ms": speedtest_result["ping"],
            "jitter_ms": speedtest_result.get("jitter", 0),
            "packet_loss_pct": packet_loss,
            "sla_compliant": (
                speedtest_result["download"] / 1_000_000 >= SLA_DOWNLOAD_MBPS and
                speedtest_result["upload"] / 1_000_000 >= SLA_UPLOAD_MBPS and
                speedtest_result["ping"] <= SLA_LATENCY_MS and
                speedtest_result.get("jitter", 0) <= SLA_JITTER_MS and
                (packet_loss <= SLA_PACKET_LOSS_PCT or packet_loss == -1.0)
            )
        }
        results.append(result)
        time.sleep(2)  # Avoid rate limiting

    # Calculate aggregates
    if results:
        avg_down = sum(r["download_mbps"] for r in results) / len(results)
        avg_up = sum(r["upload_mbps"] for r in results) / len(results)
        avg_lat = sum(r["latency_ms"] for r in results) / len(results)
        compliant_count = sum(1 for r in results if r["sla_compliant"])
        compliance_pct = (compliant_count / len(results)) * 100

        print(f"\n=== Benchmark Results for {args.space} ===")
        print(f"Average Download: {avg_down:.2f} Mbps (SLA: {SLA_DOWNLOAD_MBPS} Mbps)")
        print(f"Average Upload: {avg_up:.2f} Mbps (SLA: {SLA_UPLOAD_MBPS} Mbps)")
        print(f"Average Latency: {avg_lat:.2f} ms (SLA: {SLA_LATENCY_MS} ms)")
        print(f"SLA Compliance: {compliance_pct:.1f}% ({compliant_count}/{len(results)} iterations)")

        if args.output:
            with open(args.output, "w") as f:
                json.dump({"iterations": results, "aggregates": {
                    "avg_download_mbps": avg_down,
                    "avg_upload_mbps": avg_up,
                    "avg_latency_ms": avg_lat,
                    "compliance_pct": compliance_pct
                }}, f, indent=2)
            print(f"Results saved to {args.output}")
    else:
        print("ERROR: No valid benchmark results collected.", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

// powercheck.go – Validate power redundancy and UPS status in coworking spaces
// Build: go build -o powercheck powercheck.go
// Usage: ./powercheck --pdu1 192.168.1.100 --pdu2 192.168.1.101 --ups 192.168.1.102
// Requires: go get github.com/gosnmp/gosnmp/v2

package main

import (
    "encoding/json"
    "flag"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/gosnmp/gosnmp/v2"
)

const (
    // SNMP OIDs for APC UPS (standard MIB-II + APC proprietary)
    upsBatteryOID   = ".1.3.6.1.4.1.318.1.1.1.2.2.1.0" // Battery capacity %
    upsRuntimeOID   = ".1.3.6.1.4.1.318.1.1.1.2.2.3.0" // Runtime remaining in seconds
    pduStatusOID    = ".1.3.6.1.4.1.318.1.1.12.3.5.1.1.4.1" // PDU outlet status (1=on, 2=off)
    timeoutDuration = 5 * time.Second
)

// PowerStatus holds aggregated power redundancy results
type PowerStatus struct {
    PDU1Online       bool
    PDU2Online       bool
    Redundant        bool
    UPSBatteryPct    int
    UPSRuntimeMins   int
    LastChecked      time.Time
}

func checkSNMPDevice(ip string, oid string, community string) (int, error) {
    // Configure SNMP client
    params := &gosnmp.GoSNMP{
        Target:    ip,
        Port:      161,
        Community: community,
        Version:   gosnmp.Version2c,
        Timeout:   timeoutDuration,
        Retries:   2,
    }

    err := params.Connect()
    if err != nil {
        return -1, fmt.Errorf("failed to connect to %s: %w", ip, err)
    }
    defer params.Close()

    // Query OID
    result, err := params.Get([]string{oid})
    if err != nil {
        return -1, fmt.Errorf("failed to query OID %s on %s: %w", oid, ip, err)
    }

    if len(result.Variables) == 0 {
        return -1, fmt.Errorf("no variables returned from %s", ip)
    }

    // Parse integer value from result
    switch val := result.Variables[0].Value.(type) {
    case int:
        return val, nil
    case uint64:
        return int(val), nil
    case uint32:
        return int(val), nil
    default:
        return -1, fmt.Errorf("unexpected value type %T for OID %s", val, oid)
    }
}

func checkPDU(ip string, community string) (bool, error) {
    status, err := checkSNMPDevice(ip, pduStatusOID, community)
    if err != nil {
        return false, fmt.Errorf("PDU %s check failed: %w", ip, err)
    }
    // Status 1 = on, 2 = off, 3 = rebooting
    return status == 1, nil
}

func checkUPS(ip string, community string) (int, int, error) {
    batteryPct, err := checkSNMPDevice(ip, upsBatteryOID, community)
    if err != nil {
        return -1, -1, fmt.Errorf("UPS battery check failed: %w", err)
    }

    runtimeSec, err := checkSNMPDevice(ip, upsRuntimeOID, community)
    if err != nil {
        return batteryPct, -1, fmt.Errorf("UPS runtime check failed: %w", err)
    }

    return batteryPct, runtimeSec / 60, nil // Convert seconds to minutes
}

func main() {
    var (
        pdu1IP      string
        pdu2IP      string
        upsIP       string
        community   string
        outputJSON  string
    )

    flag.StringVar(&pdu1IP, "pdu1", "", "IP address of primary PDU (circuit 1)")
    flag.StringVar(&pdu2IP, "pdu2", "", "IP address of secondary PDU (circuit 2)")
    flag.StringVar(&upsIP, "ups", "", "IP address of UPS device")
    flag.StringVar(&community, "community", "public", "SNMP community string")
    flag.StringVar(&outputJSON, "output", "power_status.json", "Output JSON file path")
    flag.Parse()

    if pdu1IP == "" || pdu2IP == "" {
        log.Fatal("ERROR: Both --pdu1 and --pdu2 are required to check redundancy")
    }

    status := PowerStatus{
        LastChecked: time.Now(),
    }

    // Check PDU 1
    pdu1Online, err := checkPDU(pdu1IP, community)
    if err != nil {
        log.Printf("WARNING: %v", err)
        status.PDU1Online = false
    } else {
        status.PDU1Online = pdu1Online
        log.Printf("PDU 1 (%s) online: %v", pdu1IP, pdu1Online)
    }

    // Check PDU 2
    pdu2Online, err := checkPDU(pdu2IP, community)
    if err != nil {
        log.Printf("WARNING: %v", err)
        status.PDU2Online = false
    } else {
        status.PDU2Online = pdu2Online
        log.Printf("PDU 2 (%s) online: %v", pdu2IP, pdu2Online)
    }

    // Determine redundancy
    status.Redundant = status.PDU1Online && status.PDU2Online
    log.Printf("Power redundancy: %v", status.Redundant)

    // Check UPS if provided
    if upsIP != "" {
        battery, runtime, err := checkUPS(upsIP, community)
        if err != nil {
            log.Printf("WARNING: %v", err)
        } else {
            status.UPSBatteryPct = battery
            status.UPSRuntimeMins = runtime
            log.Printf("UPS battery: %d%%, runtime: %d minutes", battery, runtime)
        }
    } else {
        log.Printf("INFO: No UPS IP provided, skipping UPS check")
    }

    // Save results to JSON
    file, err := os.Create(outputJSON)
    if err != nil {
        log.Fatalf("ERROR: Failed to create output file: %v", err)
    }
    defer file.Close()

    encoder := json.NewEncoder(file)
    encoder.SetIndent("", "  ")
    if err := encoder.Encode(status); err != nil {
        log.Fatalf("ERROR: Failed to write JSON output: %v", err)
    }

    log.Printf("Results saved to %s", outputJSON)
}

// tco_calculator.ts – Calculate total cost of ownership for European coworking spaces
// Requires: npm install @types/node zod
// Usage: ts-node tco_calculator.ts --input space_config.json --team-size 4 --months 12

import { readFileSync, writeFileSync } from "fs";
import { parseArgs } from "util";
import { z } from "zod";

// Validation schema for coworking space configuration
const SpaceConfigSchema = z.object({
  name: z.string().min(1, "Space name is required"),
  city: z.string().min(1, "City is required"),
  membershipPlans: z.array(z.object({
    name: z.string(),
    monthlyCostEur: z.number().positive("Monthly cost must be positive"),
    includedDesks: z.number().int().positive(),
    meetingRoomHours: z.number().int().nonnegative(),
    printCredits: z.number().int().nonnegative(),
  })),
  hiddenCosts: z.object({
    parkingMonthlyEur: z.number().nonnegative(),
    coffeeMonthlyEur: z.number().nonnegative(),
    printingPerPageEur: z.number().nonnegative(),
    vatPct: z.number().min(0).max(100),
    meetingRoomHourlyEur: z.number().nonnegative(),
  }),
  teamUsage: z.object({
    avgPagesPerMonth: z.number().int().nonnegative(),
    meetingRoomHoursPerMonth: z.number().nonnegative(),
    parkingSpacesNeeded: z.number().int().nonnegative(),
  }),
});

type SpaceConfig = z.infer;

interface TCOResult {
  spaceName: string;
  city: string;
  planName: string;
  teamSize: number;
  months: number;
  membershipCostEur: number;
  hiddenCostsEur: number;
  vatEur: number;
  totalCostEur: number;
  costPerDeskPerMonthEur: number;
}

function calculateTCO(config: SpaceConfig, planName: string, teamSize: number, months: number): TCOResult {
  const plan = config.membershipPlans.find(p => p.name === planName);
  if (!plan) {
    throw new Error(`Plan ${planName} not found in space ${config.name}`);
  }

  // Validate team size fits plan
  const desksNeeded = Math.ceil(teamSize / 1.5); // Assume 1.5 people per desk for hot-desking
  if (plan.includedDesks < desksNeeded) {
    throw new Error(`Plan ${planName} only includes ${plan.includedDesks} desks, need ${desksNeeded}`);
  }

  // Calculate base membership cost
  const membershipCost = plan.monthlyCostEur * months;

  // Calculate hidden costs
  const parkingCost = config.hiddenCosts.parkingMonthlyEur * months * config.teamUsage.parkingSpacesNeeded;
  const coffeeCost = config.hiddenCosts.coffeeMonthlyEur * teamSize * months;
  const printingCost = config.hiddenCosts.printingPerPageEur * config.teamUsage.avgPagesPerMonth * months;
  const meetingRoomOverageHours = Math.max(0, config.teamUsage.meetingRoomHoursPerMonth - plan.meetingRoomHours);
  const meetingRoomCost = meetingRoomOverageHours * config.hiddenCosts.meetingRoomHourlyEur * months;

  const totalHiddenCosts = parkingCost + coffeeCost + printingCost + meetingRoomCost;

  // Calculate VAT
  const subtotal = membershipCost + totalHiddenCosts;
  const vat = subtotal * (config.hiddenCosts.vatPct / 100);

  const totalCost = subtotal + vat;

  return {
    spaceName: config.name,
    city: config.city,
    planName: plan.name,
    teamSize,
    months,
    membershipCostEur: membershipCost,
    hiddenCostsEur: totalHiddenCosts,
    vatEur: vat,
    totalCostEur: totalCost,
    costPerDeskPerMonthEur: totalCost / (desksNeeded * months),
  };
}

function main() {
  const { values } = parseArgs({
    options: {
      input: { type: "string", short: "i", demandOption: true },
      teamSize: { type: "number", short: "t", demandOption: true },
      months: { type: "number", short: "m", default: 12 },
      output: { type: "string", short: "o", default: "tco_results.json" },
    },
  });

  // Read and validate input config
  let config: SpaceConfig;
  try {
    const fileContent = readFileSync(values.input, "utf-8");
    const parsed = JSON.parse(fileContent);
    config = SpaceConfigSchema.parse(parsed);
    console.log(`Loaded config for ${config.name} in ${config.city}`);
  } catch (err) {
    if (err instanceof z.ZodError) {
      console.error("ERROR: Invalid config file:", err.errors);
    } else {
      console.error("ERROR: Failed to read config file:", err);
    }
    process.exit(1);
  }

  // Calculate TCO for each plan
  const results: TCOResult[] = [];
  for (const plan of config.membershipPlans) {
    try {
      const tco = calculateTCO(config, plan.name, values.teamSize, values.months);
      results.push(tco);
      console.log(`Plan ${plan.name}: Total €${tco.totalCostEur.toFixed(2)} (€${tco.costPerDeskPerMonthEur.toFixed(2)}/desk/month)`);
    } catch (err) {
      console.error(`ERROR calculating TCO for plan ${plan.name}:`, err);
    }
  }

  // Save results
  try {
    writeFileSync(values.output, JSON.stringify(results, null, 2));
    console.log(`Results saved to ${values.output}`);
  } catch (err) {
    console.error("ERROR: Failed to write output file:", err);
    process.exit(1);
  }

  // Print cheapest plan
  if (results.length > 0) {
    const cheapest = results.reduce((prev, curr) => prev.totalCostEur < curr.totalCostEur ? prev : curr);
    console.log(`\nCheapest plan: ${cheapest.planName} at €${cheapest.totalCostEur.toFixed(2)} total`);
  }
}

if (require.main === module) {
  main();
}

python coworking_netbench.py --city "Berlin" --space "Factory Berlin" --iterations 21 --output factory_berlin_peak.json

go build -o powercheck powercheck.go && ./powercheck --pdu1 10.0.0.100 --pdu2 10.0.0.101 --ups 10.0.0.102 --output factory_power.json

ts-node tco_calculator.ts --input factory_berlin_config.json --team-size 4 --months 12 --output factory_tco.json