Extra Particle(Classic/Quantum) tiny srf-sim for palindrome luyh rotational variable(s) complex interpolation dynamics

ce_mmap_hqi.py

# ce_mmap_hqi.py - Extra Particle(Classic/Quantum) tiny srf-sim for palindrome
#                  luyh rotational variable(s) complex interpolation dynamics
#______________________________________________________________________________

import tempfile
import struct
import cmath
import math
import mmap
import time
import os
#______________________________________________________________________________
BUSH = b'RENOWNED'
HDR_FMT = '>Q'
DOUBLE_FMT = '>d'
HDR_SZ = struct.calcsize(HDR_FMT)
REC_HDR_FMT = 'BQ'
REC_HDR_SZ = struct.calcsize(REC_HDR_FMT)
REG_ENTRY_HDR_FMT = '>H'
REG_ENTRY_HDR_SZ = struct.calcsize(REG_ENTRY_HDR_FMT)
REG_FRAME_META_FMT = '>BQQQII'
REG_FRAME_META_SZ = struct.calcsize(REG_FRAME_META_FMT)
#______________________________________________________________________________

def pack_rec_header(rec_type: int, body_len: int) -> bytes:
    
    return struct.pack(REC_HDR_FMT, rec_type, body_len)
#______________________________________________________________________________

def unpack_rec_header(b: bytes) -> tuple:
    
    return struct.unpack(REC_HDR_FMT, b)
#______________________________________________________________________________

def pack_u64(x: int) -> bytes:
    
    return struct.pack(">Q", x)
#______________________________________________________________________________

def unpack_u64(b: bytes) -> int:
    
    return struct.unpack(">Q", b)[0]
#______________________________________________________________________________

def pack_u32(x: int) -> bytes:
    
    return struct.pack(">I", x)
#______________________________________________________________________________

def unpack_u32(b: bytes) -> int:
    
    return struct.unpack(">I", b)[0]
#______________________________________________________________________________

def pack_u16(x: int) -> bytes:
    
    return struct.pack(">H", x)
#______________________________________________________________________________

def unpack_u16(b: bytes) -> int:
    
    return struct.unpack(">H", b)[0]
#______________________________________________________________________________
#//////////////////////////////////////////////////////////////////////////////

class BinStore:
    
    def __init__(self, path: str, initial_size: int=8_000_000):
        
        self.path, mm = path, None
        init = os.path.exists(path)
        self.fd = os.open(path, os.O_RDWR | os.O_CREAT)
        if init:
            os.ftruncate(self.fd, initial_size)
            mm = mmap.mmap(self.fd, 0)
            try:
                mm[0:8] = BUSH
                mm[8:16] = pack_u64(16)
                mm[16:24] = pack_u64(0)
            finally:
                mm.close()
        self.mm = mmap.mmap(self.fd, 0)
        self.size = self.mm.size()
        self.write_pos = self._find_end()
#______________________________________________________________________________

    def _find_end(self) -> int:
        
        i = 24
        while i+REC_HDR_SZ <= self.size:
            hdr = self.mm[i:i+REC_HDR_SZ]
            if hdr == b'\x00'*REC_HDR_SZ:
                return i
            rtype, blen = unpack_rec_header(hdr)
            i+=REC_HDR_SZ+blen
        return i
#______________________________________________________________________________

    def close(self):
        
        try:
            self.mm.flush()
            self.mm.close()
        finally:
            os.close(self.fd)
#______________________________________________________________________________

    def _ensure_capacity(self, needed: int):
        
        if self.write_pos+needed <= self.size:
            return
        new_size = max(self.size*2, self.write_pos+needed+1024)
        self.mm.close()
        os.ftruncate(self.fd, new_size)
        self.mm = mmap.mmap(self.fd, 0)
        self.size = new_size
#______________________________________________________________________________

    def append_record(self, rec_type: int, body: bytes) -> int:
        
        hdr = pack_rec_header(rec_type, len(body))
        rec = hdr+body
        off = self.write_pos
        self._ensure_capacity(len(rec))
        self.mm[off:off+len(rec)] = rec
        self.write_pos+=len(rec)
        if rec_type == 1: self.mm[16:24] = pack_u64(off)
        return off
#______________________________________________________________________________

    def read_record_header(self, offset: int) -> tuple:
        
        hdr = self.mm[offset:offset+REC_HDR_SZ]
        return unpack_rec_header(hdr)
#______________________________________________________________________________

    def read_record_body(self, offset: int, blen: int) -> bytes:
        
        return bytes(self.mm[offset+REC_HDR_SZ: offset+REC_HDR_SZ+blen])
#______________________________________________________________________________

    def read_registry_offset(self) -> int:
        
        return unpack_u64(self.mm[16:24])
#______________________________________________________________________________

    def read_BUSH(self) -> bytes:
        
        return bytes(self.mm[0:8])
#______________________________________________________________________________
#//////////////////////////////////////////////////////////////////////////////

class Base:
    __slots__ = ('_id',)
    
    def __init__(self, _id: str):
        self._id = _id
#______________________________________________________________________________
#//////////////////////////////////////////////////////////////////////////////

class ExtraParticle(Base):
    
    __slots__ = Base.__slots__ + ('store_path', 'rec_off')
    
    def __init__(self, _id: str, store_path: str, rec_off: int=0):
        
        super().__init__(_id)
        self.store_path, self.rec_off = store_path, rec_off
#______________________________________________________________________________

    def read(self) -> list:
        
        if self.rec_off == 0:
            return None
        s = BinStore(self.store_path)
        rtype, blen = s.read_record_header(self.rec_off)
        if rtype != 3:
            s.close()
            raise ValueError('expected extra record...')
        body = s.read_record_body(self.rec_off, blen)
        s.close()
        return _unpack_complex_vec(body)
#______________________________________________________________________________
#//////////////////////////////////////////////////////////////////////////////

class SystemState(Base):
    
    __slots__ = Base.__slots__ + ('store_path', 'rec_off', 'dim')
    
    def __init__(self, _id: str, store_path: str, rec_off: int):
        
        super().__init__(_id)
        self.store_path = store_path
        self.rec_off = rec_off
        s = BinStore(self.store_path)
        rtype, blen = s.read_record_header(self.rec_off)
        if rtype != 2:
            s.close()
            raise ValueError("expected system record")
        body = s.read_record_body(self.rec_off, blen)
        s.close()
        vec = _unpack_complex_vec(body)
        self.dim = len(vec)
#______________________________________________________________________________

    def read_vec(self) -> list:
        
        s = BinStore(self.store_path)
        rtype, blen = s.read_record_header(self.rec_off)
        body = s.read_record_body(self.rec_off, blen)
        s.close()
        return _unpack_complex_vec(body)
#______________________________________________________________________________
#//////////////////////////////////////////////////////////////////////////////

class ReferenceFrame(Base):
    
    __slots__ = Base.__slots__ + ('store_path', 'frame_id', 'has_joint', 'joint_off',
                                  'system_rec', 'extra_rec', 'sys_dim', 'extra_dim',
                                  'children')
    
    def __init__(self, _id: str, store_path: str, frame_id: str,
                 has_joint: bool, joint_off: int, system_rec: int,
                 extra_rec: int, sys_dim: int, extra_dim: int):
                     
        super().__init__(_id)
        self.store_path, self.frame_id, self.has_joint = store_path, frame_id, has_joint
        self.joint_off, self.system_rec, self.extra_rec = joint_off, system_rec, extra_rec
        self.sys_dim, self.extra_dim = sys_dim, extra_dim
        self.children: List['ReferenceFrame'] = []
#______________________________________________________________________________

    def add_child(self, child: 'ReferenceFrame'):
        
        self.children.append(child)
#______________________________________________________________________________

