Separate module for XRay classes

games/stalkeranomaly/XRIO.py

@@ -0,0 +1,148 @@
+# -*- encoding: utf-8 -*-
+from __future__ import annotations
+
+import io
+import struct
+from typing import Optional, Tuple
+
+from .XRMath import IVec3
+
+
+class XRReader:
+    def __init__(self, buffer: bytes):
+        self._buffer = buffer
+        self._pos = 0
+
+    def __len__(self) -> int:
+        return len(self._buffer)
+
+    def _read(self, size: int) -> Tuple[bytes, int]:
+        pos = min(len(self._buffer), self._pos + size)
+        buffer = self._buffer[self._pos : pos]
+        return (buffer, pos)
+
+    def read(self, size: int = -1) -> bytes:
+        if size < 0:
+            size = len(self._buffer)
+        if len(self._buffer) <= self._pos:
+            return b""
+        (buffer, pos) = self._read(size)
+        self._pos = pos
+        return buffer
+
+    def peek(self, size: int = -1) -> bytes:
+        if size < 0:
+            size = len(self._buffer)
+        if len(self._buffer) <= self._pos:
+            return b""
+        (buffer, pos) = self._read(size)
+        return buffer
+
+    def seek(self, pos: int, whence: int = io.SEEK_SET) -> int:
+        if whence == 0:
+            if pos < 0:
+                raise ValueError(f"negative seek position {pos}")
+            self._pos = pos
+        elif whence == 1:
+            self._pos = max(0, self._pos + pos)
+        elif whence == 2:
+            self._pos = max(0, len(self._buffer) + pos)
+        else:
+            raise ValueError("unsupported whence value")
+        return self._pos
+
+    def elapsed(self) -> int:
+        return len(self._buffer) - self._pos
+
+    def eof(self) -> bool:
+        return self.elapsed() <= 0
+
+    def u8(self) -> int:
+        return int(struct.unpack("<B", self.read(1))[0])
+
+    def s8(self) -> int:
+        return int(struct.unpack("<b", self.read(1))[0])
+
+    def u16(self) -> int:
+        return int(struct.unpack("<H", self.read(2))[0])
+
+    def s16(self) -> int:
+        return int(struct.unpack("<h", self.read(2))[0])
+
+    def u32(self) -> int:
+        return int(struct.unpack("<I", self.read(4))[0])
+
+    def s32(self) -> int:
+        return int(struct.unpack("<i", self.read(4))[0])
+
+    def u64(self) -> int:
+        return int(struct.unpack("<Q", self.read(8))[0])
+
+    def s64(self) -> int:
+        return int(struct.unpack("<q", self.read(8))[0])
+
+    def bool(self) -> bool:
+        return bool(struct.unpack("<?", self.read(1))[0])
+
+    def float(self) -> float:
+        return float(struct.unpack("<f", self.read(4))[0])
+
+    def str(self) -> str:
+        chars = bytearray()
+        while not self.eof():
+            c = self.read(1)
+            if c == b"\x00":
+                return str(chars, "utf-8")
+            else:
+                chars += c
+        return ""
+
+    def fvec3(self) -> IVec3:
+        (f1, f2, f3) = struct.unpack("<fff", self.read(12))
+        return IVec3(f1, f2, f3)
+
+
+class XRStream(XRReader):
+    def __init__(self, buffer: bytes):
+        super().__init__(buffer)
+        self.last_pos: int = 0
+
+    def find_chunk(self, id: int) -> Optional[int]:
+        dw_type = 0
+        dw_size = 0
+        success = False
+        if self.last_pos != 0:
+            self.seek(self.last_pos)
+            dw_type = self.u32()
+            dw_size = self.u32()
+            if (dw_type & (~(1 << 31))) == id:
+                success = True
+
+        if not success:
+            self.seek(0)
+            while not self.eof():
+                dw_type = self.u32()
+                dw_size = self.u32()
+                if (dw_type & (~(1 << 31))) == id:
+                    success = True
+                    break
+                else:
+                    self.seek(dw_size, io.SEEK_CUR)
+
+            if not success:
+                self.last_pos = 0
+                return None
+
+        if (self._pos + dw_size) < len(self._buffer):
+            self.last_pos = self._pos + dw_size
+        else:
+            self.last_pos = 0
+
+        return dw_size
+
+    def open_chunk(self, id: int) -> Optional[XRStream]:
+        size = self.find_chunk(id)
+        if size and size != 0:
+            data = self.read(size)
+            return XRStream(data)
+        return None

