velocitatem/PHANTOM
1
0
Fork 0
mirror of https://github.com/velocitatem/PHANTOM.git synced 2026-05-31 08:33:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

changed to new test method for singificance

Browse Source
This commit is contained in:
Daniel Rosel
2026-03-08 13:53:31 +01:00
parent 4b89b64674
commit cc24ac72f7
8 changed files with 162 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
sim/rl/behavior_loader/loader.py
View File

@@ -2,6 +2,7 @@ import os
import json
from pydantic import BaseModel as Base
class PayloadModel(Base):
sessionId: str
experimentId: str | None
@@ -13,6 +14,7 @@ class PayloadModel(Base):
userAgent: str
ts: str
class ValueModel(Base):
payload: PayloadModel
encoding: str
@@ -20,6 +22,7 @@ class ValueModel(Base):
schemaId: int
size: int
class InteractionModel(Base):
partitionID: int
offset: int
@@ -30,14 +33,17 @@ class InteractionModel(Base):
key: dict
value: ValueModel
def _is_admin(page: str | None) -> bool:
return page is not None and page.startswith("/admin/")
class Loader:
def __init__(self, src_dir: str):
self.src_dir = src_dir
self.entries = os.listdir(src_dir)
if not self.entries: raise ValueError("empty directory")
if not self.entries:
raise ValueError("empty directory")
self.data = self._load_sessions()
def _load_sessions(self) -> dict:
@@ -55,16 +61,21 @@ class Loader:
def get_entries(self) -> tuple[list[str], int]:
return self.entries, len(self.entries)
class AgentLoader(Loader):
def _load_sessions(self) -> dict:
sessions = {}
for entry in self.entries:
with open(f"{self.src_dir}/{entry}/int.json") as f:
path = f"{self.src_dir}/{entry}/int.json"
if not os.path.isfile(path):
continue
with open(path) as f:
raw = json.load(f)
ints = [PayloadModel(**i) for i in raw]
sessions[entry] = [i for i in ints if not _is_admin(i.page)]
return sessions
class JointLoader:
def __init__(self, human_dir: str, agent_dir: str):
self.human_loader = Loader(human_dir)
@@ -74,10 +85,14 @@ class JointLoader:
def _merge(self) -> dict:
return {
**{f"human_{sid}": [e.value.payload for e in evts]
for sid, evts in self.human_loader.get_data().items()},
**{f"agent_{sid}": evts
for sid, evts in self.agent_loader.get_data().items()}
**{
f"human_{sid}": [e.value.payload for e in evts]
for sid, evts in self.human_loader.get_data().items()
},
**{
f"agent_{sid}": evts
for sid, evts in self.agent_loader.get_data().items()
},
}
def get_data(self) -> dict:
@@ -86,12 +101,17 @@ class JointLoader:
def get_entries(self) -> tuple[list[str], int]:
return self.entries, len(self.entries)
if __name__ == "__main__":
agent_dir = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/agents/collected_data/"
human_dir = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/"
human_dir = (
"/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/"
)
for name, cls, path in [("agent", AgentLoader, agent_dir),
("human", Loader, human_dir),
("joint", lambda d: JointLoader(human_dir, d), agent_dir)]:
for name, cls, path in [
("agent", AgentLoader, agent_dir),
("human", Loader, human_dir),
("joint", lambda d: JointLoader(human_dir, d), agent_dir),
]:
ldr = cls(path) if name != "joint" else cls(agent_dir)
print(f"Loaded {len(ldr.get_entries()[0])} {name} sessions")