"""Audit existing application telemetry and original Journals; never opens the game or writes its data."""
import argparse
import json
import math
from pathlib import Path
from datetime import datetime,timedelta
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('--report', required=True, type=Path)
p.add_argument('--journal', required=True, type=Path)
p.add_argument('--output', required=True, type=Path)
a = p.parse_args()
report = json.loads(a.report.read_text(encoding='utf-8-sig'))
start = datetime.fromisoformat(report['start'])
end = datetime.fromisoformat(report['completedAt'])
rows = 0
first = None
begin = []
commits = []
violations = []
audit_end=end
if report.get('actualArrivalAt'):
arrival_clock=datetime.fromisoformat(report['actualArrivalAt'])
if arrival_clock>end+timedelta(seconds=5):
violations.append('Journal arrival clock exceeded the bounded five-second lead')
else:
audit_end=max(end,arrival_clock)
maximum = 0
maximum_axis = 0
maximum_scoop_ratio = 0
minimum_radius_ratio = math.inf
full_power_refill_seconds = 0
full_power_peak_heat = 0
previous_full_power_at = None
refills = {}
orbit_throttles = set()
first_control = {}
input_records = []
latest_inputs = None
cooldown_turns = {}
charge_inputs = []
charge_recovery_frames = 0
cancel_episodes = 0
previous_cancel = False
first_eligible = {}
def clear_target(frame):
pos, dest = frame['position'], frame['targetPosition']
delta = [dest[k]-pos[k] for k in ('x','y','z')]
square = sum(v*v for v in delta)
for body in frame.get('obstacles', []) + [frame['currentStar']]:
centre = body['centre']
projection = min(1, max(0, sum((centre[k]-pos[k])*delta[i] for i,k in enumerate(('x','y','z'))) / square))
distance = sum((pos[k]+projection*delta[i]-centre[k])**2 for i,k in enumerate(('x','y','z')))
if distance <= (body['exclusionRadius']*1.05)**2:
return False
return True
with Path(str(a.report) + '.jsonl').open(encoding='utf-8') as source:
for line in source:
item = json.loads(line)
if item.get('kind') == 'application-charge-input':
charge_inputs.append(item)
if item['request']['heatPercent']>=90 or item['request']['predictedPeak']>95:
violations.append('Original desired-FSD input exceeded software compatibility thermal bounds')
continue
if item.get('kind') == 'application-inputs':
latest_inputs = item
input_records.append(item)
continue
if item.get('kind') != 'application-frame':
continue
frame = item['frame']
rows += 1
first = first or frame
maximum = max(maximum, frame.get('heatPercent') or 0)
star = frame.get('currentStar')
if star:
radius = math.sqrt(sum((frame['position'][k]-star['centre'][k])**2 for k in ('x','y','z')))
minimum_radius_ratio = min(minimum_radius_ratio, radius/star['exclusionRadius'])
decision = item.get('decision')
# FsdPhase.Jumping is 3. A departing frame may already see the next
# Journal fuel total while still using the previous system geometry.
if frame.get('adaptiveFlightControl') and frame.get('refuellingToFull') and frame.get('fsd') != 3:
refill = refills.setdefault(str(frame['systemAddress']), dict(minimum=frame['fuelMain'],maximum=frame['fuelMain'],capacity=frame['fuelTankCapacity']))
refill['maximum'] = max(refill['maximum'], frame['fuelMain'])
if str(frame['systemAddress']) in refills and frame.get('fuelMain') is not None:
refill=refills[str(frame['systemAddress'])]
refill['maximum']=max(refill['maximum'],frame['fuelMain'])
if frame.get('scoopRate') is not None and (frame.get('scoopMaximumRate') or 0) > 0:
maximum_scoop_ratio = max(maximum_scoop_ratio, frame['scoopRate']/frame['scoopMaximumRate'])
at = datetime.fromisoformat(item['at'])
full_power = (frame.get('refuellingToFull') and frame.get('scoopActive')
and frame['fuelMain'] < frame['fuelTankCapacity']-.02
and (frame.get('scoopMaximumRate') or 0) > 0
and (frame.get('scoopRate') or 0) >= frame['scoopMaximumRate']*.98)
if full_power:
full_power_peak_heat = max(full_power_peak_heat, frame['heatPercent'])
if previous_full_power_at is not None:
full_power_refill_seconds += min(.3, max(0, (at-previous_full_power_at).total_seconds()))
previous_full_power_at = at
else:
previous_full_power_at = None
if not decision:
continue
if decision.get('coolingRecovery'):
charge_recovery_frames += 1
if decision['cancelCharging'] and not previous_cancel:
cancel_episodes += 1
previous_cancel = decision['cancelCharging']
applied = item.get('applied')
if applied and applied['sequence'] == frame['sequence']:
for axis in ('yaw','pitch'):
if applied.get(axis) is not None:
maximum_axis = max(maximum_axis,abs(applied[axis]))
if abs(applied[axis]) > 1.000001:
violations.append('Native steering exceeded game axis range')
if applied.get('predictedPeak') is not None and decision['beginCharging'] and applied['predictedPeak'] >= 100:
violations.append('Native charge forecast would exceed 100 percent')
if '高功率' in decision['state'] or '圆弧' in decision['state']:
orbit_throttles.add(applied.get('throttle'))
first_control.setdefault(str(frame['systemAddress']),dict(at=item['at'],state=decision['state']))
if frame.get('adaptiveFlightControl'):
fuel_ready = frame['fuelMain'] >= frame['fuelNeededToNextRefuel']
if decision['beginCharging'] and not fuel_ready:
violations.append('Charging began before the persistent fuel target was met')
if decision['beginCharging'] and str(frame['systemAddress']) in refills:
refill = refills[str(frame['systemAddress'])]
refill['departedFuel'] = frame['fuelMain']
if frame['fuelMain'] < refill['capacity']-.02:
violations.append('A required refill departed before filling the tank')
peak = decision.get('predictedPeakHeat')
radial = [frame['position'][k]-frame['currentStar']['centre'][k] for k in ('x','y','z')]
outward = sum(radial[i]*frame['forward'][k] for i,k in enumerate(('x','y','z'))) >= 0
eligible = (frame['fsd']==0 and frame['canBeginCharge'] and fuel_ready and frame.get('nativeAllowsCharging')
and peak is not None and peak <= 95 and clear_target(frame)
and (frame['heatPercent']<90 or not frame['outwardDepartureRequired'] or sum(radial[i]*(frame['targetPosition'][k]-frame['position'][k]) for i,k in enumerate(('x','y','z')))>=0)
and not decision.get('coolingRecovery'))
if eligible and not decision['beginCharging']:
violations.append('An eligible clear/cool/fuelled frame postponed FSD charging')
if eligible:
first_eligible.setdefault(str(frame['systemAddress']),dict(at=item['at'],chargeRequested=decision['beginCharging']))
if frame['fsd']==4 and fuel_ready and clear_target(frame) and frame['heatPercent']<97:
direction=decision.get('direction')
target=[frame['targetPosition'][k]-frame['position'][k] for k in ('x','y','z')]
length=math.sqrt(sum(v*v for v in target))
dot=sum(direction[k]*target[i]/length for i,k in enumerate(('x','y','z'))) if direction else -1
if dot<math.cos(math.radians(.01)):
violations.append('A clear fuelled cooldown frame postponed turning to the target')
if decision['beginCharging'] or decision['permitJump']:
violations.append('FSD requested before cooldown ended')
forward_dot=sum(frame['forward'][k]*target[i]/length for i,k in enumerate(('x','y','z')))
error=math.degrees(math.acos(min(1,max(-1,forward_dot))))
record=cooldown_turns.setdefault(str(frame['systemAddress']),dict(firstAt=item['at'],firstError=error,frames=0))
record.update(lastAt=item['at'],lastError=error,frames=record['frames']+1)
record = dict(at=item['at'], inSupercruise=frame['inSupercruise'],
speed=frame['speedMetresPerSecond'], fsd=frame['fsd'],
throttle=decision['throttle'], heat=frame['heatPercent'], sequence=frame['sequence'])
if decision['beginCharging'] and not decision['enterSupercruise']:
begin.append(record)
if decision['throttle'] != 1:
violations.append('Hyperspace charge did not request concurrent full throttle')
if decision['beginCharging'] or decision['permitJump']:
flags = item.get('journalContext', {}).get('statusFlags')
if flags is not None and flags & ((1 << 2) | (1 << 6) | (1 << 9)):
violations.append('Charging or commit requested while game Status still reported deployed equipment')
if latest_inputs is not None:
inputs = latest_inputs['state']
age = (at-datetime.fromisoformat(inputs['observedAt'])).total_seconds()
if any(inputs[k] for k in ('hardpoints','cargoScoop','landingGear')) or not -.1 <= age <= .3:
violations.append('Charging or commit lacked fresh retracted native-input confirmation')
if decision['permitJump']:
f = frame['forward']
pos = frame['position']
dest = frame['targetPosition']
ray = [dest[k] - pos[k] for k in ('x', 'y', 'z')]
norm = math.sqrt(sum(v*v for v in ray))
dot = sum(f[k] * ray[i] for i, k in enumerate(('x', 'y', 'z'))) / norm
record['alignmentDegrees'] = math.degrees(math.acos(min(1, max(-1, dot))))
commits.append(record)
if decision['throttle'] != 1 or record['alignmentDegrees'] > 1.001 or decision['predictedPeakHeat'] >= 100:
violations.append('Jump commit throttle, alignment or thermal guard failed')
events = []
with a.journal.open(encoding='utf-8-sig') as source:
for line in source:
item = json.loads(line)
at = datetime.fromisoformat(item['timestamp'].replace('Z', '+00:00'))
if start.replace(microsecond=0) <= at <= audit_end and item['event'] in (
'StartJump', 'FSDJump', 'SupercruiseEntry', 'SupercruiseExit', 'FuelScoop', 'HeatWarning', 'HeatDamage'):
events.append({k: item[k] for k in ('timestamp', 'event', 'JumpType', 'StarSystem', 'SystemAddress', 'JumpDist', 'FuelUsed', 'FuelLevel', 'Scooped', 'Total') if k in item})
arrivals = [v for v in events if v['event'] == 'FSDJump']
normal_direct = bool(first and not first['inSupercruise'] and first['speedMetresPerSecond'] < .01 and begin
and not any(v['event'] == 'SupercruiseEntry' for v in events))
passed = report['passed'] and bool(arrivals) and arrivals[-1]['SystemAddress'] == report['expectedSystem'] and not violations
result = dict(passed=passed, applicationReport=str(a.report.resolve()), journal=str(a.journal.resolve()),
firstFrame=first, frameRecords=rows, observedPeakHeat=maximum, chargeBegins=begin,
commits=commits, events=events, normalSpaceZeroSpeedDirect=normal_direct,
maximumNativeAxis=maximum_axis,maximumScoopRateRatio=maximum_scoop_ratio,refills=refills,
minimumRadiusRatio=minimum_radius_ratio if math.isfinite(minimum_radius_ratio) else None,
fullPowerRefillSeconds=full_power_refill_seconds,fullPowerPeakHeat=full_power_peak_heat,
orbitThrottleCommands=sorted(v for v in orbit_throttles if v is not None),firstControlBySystem=first_control,
inputConfirmations=len(input_records), firstRetractedInputs=input_records[0] if input_records else None,
clearFuelledCooldownTurns=cooldown_turns, originalDesiredChargeInputs=charge_inputs,
firstEligibleChargeBySystem=first_eligible, chargeCancelEpisodes=cancel_episodes, coolingRecoveryFrames=charge_recovery_frames,
violations=violations, journalWindowEnd=audit_end.isoformat(), gameAccess=False, screenUse=False, journalWrites=0)
if a.output.exists():
raise FileExistsError('Use a new audit output')
a.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
print(json.dumps(dict(passed=passed, jumps=len(arrivals), observedPeakHeat=maximum,
normalSpaceZeroSpeedDirect=normal_direct, violations=violations), ensure_ascii=False))
"""Audit existing application telemetry and original Journals; never opens the game or writes its data."""
import argparse
import json
import math
from pathlib import Path
from datetime import datetime,timedelta
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('--report', required=True, type=Path)
p.add_argument('--journal', required=True, type=Path)
p.add_argument('--output', required=True, type=Path)
a = p.parse_args()
report = json.loads(a.report.read_text(encoding='utf-8-sig'))
start = datetime.fromisoformat(report['start'])
end = datetime.fromisoformat(report['completedAt'])
rows = 0
first = None
begin = []
commits = []
violations = []
audit_end=end
if report.get('actualArrivalAt'):
arrival_clock=datetime.fromisoformat(report['actualArrivalAt'])
if arrival_clock>end+timedelta(seconds=5):
violations.append('Journal arrival clock exceeded the bounded five-second lead')
else:
audit_end=max(end,arrival_clock)
maximum = 0
maximum_axis = 0
maximum_scoop_ratio = 0
minimum_radius_ratio = math.inf
full_power_refill_seconds = 0
full_power_peak_heat = 0
previous_full_power_at = None
refills = {}
orbit_throttles = set()
first_control = {}
input_records = []
latest_inputs = None
cooldown_turns = {}
charge_inputs = []
charge_recovery_frames = 0
cancel_episodes = 0
previous_cancel = False
first_eligible = {}
def clear_target(frame):
pos, dest = frame['position'], frame['targetPosition']
delta = [dest[k]-pos[k] for k in ('x','y','z')]
square = sum(v*v for v in delta)
for body in frame.get('obstacles', []) + [frame['currentStar']]:
centre = body['centre']
projection = min(1, max(0, sum((centre[k]-pos[k])*delta[i] for i,k in enumerate(('x','y','z'))) / square))
distance = sum((pos[k]+projection*delta[i]-centre[k])**2 for i,k in enumerate(('x','y','z')))
if distance <= (body['exclusionRadius']*1.05)**2:
return False
return True
with Path(str(a.report) + '.jsonl').open(encoding='utf-8') as source:
for line in source:
item = json.loads(line)
if item.get('kind') == 'application-charge-input':
charge_inputs.append(item)
if item['request']['heatPercent']>=90 or item['request']['predictedPeak']>95:
violations.append('Original desired-FSD input exceeded software compatibility thermal bounds')
continue
if item.get('kind') == 'application-inputs':
latest_inputs = item
input_records.append(item)
continue
if item.get('kind') != 'application-frame':
continue
frame = item['frame']
rows += 1
first = first or frame
maximum = max(maximum, frame.get('heatPercent') or 0)
star = frame.get('currentStar')
if star:
radius = math.sqrt(sum((frame['position'][k]-star['centre'][k])**2 for k in ('x','y','z')))
minimum_radius_ratio = min(minimum_radius_ratio, radius/star['exclusionRadius'])
decision = item.get('decision')
# FsdPhase.Jumping is 3. A departing frame may already see the next
# Journal fuel total while still using the previous system geometry.
if frame.get('adaptiveFlightControl') and frame.get('refuellingToFull') and frame.get('fsd') != 3:
refill = refills.setdefault(str(frame['systemAddress']), dict(minimum=frame['fuelMain'],maximum=frame['fuelMain'],capacity=frame['fuelTankCapacity']))
refill['maximum'] = max(refill['maximum'], frame['fuelMain'])
if str(frame['systemAddress']) in refills and frame.get('fuelMain') is not None:
refill=refills[str(frame['systemAddress'])]
refill['maximum']=max(refill['maximum'],frame['fuelMain'])
if frame.get('scoopRate') is not None and (frame.get('scoopMaximumRate') or 0) > 0:
maximum_scoop_ratio = max(maximum_scoop_ratio, frame['scoopRate']/frame['scoopMaximumRate'])
at = datetime.fromisoformat(item['at'])
full_power = (frame.get('refuellingToFull') and frame.get('scoopActive')
and frame['fuelMain'] < frame['fuelTankCapacity']-.02
and (frame.get('scoopMaximumRate') or 0) > 0
and (frame.get('scoopRate') or 0) >= frame['scoopMaximumRate']*.98)
if full_power:
full_power_peak_heat = max(full_power_peak_heat, frame['heatPercent'])
if previous_full_power_at is not None:
full_power_refill_seconds += min(.3, max(0, (at-previous_full_power_at).total_seconds()))
previous_full_power_at = at
else:
previous_full_power_at = None
if not decision:
continue
if decision.get('coolingRecovery'):
charge_recovery_frames += 1
if decision['cancelCharging'] and not previous_cancel:
cancel_episodes += 1
previous_cancel = decision['cancelCharging']
applied = item.get('applied')
if applied and applied['sequence'] == frame['sequence']:
for axis in ('yaw','pitch'):
if applied.get(axis) is not None:
maximum_axis = max(maximum_axis,abs(applied[axis]))
if abs(applied[axis]) > 1.000001:
violations.append('Native steering exceeded game axis range')
if applied.get('predictedPeak') is not None and decision['beginCharging'] and applied['predictedPeak'] >= 100:
violations.append('Native charge forecast would exceed 100 percent')
if '高功率' in decision['state'] or '圆弧' in decision['state']:
orbit_throttles.add(applied.get('throttle'))
first_control.setdefault(str(frame['systemAddress']),dict(at=item['at'],state=decision['state']))
if frame.get('adaptiveFlightControl'):
fuel_ready = frame['fuelMain'] >= frame['fuelNeededToNextRefuel']
if decision['beginCharging'] and not fuel_ready:
violations.append('Charging began before the persistent fuel target was met')
if decision['beginCharging'] and str(frame['systemAddress']) in refills:
refill = refills[str(frame['systemAddress'])]
refill['departedFuel'] = frame['fuelMain']
if frame['fuelMain'] < refill['capacity']-.02:
violations.append('A required refill departed before filling the tank')
peak = decision.get('predictedPeakHeat')
radial = [frame['position'][k]-frame['currentStar']['centre'][k] for k in ('x','y','z')]
outward = sum(radial[i]*frame['forward'][k] for i,k in enumerate(('x','y','z'))) >= 0
eligible = (frame['fsd']==0 and frame['canBeginCharge'] and fuel_ready and frame.get('nativeAllowsCharging')
and peak is not None and peak <= 95 and clear_target(frame)
and (frame['heatPercent']<90 or not frame['outwardDepartureRequired'] or sum(radial[i]*(frame['targetPosition'][k]-frame['position'][k]) for i,k in enumerate(('x','y','z')))>=0)
and not decision.get('coolingRecovery'))
if eligible and not decision['beginCharging']:
violations.append('An eligible clear/cool/fuelled frame postponed FSD charging')
if eligible:
first_eligible.setdefault(str(frame['systemAddress']),dict(at=item['at'],chargeRequested=decision['beginCharging']))
if frame['fsd']==4 and fuel_ready and clear_target(frame) and frame['heatPercent']<97:
direction=decision.get('direction')
target=[frame['targetPosition'][k]-frame['position'][k] for k in ('x','y','z')]
length=math.sqrt(sum(v*v for v in target))
dot=sum(direction[k]*target[i]/length for i,k in enumerate(('x','y','z'))) if direction else -1
if dot<math.cos(math.radians(.01)):
violations.append('A clear fuelled cooldown frame postponed turning to the target')
if decision['beginCharging'] or decision['permitJump']:
violations.append('FSD requested before cooldown ended')
forward_dot=sum(frame['forward'][k]*target[i]/length for i,k in enumerate(('x','y','z')))
error=math.degrees(math.acos(min(1,max(-1,forward_dot))))
record=cooldown_turns.setdefault(str(frame['systemAddress']),dict(firstAt=item['at'],firstError=error,frames=0))
record.update(lastAt=item['at'],lastError=error,frames=record['frames']+1)
record = dict(at=item['at'], inSupercruise=frame['inSupercruise'],
speed=frame['speedMetresPerSecond'], fsd=frame['fsd'],
throttle=decision['throttle'], heat=frame['heatPercent'], sequence=frame['sequence'])
if decision['beginCharging'] and not decision['enterSupercruise']:
begin.append(record)
if decision['throttle'] != 1:
violations.append('Hyperspace charge did not request concurrent full throttle')
if decision['beginCharging'] or decision['permitJump']:
flags = item.get('journalContext', {}).get('statusFlags')
if flags is not None and flags & ((1 << 2) | (1 << 6) | (1 << 9)):
violations.append('Charging or commit requested while game Status still reported deployed equipment')
if latest_inputs is not None:
inputs = latest_inputs['state']
age = (at-datetime.fromisoformat(inputs['observedAt'])).total_seconds()
if any(inputs[k] for k in ('hardpoints','cargoScoop','landingGear')) or not -.1 <= age <= .3:
violations.append('Charging or commit lacked fresh retracted native-input confirmation')
if decision['permitJump']:
f = frame['forward']
pos = frame['position']
dest = frame['targetPosition']
ray = [dest[k] - pos[k] for k in ('x', 'y', 'z')]
norm = math.sqrt(sum(v*v for v in ray))
dot = sum(f[k] * ray[i] for i, k in enumerate(('x', 'y', 'z'))) / norm
record['alignmentDegrees'] = math.degrees(math.acos(min(1, max(-1, dot))))
commits.append(record)
if decision['throttle'] != 1 or record['alignmentDegrees'] > 1.001 or decision['predictedPeakHeat'] >= 100:
violations.append('Jump commit throttle, alignment or thermal guard failed')
events = []
with a.journal.open(encoding='utf-8-sig') as source:
for line in source:
item = json.loads(line)
at = datetime.fromisoformat(item['timestamp'].replace('Z', '+00:00'))
if start.replace(microsecond=0) <= at <= audit_end and item['event'] in (
'StartJump', 'FSDJump', 'SupercruiseEntry', 'SupercruiseExit', 'FuelScoop', 'HeatWarning', 'HeatDamage'):
events.append({k: item[k] for k in ('timestamp', 'event', 'JumpType', 'StarSystem', 'SystemAddress', 'JumpDist', 'FuelUsed', 'FuelLevel', 'Scooped', 'Total') if k in item})
arrivals = [v for v in events if v['event'] == 'FSDJump']
normal_direct = bool(first and not first['inSupercruise'] and first['speedMetresPerSecond'] < .01 and begin
and not any(v['event'] == 'SupercruiseEntry' for v in events))
passed = report['passed'] and bool(arrivals) and arrivals[-1]['SystemAddress'] == report['expectedSystem'] and not violations
result = dict(passed=passed, applicationReport=str(a.report.resolve()), journal=str(a.journal.resolve()),
firstFrame=first, frameRecords=rows, observedPeakHeat=maximum, chargeBegins=begin,
commits=commits, events=events, normalSpaceZeroSpeedDirect=normal_direct,
maximumNativeAxis=maximum_axis,maximumScoopRateRatio=maximum_scoop_ratio,refills=refills,
minimumRadiusRatio=minimum_radius_ratio if math.isfinite(minimum_radius_ratio) else None,
fullPowerRefillSeconds=full_power_refill_seconds,fullPowerPeakHeat=full_power_peak_heat,
orbitThrottleCommands=sorted(v for v in orbit_throttles if v is not None),firstControlBySystem=first_control,
inputConfirmations=len(input_records), firstRetractedInputs=input_records[0] if input_records else None,
clearFuelledCooldownTurns=cooldown_turns, originalDesiredChargeInputs=charge_inputs,
firstEligibleChargeBySystem=first_eligible, chargeCancelEpisodes=cancel_episodes, coolingRecoveryFrames=charge_recovery_frames,
violations=violations, journalWindowEnd=audit_end.isoformat(), gameAccess=False, screenUse=False, journalWrites=0)
if a.output.exists():
raise FileExistsError('Use a new audit output')
a.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
print(json.dumps(dict(passed=passed, jumps=len(arrivals), observedPeakHeat=maximum,
normalSpaceZeroSpeedDirect=normal_direct, violations=violations), ensure_ascii=False))
