# GPlay Central AI — Python SDK Integration Blueprint Panduan integrasi resmi GPlay Central AI Engine untuk ekosistem **Python (FastAPI / Django / Flask / Data Science)**. --- ## 1. Instalasi Dependensi ```bash pip install requests pydantic ``` --- ## 2. Kredensial Environment (`.env`) ```env GPLAY_AI_BASE_URL=http://localhost:5000/api/v1 GPLAY_AI_API_KEY=gplay_kunci_api_tenant_anda ``` --- ## 3. Official Python Client (`gplay_client.py`) ```python import os import requests from typing import Dict, Any, Optional class GPlayAiClient: def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None): self.api_key = api_key or os.getenv("GPLAY_AI_API_KEY", "") self.base_url = base_url or os.getenv("GPLAY_AI_BASE_URL", "http://localhost:5000/api/v1") self.headers = { "X-API-KEY": self.api_key, "Content-Type": "application/json" } def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}" response = requests.post(url, json=payload, headers=self.headers) response.raise_for_status() return response.json() # --- Phase 8: Next-Gen AI Modules --- def code_assistant(self, source_code: str, language: str = "python", task: str = "audit_and_unit_test") -> Dict[str, Any]: """AI Coding Assistant, Bug Scanner & Unit Test Auto-Generator""" return self._post("/devops/code-assistant", { "sourceCode": source_code, "language": language, "task": task }) def threat_analyzer(self, target_endpoint: str, traffic_log: str, attack_vector: str = "DDoS_AND_RANSOMWARE") -> Dict[str, Any]: """IT System Security & CyberDefense Threat Analyzer""" return self._post("/security/threat-analyzer", { "targetEndpoint": target_endpoint, "trafficLog": traffic_log, "attackVector": attack_vector }) def analyze_video_safety(self, stream_url: str, location: str = "Factory Floor", rule: str = "K3_PPE_FIRE") -> Dict[str, Any]: """Multimodal Video Analytics & CCTV HSE Safety Inspector""" return self._post("/vision/analyze-video", { "videoStreamUrl": stream_url, "location": location, "safetyRule": rule }) def speech_intelligence(self, audio_url: str, agent_name: str, phone: str, language: str = "id-ID") -> Dict[str, Any]: """Voice Intelligence & Call Center Speech Quality Engine""" return self._post("/audio/speech-intelligence", { "audioUrl": audio_url, "agentName": agent_name, "customerPhone": phone, "language": language }) def predict_turnover(self, employee_id: str, dept: str, tenure: int, overtime_hrs: int) -> Dict[str, Any]: """Predictive HR Attrition & Employee Retention AI""" return self._post("/hr/predict-turnover", { "employeeId": employee_id, "department": dept, "tenureMonths": tenure, "recentOvertimeHours": overtime_hrs }) def dispute_analyzer(self, contract_id: str, title: str, expiry_date: str) -> Dict[str, Any]: """Legal Contract Expiry & Dispute Settlement AI""" return self._post("/legal/dispute-analyzer", { "contractId": contract_id, "contractTitle": title, "expiryDate": expiry_date }) def detect_ai_plagiarism(self, text: str, sensitivity: float = 0.85) -> Dict[str, Any]: """AI Content & Plagiarism Detector Engine (Milestone 83)""" return self._post("/content/detect-ai-plagiarism", { "text": text, "sensitivity": sensitivity }) def audit_smart_contract(self, source_code: str, language: str = "solidity") -> Dict[str, Any]: """Web3 & Smart Contract Auditor Engine (Milestone 84)""" return self._post("/web3/audit-smart-contract", { "sourceCode": source_code, "language": language }) def send_slack_alert(self, channel: str, event_title: str, alert_level: str = "critical", payload: dict = None) -> Dict[str, Any]: """Enterprise Workspace Slack/Teams AI Alerting (Milestone 85)""" return self._post("/workspace/slack-notifier", { "channel": channel, "eventTitle": event_title, "alertLevel": alert_level, "payload": payload or {} }) def detect_api_anomaly(self, api_key_id: str, request_rate_per_min: int, endpoint_target: str = "/api/v1/data/query") -> Dict[str, Any]: """Real-Time API Usage Anomaly Detector (Milestone 91)""" return self._post("/security/api-anomaly-detector", { "apiKeyId": api_key_id, "requestRatePerMin": request_rate_per_min, "endpointTarget": endpoint_target }) def shield_prompt_injection(self, prompt: str, context_documents: list = None) -> Dict[str, Any]: """Real-Time Indirect Prompt Injection Shield (Milestone 104)""" return self._post("/guardrails/prompt-injection-shield", { "prompt": prompt, "contextDocuments": context_documents or [] }) def check_invoice_tamper(self, document_hash: str, e_meterai_serial: str) -> Dict[str, Any]: """Deep Document Forgery & Digital Stamp Authenticator (Milestone 114)""" return self._post("/finsec/invoice-tamper-check", { "documentHash": document_hash, "eMeteraiSerial": e_meterai_serial }) # --- Category N: AIDC Power Infrastructure & BESS Suite --- def optimize_aidc_power(self, it_compute_load_kw: float, cooling_facility_load_kw: float, cooling_method: str = "DIRECT_LIQUID_COOLING", facility_name: str = "Jakarta AIDC") -> Dict[str, Any]: """AIDC Power Infrastructure & PUE/Thermal Optimizer (Milestone 115)""" return self._post("/energy/aidc-power-optimizer", { "itComputeLoadKw": it_compute_load_kw, "coolingFacilityLoadKw": cooling_facility_load_kw, "coolingMethod": cooling_method, "facilityName": facility_name }) def manage_bess_microgrid(self, cluster_id: str, current_soc_pct: float, target_discharge_kw: float = 0.0, grid_tariff_mode: str = "PEAK_HOUR") -> Dict[str, Any]: """Grid-Scale BESS & Smart Micro-Grid EMS Controller (Milestone 116)""" return self._post("/energy/bess-microgrid-ems", { "clusterId": cluster_id, "currentSocPct": current_soc_pct, "targetDischargeKw": target_discharge_kw, "gridTariffMode": grid_tariff_mode }) def get_hardware_supply_chain(self, query_category: str = "BESS_BATTERY_STORAGE", include_patents: bool = True) -> Dict[str, Any]: """Central Bank AI Hardware Supply Chain KB (Milestone 117)""" return self._post("/intel/hardware-supply-chain", { "queryCategory": query_category, "includePatents": include_patents }) def ingest_power_telemetry(self, device_id: str, device_source: str = "PVSPECTRA_SOLAR", voltage_v: float = 480.0, current_a: float = 84.5, thd_noise_pct: float = 1.8, noise_attenuation_pct: float = 91.8) -> Dict[str, Any]: """Industrial IoT Edge Power & PVSpectra Dual-Probe Telemetry (Milestone 118 - IMPI No. 5960)""" return self._post("/iot/server-power-telemetry", { "deviceId": device_id, "deviceSource": device_source, "voltageV": voltage_v, "currentA": current_a, "thdNoisePct": thd_noise_pct, "noiseAttenuationPct": noise_attenuation_pct }) ``` --- ## 4. Contoh Penggunaan ```python from gplay_client import GPlayAiClient client = GPlayAiClient(api_key="gplay_YOUR_SECRET_KEY") # 1. Audit Source Code audit_res = client.code_assistant( source_code="def query_user(id): return db.execute(f'SELECT * FROM users WHERE id={id}')", language="python" ) print("Code Audit Result:", audit_res["data"]["codeAudit"]) # 2. Prediksi Retensi Karyawan hr_res = client.predict_turnover( employee_id="EMP-9021", dept="AI Research", tenure=24, overtime_hrs=40 ) print("Flight Risk Score:", hr_res["data"]["predictiveAttrition"]["flightRiskScorePercentage"], "%") ```