def _pack_complex_vec(vec: list) -> bytes:
    
    out = bytearray()
    out+=pack_u32(len(vec))
    for z in vec:
        out+=struct.pack(DOUBLE_FMT, float(z.real))
        out+=struct.pack(DOUBLE_FMT, float(z.imag))
    return bytes(out)
#______________________________________________________________________________

def _unpack_complex_vec(body: bytes) -> list:
    
    cnt, vec, off = unpack_u32(body[0:4]), [], 4
    for _ in range(cnt):
        re = struct.unpack(DOUBLE_FMT, body[off:off+8])[0]; off+=8
        im = struct.unpack(DOUBLE_FMT, body[off:off+8])[0]; off+=8
        vec.append(complex(re, im))
    return vec
#______________________________________________________________________________

def pack_registry(reg: list) -> bytes:
    
    out = bytearray()
    for fid, meta in reg.items():
        fid_b = fid.encode('utf-8')
        out+=pack_u16(len(fid_b))
        out+=fid_b
        has_joint = 1 if meta.get('joint_off', 0) else 0
        out+=struct.pack(REG_FRAME_META_FMT, has_joint, meta.get('joint_off', 0),
                         meta.get('system_off', 0), meta.get('extra_off', 0),
                         meta.get('sys_dim', 0), meta.get('extra_dim', 0))
    return bytes(out)
#______________________________________________________________________________

def unpack_registry(body: bytes) -> dict:
    
    i, reg = 0, {}
    while i < len(body):
        fid_len = unpack_u16(body[i:i+2]); i+=2
        fid = body[i:i+fid_len].decode('utf-8'); i+=fid_len
        has_joint = body[i]; i+=1
        joint_off = unpack_u64(body[i:i+8]); i+=8
        system_off = unpack_u64(body[i:i+8]); i+=8
        extra_off = unpack_u64(body[i:i+8]); i+=8
        sys_dim = unpack_u32(body[i:i+4]); i+=4
        extra_dim = unpack_u32(body[i:i+4]); i+=4
        reg[fid] = {'joint_off':joint_off, 'system_off':system_off,
                    'extra_off':extra_off, 'sys_dim':sys_dim,
                    'extra_dim':extra_dim}
    return reg
#______________________________________________________________________________

def mat_vec(U: list, v: list) -> list:
    
    return [sum(ui[j]*v[j] for j in range(len(v))) for ui in U]
#______________________________________________________________________________

def dagger(U: list) -> list:
    
    return [list(col) for col in zip(*[[complex(x).conjugate() for x in row] for row in U])]
#______________________________________________________________________________

def is_unitary(U: list, eps: float=1e-9) -> bool:
    
    n, ut = len(U), dagger(U)
    prod = [[sum(U[i][k]*ut[k][j] for k in range(n)) for j in range(n)] for i in range(n)]
    for i in range(n):
        for j in range(n):
            target = 1.0 if i == j else 0.0
            if abs(prod[i][j]-target) > eps:
                return False
    return True
#______________________________________________________________________________

def pure_state_density(vec: list) -> list:
    
    return [[vec[i]*complex(vec[j]).conjugate() for j in range(len(vec))] for i in range(len(vec))]
#______________________________________________________________________________

def partial_trace(rho: list, sys_dim: int, extra_dim: int, trace_over: str) -> list:
    
    if trace_over == 'extra':
        res = [[0+0j for _ in range(sys_dim)] for _ in range(sys_dim)]
        for a in range(sys_dim):
            for b in range(sys_dim):
                s = 0+0j
                for k in range(extra_dim):
                    i, j = a*extra_dim+k, b*extra_dim+k
                    s+=rho[i][j]
                res[a][b] = s
        return res
    else:
        res = [[0+0j for _ in range(extra_dim)] for _ in range(extra_dim)]
        for a in range(extra_dim):
            for b in range(extra_dim):
                s = 0+0j
                for k in range(sys_dim):
                    i, j = k*extra_dim+a, k*extra_dim+b
                    s+=rho[i][j]
                res[a][b] = s
        return res
#______________________________________________________________________________

