cache.py

@@ -1,3 +1,4 @@
+import datetime
 import lzma
 import pathlib
 import time
@@ -10,16 +11,20 @@ import urllib.parse
 if typing.TYPE_CHECKING:
 	import httpx._types
 
+ONE_DAY = datetime.timedelta(days=1)
+
 CACHE_DIR = pathlib.Path(__file__).parent / 'cache'
 client = httpx.Client(transport=httpx.HTTPTransport(http2=True, retries=2), timeout=10)
 
 @typing.overload
-def get(url: str, *, json: typing.Literal[True]=True, headers: httpx._types.HeaderTypes|None=None) -> typing.Any:
+def get(url: str, *, json: typing.Literal[True]=True, headers: httpx._types.HeaderTypes|None=None,
+		expiry: datetime.timedelta=datetime.timedelta(minutes=10)) -> typing.Any:
 	...
 @typing.overload
-def get(url: str, *, json: typing.Literal[False], headers: httpx._types.HeaderTypes|None=None) -> str:
+def get(url: str, *, json: typing.Literal[False], headers: httpx._types.HeaderTypes|None=None,
+		expiry: datetime.timedelta=datetime.timedelta(minutes=10)) -> str:
 	...
-def get(url: str, *, json=True, headers=None) -> typing.Any:
+def get(url: str, *, json=True, headers=None, expiry=datetime.timedelta(minutes=10)) -> typing.Any:
 	parsed = urllib.parse.urlparse(url)
 	assert parsed.hostname is not None
 	cache_filename = urllib.parse.quote(parsed.path.removeprefix('/'), safe='')
@@ -29,7 +34,7 @@ def get(url: str, *, json=True, headers=None) -> typing.Any:
 	cache_path = CACHE_DIR / parsed.hostname / cache_filename
 
 	try:
-		if cache_path.stat().st_mtime > time.time() - 600: # less than 10 minutes old
+		if cache_path.stat().st_mtime > time.time() - expiry.total_seconds(): # less than 10 minutes old
 			with lzma.open(cache_path, 'rb') as f:
 				if json:
 					return cbor2.load(f)

company.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import collections
-import concurrent.futures
+import datetime
 import sys
 import typing
 
@@ -11,26 +11,16 @@ import integration
 def main() -> None:
 	code = sys.argv[1]
 	company: Company = cache.get('https://rest.fnar.net/company/code/' + code)
+	planets = {cp['PlanetName'] for cp in company['Planets']}
 
 	cogc_planets: dict[str, list[str]] = collections.defaultdict(list)
-	with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
-		futures: list[concurrent.futures.Future[Planet]] = [
-				executor.submit(cache.get, 'https://rest.fnar.net/planet/' + planet['PlanetNaturalId'])
-				for planet in company['Planets']]
-		for future in futures:
-			planet = future.result()
-			cogc = None
-			cogcs = planet['COGCPrograms']
-			if len(cogcs) > 1:
-				cogcs.sort(key=lambda c: c['StartEpochMs'], reverse=True)
-				cogc = cogcs[1]['ProgramType']
-				if cogc is not None:
-					cogc = cogc.removeprefix('ADVERTISING_')
-			print(planet['PlanetName'], cogc)
+	for planet, cogc in iter_planet_cogc():
+		if planet['PlanetName'] in planets:
+			print(planet['PlanetName'], cogc, sep='\t\t')
 			if cogc is not None:
 				cogc_planets[cogc].append(planet['PlanetName'])
 
-	buildings: typing.Sequence[Building] = cache.get('https://rest.fnar.net/building/allbuildings')
+	buildings: typing.Sequence[Building] = cache.get('https://rest.fnar.net/building/allbuildings', expiry=cache.ONE_DAY)
 	experts: dict[str, str] = {}
 	for building in buildings:
 		for recipe in building['Recipes']:
@@ -39,9 +29,25 @@ def main() -> None:
 	print()
 
 	company_report = integration.pmmg_monthly_report()[company['CompanyId']]
-	for mat in company_report.keys():
+	for mat, production in company_report.items():
 		expertise = experts.get(mat)
