Best Best Coworking Spaces Harvest in 2026: For Every Budget

#!/usr/bin/env python3
'''
Network benchmarking tool used to evaluate Harvest coworking space connectivity.
Tests download/upload speed, latency to major cloud providers, packet loss, and jitter.
Requires: iperf3, speedtest-cli, boto3 (optional for AWS tests)
Run: sudo python3 harvest_netbench.py --space 'Space Name' --output results.json
'''
import argparse
import json
import subprocess
import sys
import time
from datetime import datetime
from typing import Dict, List, Optional

# Constants for tested cloud endpoints
AWS_REGIONS = ['us-east-1a', 'us-central-1a']
AZURE_REGIONS = ['eastus', 'centralus']
IPERF3_SERVER = 'iperf3.deploy.cloud.hosting'  # Public iperf3 server hosted in Huntsville, AL

def run_subprocess(cmd: List[str], timeout: int = 30) -> Optional[str]:
    '''Run a shell command with error handling and timeout.'''
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=timeout
        )
        if result.returncode != 0:
            cmd_str = ' '.join(cmd)
            print(f'Warning: Command {cmd_str} failed: {result.stderr.strip()}', file=sys.stderr)
            return None
        return result.stdout.strip()
    except subprocess.TimeoutExpired:
        cmd_str = ' '.join(cmd)
        print(f'Warning: Command {cmd_str} timed out after {timeout}s', file=sys.stderr)
        return None
    except Exception as e:
        cmd_str = ' '.join(cmd)
        print(f'Error running {cmd_str}: {str(e)}', file=sys.stderr)
        return None

def test_speedtest() -> Dict:
    '''Run speedtest-cli and return parsed results.'''
    output = run_subprocess(['speedtest-cli', '--json', '--secure'])
    if not output:
        return {'error': 'speedtest-cli failed'}
    try:
        data = json.loads(output)
        return {
            'download_mbps': round(data['download'] / 1_000_000, 2),
            'upload_mbps': round(data['upload'] / 1_000_000, 2),
            'server': data['server']['name'],
            'latency_ms': data['ping']
        }
    except (json.JSONDecodeError, KeyError) as e:
        return {'error': f'Failed to parse speedtest output: {str(e)}'}

def test_iperf3(duration: int = 10) -> Dict:
    '''Run iperf3 test to public Huntsville server.'''
    output = run_subprocess([
        'iperf3', '-c', IPERF3_SERVER, '-t', str(duration), '-J'
    ])
    if not output:
        return {'error': 'iperf3 failed'}
    try:
        data = json.loads(output)
        return {
            'download_mbps': round(data['end']['sum_received']['bits_per_second'] / 1_000_000, 2),
            'upload_mbps': round(data['end']['sum_sent']['bits_per_second'] / 1_000_000, 2),
            'retransmits': data['end']['sum_sent']['retransmits'],
            'p99_latency_ms': round(data['end']['streams'][0]['sender']['p99_latency'] / 1000, 2)
        }
    except (json.JSONDecodeError, KeyError) as e:
        return {'error': f'Failed to parse iperf3 output: {str(e)}'}

def test_cloud_latency(regions: List[str], count: int = 10) -> Dict:
    '''Ping cloud regions to measure latency.'''
    results = {}
    for region in regions:
        # Map region to public endpoint
        if 'us-east-1' in region:
            endpoint = 'ec2.us-east-1.amazonaws.com'
        elif 'us-central-1' in region:
            endpoint = 'ec2.us-central-1.amazonaws.com'
        else:
            continue
        latencies = []
        for _ in range(count):
            output = run_subprocess(['ping', '-c', '1', '-W', '2', endpoint])
            if output:
                # Parse ping output for time=
                for line in output.split('
'):
                    if 'time=' in line:
                        lat = float(line.split('time=')[1].split(' ')[0])
                        latencies.append(lat)
        if latencies:
            sorted_lats = sorted(latencies)
            results[region] = {
                'min_ms': min(latencies),
                'max_ms': max(latencies),
                'avg_ms': round(sum(latencies)/len(latencies), 2),
                'p99_ms': sorted_lats[int(len(latencies)*0.99)]
            }
    return results

def main():
    parser = argparse.ArgumentParser(description='Harvest Coworking Network Benchmark')
    parser.add_argument('--space', required=True, help='Name of coworking space being tested')
    parser.add_argument('--output', default='netbench_results.json', help='Output JSON file')
    parser.add_argument('--duration', type=int, default=10, help='iperf3 test duration in seconds')
    args = parser.parse_args()

    print(f'Starting network benchmark for {args.space} at {datetime.now().isoformat()}')

    results = {
        'space': args.space,
        'timestamp': datetime.now().isoformat(),
        'speedtest': test_speedtest(),
        'iperf3': test_iperf3(args.duration),
        'cloud_latency': test_cloud_latency(AWS_REGIONS + AZURE_REGIONS)
    }

    with open(args.output, 'w') as f:
        json.dump(results, f, indent=2)
    print(f'Results saved to {args.output}')

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
'''
Power and charging benchmark for Harvest coworking spaces.
Measures AC outlet voltage, USB-C PD charging rate, UPS backup duration, and desk power capacity.
Requires: psutil, pyusb (optional for USB PD tests)
Run: sudo python3 harvest_powerbench.py --desk 12 --space 'The Launch Pad'
'''
import argparse
import json
import os
import subprocess
import sys
import time
from datetime import datetime
from typing import Dict, Optional

try:
    import psutil
except ImportError:
    print('Error: psutil is required. Install with pip install psutil', file=sys.stderr)
    sys.exit(1)

# PDO (Power Delivery Object) constants for USB-C PD testing
USBPD_VOLTAGES = [5, 9, 12, 15, 20]  # Standard USB-C PD voltages in volts
MIN_CHARGE_RATE_W = 45  # Minimum acceptable charging rate for a 2026 ultrabook

def run_subprocess(cmd: list, timeout: int = 10) -> Optional[str]:
    '''Run shell command with error handling.'''
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=timeout
        )
        if result.returncode != 0:
            cmd_str = ' '.join(cmd)
            print(f'Warning: {cmd_str} failed: {result.stderr.strip()}', file=sys.stderr)
            return None
        return result.stdout.strip()
    except subprocess.TimeoutExpired:
        cmd_str = ' '.join(cmd)
        print(f'Warning: {cmd_str} timed out', file=sys.stderr)
        return None
    except Exception as e:
        print(f'Error: {str(e)}', file=sys.stderr)
        return None

def get_ac_power_status() -> Dict:
    '''Get AC power status using psutil.'''
    try:
        battery = psutil.sensors_battery()
        if not battery:
            return {'error': 'No battery detected (desktop system?)'}
        return {
            'plugged_in': battery.power_plugged,
            'percent': battery.percent,
            'secs_left': battery.secsleft if battery.secsleft != psutil.POWER_TIME_UNLIMITED else -1
        }
    except Exception as e:
        return {'error': f'Failed to get battery status: {str(e)}'}

def test_usb_pd(desk_id: str) -> Dict:
    '''Test USB-C PD charging rate (requires uhubctl for USB hub control).'''
    # Check if uhubctl is installed
    if not run_subprocess(['which', 'uhubctl']):
        return {'error': 'uhubctl not installed, skipping USB PD test'}

    # Enable USB port to trigger PD negotiation
    run_subprocess(['uhubctl', '-a', 'on', '-l', '1-1'])  # Assume first USB hub
    time.sleep(2)  # Wait for PD negotiation

    # Read PD info from sysfs (Linux only)
    pd_path = '/sys/class/usb_power_delivery/usbpd0'
    if not os.path.exists(pd_path):
        return {'error': 'USB PD sysfs path not found'}

    try:
        with open(f'{pd_path}/sink_capability') as f:
            caps = f.read().strip().split('
')
        # Parse voltage and current from capability
        charge_w = 0
        for cap in caps:
            if 'Voltage:' in cap and 'Current:' in cap:
                v = float(cap.split('Voltage:')[1].split('V')[0].strip())
                i = float(cap.split('Current:')[1].split('A')[0].strip())
                w = v * i
                if w > charge_w:
                    charge_w = w
        return {
            'max_charge_w': round(charge_w, 2),
            'supports_20v': 20 in [v for v in USBPD_VOLTAGES if str(v) in str(caps)],
            'desk_id': desk_id
        }
    except Exception as e:
        return {'error': f'Failed to read USB PD info: {str(e)}'}

def test_ups_backup() -> Dict:
    '''Test UPS backup duration by unplugging AC (simulated via battery check).'''
    # Note: Real UPS test requires physically unplugging AC, this is a simulated check
    initial = get_ac_power_status()
    if initial.get('error'):
        return initial
    if not initial['plugged_in']:
        return {'error': 'Already unplugged, plug in AC first'}

    print('Unplug AC power now to test UPS backup (you have 10 seconds to unplug)...')
    time.sleep(10)
    after = get_ac_power_status()
    if after['plugged_in']:
        return {'error': 'AC power still connected, UPS test failed'}

    # Measure time until system suspends (simulated: wait 60s then check)
    start = time.time()
    time.sleep(60)
    final = get_ac_power_status()
    backup_secs = time.time() - start
    return {
        'ups_backup_secs': round(backup_secs, 2),
        'battery_percent_drop': initial['percent'] - final['percent']
    }

def main():
    parser = argparse.ArgumentParser(description='Harvest Coworking Power Benchmark')
    parser.add_argument('--space', required=True, help='Coworking space name')
    parser.add_argument('--desk', required=True, help='Desk number being tested')
    parser.add_argument('--output', default='powerbench_results.json', help='Output JSON file')
    args = parser.parse_args()

    print(f'Starting power benchmark for Desk {args.desk} at {args.space}')

    results = {
        'space': args.space,
        'desk_id': args.desk,
        'timestamp': datetime.now().isoformat(),
        'ac_power': get_ac_power_status(),
        'usb_pd': test_usb_pd(args.desk),
        'ups_backup': test_ups_backup()
    }

    with open(args.output, 'w') as f:
        json.dump(results, f, indent=2)
    print(f'Results saved to {args.output}')

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
'''
GPU availability and performance benchmark for Harvest coworking spaces.
Checks for dedicated GPUs, measures inference speed for 7B LLMs, and VRAM availability.
Requires: torch, transformers, nvidia-ml-py3
Run: python3 harvest_gpubench.py --space 'TechHub Harvest' --model 'mistral-7b-v0.3'
'''
import argparse
import json
import subprocess
import sys
import time
from datetime import datetime
from typing import Dict, Optional

try:
    import torch
    from transformers import AutoTokenizer, AutoModelForCausalLM
except ImportError:
    print('Error: torch and transformers required. Install with pip install torch transformers', file=sys.stderr)
    sys.exit(1)

try:
    import pynvml
    pynvml.nvmlInit()
    NVML_AVAILABLE = True
except ImportError:
    print('Warning: pynvml not installed, limited GPU info available', file=sys.stderr)
    NVML_AVAILABLE = False
except Exception as e:
    print(f'Warning: NVML init failed: {str(e)}', file=sys.stderr)
    NVML_AVAILABLE = False

def run_subprocess(cmd: list, timeout: int = 30) -> Optional[str]:
    '''Run shell command with error handling.'''
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=timeout
        )
        if result.returncode != 0:
            cmd_str = ' '.join(cmd)
            print(f'Warning: {cmd_str} failed: {result.stderr.strip()}', file=sys.stderr)
            return None
        return result.stdout.strip()
    except subprocess.TimeoutExpired:
        cmd_str = ' '.join(cmd)
        print(f'Warning: {cmd_str} timed out', file=sys.stderr)
        return None
    except Exception as e:
        print(f'Error: {str(e)}', file=sys.stderr)
        return None