games/stalkeranomaly/XRMath.py

@@ -0,0 +1,49 @@
+# -*- encoding: utf-8 -*-
+
+
+class IVec3:
+    def __init__(self, x, y, z):
+        self.x = x
+        self.y = y
+        self.z = z
+
+    def __str__(self) -> str:
+        return f"{self.x}, {self.y}, {self.z}"
+
+    def set(self, x, y, z):
+        self.x = x
+        self.y = y
+        self.z = z
+
+
+class IVec4(IVec3):
+    def __init__(self, x, y, z, w):
+        super().__init__(x, y, z)
+        self.w = w
+
+    def __str__(self) -> str:
+        return f"{self.x}, {self.y}, {self.z}, f{self.w}"
+
+    def set(self, x, y, z, w):
+        super().set(x, y, z)
+        self.w = w
+
+
+class IFlag:
+    def __init__(self, flag: int):
+        self._flag = flag
+
+    def __str__(self) -> str:
+        return str(self._flag)
+
+    def assign(self, mask):
+        self._flag = mask
+
+    def has(self, mask) -> bool:
+        return bool((self._flag & mask) == mask)
+
+    def set(self, mask):
+        self._flag |= mask
+
+    def remove(self, mask):
+        self._flag &= ~mask

games/stalkeranomaly/XRNET.py

@@ -0,0 +1,47 @@
+# -*- encoding: utf-8 -*-
+
+from .XRIO import XRReader
+from .XRMath import IVec3, IVec4
+
+
+class XRNETState:
+    def __init__(self):
+        self.position = IVec3(0.0, 0.0, 0.0)
+        self.quaternion = IVec4(0.0, 0.0, 0.0, 0.0)
+        self.enabled = False
+
+    def read(self, reader: XRReader, fmin: IVec3, fmax: IVec3):
+        self.position = self.fvec_q8(reader, fmin, fmax)
+        self.quaternion = self.fqt_q8(reader)
+        self.enabled = bool(reader.u8())
+
+    def clamp(self, val: float, low: float, high: float) -> float:
+        if val < low:
+            return low
+        elif val > high:
+            return high
+        else:
+            return val
+
+    def fvec_q8(self, reader: XRReader, fmin: IVec3, fmax: IVec3):
+        vec = IVec3(0.0, 0.0, 0.0)
+        vec.x = self.f_q8(reader, fmin.x, fmax.x)
+        vec.y = self.f_q8(reader, fmin.y, fmax.y)
+        vec.z = self.f_q8(reader, fmin.z, fmax.z)
+        vec.x = self.clamp(vec.x, fmin.x, fmax.x)
+        vec.y = self.clamp(vec.x, fmin.y, fmax.y)
+        vec.z = self.clamp(vec.z, fmin.z, fmax.z)
+
+    def fqt_q8(self, reader: XRReader):
+        vec = IVec4(0.0, 0.0, 0.0, 0.0)
+        vec.x = self.f_q8(reader, -1.0, 1.0)
+        vec.y = self.f_q8(reader, -1.0, 1.0)
+        vec.z = self.f_q8(reader, -1.0, 1.0)
+        vec.w = self.f_q8(reader, -1.0, 1.0)
+        vec.x = self.clamp(vec.x, -1.0, 1.0)
+        vec.y = self.clamp(vec.y, -1.0, 1.0)
+        vec.z = self.clamp(vec.z, -1.0, 1.0)
+        vec.w = self.clamp(vec.w, -1.0, 1.0)
+
+    def f_q8(self, reader: XRReader, fmin: float, fmax: float):
+        return (float(reader.u8()) / 255.0) * (fmax - fmin) + fmin

games/stalkeranomaly/XRObject.py

