{ "cells": [ { "cell_type": "code", "execution_count": 2, "id": "ca1eef8e", "metadata": {}, "outputs": [], "source": [ "#!/usr/bin/env python3\n", "import sys\n", "\n", "\n", "# ansi codes to produce a colored rotation matrix\n", "class bcolors:\n", " MAGENTA = '\\033[95m'\n", " BLUE = '\\033[94m'\n", " GREEN = '\\033[92m'\n", " YELLOW = '\\033[93m'\n", " CYAN = '\\033[96m'\n", " WHITE = '\\033[97m'\n", " RED = '\\033[91m'\n", " ENDC = '\\033[0m'\n", " BOLD = '\\033[1m'\n", " UNDERLINE = '\\033[4m'\n", "\n", "\n", "# compute bwt and show the rotation matrix\n", "# the last input symbol must be unique in the input string\n", "# to avoid issues with sorting the rotations \n", "def bwt(s,show=True,color=False):\n", " assert s.count(s[-1])==1, \"L'ultimo carattere deve apparire una volta sola\"\n", " # build array of rotations, O(n^2) space (ugly!)\n", " rot = []\n", " for i in range(len(s)):\n", " rot.append(s[i:]+ s[:i])\n", " # sort array of rotations, O(n^2 \\log n) time (ugly!) \n", " rot.sort()\n", " bwt = \"\"\n", " eos = rowindex = 0\n", " # scan sorted rotations\n", " for x in rot:\n", " bwt += x[-1]\n", " if x[-1]==s[-1]: eos = rowindex\n", " rowindex += 1\n", " if show:\n", " if color:\n", " print(\"%s%2d%s %s %s|%s|%s\" %\n", " (bcolors.BOLD+bcolors.WHITE,len(s)-x.index(s[-1])-1,bcolors.ENDC\n", " ,x[:-1], bcolors.BOLD+bcolors.GREEN, x[-1],bcolors.ENDC ))\n", " else:\n", " print(\"%2d %s |%s|\" %(len(s)-x.index(s[-1]),x[:-1], x[-1]))\n", " return bwt, eos\n", "\n", "\n", "# compute the rank[] array such that rank[i]\n", "# is the number of occurrence of L[i] in L[0...i-1]\n", "# also compute the array occ[] containing the\n", "# # of occurrences of each char \n", "def compute_rank(s):\n", " rank = []\n", " occ = [0]*256\n", " for c in s:\n", " rank.append(occ[ord(c)])\n", " occ[ord(c)] += 1\n", " return rank, occ\n", "\n", "# compute array first[] such that first[i] is the\n", "# position of the first occurrence of char i in the \n", "# first column. note this is an encoding of column F\n", "def computeF(occ):\n", " first = [0]*256\n", " for i in range(1,256):\n", " first[i] = first[i-1] + occ[i-1]\n", " return first\n", "\n", "\n", "# invert bwt (parameter s) and the last position in s\n", "# of the last char of the original text (parameter j)\n", "def unbwt(s,j):\n", " rank,numOcc = compute_rank(s)\n", " startF = computeF(numOcc)\n", " text = \"\" \n", " for i in range(len(s)):\n", " text = s[j] + text\n", " j = startF[ord(s[j])] + rank[j]\n", " return text\n" ] }, { "cell_type": "code", "execution_count": 3, "id": "37a15f76", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\u001b[1m\u001b[97m 5\u001b[0m .miss.missingswis \u001b[1m\u001b[92m|s|\u001b[0m\n", "\u001b[1m\u001b[97m10\u001b[0m .missingswiss.mis \u001b[1m\u001b[92m|s|\u001b[0m\n", "\u001b[1m\u001b[97m17\u001b[0m gswiss.miss.missi \u001b[1m\u001b[92m|n|\u001b[0m\n", "\u001b[1m\u001b[97m15\u001b[0m ingswiss.miss.mis \u001b[1m\u001b[92m|s|\u001b[0m\n", "\u001b[1m\u001b[97m 2\u001b[0m iss.miss.missings \u001b[1m\u001b[92m|w|\u001b[0m\n", "\u001b[1m\u001b[97m 7\u001b[0m iss.missingswiss. \u001b[1m\u001b[92m|m|\u001b[0m\n", "\u001b[1m\u001b[97m12\u001b[0m issingswiss.miss. \u001b[1m\u001b[92m|m|\u001b[0m\n", "\u001b[1m\u001b[97m 6\u001b[0m miss.missingswiss \u001b[1m\u001b[92m|.|\u001b[0m\n", "\u001b[1m\u001b[97m11\u001b[0m missingswiss.miss \u001b[1m\u001b[92m|.|\u001b[0m\n", "\u001b[1m\u001b[97m16\u001b[0m ngswiss.miss.miss \u001b[1m\u001b[92m|i|\u001b[0m\n", "\u001b[1m\u001b[97m 4\u001b[0m s.miss.missingswi \u001b[1m\u001b[92m|s|\u001b[0m\n", "\u001b[1m\u001b[97m 9\u001b[0m s.missingswiss.mi \u001b[1m\u001b[92m|s|\u001b[0m\n", "\u001b[1m\u001b[97m14\u001b[0m singswiss.miss.mi \u001b[1m\u001b[92m|s|\u001b[0m\n", "\u001b[1m\u001b[97m 3\u001b[0m ss.miss.missingsw \u001b[1m\u001b[92m|i|\u001b[0m\n", "\u001b[1m\u001b[97m 8\u001b[0m ss.missingswiss.m \u001b[1m\u001b[92m|i|\u001b[0m\n", "\u001b[1m\u001b[97m13\u001b[0m ssingswiss.miss.m \u001b[1m\u001b[92m|i|\u001b[0m\n", "\u001b[1m\u001b[97m 0\u001b[0m swiss.miss.missin \u001b[1m\u001b[92m|g|\u001b[0m\n", "\u001b[1m\u001b[97m 1\u001b[0m wiss.miss.missing \u001b[1m\u001b[92m|s|\u001b[0m\n", "16\n" ] } ], "source": [ "# compute BWT, showing the bwt matrix and using colors\n", "b,eos = bwt(\"swiss.miss.missing\",True,True)\n", "print(eos)" ] }, { "cell_type": "code", "execution_count": 4, "id": "f245fa45", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Recovered: \u001b[1mswiss.miss.missing\u001b[0m\n" ] } ], "source": [ "# invert bwt given row index containing the original string\n", "print(\"Recovered:\", bcolors.BOLD+unbwt(b,eos)+bcolors.ENDC)" ] }, { "cell_type": "code", "execution_count": 5, "id": "a93886bd", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[98, 1, 97, 5, 98, 3, 97, 3, 98, 20, 97, 129, 71]\n" ] } ], "source": [ "# convert to base 128 big endian form\n", "def tobase128(n):\n", " assert(n>=0), \"input must be non-negative\"\n", " if n==0: \n", " return [0] # special case to simplify the while loop \n", " a = []\n", " while n>0:\n", " nextd = n%128 # remainder\n", " a.insert(0,nextd) # insert at the beginning of the list\n", " n = n//128 # integer division\n", " return a\n", "\n", "# compute VB coding of n, big endian\n", "def vb_encode(n):\n", " a = tobase128(n)\n", " for i in range(len(a)-1):\n", " a[i] += 128 # turning on the most significant bit\n", " return a\n", " \n", " \n", "# run lenght encoding\n", "def run_length_enc(input_string):\n", " if not input_string:\n", " return []\n", "\n", " encoded_string = []\n", " count = 1\n", " length = len(input_string)\n", " for i in range(1, length + 1): # Go until length + 1 to handle last group\n", " if i < length and input_string[i] == input_string[i - 1]:\n", " count += 1\n", " else:\n", " encoded_string.append(ord(input_string[i - 1]))\n", " encoded_string += vb_encode(count-1)\n", " count = 1 # Reset count for the next symbol\n", " return bytes(encoded_string)\n", "\n", "# run length decoding\n", "def run_length_dec(bytes):\n", " i=0\n", " s=\"\"\n", " while(i127): # continuation bit on\n", " n = n*128+(bytes[i]-128)\n", " i +=1\n", " # final byte \n", " n = n*128+bytes[i]\n", " s += s[-1]*(n)\n", " i+=1\n", " return s\n", "\n", "\n", "s= \"bbaaaaaabbbbaaaabbbbbbbbbbbbbbbbbbbbb\"+200*\"a\" \n", "sr = run_length_enc(s)\n", "print(list(sr))" ] }, { "cell_type": "code", "execution_count": 6, "id": "246adec3", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[98, 1, 97, 5, 98, 3, 97, 3, 98, 20, 97, 129, 71]\n" ] }, { "data": { "text/plain": [ "True" ] }, "execution_count": 6, "metadata": {}, "output_type": "execute_result" } ], "source": [ "z = run_length_dec(sr)\n", "print(list(run_length_enc(z)))\n", "s==z" ] }, { "cell_type": "code", "execution_count": 7, "id": "e3968afd", "metadata": {}, "outputs": [], "source": [ "# compress a file using bwt+rle+vbyte\n", "# return a bytes object with the compressed bwt\n", "# and the position of the eos in the BWT (necessary for inversion)\n", "def bzip(filename):\n", " with open(filename, 'r', encoding='utf-8') as file:\n", " content = file.read()+\"\\0\" # add eof char\n", " bwtf,eos = bwt(content,False)\n", " bwtz = run_length_enc(bwtf)\n", " print(\"Input file length:\", len(content), \"Output length:\", len(bwtz), \"position:\", eos)\n", " return bwtz,eos" ] }, { "cell_type": "code", "execution_count": 8, "id": "fd350883", "metadata": {}, "outputs": [], "source": [ "# uncompress a compressed bwt doing unrle+unbwt\n", "def bunzip(bwtz,eos):\n", " # invert run_length_encoding\n", " urle = run_length_dec(bwtz)\n", " #invert bwt\n", " t = unbwt(urle,eos)\n", " print(\"Uncompressed lenght:\", len(t)-1)\n", " return t[:-1] # remove the eos symbol " ] }, { "cell_type": "code", "execution_count": 9, "id": "a9c2e6bd", "metadata": {}, "outputs": [], "source": [ "# check if the content of filename is equal to string s\n", "def check_file(filename,s):\n", " with open(filename, 'r', encoding='utf-8') as file:\n", " content = file.read()\n", " return content==s" ] }, { "cell_type": "code", "execution_count": 10, "id": "f5ea5354", "metadata": {}, "outputs": [ { "ename": "FileNotFoundError", "evalue": "[Errno 2] No such file or directory: 'grammar.lsp'", "output_type": "error", "traceback": [ "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m", "\u001b[0;31mFileNotFoundError\u001b[0m Traceback (most recent call last)", "\u001b[0;32m/tmp/ipykernel_87459/1956736435.py\u001b[0m in \u001b[0;36m\u001b[0;34m\u001b[0m\n\u001b[1;32m 1\u001b[0m \u001b[0;31m# compress file grammar.lsp\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0mz\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0meos\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mbzip\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"grammar.lsp\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m", "\u001b[0;32m/tmp/ipykernel_87459/4195308260.py\u001b[0m in \u001b[0;36mbzip\u001b[0;34m(filename)\u001b[0m\n\u001b[1;32m 3\u001b[0m \u001b[0;31m# and the position of the eos in the BWT (necessary for inversion)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 4\u001b[0m \u001b[0;32mdef\u001b[0m \u001b[0mbzip\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfilename\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 5\u001b[0;31m \u001b[0;32mwith\u001b[0m \u001b[0mopen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfilename\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m'r'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mencoding\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'utf-8'\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0mfile\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m 6\u001b[0m \u001b[0mcontent\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mfile\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mread\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m+\u001b[0m\u001b[0;34m\"\\0\"\u001b[0m \u001b[0;31m# add eof char\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 7\u001b[0m \u001b[0mbwtf\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0meos\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mbwt\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcontent\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;32mFalse\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n", "\u001b[0;31mFileNotFoundError\u001b[0m: [Errno 2] No such file or directory: 'grammar.lsp'" ] } ], "source": [ "# compress file grammar.lsp\n", "z,eos = bzip(\"grammar.lsp\")" ] }, { "cell_type": "code", "execution_count": null, "id": "6f2cbc2a", "metadata": {}, "outputs": [], "source": [ "# recover original content \n", "s = bunzip(z,eos)" ] }, { "cell_type": "code", "execution_count": null, "id": "e771671f", "metadata": {}, "outputs": [], "source": [ "# check we got the original input \n", "check_file(\"grammar.lsp\",s)" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.12" } }, "nbformat": 4, "nbformat_minor": 5 }