def get_gpu_info() -> Dict:
    '''Get GPU info using nvidia-smi or torch.'''
    info = {}
    # Try nvidia-smi first
    smi_output = run_subprocess(['nvidia-smi', '--query-gpu=name,memory.total,memory.free,utilization.gpu', '--format=csv,noheader,nounits'])
    if smi_output:
        parts = smi_output.split(', ')
        info = {
            'name': parts[0],
            'total_vram_mb': int(parts[1]),
            'free_vram_mb': int(parts[2]),
            'utilization_pct': int(parts[3])
        }
    elif torch.cuda.is_available():
        # Fallback to torch
        info = {
            'name': torch.cuda.get_device_name(0),
            'total_vram_mb': torch.cuda.get_device_properties(0).total_mem // (1024 ** 2),
            'free_vram_mb': torch.cuda.mem_get_info()[0] // (1024 ** 2),
            'utilization_pct': 0  # Torch doesn't expose utilization directly
        }
    else:
        info['error'] = 'No CUDA GPU detected'
    return info

def test_llm_inference(model_name: str, prompt: str = 'Write a Python function to sort a list of integers.') -> Dict:
    '''Test 7B LLM inference speed on available GPU.'''
    if not torch.cuda.is_available():
        return {'error': 'No GPU available for inference'}

    try:
        print(f'Loading model {model_name}...')
        tokenizer = AutoTokenizer.from_pretrained(model_name)
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map='auto'
        )

        inputs = tokenizer(prompt, return_tensors='pt').to('cuda')
        # Warmup run
        model.generate(**inputs, max_new_tokens=10)

        # Timed run
        start = time.time()
        outputs = model.generate(**inputs, max_new_tokens=50)
        elapsed = time.time() - start

        decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
        return {
            'model': model_name,
            'inference_time_sec': round(elapsed, 2),
            'tokens_per_sec': round(50 / elapsed, 2),
            'output_preview': decoded[:200]
        }
    except Exception as e:
        return {'error': f'Inference failed: {str(e)}'}

def main():
    parser = argparse.ArgumentParser(description='Harvest Coworking GPU Benchmark')
    parser.add_argument('--space', required=True, help='Coworking space name')
    parser.add_argument('--model', default='mistralai/Mistral-7B-v0.3', help='7B LLM model to test')
    parser.add_argument('--output', default='gpubench_results.json', help='Output JSON file')
    args = parser.parse_args()

    print(f'Starting GPU benchmark for {args.space} at {datetime.now().isoformat()}')

    results = {
        'space': args.space,
        'timestamp': datetime.now().isoformat(),
        'gpu_info': get_gpu_info(),
        'llm_inference': test_llm_inference(args.model)
    }

    with open(args.output, 'w') as f:
        json.dump(results, f, indent=2)
    print(f'Results saved to {args.output}')

if __name__ == '__main__':
    main()

ping -c 100 ec2.us-east-1.amazonaws.com | grep 'packet loss'

cat /sys/class/power_supply/BAT0/power_now | awk '{print $1/1000000 "W"}'

nvidia-smi --query-gpu=memory.free --format=csv