@@ -0,0 +1,277 @@
+# -*- encoding: utf-8 -*-
+
+import io
+from typing import List
+from enum import IntFlag
+
+from .XRIO import XRReader
+from .XRMath import IVec3, IFlag
+from .XRNET import XRNETState
+
+
+class XRFlag(IntFlag):
+    CHUNK_ALIFE = 0x0
+    CHUNK_SPAWN = 0x1
+    CHUNK_OBJECT = 0x2
+    CHUNK_GAME_TIME = 0x5
+    CHUNK_REGISTRY = 0x9
+
+    SPAWN_VERSION = 1 << 5
+
+    MSG_UPDATE = 0x0
+    MSG_SPAWN = 0x1
+
+    TRADER_INFINITE_AMMO = 0x1
+
+
+class XRAbstract:
+    def __init__(self, name: str = ""):
+        self._valid = False
+        self.name = name
+        self.name_replace = ""
+        self.rp = 0xFE
+        self.position = IVec3(0.0, 0.0, 0.0)
+        self.angle = IVec3(0.0, 0.0, 0.0)
+        self.respawn_time = 0
+        self.id = 0xFFFF
+        self.id_parent = 0xFFFF
+        self.id_phantom = 0xFFFF
+        self.flags = IFlag(0)
+        self.version = 0
+        self.game_type = IFlag(0)
+        self.script_version = 0
+        self.client_data: List[int] = []
+        self.spawn_id = 0
+        self.ini_str = ""
+
+    def read_spawn(self, reader: XRReader):
+        spawn = reader.u16()
+        if spawn != XRFlag.MSG_SPAWN:
+            self._valid = False
+            return
+        self.name = reader.str()
+        self.name2 = reader.str()
+        reader.seek(1, io.SEEK_CUR)  # temp_gt
+        self.rp = reader.u8()
+        self.position = reader.fvec3()
+        self.angle = reader.fvec3()
+        self.respawn_time = reader.u16()
+        self.id = reader.u16()
+        self.id_parent = reader.u16()
+        self.id_phantom = reader.u16()
+        self.flags = IFlag(reader.u16())
+        if self.flags.has(XRFlag.SPAWN_VERSION):
+            self.version = reader.u16()
+        if self.version == 0:
+            reader._pos -= 2
+            return
+        if self.version > 120:
+            self.game_type.set(reader.u16())
+        if self.version > 69:
+            self.script_version = reader.u16()
+        if self.version > 70:
+            cl_size = 0
+            if self.version > 93:
+                cl_size = reader.u16()
+            else:
+                cl_size = reader.u8()
+            if cl_size > 0:
+                for x in range(cl_size):
+                    data = reader.u8()
+                    self.client_data.append(data)
+        if self.version > 79:
+            self.spawn_id = reader.u16()
+        self._valid = True
+
+    def __bool__(self):
+        return self._valid
+
+
+class XRVisual:
+    def __init__(self):
+        self.visual_name = ""
+        self.startup_animation = ""
+        self.flags = IFlag(0)
+
+    def read_visual(self, reader: XRReader, version: int):
+        self.visual_name = reader.str()
+        self.flags = reader.u8()
+
+
+class XRBoneData:
+    def __init__(self):
+        self.bones_mask = -1
+        self.root_bone = 0
+        self.min = IVec3(0.0, 0.0, 0.0)
+        self.max = IVec3(0.0, 0.0, 0.0)
+        self.bones = []
+
+    def load(self, reader: XRReader):
+        self.bones_mask = reader.u64()
+        self.root_bone = reader.u16()
+        self.min = reader.fvec3()
+        self.max = reader.fvec3()
+        bones_count = reader.u16()
+        for x in range(bones_count):
+            bone = XRNETState()
+            bone.read(reader, self.min, self.max)
+            self.bones.append(bone)
+
+
+class XRSkeleton:
+    def __init__(self):
+        self.source_id = -1
+        self.saved_bones = XRBoneData()
+
+    def read_state(self, reader: XRReader):
+        self.visual_animation = reader.str()
+        flags = IFlag(reader.u8())
+        self.source_id = reader.u16()
+        if flags.has(4):
+            self.saved_bones.load(reader)
+
+
+class XRObject(XRAbstract):
+    def __init__(self):
+        super().__init__()
+        self.graph_id = 0
+        self.distance = 0.0
+        self.direct_control = True
+        self.node_id = 0
+        self.story_id = -1
+        self.spawn_story_id = -1
+
+    def read_spawn(self, reader: XRReader):
+        super().read_spawn(reader)
+        size = reader.u16()
+        if size <= 2:
+            self._valid = False
+            return
+        self.read_state(reader)
+
+    def read_state(self, reader: XRReader):
+        self.graph_id = reader.u16()
+        self.distance = reader.float()
+        self.direct_control = bool(reader.u32())
+        self.node_id = reader.u32()
+        self.flags.set(reader.u32())
+        self.ini_str = reader.str()
+        self.story_id = reader.u32()
+        self.spawn_story_id = reader.u32()
+
+    def read_update(self, reader: XRReader):
+        update = reader.u16()
+        if update != XRFlag.MSG_UPDATE:
+            self._valid = False
+            return
+
+
+class XRDynamicObject(XRObject):
+    def __init__(self):
+        super().__init__()
+        self.time_id = -1
+        self.switch_counter = 0
+
+
+class XRDynamicObjectVisual(XRDynamicObject, XRVisual):
+    def read_state(self, reader: XRReader):
+        XRDynamicObject.read_state(self, reader)
+        if self.version > 31:
+            XRVisual.read_visual(self, reader, self.version)
+
+
+class XRCreatureAbstract(XRDynamicObjectVisual):
+    def __init__(self):
+        super().__init__()
+        self.team = 0
+        self.squad = 0
+        self.group = 0
+        self.health = 1.0
+        self.dynamic_out = []
+        self.dynamic_in = []
+        self.killer_id = -1
+        self.death_time = 0
+
+    def read_state(self, reader: XRReader):
+        super().read_state(reader)
+        self.team = reader.u8()
+        self.squad = reader.u8()
+        self.group = reader.u8()
+        self.health = reader.float() * 100
+
+        for x in range(reader.u32()):
+            _id = reader.u16()
+            self.dynamic_out.append(_id)
+
+        for x in range(reader.u32()):
+            _id = reader.u16()
+            self.dynamic_in.append(_id)
+
+        self.killer_id = reader.u16()
+        self.death_time = reader.u64()
+
+
+class XRTraderAbstract:
+    def __init__(self):
+        self.money = 0
+        self.max_item_mass = 0
+        self.character_name = ""
+        self.character_name_str = ""
+        self.character_profile = ""
+        self.specific_character = ""
+        self.community_index = -1
+        self.rank = -1  # -MAX_INT?
+        self.reputation = -1  # -MAX_INT?
+        self.dead_body_can_take = True
+        self.dead_body_closed = False
+        self.trader_flags = IFlag(0)
+        self.trader_flags.remove(XRFlag.TRADER_INFINITE_AMMO)
+
+    def read_state(self, reader: XRReader):
+        self.money = reader.u32()
+        self.specific_character = reader.str()
+        self.trader_flags.assign(reader.u32())
+        self.character_profile = reader.str()
+        self.community_index = reader.s32()
+        self.rank = reader.s32()
+        self.reputation = reader.s32()
+        self.character_name_str = reader.str()
+        self.dead_body_can_take = reader.u8() == 1
+        self.dead_body_closed = reader.u8() == 1
+
+
+class XRCreatureActor(XRCreatureAbstract, XRTraderAbstract, XRSkeleton):
+    def __init__(self):
+        XRCreatureAbstract.__init__(self)
+        XRTraderAbstract.__init__(self)
+        XRSkeleton.__init__(self)
+        self.state = 0
+        self.acceleration = IVec3(0.0, 0.0, 0.0)
+        self.velocity = IVec3(0.0, 0.0, 0.0)
+        self.radiation = 0.0
+        self.weapon = 0
+        self.num_items = 0
+        self.holder_id = -1
+
+    def read_spawn(self, reader: XRReader):
+        XRCreatureAbstract.read_spawn(self, reader)
+
+    def read_state(self, reader: XRReader):
+        XRCreatureAbstract.read_state(self, reader)
+        XRTraderAbstract.read_state(self, reader)
+        XRSkeleton.read_state(self, reader)
+        self.holder_id = reader.u16()
+
+    def read_update(self, reader: XRReader):
+        XRCreatureAbstract.read_update(self, reader)
+        # XRTraderAbstract.read_update(self, reader)  # Future?
+        self.state = reader.u16()
+        # acceleration (r_sdir)
+        reader.seek(2, io.SEEK_CUR)  # u16
+        reader.seek(4, io.SEEK_CUR)  # float
+        # velocity (r_sdir)
+        reader.seek(2, io.SEEK_CUR)  # u16
+        reader.seek(4, io.SEEK_CUR)  # float
+        self.radiation = reader.float()
+        self.weapon = reader.u8()
+        self.num_items = reader.u16()

