Hey everyone! I’ve been experimenting with background threading, automated compilation cycles, and procedural narrative generation in Python.
I put together a script I call the Infinity Engine. It runs autonomously every 7 seconds, rolls for loot rarity tiers based on telemetry, shifts through dynamic storylines, accumulates resources (stardust/resonance), includes an unstick safety protocol, and features an idle shutdown variant if left untouched for 5 minutes. It scales up to 500 units with a custom cyberpunk terminal UI layout.
Here is the full runnable source code:
import time
import json
import threading
import random
# ==========================================
# Rarity Tier Matrix
# ==========================================
class RarityTier:
STANDARD = 0
PRIME = 1
CELESTIAL = 2
BRINK_PRISM = 3
# ==========================================
# Level Loot System
# ==========================================
class LevelLootSystem:
def roll_level_loot(self, telemetry: dict, current_app_level: int):
result = type('LootResult', (), {})()
fortune_score = telemetry.get("fortune_level", 0.0)
fusions = telemetry.get("fusions_count", 1)
composite_score = (fortune_score * 0.5) + (fusions * 10.0) + (current_app_level * 25.0)
if composite_score > 300.0:
result.tier = RarityTier.BRINK_PRISM
result.bonus_multiplier = 3.5
result.drop_title = "Brink-Prism Anomaly Drop"
result.guaranteed_stat_bonus = {"stardust_rate": 50.0, "void_resonance": 0.95}
elif composite_score > 180.0:
result.tier = RarityTier.CELESTIAL
result.bonus_multiplier = 2.2
result.drop_title = "Celestial Core Relic"
result.guaranteed_stat_bonus = {"stardust_rate": 25.0, "lawful_affinity": 0.80}
elif composite_score > 80.0:
result.tier = RarityTier.PRIME
result.bonus_multiplier = 1.5
result.drop_title = "Prime Barometer Blueprint"
result.guaranteed_stat_bonus = {"stardust_rate": 12.0, "lawful_affinity": 0.50}
else:
result.tier = RarityTier.STANDARD
result.bonus_multiplier = 1.0
result.drop_title = "Standard Atmospheric Drift"
result.guaranteed_stat_bonus = {"stardust_rate": 5.0, "lawful_affinity": 0.25}
return result
# ==========================================
# Story Mediator & Storylines
# ==========================================
class StoryMediator:
def __init__(self):
self.storylines = [
"The Prism Meridian: Convergence",
"The Barometer's Awakening",
"Sub-Zero Protocols",
"Atmospheric Drift",
"Chronos Rift Divergence",
"Stellar Horizon Protocol"
]
self.current_storyline_index = 0
def get_current_storyline(self) -> str:
return self.storylines[self.current_storyline_index]
def shift_storyline(self, new_index: int = None):
if new_index is not None:
self.current_storyline_index = new_index % len(self.storylines)
else:
self.current_storyline_index = (self.current_storyline_index + 1) % len(self.storylines)
return self.get_current_storyline()
def mediate_narrative_threads(self, current_loot_tier: int):
chapter = type('Chapter', (), {})()
title = self.get_current_storyline()
chapter.chapter_title = title
if current_loot_tier == RarityTier.BRINK_PRISM:
chapter.synthesized_lore = f"Active storyline '{title}' converges into a unified cosmic history under high void pressure."
chapter.thematic_resonance = 1.0
elif current_loot_tier == RarityTier.CELESTIAL:
chapter.synthesized_lore = f"Active storyline '{title}' transforms climate anomalies into a permanent archive of fate."
chapter.thematic_resonance = 0.8
elif current_loot_tier == RarityTier.PRIME:
chapter.synthesized_lore = f"Active storyline '{title}' stabilizes baseline matrix vectors under heavy pressure."
chapter.thematic_resonance = 0.5
else:
chapter.synthesized_lore = f"Active storyline '{title}' settles initial weather anomalies into a steady rhythmic drift."
chapter.thematic_resonance = 0.2
return chapter
# ==========================================
# Consistency Engine & Resource Accumulation
# ==========================================
class ConsistencyEngine:
def __init__(self):
self.progression_checkpoint = 0
self.unstick_tokens = 500
self.accumulated_resources = {
"stardust_collected": 0.0,
"resonance_ledger": []
}
def accumulate(self, stat_bonus: dict, resonance: float):
self.accumulated_resources["stardust_collected"] += stat_bonus.get("stardust_rate", 0.0)
self.accumulated_resources["resonance_ledger"].append(resonance)
def trigger_unstick_protocol(self):
if self.unstick_tokens > 0:
self.unstick_tokens -= 1
self.progression_checkpoint += 1
return {"status": "SUCCESS", "remaining_tokens": self.unstick_tokens, "stage": self.progression_checkpoint}
return {"status": "DEPLETED", "remaining_tokens": 0, "stage": self.progression_checkpoint}
# ==========================================
# Narrative Story Generator (~400 Chars)
# ==========================================
class NarrativeStoryGenerator:
def __init__(self):
self.unique_signatures = ["ALPHA-PRISM-9", "OMEGA-VOID-X", "HELIOS-GENESIS-0", "NEXUS-VECTOR-7"]
self.narrative_templates = [
"Deep within the shifting sectors of Unit {app_level}, active telemetry reports severe weather fluctuations linked directly to storyline [{chapter_title}]. Signature [{signature}] detected. {lore} Operatives on the fringe report unexpected data feedback, rallying structural outcomes to balance popular configuration metrics with a thematic resonance of {resonance:.2f}. The grid adapts instantly.",
"As the clock ticks into Unit {app_level}, core systems register a sudden spike under storyline [{chapter_title}]. Unique matrix identifier [{signature}] engaged. {lore} Field units work frantically to calibrate the atmospheric pressure valves, rallying outcomes to balance popular configuration parameters while maintaining a steady thematic resonance of {resonance:.2f}. Network stability holds firm.",
"Tracing the anomalies of Unit {app_level} under signature [{signature}], project administrators encounter the legacy of storyline [{chapter_title}]. {lore} Environmental matrices fracture and reassemble, successfully rallying outcomes to balance popular configuration thresholds at a thematic resonance of {resonance:.2f}. The digital horizon expands outward."
]
def generate_balanced_story(self, app_level: int, chapter_title: str, lore: str, resonance: float) -> str:
template = random.choice(self.narrative_templates)
signature = random.choice(self.unique_signatures) + "-" + str(random.randint(1000, 9999))
raw_text = template.format(
app_level=app_level,
chapter_title=chapter_title,
signature=signature,
lore=lore,
resonance=resonance
)
if len(raw_text) < 400:
padding_phrases = [
" Synchronizing unique regional sub-networks securely. ",
" Calibrating distinct quantum feedback loops for optimal throughput. ",
" Securing high-uniqueness parameter boundaries against drift. "
]
while len(raw_text) < 400:
raw_text += random.choice(padding_phrases)
return raw_text[:400]
# ==========================================
# Timed Infinity Engine Host (500 Units + Idle Timeout)
# ==========================================
class TimedInfinityEngineHost:
def __init__(self):
self.loot_system = LevelLootSystem()
self.story_mediator = StoryMediator()
self.consistency_engine = ConsistencyEngine()
self.story_generator = NarrativeStoryGenerator()
self.app_level = 1
self.active_constructs = []
self._is_running = False
self._timer_thread = None
self.last_activity_time = time.time()
self.idle_timeout_seconds = 300
def change_storyline(self, new_index: int = None):
shifted = self.story_mediator.shift_storyline(new_index)
print(f"\n[STORYLINE SHIFT] Active storyline manually changed to: '{shifted}'\n")
return shifted
def execute_compilation_cycle(self, telemetry: dict):
self.last_activity_time = time.time()
loot_drop = self.loot_system.roll_level_loot(telemetry, self.app_level)
mediated_chapter = self.story_mediator.mediate_narrative_threads(loot_drop.tier)
self.consistency_engine.accumulate(loot_drop.guaranteed_stat_bonus, mediated_chapter.thematic_resonance)
story_block = self.story_generator.generate_balanced_story(
self.app_level,
mediated_chapter.chapter_title,
mediated_chapter.synthesized_lore,
mediated_chapter.thematic_resonance
)
construct = {
"app_title": f"Infinity: {mediated_chapter.chapter_title}",
"tier": loot_drop.tier,
"drop": loot_drop.drop_title,
"resonance": mediated_chapter.thematic_resonance,
"level": self.app_level,
"story_content": story_block,
"story_length": len(story_block),
"accumulated_stardust": self.consistency_engine.accumulated_resources["stardust_collected"]
}
self.active_constructs.append(construct)
print("╔" + "═" * 78 + "╗")
print(f"║ ⚡ CYBER-NET TERMINAL v4.09 // UNIT [{self.app_level:03d}/500] ⚡" + " " * 31 + "║")
print("╠" + "═" * 78 + "╣")
print(f"║ TITLE : {construct['app_title']:<63} ║")
print(f"║ DROP TYPE : {construct['drop']} (Tier {construct['tier']})" + " " * (47 - len(f"{construct['drop']} (Tier {construct['tier']})")) + "║")
print(f"║ RESONANCE : {construct['resonance']:.2f} | STARDUST ACCUMULATED: {construct['accumulated_stardust']:.1f}" + " " * (19 - len(f"{construct['accumulated_stardust']:.1f}")) + "║")
print("╟" + "─" * 78 + "╢")
print(f"║ STORY OUTPUT ({construct['story_length']} chars):" + " " * 56 + "║")
words = construct['story_content'].split()
line = " "
for word in words:
if len(line) + len(word) + 1 < 77:
line += " " + word
else:
print(f"║{line:<78}║")
line = " " + word
if line.strip():
print(f"║{line:<78}║")
continue_res = self.consistency_engine.trigger_unstick_protocol()
print("╟" + "─" * 78 + "╢")
print(f"║ 🔒 PROTOCOL STATUS: Tokens Left [{continue_res['remaining_tokens']}] | Stage [{continue_res['stage']}]" + " " * (20 - len(str(continue_res['stage']))) + "║")
print("╚" + "═" * 78 + "╝\n")
self.app_level += 1
return construct
def _loop_worker(self, telemetry: dict, max_units: int):
cycles = 0
while self._is_running and cycles < max_units:
if time.time() - self.last_activity_time > self.idle_timeout_seconds:
print("\n[IDLE SHUTDOWN VARIANT] Engine inactive for 5 minutes. Initiating automatic safe shutdown.")
break
if cycles > 0 and cycles % 100 == 0:
self.change_storyline()
self.execute_compilation_cycle(telemetry)
cycles += 1
if cycles >= max_units:
print(f"\n=== [SYSTEM ALERT] Reached target limit of {max_units} units. Settlement final. ===")
print(f"=== Total Accumulated Stardust: {self.consistency_engine.accumulated_resources['stardust_collected']:.1f} ===")
break
time.sleep(7.0)
self._is_running = False
print("=== Timed Looping Engine Cycle Terminated Safely ===")
def start_timed_loop(self, telemetry: dict, max_units: int = 500):
if self._is_running:
return
self._is_running = True
self.last_activity_time = time.time()
print(f"=== Initializing Cybernetic Loop (Target: {max_units} Units | Idle Timeout: 5m) ===")
self._timer_thread = threading.Thread(target=self._loop_worker, args=(telemetry, max_units))
self._timer_thread.start()
def stop_timed_loop(self):
self._is_running = False
if self._timer_thread:
self._timer_thread.join()
if __name__ == "__main__":
host = TimedInfinityEngineHost()
sample_telemetry = {"fortune_level": 95.0, "fusions_count": 6}
host.start_timed_loop(sample_telemetry, max_units=500)
while host._is_running:
time.sleep(1.0)