Hier ist jetzt einmal das komplette Script mit Datentypen in "calc_1". Ich habe die Datenmenge durch eine Anpassung in Zeile 121 auch etwas reduziert, aber es hält nach wie vor nicht an und "0" ist die einzige Ausgabe:
Code: Alles auswählen
import json
import csv
import math
import os
import psutil
import inspect
systems: list = []
systems_dic: dict = {}
stats: list = []
stats_dic: dict = {}
listings: list = []
listings_dic: dict = {}
def print_mem_info():
curframe = inspect.currentframe()
calframe = inspect.getouterframes(curframe, 2)
print("caller:", calframe[1][3])
print("systems, stats, listings")
print(type(systems), len(systems), type(systems_dic), len(systems_dic), sep=" , ")
print(type(stats), len(stats), type(stats_dic), len(stats_dic), sep=" , ")
print(
type(listings), len(listings), type(listings_dic), len(listings_dic), sep=" , "
)
mybytes = psutil.Process(os.getpid()).memory_info()[0]
print(mybytes / (1024.0 * 1024.0), "mb", mybytes, "b")
print()
def get_id_dict(d: dict):
mid = d["id"]
if mid is not None:
try:
return int(mid)
except ValueError:
pass
print("ValueError:", d)
return None
def get_id_list(l: list, i: int = 0):
mid = l[i]
if mid is not None:
try:
return int(mid)
except ValueError:
pass
print("ValueError:", l)
return None
def parse_systems():
global systems, systems_dic
with open("systems_populated.json") as f:
systems = json.load(f)
for s in systems:
n = get_id_dict(s)
if n is not None:
systems_dic[n] = s
print_mem_info()
headers: list
with open("systems_recently.csv") as f:
csv_reader = csv.reader(f, delimiter=",")
i: int = 0
for row in csv_reader:
if i == 0:
headers = row
else:
n = get_id_list(row)
if n is not None and n not in systems_dic:
e = dict(zip(headers, row))
systems.append(e)
systems_dic[n] = e
i += 1
print_mem_info()
def get_dist_to_sun(s: dict):
try:
x = s["x"]
y = s["y"]
z = s["z"]
if None not in (x, y, z):
x = float(x)
y = float(y)
z = float(z)
tosun: float = math.sqrt(x * x + y * y + z * z)
return tosun
except ValueError:
print("ValueError:", s)
return -1.0
def get_dist(s1: dict, s2: dict):
try:
x = s1["x"]
y = s1["y"]
z = s1["z"]
a = s2["x"]
b = s2["y"]
c = s2["z"]
if None not in (x, y, z, a, b, c):
x = float(x)
y = float(y)
z = float(z)
a = float(a)
b = float(b)
c = float(c)
dist: float = math.sqrt(
math.pow(x - a, 2) + math.pow(y - b, 2) + math.pow(z - c, 2)
)
return dist
except ValueError:
print("ValueError:", s1, s2)
return -1.0
def grep_systems():
global systems, systems_dic
systems = list(filter(lambda s: 0 <= get_dist_to_sun(s) <= 50, systems))
systems_dic.clear()
for s in systems:
n = get_id_dict(s)
systems_dic[n] = s
print_mem_info()
def parse_stats():
global stats, stats_dic
with open("stations.json") as f:
stats = json.load(f)
print_mem_info()
stats = list(filter(lambda sta: sta["system_id"] in systems_dic, stats))
stats_dic.clear()
for s in stats:
n = get_id_dict(s)
stats_dic[n] = s
print_mem_info()
def grep_stats():
global systems, systems_dic, stats, stats_dic
stats = list(
filter(
lambda sta: sta["max_landing_pad_size"] == "L"
and sta["distance_to_star"] is not None
and 0 < int(sta["distance_to_star"]) <= 5000,
stats,
)
)
stats_dic.clear()
for s in stats:
n = get_id_dict(s)
stats_dic[n] = s
myset: set = set()
systems.clear()
for s in stats:
myid = s["system_id"]
if myid not in myset:
myset.add(myid)
systems.append(systems_dic[myid])
systems_dic.clear()
for s in systems:
n = get_id_dict(s)
systems_dic[n] = s
print_mem_info()
def parse_listings():
global listings
headers: list
with open("listings.csv") as f:
csv_reader = csv.reader(f, delimiter=",")
i: int = 0
for row in csv_reader:
if i == 0:
headers = row
else:
e = dict(zip(headers, row))
listings.append(e)
i += 1
print_mem_info()
myset: set = set()
mylist: list = []
for l in listings:
if l["id"] not in myset:
myset.add(l["id"])
mylist.append(l)
listings = mylist
print_mem_info()
def grep_listings():
global systems, systems_dic, stats, stats_dic, listings, listings_dic
for l in listings:
myid = int(l["station_id"])
if myid in stats_dic:
if myid not in listings_dic:
listings_dic[myid] = [l]
else:
listings_dic[myid].append(l)
print_mem_info()
listings.clear()
print_mem_info()
stats = list(
filter(
lambda sta: sta["id"] in listings_dic,
stats,
)
)
stats_dic.clear()
for s in stats:
n = get_id_dict(s)
stats_dic[n] = s
myset: set = set()
systems.clear()
for s in stats:
myid = int(s["system_id"])
if myid not in myset:
myset.add(myid)
systems.append(systems_dic[myid])
systems_dic.clear()
for s in systems:
n = get_id_dict(s)
systems_dic[n] = s
print_mem_info()
def get_best_listing(a: list, b: list):
max_pro = 0
max_l1 = None
max_l2 = None
for l1 in a:
for l2 in b:
cid1 = int(l1["commodity_id"])
cid2 = int(l2["commodity_id"])
buyp = int(l1["buy_price"])
sellp = int(l2["sell_price"])
if cid1 == cid2 and buyp > 0 and sellp > buyp:
if sellp - buyp > max_pro:
max_pro = sellp - buyp
max_l1 = l1
max_l2 = l2
if max_pro > 0:
return (max_l1, max_l2, max_pro)
return None
def get_best_loop(a: list, b: list):
max_a = get_best_listing(a, b)
max_b = get_best_listing(b, a)
if max_a is not None and max_b is not None:
return (max_a, max_b, max_a[2] + max_b[2])
return None
def calc_1():
max: int = 0
max_loop: set = None
keys: list = list(listings_dic.keys())
for i in range(len(keys)):
key_1: int = keys[i]
a: list = listings_dic[key_1]
b: dict = stats_dic[key_1]
c: dict = systems_dic[b["system_id"]]
if i % 100 == 0:
print(i)
for j in range(i + 1, len(keys)):
key_2: int = keys[j]
a2: list = listings_dic[key_2]
b2: dict = stats_dic[key_2]
c2: dict = systems_dic[b2["system_id"]]
l: set = get_best_loop(a, a2)
if l is not None and l[2] > max:
max = l[2]
max_loop = l
print(max_loop)
print(systems_dic[stats_dic[max_loop[0][0]["station_id"]]["system_id"]]["name"])
print(stats_dic[max_loop[0][0]["station_id"]]["name"])
print(systems_dic[stats_dic[max_loop[0][1]["station_id"]]["system_id"]]["name"])
print(stats_dic[max_loop[0][1]["station_id"]]["name"])
print(systems_dic[stats_dic[max_loop[1][0]["station_id"]]["system_id"]]["name"])
print(stats_dic[max_loop[1][0]["station_id"]]["name"])
print(systems_dic[stats_dic[max_loop[1][1]["station_id"]]["system_id"]]["name"])
print(stats_dic[max_loop[1][1]["station_id"]]["name"])
print(max_loop[0][0]["commodity_id"])
print(max_loop[0][1]["commodity_id"])
print(max_loop[1][0]["commodity_id"])
print(max_loop[1][1]["commodity_id"])
parse_systems()
grep_systems()
parse_stats()
grep_stats()
parse_listings()
grep_listings()
calc_1()