def is_classical_density(rho: list, eps: float=1e-9) -> bool:
    
    n = len(rho)
    for i in range(n):
        for j in range(n):
            if i != j and abs(rho[i][j]) > eps:
                return False
    return True
#______________________________________________________________________________
#//////////////////////////////////////////////////////////////////////////////

class Interpreter:
    
    __slots__ = ('store_path', 'attached_frame_id')
    
    def __init__(self, store_path: str):
        
        self.store_path, self.attached_frame_id = store_path, None
#______________________________________________________________________________

    def attach(self, frame_id: str):
        
        self.attached_frame_id = frame_id
#______________________________________________________________________________

    def detach(self):
        
        self.attached_frame_id = None
#______________________________________________________________________________

    def is_extra_classical(self) -> bool:
        
        if self.attached_frame_id is None:
            raise RuntimeError('not attached')
        s = BinStore(self.store_path)
        reg_off = s.read_registry_offset()
        if reg_off == 0:
            s.close()
            raise RuntimeError('no registry')
        rtype, blen = s.read_record_header(reg_off)
        reg = unpack_registry(s.read_record_body(reg_off, blen))
        info = reg.get(self.attached_frame_id)
        if info is None:
            s.close()
            raise RuntimeError('frame not in registry')
        if info['joint_off'] != 0:
            rec = s.read_record_body(info['joint_off'], s.read_record_header(info['joint_off'])[1])
            s.close()
            rho = pure_state_density(_unpack_complex_vec(rec))
            red_extra = partial_trace(rho, info['sys_dim'], info['extra_dim'], trace_over='sys')
            return is_classical_density(red_extra)
        else:
            s.close()
            return info['extra_off'] == 0
#______________________________________________________________________________

    def apply_unitary_system(self, U: list) -> int:
        
        if self.attached_frame_id is None:
            raise RuntimeError('not attached')
        s = BinStore(self.store_path)
        reg_off = s.read_registry_offset()
        if reg_off == 0:
            s.close()
            raise RuntimeError('no registry')
        rtype, blen = s.read_record_header(reg_off)
        reg = unpack_registry(s.read_record_body(reg_off, blen))
        info = reg.get(self.attached_frame_id)
        if info is None:
            s.close()
            raise RuntimeError('frame not in registry')
        if info['joint_off'] != 0:
            joint_body = s.read_record_body(info['joint_off'], s.read_record_header(info['joint_off'])[1])
            joint_vec = _unpack_complex_vec(joint_body)
            sys_dim = info['sys_dim']; extra_dim = info['extra_dim']
            if len(U) != sys_dim or not is_unitary(U):
                s.close()
                raise ValueError('unitary mismatch')
            new_joint = [0+0j]*(sys_dim*extra_dim)
            for i in range(sys_dim):
                for j in range(extra_dim):
                    acc = 0+0j
                    for k in range(sys_dim): acc+=U[i][k]*joint_vec[k*extra_dim+j]
                    new_joint[i*extra_dim+j] = acc
            norm = math.sqrt(sum(abs(x)**2 for x in new_joint))
            newn = [x/norm for x in new_joint]
            body_new = _pack_complex_vec(newn)
            new_off = s.append_record(4, body_new)
            reg2 = dict(reg)
            reg2[self.attached_frame_id] = dict(info)
            reg2[self.attached_frame_id]['joint_off'] = new_off
            body_reg = pack_registry(reg2)
            s.append_record(1, body_reg)
            s.close()
            return new_off
        else:
            sys_body = s.read_record_body(info['system_off'], s.read_record_header(info['system_off'])[1])
            vec = _unpack_complex_vec(sys_body)
            if len(U) != len(vec) or not is_unitary(U):
                s.close()
                raise ValueError('unitary mismatch')
            new = mat_vec(U, vec)
            norm = math.sqrt(sum(abs(x)**2 for x in new))
            newn = [x/norm for x in new]
            body_new = _pack_complex_vec(newn)
            new_off = s.append_record(2, body_new)
            reg2 = dict(reg)
            reg2[self.attached_frame_id] = dict(info)
            reg2[self.attached_frame_id]['system_off'] = new_off
            body_reg = pack_registry(reg2)
            s.append_record(1, body_reg)
            s.close()
            return new_off