games/stalkeranomaly/XRSave.py

@@ -0,0 +1,152 @@
+# -*- encoding: utf-8 -*-
+
+import io
+import struct
+import lzokay
+from pathlib import Path
+from typing import Optional
+from datetime import datetime
+
+from .XRObject import XRCreatureActor, XRFlag
+from .XRIO import XRReader, XRStream
+
+
+class XRSave:
+    filepath: Path
+    player: XRCreatureActor
+
+    _factions = {
+        0: "Loner",
+        1: "Monster",
+        2: "Trader",
+        3: "Army",
+        4: "Sin",
+        5: "Bandit",
+        6: "Duty",
+        7: "Ecologist",
+        8: "Freedom",
+        9: "Mercenary",
+        10: "Army",
+        11: "Monolith",
+        12: "Sin",
+        13: "Loner",
+        14: "Zombified",
+        15: "Clear Sky",
+        16: "UNISG",
+        17: "Renegade",
+        18: "Loner",
+        19: "Bandit",
+        20: "Duty",
+        21: "Freedom",
+        22: "Clear Sky",
+        23: "Ecologist",
+        24: "Mercenary",
+        25: "Military",
+        26: "Monolith",
+        27: "Zombified",
+        28: "Sin",
+        29: "UNISG",
+        30: "Renegade",
+        31: "Participant",
+    }
+
+    _ranks = {
+        1999: "Novice",
+        3999: "Trainee",
+        6999: "Experienced",
+        9999: "Professional",
+        14999: "Veteran",
+        20999: "Expert",
+        27999: "Master",
+        "max": "Legend",
+    }
+
+    _reputation = {
+        -2000: "Terrible",
+        -1500: "Really Bad",
+        -1000: "Very Bad",
+        -500: "Bad",
+        499: "Netural",
+        999: "Good",
+        1499: "Very Good",
+        1999: "Really Good",
+        "max": "Excellent",
+    }
+
+    def __init__(self, filepath: Path):
+        self.filepath = filepath
+        self.fetchInfo()
+        with open(filepath, "rb") as file:
+            stream = self.readFile(file)
+            if stream:
+                self.readObject(stream)
+
+    def fetchInfo(self):
+        save_clean = self.filepath.name.split(".scop", 1)[0]
+        save_split = save_clean.split(" - ", 1)
+        self.user = save_split[0]
+        save_end = save_split[1].split("_", 1)
+        if len(save_end) > 1:
+            self.save_fmt = f"{save_end[0]} (#{save_end[1]})".title()
+        else:
+            self.save_fmt = f"{save_end[0]}".title()
+        self.time = self.filepath.stat().st_mtime
+        self.time_date = datetime.fromtimestamp(self.time)
+        self.time_fmt = self.time_date.strftime("%I:%M %m/%d/%Y")
+
+    def readFile(self, file) -> Optional[XRStream]:
+        size = self.filepath.stat().st_size
+        if size < 8:
+            return None
+
+        (start, version, source) = struct.unpack("@iii", file.read(12))
+        if (start == -1) and (version >= 6):
+            file.seek(12)
+            data = file.read(size - 12)
+            return XRStream(lzokay.decompress(data, source))
+
+        return None
+
+    def readObject(self, stream: XRStream):
+        chunk = stream.open_chunk(XRFlag.CHUNK_OBJECT)
+        if chunk:
+            chunk.seek(4, io.SEEK_CUR)  # obj_count
+            count_spawn = chunk.u16()
+            spawn = XRReader(chunk.read(count_spawn))
+            actor = XRCreatureActor()
+            actor.read_spawn(spawn)
+            count_update = chunk.u16()
+            update = XRReader(chunk.read(count_update))
+            actor.read_update(update)
+            if actor:
+                self.player = actor
+        return None
+
+    def getFaction(self) -> str:
+        player = self.player
+        if player:
+            player_faction = player.community_index
+            for faction in self._factions:
+                if player_faction == faction:
+                    return self._factions[faction]
+        return "Unknown"
+
+    def getRank(self) -> str:
+        player = self.player
+        if player:
+            player_rank = player.rank
+            for rank in self._ranks:
+                if isinstance(rank, int):
+                    if player_rank <= rank:
+                        return self._ranks[rank]
+        return self._ranks["max"]
+
+    def getReputation(self) -> str:
+        player = self.player
+        if player:
+            player_rep = player.reputation
+            for rep in self._reputation:
+                if isinstance(rep, int):
+                    if player_rep <= rep:
+                        return self._reputation[rep]
+        return self._reputation["max"]

games/stalkeranomaly/__init__.py

@@ -0,0 +1,20 @@
+# -*- encoding: utf-8 -*-
+# flake8: noqa
+
+from .XRIO import XRReader, XRStream
+from .XRMath import IVec3, IVec4, IFlag
+from .XRNET import XRNETState
+from .XRObject import (
+    XRFlag,
+    XRAbstract,
+    XRVisual,
+    XRBoneData,
+    XRSkeleton,
+    XRObject,
+    XRDynamicObject,
+    XRDynamicObjectVisual,
+    XRCreatureAbstract,
+    XRTraderAbstract,
+    XRCreatureActor,
+)
+from .XRSave import XRSave