c=0xffff

for b in d:

c^=b<<8

for _ in range(8): c=((c<<1)^0x1021)&0xffff if c&0x8000 else (c<<1)&0xffff

if not _t:

for b in range(256):

c=b<<24

for _ in range(8): c=((c<<1)^0x1edc6f41)&0xffffffff if c&0x80000000 else (c<<1)&0xffffffff

_t.append(c)

crc=0

for byte in buf: crc=((crc<<8)&0xffffffff)^_t[((crc>>24)^byte)&0xff]

# seq=None -> allocate a fresh transport seq; pass an explicit seq to share one

# across a marker+block pair (matches the official app).

if seq is None: seq=_nextseq()

body=pb+crc16(pb); tot=max(1,-(-len(body)//CHUNK)); out=[];off=0

for i in range(tot):

ch=body[off:off+CHUNK];off+=len(ch)

out.append(bytes([0xaa,0x21,seq,len(ch),tot,i+1,sid,flag])+ch)

"""Unpack the OTA container into its component segments. Returns a list of

if len(img) < 0x40:

raise ValueError("file is too small to be a firmware image")

n = struct.unpack_from('<I', img, 8)[0]

if not (0 < n <= 64):

raise ValueError(f"implausible component count {n} (corrupt header?)")

segs = []

for i in range(n):

eid, off, size, crc = struct.unpack_from('<IIII', img, 0x40 + i*16)

sub = img[off:off+128]

if len(sub) < 128:

raise ValueError(f"segment {i} subheader runs past end of file")

ps = struct.unpack_from('<I', sub, 8)[0]

fn = sub[48:128].split(b'\0')[0].decode('latin1')

segs.append({'eid':eid,'off':off,'size':size,'crc':crc,'sub':sub,'ps':ps,'fn':fn})

"""Sanity-check that the file looks like a flashable G2 firmware image. Returns

segs = parse_firmware_segments(img)

names = [s['fn'] for s in segs]

if len(segs) != 5 and len(segs) != 6:

raise ValueError(

f"expected 5-6 segments, found {len(segs)}: {names}")

if REQUIRED_SEGMENT not in names:

raise ValueError(

f"required segment {REQUIRED_SEGMENT!r} not found; segments are: {names}")

# Each component stores a CRC32C of its payload in the TOC and the sub-header

# echo (sub[12:16]); the glasses verify it on END (status 7 = CHECK_FAIL). The

# flasher streams the sub-header verbatim, so a stale stored CRC (image patched

# without recomputing checksums) guarantees a status-7 failure after a full,

# slow transfer. Catch it here instead.

for i, s in enumerate(segs):

payload = img[s['off']+128:s['off']+128+s['ps']]

calc = crc32c_msb(payload)

sub_crc = struct.unpack_from('<I', s['sub'], 12)[0]

if calc != s['crc'] or calc != sub_crc:

raise ValueError(

f"segment {i} ({s['fn']}) stored CRC32C is stale: payload computes "

f"{calc:08x} but TOC={s['crc']:08x} sub={sub_crc:08x}. The image was "

"modified without recomputing checksums — re-run the patch tool's "

"checksum fixup (and the mainApp internal preamble CRC32) before flashing.")

check_mainapp_fits_mram(img, segs)

"""Guard against an enlarged main-app image overrunning internal MRAM.

s = next((x for x in segs if x['fn'] == REQUIRED_SEGMENT), None)

if s is None: # already caught above, but be defensive

return

ps = s['ps']

pre = img[s['off']+128:s['off']+128+APP_PREAMBLE]

if len(pre) < APP_PREAMBLE:

raise ValueError("main-app payload is smaller than its 32-byte preamble")

load_addr = struct.unpack_from('<I', pre, 0x14)[0]

pre_len = struct.unpack_from('<I', pre, 0)[0] & 0xFFFFFF

if load_addr != APP_LOAD_ADDR:

raise ValueError(

f"main-app preamble load address is 0x{load_addr:08x}, expected "

f"0x{APP_LOAD_ADDR:08x}. The bootloader programs to THIS address with "

"no check — flashing would write the app to the wrong MRAM location.")

if pre_len != ps:

raise ValueError(

f"main-app preamble length (0x{pre_len:x} = {pre_len} B) != staged "

f"payload size ps (0x{ps:x} = {ps} B). The bootloader erases/programs "

"preamble-length bytes but the flasher streams ps bytes; a mismatch "

"truncates or overruns. If you resized the image, set preamble[0] "

"low-24 to the new size and re-run --recompute-checksums.")

prog_end = APP_LOAD_ADDR + ps - APP_PREAMBLE # bootloader writes payload[0x20:]

if prog_end > APP_MAX_END:

over = prog_end - APP_MAX_END

raise ValueError(

f"main-app is too large: programmed region ends at 0x{prog_end:08x}, "

f"{over} B ({over/1024:.1f} KB) past the safe ceiling 0x{APP_MAX_END:08x}. "

f"MRAM app window is 0x{APP_LOAD_ADDR:08x}..0x{OTA_FLAG_ADDR:08x} (OTA flag); "

f"end of MRAM is 0x{MRAM_END:08x}. The bootloader does NOT bounds-check "

"this, so flashing risks clobbering the OTA flag / NV or bricking the lens "

"(SWD-only recovery). Shrink the image or, if you have verified the exact "

"NV base, raise APP_MAX_END deliberately.")

headroom = APP_MAX_END - prog_end

if DEBUG:

print(f" [main-app fits: ends 0x{prog_end:08x}, "

"""Rewrite an EVENOTA image's stored checksums in place to match its current

with open(path,'rb') as fh: data=bytearray(fh.read())

segs=parse_firmware_segments(bytes(data)) # validates structure / offsets

changed=0

for i,s in enumerate(segs):

off=s['off']; ps=s['ps']; fn=s['fn']

# The mainApp internal preamble CRC32 lives INSIDE the payload, so fix it

# before computing the component CRC32C (which covers the whole payload).

if fn.endswith('s200_firmware_ota.bin'):

pre=zlib.crc32(bytes(data[off+128+8:off+128+ps]))&0xffffffff

old=struct.unpack_from('<I',data,off+128+4)[0]

if old!=pre:

struct.pack_into('<I',data,off+128+4,pre); changed+=1

print(f" [{i}] {fn}: preamble crc32 {old:08x} -> {pre:08x}")

payload=bytes(data[off+128:off+128+ps])

crc=crc32c_msb(payload)

old_toc=struct.unpack_from('<I',data,0x40+i*16+12)[0]

old_sub=struct.unpack_from('<I',data,off+12)[0]

if old_toc!=crc: struct.pack_into('<I',data,0x40+i*16+12,crc); changed+=1

if old_sub!=crc: struct.pack_into('<I',data,off+12,crc); changed+=1

if old_toc!=crc or old_sub!=crc:

print(f" [{i}] {fn}: component crc32c -> {crc:08x} (was TOC={old_toc:08x} sub={old_sub:08x})")

else:

print(f" [{i}] {fn}: component crc32c {crc:08x} (ok)")

if not changed:

print(f"{path}: already consistent, no changes"); return 0

st=os.stat(path); d=os.path.dirname(os.path.abspath(path))

fd,tmp=tempfile.mkstemp(dir=d, prefix='.g2ck_')

try:

with os.fdopen(fd,'wb') as fh: fh.write(data)

os.chmod(tmp, st.st_mode)

os.replace(tmp, path)

except BaseException:

try: os.unlink(tmp)

except OSError: pass

raise

print(f"updated {path}: {changed} checksum field(s) rewritten")

def __init__(self, base, token):

self.base=base.rstrip('/'); self.token=token

self.notes=queue.Queue() # (char_uuid_lower, bytes)

self._stop=False

def _req(self,path,obj=None,method=None):

data=json.dumps(obj).encode() if obj is not None else None

r=urllib.request.Request(self.base+path,data=data,method=method or ('POST' if data else 'GET'))

if self.token: r.add_header('Authorization','Bearer '+self.token)

if data: r.add_header('Content-Type','application/json')

with urllib.request.urlopen(r,timeout=30) as resp: return resp.read()

def status(self): return json.loads(self._req('/status'))

def connect(self,a): return self._req('/connect',{"address":a})

def discover(self,a): return self._req('/discover',{"address":a})

def services(self,a): return json.loads(self._req('/services/'+a))

def notify(self,a,svc,ch,en=True): return self._req('/notify',{"address":a,"service":svc,"characteristic":ch,"enable":en})

def write(self,a,svc,ch,hexdata,wtype=1): return self._req('/write',{"address":a,"service":svc,"characteristic":ch,"data":hexdata,"writeType":wtype})

def start_ws(self):

import websocket

self.ws_open=False

def on_open(ws): self.ws_open=True

def on_close(ws,*a): self.ws_open=False

def on_msg(ws,msg):

try:

m=json.loads(msg)

if m.get('type')=='notification':

d=m['data']; self.notes.put((d.get('characteristic','').lower(), bytes.fromhex(d.get('data',''))))

except Exception: pass

hdr=['Authorization: Bearer '+self.token] if self.token else None

self.ws=websocket.WebSocketApp(self.base.replace('http','ws'),header=hdr,on_open=on_open,on_close=on_close,on_message=on_msg)

threading.Thread(target=lambda: self.ws.run_forever(reconnect=3),daemon=True).start()

def ka():

while not self._stop:

try: self.ws.send("ping")

except Exception: pass

time.sleep(1.5)

"""Per-lens view onto a shared DroidBridge. Notifications arrive on the

def __init__(self, bridge, address):

self.br=bridge; self.address=address; self.notes=bridge.notes

def status(self):

try: return f"droidbridge {self.br.base}: {self.br.status()}"

except Exception as e: return f"droidbridge {self.br.base}: status failed ({e})"

def connect(self):

# fresh GATT state: stale notify/connection state makes begin time out

try: self.br._req('/disconnect',{"address":self.address})

except Exception: pass

time.sleep(2)

self.br.connect(self.address); time.sleep(2)

def discover(self):

self.br.discover(self.address)

for _ in range(20):

try:

s=self.br.services(self.address)

if s and s.get('services'): return True

except Exception: pass

time.sleep(1)

return False

def set_notify(self, svc, ch, enable=True):

self.br.notify(self.address, svc, ch, enable)

def write(self, svc, ch, hexdata, wtype=1):

if LOSS_RATE>0 and ch==DATA[1] and random.random()<LOSS_RATE:

return # debug: simulate a dropped write-without-response fragment

self.br.write(self.address, svc, ch, hexdata, wtype)

def close(self):

try: self.br._req('/disconnect',{"address":self.address})

"""Pick a scanned arm for the requested address/side. `devs` is bleak's

want=_norm_addr(address); want_tail=want[-6:]

side_letter={'left':'L','right':'R'}.get(side or '')

def name_of(d,adv): return adv.local_name or d.name or ""

# 1) exact address match (Linux MAC, or macOS UUID connection strings)

for a,(d,adv) in devs.items():

if _norm_addr(a)==want: return d

# 2) MAC tail embedded in the advertised name (the macOS case for a MAC string)

for a,(d,adv) in devs.items():

m=G2_NAME_RE.search(name_of(d,adv))

if m and m.group(3).lower()==want_tail and (not side_letter or m.group(2).upper()==side_letter):

return d

# 3) last resort: exactly one advertised arm on the requested side

if side_letter:

cands=[d for a,(d,adv) in devs.items()

if (lambda m: m and m.group(2).upper()==side_letter)(G2_NAME_RE.search(name_of(d,adv)))]

if len(cands)==1: return cands[0]

"""Direct connection over this machine's Bluetooth radio via bleak. bleak is

def __init__(self, address, address_type=None, side=None):

self.address=address; self.address_type=address_type; self.side=side

self.notes=queue.Queue()

self.client=None

self._loop=asyncio.new_event_loop()

self._thr=threading.Thread(target=self._run_loop, daemon=True)

self._thr.start()

def _run_loop(self):

asyncio.set_event_loop(self._loop); self._loop.run_forever()

def _call(self, coro): return asyncio.run_coroutine_threadsafe(coro, self._loop).result()

def status(self):

return f"local BLE {self.side or '?'} {self.address}" + (f" ({self.address_type})" if self.address_type else "")

def connect(self):

from bleak import BleakClient, BleakScanner

async def _c():

devs=await BleakScanner.discover(timeout=SCAN_TIMEOUT, return_adv=True)

dev=match_scanned_device(devs, self.address, self.side)

if dev is None:

seen=[f"{(adv.local_name or d.name)!r}@{a}" for a,(d,adv) in devs.items()

if G2_NAME_RE.search(adv.local_name or d.name or "")]

raise RuntimeError(

f"could not find G2 {self.side or ''} ({self.address}) in a {SCAN_TIMEOUT}s scan. "

f"G2 arms advertising now: {seen or 'none'}. "

"The arm must be powered on and NOT connected to the phone (close the Even app / "

"turn off the phone's Bluetooth) so it advertises for a direct connection.")

self.client=BleakClient(dev)

await self.client.connect()

self._call(_c())

def discover(self):

# bleak performs service discovery during connect()

return bool(self.client and self.client.is_connected)

def _on_note(self, fallback_ch):

fb=fallback_ch.lower()

def cb(sender, data):

try: u=sender.uuid.lower()

except Exception: u=fb

self.notes.put((u, bytes(data)))

return cb

def set_notify(self, svc, ch, enable=True):

async def _n():

if enable: await self.client.start_notify(ch, self._on_note(ch))

else: await self.client.stop_notify(ch)

self._call(_n())

def write(self, svc, ch, hexdata, wtype=1):

data=bytes.fromhex(hexdata)

if LOSS_RATE>0 and ch==DATA[1] and random.random()<LOSS_RATE:

return # debug: simulate a dropped write-without-response fragment

# Android writeType 1 == WRITE_TYPE_NO_RESPONSE; anything else => with response.

response=(wtype!=1)

async def _w(): await self.client.write_gatt_char(ch, data, response=response)

self._call(_w())

def close(self):

async def _d():

if self.client and self.client.is_connected:

await self.client.disconnect()

try: self._call(_d())

except Exception: pass

"""unwrap an aa12 reply envelope -> (sid, pb). pb is [opcode,status,...]."""

if len(frame)>=10 and frame[0]==0xaa and frame[1]==0x12:

ln=frame[3]; sid=frame[6]; pb=frame[8:8+max(0,ln-2)]

return sid, pb

ch_uuid=ch_uuid.lower(); deadline=time.time()+timeout

while time.time()<deadline:

try: ch,frame=tp.notes.get(timeout=max(0.1,deadline-time.time()))

except queue.Empty: break

sid,pb=parse_rx(frame)

if DEBUG: print(f" [rx ...{ch[-4:]} sid=0x{sid:02x} pb={pb.hex()}]" if sid is not None else f" [rx ...{ch[-4:]} {frame.hex()}]")

if ch==ch_uuid and len(pb)>=2 and pb[0]==want_op: return pb[1]

"""Send one 4 KB block as marker+data sharing a single envelope seq (matches the

seq=_nextseq()

while not tp.notes.empty(): tp.notes.get()

for f in ctrl_frames(0x02, seq=seq): tp.write(DATA[0],DATA[1],f.hex(),1) # marker: reset rx offset

dfr=data_frames(blk, seq=seq)

if fault=='frag':

drop=len(dfr)//2

print(f" [FAULT] dropping data frag {drop}/{len(dfr)} to force a block NAK")

dfr=[f for k,f in enumerate(dfr) if k!=drop]

for f in dfr: tp.write(DATA[0],DATA[1],f.hex(),1) # 4 KB payload

if fault=='ack':

print(" [FAULT] block sent; swallowing its ack to simulate ack-loss -> timeout")

try: wait_ack(tp, 0x02, DATA[2], timeout=BLOCK_ACK_TIMEOUT)

except TimeoutError: pass

raise TimeoutError("injected ack-loss")

"""One pass over a component: FILE_CHECK, all 4 KB blocks, END. Returns the END

sub=seg['sub']; ps=seg['ps']; fn=seg['fn']

payload=img[seg['off']+128:seg['off']+128+ps]

st=send_data_msg(tp, ctrl_frames(0x01, sub), 0x01) # FILE_CHECK / OTA_TRANSMIT_INFORMATION

if st: raise RuntimeError(f"FILE_CHECK rejected status={st} ({rsp_name(st)})")

if not allowed("flash", stop_before):

return None

nb=-(-len(payload)//4096)

resends=0

for b in range(nb):

blk=payload[b*4096:(b+1)*4096]

for tries in range(BLOCK_NAK_RETRIES):

fault=None

if comp_index==FAULT_COMPONENT and b==FAULT_BLOCK and not _fault_fired[0]:

_fault_fired[0]=True; fault=FAULT_MODE or None

try:

st=send_block(tp, blk, fault=fault)

except TimeoutError:

print(f" block {b}/{nb} ACK TIMEOUT after {BLOCK_ACK_TIMEOUT}s")

raise # -> whole-component re-flash (clean reset)

if st==0: break

resends+=1

print(f" block {b}/{nb} NAK={st} ({rsp_name(st)}) resend {tries+1}/{BLOCK_NAK_RETRIES}")

else:

raise RuntimeError(f"block {b} NAK'd {BLOCK_NAK_RETRIES}x")

if b%100==0 or b==nb-1: print(f" {fn}: block {b+1}/{nb}")

print(f" {fn}: data phase done, {resends} block resend(s); sending END")

"""Flash a component, re-flashing the whole thing if END verification fails or a

fn=seg['fn']; ps=seg['ps']

payload=img[seg['off']+128:seg['off']+128+ps]

print(f"[{i}] {fn} ({ps}B crc32c=0x{crc32c_msb(payload):08x})")

for attempt in range(COMPONENT_RETRIES):

if attempt: print(f" re-flash attempt {attempt+1}/{COMPONENT_RETRIES}")

try:

end_st=flash_component(tp, seg, img, stop_before, comp_index=i)

except (TimeoutError, RuntimeError) as e:

print(f" block phase failed: {e}")

end_st=-1

if end_st is None: return None

if end_st in END_OK:

print(f" END verify OK ({fn}) status={end_st} ({rsp_name(end_st)})")

return 0

if end_st>=0:

print(f" END verify FAILED status={end_st} ({rsp_name(end_st)})")

while not tp.notes.empty(): tp.notes.get() # settle before the next attempt

time.sleep(1.5)

_reset_seq()

print("status:", tp.status())

tp.connect()

ok=tp.discover()

print("discovery:", "ok" if ok else "FAILED")

if not ok:

raise RuntimeError("service discovery failed")

if not allowed("heartbeat", stop_before):

print(f"[stop-before={stop_before}] discovery only; no writes."); return

# enable notifications on the data + ctrl notify chars (give CCCD time to take)

tp.set_notify(DATA[0],DATA[2],True)

tp.set_notify(CTRL[0],CTRL[2],True)

time.sleep(2.5)

while not tp.notes.empty(): tp.notes.get()

if not allowed("file_check", stop_before):

print(f"[stop-before={stop_before}] notify set up; stopping before FILE_CHECK."); return

# ---- (gated) actual flash ----

# heartbeat keepalive on CTRL channel during the transfer (~12s, like the app)

hb_stop=threading.Event()

def hb_loop():

while not hb_stop.wait(12):

try:

for f in frames(0x80, bytes.fromhex("080e10266a00")): tp.write(CTRL[0],CTRL[1],f.hex(),1)

except Exception: pass

threading.Thread(target=hb_loop,daemon=True).start()

N=len(segs)

print(f"flashing {len(img)}B, {N} components")

while not tp.notes.empty(): tp.notes.get() # drain stale notifications

bst=send_data_msg(tp, ctrl_frames(0x00), 0x00)

print(f"begin ack {bst} ({rsp_name(bst)})")

if bst not in END_OK: print(f" WARNING: unexpected begin status {bst} ({rsp_name(bst)}); continuing")

t_start=time.time()

try:

for i,seg in enumerate(segs):

if flash_component_with_retry(tp, i, seg, img, stop_before) is None:

print(f"[stop-before={stop_before}] FILE_CHECK acked; stopping before data blocks.")

return

finally:

hb_stop.set()

"""Parse g2://<method>?... into a dict. Methods: 'local', 'droidbridge'."""

u=urllib.parse.urlparse(raw.strip())

if u.scheme!='g2':

raise ValueError(f"connection string must start with 'g2://', got {raw!r}")

method=u.netloc.lower()

q={k:v[0] for k,v in urllib.parse.parse_qs(u.query).items()}

left=q.get('left'); right=q.get('right')

if not left or not right:

raise ValueError("connection string must include left= and right=")

if method=='local':

at=(q.get('addressType') or '').lower() or None

if at not in (None,'public','random'):

raise ValueError("addressType must be 'public' or 'random'")

return {'method':'local','left':left,'right':right,'address_type':at}

if method=='droidbridge':

phone=q.get('phone'); port=q.get('port'); token=q.get('token','')

if not phone or not port:

raise ValueError("droidbridge connection string must include phone= and port=")

return {'method':'droidbridge','left':left,'right':right,

'base':f"http://{phone}:{port}",'token':token}

print("="*72)

print("WARNING: flashing a custom firmware will VOID your G2's warranty and")

print("carries a real risk of BRICKING the device. Proceed only if you")

print("understand and accept that risk.")

print("="*72)

if skip:

print('--my-warranty-is-void supplied; skipping interactive confirmation.')

return

try:

resp=input(f'Type "{WARRANTY_PHRASE}" to continue: ')

except (EOFError, KeyboardInterrupt):

print("\nAborted."); sys.exit(1)

if resp.strip()!=WARRANTY_PHRASE:

global DEBUG, COMPONENT_RETRIES, BLOCK_NAK_RETRIES

p=argparse.ArgumentParser(description="Flash firmware onto Even Realities G2 glasses.")

p.add_argument('-c','--connection',

help="connection string: g2://droidbridge?phone=..&port=..&token=..&left=..&right=.. "

"or g2://local?left=..&right=..&addressType=public|random")

p.add_argument('-f','--firmware', help="path to the firmware image to flash")

p.add_argument('--recompute-checksums', metavar='IMAGE',

help="rewrite IMAGE's stored checksums in place to match its payloads "

"(component CRC32C + mainApp preamble CRC32), then exit. Use after a "

"length-preserving binary patch; does not connect or flash.")

p.add_argument('--lens', choices=['left','right','both'], default='both',

help="which lens to flash (default: both)")

p.add_argument('--stop-before', choices=STAGES, default='done',

help="stop before this stage (dry-run gating; default: done = full flash)")

p.add_argument('--my-warranty-is-void', action='store_true',

help="skip the interactive warranty confirmation (for automation)")

p.add_argument('--component-retries', type=int, default=COMPONENT_RETRIES,

help=f"whole-component re-flash attempts on END fail/timeout (default {COMPONENT_RETRIES})")

p.add_argument('--block-nak-retries', type=int, default=BLOCK_NAK_RETRIES,

help=f"in-place block resends on an explicit NAK (default {BLOCK_NAK_RETRIES})")

p.add_argument('--debug', action='store_true', help="print received frames")

args=p.parse_args(argv)

DEBUG=args.debug

COMPONENT_RETRIES=args.component_retries

BLOCK_NAK_RETRIES=args.block_nak_retries

if args.recompute_checksums:

try:

recompute_checksums(args.recompute_checksums)

except (OSError, ValueError) as e:

print(f"recompute-checksums failed: {e}"); sys.exit(1)

return

if not args.connection or not args.firmware:

p.error("-c/--connection and -f/--firmware are required (unless --recompute-checksums)")

try:

conn=parse_connection_string(args.connection)

except ValueError as e:

p.error(str(e))

try:

with open(args.firmware,'rb') as fh: img=fh.read()

except OSError as e:

print(f"cannot read firmware: {e}"); sys.exit(1)

try:

segs=validate_firmware(img)

except ValueError as e:

print(f"firmware validation failed: {e}"); sys.exit(1)

print(f"firmware ok: {args.firmware} ({len(img)}B, {len(segs)} segments: {[s['fn'] for s in segs]})")

confirm_warranty(args.my_warranty_is_void)

sides=['left','right'] if args.lens=='both' else [args.lens]

bridge=None

if conn['method']=='droidbridge':

bridge=Bridge(conn['base'], conn['token']); bridge.start_ws(); time.sleep(2)

failures=[]

for side in sides:

addr=conn[side]

print(f"\n=== flashing {side} lens ({addr}) ===")

if conn['method']=='local':

tp=LocalBleTransport(addr, conn.get('address_type'), side=side)

else:

tp=DroidBridgeTransport(bridge, addr)

try:

flash_lens(tp, img, segs, args.stop_before)

except Exception as e:

print(f"!!! {side} lens flash failed: {e}")

failures.append(side)

finally:

tp.close()

if bridge is not None:

bridge._stop=True

if failures:

print(f"\nFAILED lenses: {failures}"); sys.exit(1)