#______________________________________________________________________________

def create_initial_store(path: str) -> None:
    
    s, sysA, joint = BinStore(path), [1+0j, 0+0j], []
    off_sysA = s.append_record(2, _pack_complex_vec(sysA))
    for i in range(2):
        for j in range(2): joint.append(1/math.sqrt(2) if i == j else 0+0j)
    off_jointQ = s.append_record(4, _pack_complex_vec(joint))
    reg = {'A': {'joint_off':0, 'system_off':off_sysA, 'extra_off':0, 'sys_dim':2, 'extra_dim':1},
           'Q': {'joint_off':off_jointQ, 'system_off':0, 'extra_off':0, 'sys_dim':2, 'extra_dim':2},}
    s.append_record(1, pack_registry(reg))
    s.close()
#______________________________________________________________________________

def measure(store_path: str, frame_id: str, out_path: str):
    
    interp = Interpreter(store_path)
    interp.attach(frame_id)
    res = interp.is_extra_classical()
    with open(out_path, 'w') as f: f.write('classical\n' if res else 'quantum\n')
#______________________________________________________________________________

def apply_then_measure(store_path: str, frame_id: str, out_path: str):
    
    interp = Interpreter(store_path)
    interp.attach(frame_id)
    H = [[1/math.sqrt(2), 1/math.sqrt(2)], [1/math.sqrt(2), -1/math.sqrt(2)]]
    interp.apply_unitary_system([[complex(x) for x in row] for row in H])
    res = interp.is_extra_classical()
    with open(out_path, 'w') as f: f.write('classical\n' if res else 'quantum\n')
#______________________________________________________________________________

# q8p.py
#___________________________________________________________________________________

class AA:
    
    slots = ('_u_tlr', '_u_srv')

    def __init__(self):
        pass
#___________________________________________________________________________________

    def set_aa_short_bus(self, vhs_com: tuple) -> list:
        
        mtx, vhs, chs, chn = vhs_com[0], vhs_com[1], vhs_com[2], vhs_com[3]
        t = time.ctime().replace(' ', '').split(':')
        t = math.log(int(f'{"".join([c for c in t if c.isdigit()])}{t[1]}')*vhs)
        mtx_len = len(mtx)
        if mtx_len%2 != 0:
            raise Exception(f'<q8p> variance matrices length({len(mtx)}) not even')
        mtx = [self.aa_vhs_commutator(mtx[i], mtx[i+1], vhs, chs, chn) for i in range(mtx_len-1)]
        for i in range(mtx_len-1):
            print(self.solve_aa_short_bus_pvt_cycle(t, mtx[i]))
#___________________________________________________________________________________

    def solve_aa_short_bus_pvt_cycle(self, t: float, mt: tuple) -> tuple:
        
        hd, R, tlr, srv = mt[0], mt[1], mt[2], mt[3]
        if tlr < srv:
            gr = self.get_pvt_gate_range(False, t, tlr, srv)
        else:
            gr = self.get_pvt_gate_range(True, t, tlr, srv)
        if gr+hd > abs(cmath.sqrt(R.real)*cmath.sqrt(R.imag))**hd/gr:
            return math.floor(abs((t+gr)*cmath.cos(R))+tlr), math.floor((gr+hd*t)/t)
        else:
            return 1, math.floor(((gr*hd)/abs(cmath.sin(R)))/t)
#___________________________________________________________________________________

    def get_pvt_gate_range(self, isRested: bool, t: float, tlr: float, srv: float) -> int:
        
        x = tlr*srv/math.sqrt(t)
        if isRested:
            return math.ceil(abs(cmath.sqrt(complex((x+t)/math.tanh(t*2), x*t)))/(tlr+srv))
        return math.ceil(abs(cmath.sqrt(complex(x+t, x*t)))/math.pi)