-		print(mat, expertise, ', '.join(cogc_planets.get(expertise, []))) # pyright: ignore[reportArgumentType, reportCallIssue]
+		print(f'{mat:3} {production["amount"]:8,.0f} {expertise or '':19}',
+				', '.join(cogc_planets.get(expertise, []))) # pyright: ignore[reportArgumentType, reportCallIssue]
+
+def iter_planet_cogc() -> typing.Iterator[tuple[Planet, Expertise | None]]:
+	all_planets: typing.Collection[Planet] = cache.get('https://universemap.taiyibureau.de/planet_data.json',
+			expiry=cache.ONE_DAY)
+	for planet in all_planets:
+		cogc = None
+		if len(cogcs := planet['COGCPrograms']) > 1:
+			cogcs.sort(key=lambda c: c['StartEpochMs'], reverse=True)
+			cogc = cogcs[1]['ProgramType']
+			if cogc is not None:
+				cogc = cogc.removeprefix('ADVERTISING_')
+		yield planet, typing.cast(Expertise | None, cogc)
+
+Expertise = typing.Literal['AGRICULTURE', 'CHEMISTRY', 'CONSTRUCTION', 'ELECTRONICS', 'FOOD_INDUSTRIES',
+	'FUEL_REFINING', 'MANUFACTURING', 'METALLURGY', 'RESOURCE_EXTRACTION']
 
 class Company(typing.TypedDict):
 	CompanyId: str
@@ -49,6 +55,7 @@ class Company(typing.TypedDict):
 
 class CompanyPlanet(typing.TypedDict):
 	PlanetNaturalId: str
+	PlanetName: str
 
 class Planet(typing.TypedDict):
 	PlanetId: str

mat_competitors.py

@@ -18,17 +18,11 @@ def main() -> None:
 	with open('www/closest.json') as f:
 		planet_ids = {planet_id for planet_id, closest_cx in json.load(f).items() if closest_cx == cx}
 
-	all_planets: typing.Collection[company.Planet] = cache.get('https://universemap.taiyibureau.de/planet_data.json')
 	planets = []
-	for planet in all_planets:
-		if planet['PlanetId'] not in planet_ids:
-			continue
-		if len(cogcs := planet['COGCPrograms']) > 1:
-			cogcs.sort(key=lambda c: c['StartEpochMs'], reverse=True)
-			cogc = cogcs[1]['ProgramType']
-			if cogc is not None and cogc.removeprefix('ADVERTISING_') == expertise:
-				planets.append(planet['PlanetName'])
 	print(len(planets), expertise, 'planets near', cx)
+	for planet, cogc in company.iter_planet_cogc():
+		if planet['PlanetId'] in planet_ids and cogc == expertise:
+			planets.append(planet['PlanetName'])
 
 	coid_code: dict[str, str] = {}
 	coid_bases = collections.defaultdict(list)
@@ -42,7 +36,7 @@ def main() -> None:
 			coid_bases[base['OwnerId']].append(planet)
 
 	coid_users: dict[str, str] = {company_id: d['Username']
-		for company_id, d in cache.get('https://pmmg-products.github.io/reports/data/knownCompanies.json').items()}
+		for company_id, d in cache.get('https://pmmg-products.github.io/reports/data/knownCompanies.json', expiry=cache.ONE_DAY).items()}
 
 	for company_id, co_production in integration.pmmg_monthly_report().items():
 		if (mat_production := co_production.get(ticker)) is None:
@@ -54,7 +48,7 @@ def main() -> None:
 			print(f'{mat_production["amount"]:10.1f}', ', '.join(bases))
 
 def iter_expertise(ticker: str) -> typing.Iterator[str]:
-	buildings: typing.Sequence[company.Building] = cache.get('https://rest.fnar.net/building/allbuildings')
+	buildings: typing.Sequence[company.Building] = cache.get('https://rest.fnar.net/building/allbuildings', expiry=cache.ONE_DAY)
 	for building in buildings:
 		for recipe in building['Recipes']:
 			for output in recipe['Outputs']: