Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
Python
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768"""A decision log reads the same compressed as it does plain.
A finished log compresses about 17 to 1 - `phi` writes all 36 feature namesagain for every candidate, and a 300-match corpus holds about five million ofthem - so 53GB of corpus becomes 3.2GB. That is the difference between keepingthe rows a refit needs and deleting them for space, which is a choice thisproject has already made the wrong way once.
Nothing writes the compressed form; `gzip` does. Everything reads it."""
from __future__ import annotations
import gzipimport jsonimport tempfileimport unittestfrom pathlib import Path
from sds.viewer import COMPRESSED_SUFFIX, DECISION_SUFFIX, _tag_of, decision_logs, iter_decisions
ROWS = [ {"seq": 0, "round": 1, "phase": "movement", "candidates": [{"phi": {"cohesion": 1.0}}]}, {"seq": 1, "round": 1, "phase": "firing", "candidates": [{"phi": {"cohesion": 0.5}}]},]
class TestCompressedLogs(unittest.TestCase): def setUp(self): self.dir = Path(tempfile.mkdtemp()) body = "".join(json.dumps(r) + "\n" for r in ROWS) self.plain = self.dir / f"tag-seed1-abcd1234-sds_North{DECISION_SUFFIX}" self.plain.write_text(body) self.gz = self.dir / f"tag-seed2-abcd5678-sds_North{COMPRESSED_SUFFIX}" with gzip.open(self.gz, "wt", encoding="utf-8") as handle: handle.write(body)
def test_the_two_forms_read_the_same_records(self): self.assertEqual(list(iter_decisions(self.plain)), list(iter_decisions(self.gz)))
def test_both_forms_are_discovered(self): self.assertEqual(decision_logs(self.dir), sorted([self.gz, self.plain]))
def test_a_corpus_may_hold_a_mix(self): # Which is the normal state part-way through a compression pass. found = decision_logs(self.dir) self.assertEqual(len(found), 2) rows = [r for log in found for r in iter_decisions(log)] self.assertEqual(len(rows), 2 * len(ROWS))
def test_the_tag_survives_the_extra_suffix(self): # `.gz` is stripped before the seat is split off, or the tag keeps it # and a compressed log stops matching its own result document. self.assertEqual(_tag_of(self.gz), "tag-seed2-abcd5678") self.assertEqual(_tag_of(self.plain), "tag-seed1-abcd1234")
def test_a_truncated_member_yields_what_it_had(self): # The plain reader has always tolerated a half-written last line. A # compressed one raises instead of returning a short read, so that is # caught too rather than losing the whole log. cut = self.dir / f"cut-seed3-abcd9999-sds_North{COMPRESSED_SUFFIX}" cut.write_bytes(self.gz.read_bytes()[:-6]) self.assertLessEqual(len(list(iter_decisions(cut))), len(ROWS))
if __name__ == "__main__": unittest.main()