#___________________________________________________________________________________

    def aa_vhs_commutator(self, vrncA: list, vrncB: list, vhs: float, chs: complex, chn: complex) -> tuple:
        
        c_ttl = (chs.real+chs.imag)+(chn.real+chn.imag)
        if sum(vrncA) < c_ttl or sum(vrncB) < c_ttl:
            raise ValueError(f'<q8p> @chs + @chn summation({c_ttl}) > listed variances')
        hd = (int(vhs*sum(vrncA)*sum(vrncB)*math.pi)*int(math.sqrt(abs(chs)+abs(chn))))*math.pi
        R = min(math.cos(abs(chs)+abs(chn)), math.sin(vhs+abs(chn)))
        t, q = cmath.tan(((chs*chn)/math.pi)+R), math.pi/2
        k = (vrncA[0]*vrncA[len(vrncA)-1]*vrncB[0]*vrncB[len(vrncB)-1])**t
        ds, s, c, r, rv = math.ceil(abs(cmath.exp(R+chs-chn)))**vhs*t, hd, 1, set(), set()
        t+=k**R
        while s > -1:
            for h in vrncA:
                t+=h-ds
                m = (t*k+s)-cmath.sqrt((t-ds)+c)-t+(ds*(c*2))
                if m.real < 0:
                    q = abs(c-t)
                    for n in vrncB:
                        w = sum([math.sin(abs(m)+q) for x in vrncB])
                        a_ds, a_ks = abs(ds), abs(cmath.cos(k))
                        if ((w/(c/2))+(R*n))-a_ds <= (vhs+c+R)/(math.pi-abs(t)+a_ks):
                            r.add(w)
                else:
                    t-=vhs
                    n = math.sin((abs(cmath.cos(t)+ds)//(c*abs(t))))
                    if n > 0.0:
                        n, cn = math.floor(n*c*abs(cmath.tan(k+m))), 0
                        lm = [complex((t*c)/(n+1)/q, -1) for x in range(n)]
                        r = list(r)
                        r.sort()
                        d = len(r)
                        if d > 0:
                            while cn < len(lm):
                                l = -(abs(lm[cn])-cn)/q
                                if cn >= d: cn = 0
                                l = l+r[cn]
                                if l > 0.0: rv.add(l)
                                cn+=1
                        r = set(r)
                a_m = abs(m)
                if a_m > 0: r.add(a_m/(abs(cmath.sqrt(chn))+math.pi))                    
            s+=-1
            c+=1
        r, rv = list(r), list(rv)
        r.sort()
        rv.sort()
        tlr = -math.log((abs(cmath.tan(t))/hd)*abs(cmath.log(cmath.sqrt(m+hd))))
        return hd, R+cmath.log(t), tlr, sum(r)/sum(rv)/cn
#___________________________________________________________________________________

def test():

    cls = AA()
    
    stackA = [0.2, 0.5, 0.8, 1.1, 1.3, 1.5]
    stackB = [0.7, 0.9, 1.1, 1.5, 1.8, 2.0]
    stackC = [0.1, 0.4, 1.2, 1.8, 1.9, 2.2]
    stackD = [0.2, 0.5, 0.8, 1.0, 1.4, 2.6]
    stackE = [0.1, 0.2, 0.6, 0.8, 1.0, 1.7]
    stackF = [0.2, 0.5, 0.8, 1.0, 1.4, 2.6]
    stackG = [0.1, 0.4, 0.9, 1.2, 1.4, 2.3]
    stackH = [0.6, 0.9, 1.2, 1.6, 1.9, 2.8]
    grid = [stackA, stackB, stackC, stackD, stackE, stackF, stackG, stackH]
    vhs = 0.2
    
    chs = complex(1.0, 0.2)
    chn = complex(1.0, 0.5)
    
    cls.set_aa_short_bus((grid, vhs, chs, chn))
    
test()