<jupyter_start><jupyter_text>Contact Tracing  UTILS<jupyter_code>import random
import string

def get_random_alphaNumeric_string(stringLength=8):
    lettersAndDigits = string.ascii_letters + string.digits
    return ''.join((random.choice(lettersAndDigits) for i in range(stringLength)))

len(get_random_alphaNumeric_string(32))
import hashlib
import hmac
from math import ceil

hash_len = 32
def hmac_sha256(key, data):
    return hmac.new(key, data, hashlib.sha256).digest()

def hkdf(length: int, ikm, salt: bytes = b"", info: bytes = b"") -> bytes:
    if len(salt) == 0:
        salt = bytes([0]*hash_len)
    prk = hmac_sha256(salt, ikm)
    t = b""
    okm = b""
    for i in range(ceil(length / hash_len)):
        t = hmac_sha256(prk, t + info + bytes([1+i]))
        okm += t
    return okm[:length]<jupyter_output><empty_output><jupyter_text>CLASSES OF IMPLIMENTATION<jupyter_code># MAKING KEYS
import datetime;



# REQUESTS TO SERVER
import requests
import json

import base64

#VIEWS
from termcolor import colored


# Class of the device of the user of app \d:D
class Device:
    def __init__(self, owner_name, device_id):
        self.device_id = device_id
        self.owner_name = owner_name
        self.__tab_daily_tracing_key = []
        
        self.__makeTracingKey() #generate __tracing_key
        self.__day_number = int(datetime.datetime.now().timestamp()/(60*60*24))
        self.__encodeDailyTracingKey() #generate __daily_tracing_key
        self.__tab_daily_tracing_key.append((self.__day_number, self.__daily_tracing_key))
        
        ##
        self.__tab_rolling_proxy_ids = []
        
        ##
        self.__get_infected = False
        
        ##
        self.__tab_infected_diag_keys = {}
        
        ###
        self.__tab_infected_diag_keys_tuples = []
        
        ###
        self.__infections_days = []
        
        
    def getOwnerName(self):
        return self.owner_name
    
    def getDeviceId(self):
        return self.device_id
    
    
    def __makeTracingKey(self, length=32):
            self.__tracing_key = get_random_alphaNumeric_string(length)
        
    def getTracingKey(self):
        return self.__tracing_key
    
    
    def __encodeDailyTracingKey(self):
        #print(self.__tracing_key)
        self.__daily_tracing_key = hkdf(length=16, ikm = bytes(self.__tracing_key, 'utf-8'), info = bytes(self.__day_number) )
        self.__tab_daily_tracing_key.append((self.__day_number, self.__daily_tracing_key))
        
        print(self.__tab_daily_tracing_key)
        
    def getDailyTraicingKey(self):
        if self.__day_number == int(datetime.datetime.now().timestamp()/(60*60*24)):
            return self.__daily_tracing_key
        
        self.__day_number = int(datetime.datetime.now().timestamp()/(60*60*24))
        
        #------update
        self.__encodeDailyTracingKey()
        
        return self.__daily_tracing_key
    
    
    def GetRollingProximityIdentifierKey(self):
        '''for now we gonna call this function to get proximity id each time'''
        
        #------update
        self.getDailyTraicingKey()
        
        
        seconds_since_start_of_day_number =  datetime.datetime.now().timestamp()% (60*60*24)
        
        self.__time_interval_number = int(seconds_since_start_of_day_number/(60*10)) 
        self.__time_interval_number_encoded = str(self.__time_interval_number).encode()
        print(self.__time_interval_number_encoded)
        
        self.__proximity_id_compete  = hmac.new(self.__daily_tracing_key, self.__time_interval_number_encoded , hashlib.sha256).digest()
        self.__proximity_id_16 = self.__proximity_id_compete[0:16]
        return self.__proximity_id_16
        
        
    def addContact(self, new_rolling_proxy_keys):
        if new_rolling_proxy_keys not in self.__tab_rolling_proxy_ids:
            self.__tab_rolling_proxy_ids.append(new_rolling_proxy_keys)
     
        
    def addContacts(self, new_rolling_proxy_keys):
        for proxy_key in new_rolling_proxy_keys:
            if proxy_key not in self.__tab_rolling_proxy_ids:
                self.__tab_rolling_proxy_ids.append(proxy_key)
        #extend
            
            
    def testedPositive(self, server_url = 'http://localhost:9000/positive' ):
        if not self.__get_infected:
            self.__get_infected = True
            
            
            
            self.__infetion_server = server_url

            
            diag_keys_list =[]
            
            for daily_tracy_key in self.__tab_daily_tracing_key:
                day_num = daily_tracy_key[0]
                diag_bytes = daily_tracy_key[1]
                
                diag_64 = base64.b64encode(diag_bytes).decode('ascii')
                
                diag_keys_list.append(
                    { 'day_num':day_num, 'diag_key': diag_64 } 
                )
            diagnosis_keys = {
                'data': diag_keys_list
            }
            
            response = requests.post(self.__infetion_server, data = json.dumps(diagnosis_keys)) 
            
            if response.status_code == 200:
                print('Success')
                print(response.text)
        
        
    def getInfectedDiagKeys(self,  server_url = 'http://localhost:9090/positive/all' ):

        self.__infetion_server = server_url
        response = requests.get(self.__infetion_server) 
        if response.status_code == 200:
            
            
            #it return some string keys so lets just chnage it for now at least
            self.__tab_infected_diag_keys = {}
            tab_infected_diag_keys_str = response.json()

            for num_day_str in sorted( tab_infected_diag_keys_str):
                if tab_infected_diag_keys_str[num_day_str] in self.__tab_daily_tracing_key and self.__get_infected==True:
                    print(self.getOwnerName(), ' Is Already infected self check from Cloud')
                    
                elif tab_infected_diag_keys_str[num_day_str] in self.__tab_daily_tracing_key and self.__get_infected==False:
                    print(self.getOwnerName(), ' Is not infected but his tracing still in Cloud as infected')
                    #later we will see if important to remove it from Cloud
                    
                elif tab_infected_diag_keys_str[num_day_str] not in self.__tab_daily_tracing_key and self.__get_infected==True:
                    self.testedPositive()
                else: 
                    self.__tab_infected_diag_keys[int(num_day_str)] = tab_infected_diag_keys_str[num_day_str]
            '''
            for num_day_str, infected_diag_key in  tab_infected_diag_keys_str.items():
                self.__tab_infected_diag_keys[int(num_day_str)] = infected_diag_key
            '''
            print('Success')
            #print(self.__tab_infected_diag_keys)

       
    def __generateRollingProximityIdentifierKey(self, infected_diag_key, time_interval_number):
        proximity_id_compete  = hmac.new(
            base64.b64decode(infected_diag_key) , 
            str(time_interval_number).encode()  , 
            hashlib.sha256).digest()
        
        proximity_id_16 = proximity_id_compete[0:16]
        return proximity_id_16
    
    def __generateAllPossibleRollingProximityIdentifierKey(self, nb = None):
        if nb is None:
            nb = len(self.__tab_infected_diag_keys)
        nb = min(len(self.__tab_infected_diag_keys), nb)    
        
        self.all_possible_rolling_proximity_key = []
        
        
        for num_day, infected_diag_key_list in list(self.__tab_infected_diag_keys.items())[-nb:]:
            self.__tab_infected_diag_keys_tuples.extend([(num_day, diag_key) for diag_key in infected_diag_key_list])
        
        self.all_possible_rolling_proximity_key = [(num_day, self.__generateRollingProximityIdentifierKey(diag_key, time_inter_num) )
                                                   for (num_day, diag_key) in self.__tab_infected_diag_keys_tuples for time_inter_num in range(144) ]
        
        #print(self.all_possible_rolling_proximity_key)
        
        
    def checkIfContactedInfectedDiagKeys(self):
        '''Here we give final situation of contact we give situation but Im not sure if we should give also the date '''
        
        #------update
        self.getInfectedDiagKeys()
        
        self.__generateAllPossibleRollingProximityIdentifierKey()
        
        
        for (num_day, gen_roll_prox_id_key ) in self.all_possible_rolling_proximity_key:
            for proxy_key in self.__tab_rolling_proxy_ids:
                if gen_roll_prox_id_key== proxy_key:
                    self.__infections_days.append(num_day)
                    
        if len(self.__infections_days)>0:
            for day in self.__infections_days :
                date =  date.fromtimestamp(day)
                print(colored('Alert ‼', 'red'), self.owner_name, "Has One contact Infected Persone at day : ", day)
                print(colored(self.owner_name+' Get a test if possible or self quarantine and and visit the hospital when facing Complications with infection', 'red'))
                #print(self.owner_name, "Has One contact infected at day : ", num_day)
        elif  self.__get_infected==True:
            print(colored('Double Alert 💥‼', 'red'), self.owner_name, " is Tested postive His infected")
            print(colored(self.owner_name+' Self qurantine and visit the hospital when facing Complications with infection', 'red'))
        else:
            
            print(colored('Ok ✔ ', 'green'), self.owner_name, " Has No contact with infected people")
            #print(self.owner_name, " Has No contact with infected people")
        
        #print(self.__tab_rolling_proxy_ids)<jupyter_output><empty_output><jupyter_text>TESTING DEVICES METHODES<jupyter_code>d1 = Device("<NAME>", 111)
#print(d1.getDailyTraicingKey())
print('Dailly tracing : ', d1.getDailyTraicingKey())
p_id = d1.GetRollingProximityIdentifierKey()
print('p_id ', p_id)
print(len(d1.GetRollingProximityIdentifierKey()))

print(d1.getTracingKey())
print(d1.getOwnerName())
b = bytes(d1.getTracingKey(), 'utf-8')
print(b)
print(len(b))<jupyter_output><KEY>
<NAME>
b'<KEY>'
32<jupyter_text>Infections methodes<jupyter_code>d1.getInfectedDiagKeys()
d1.testedPositive()
d1.checkIfContactedInfectedDiagKeys()<jupyter_output>Success
[31mDouble Alert 💥‼[0m <NAME>  is Tested postive His infected
[31m<NAME> Self qurantine and visit the hospital when facing Complications with infection[0m<jupyter_text>Pulic meeting<jupyter_code>## HERE WE GONNA DEFINE SOME METING TO KNOW PEOPLE THAT MEET .... WE SEMULATE IN PRIMITIVE WAY THE BLE OF THE DEVICE 
### We need a table where exist all devices with an Id tha semulat BLE signal so ... 

class Meeting:
    
    def __init__(self):
        
        self.__tab_proxy_contact_tracing_people = [] # contain ids of devices in same pace ther is a BLE Link ...
        self.__tab_rolling_proximity_identifiers = [] # comtains rolling proximity identifiers generated by those devices every 10 min
    
    def getAllRollingProximityIdentifiers(self):
        return self.__tab_rolling_proximity_identifiers
    
    def addContact(self, device_id, devices_list): 
        '''remember that contact tracing is focusing on privacy here id device is just to semulat near distance contact between device that have Contact Tracing active'''
        if device_id not in self.__tab_proxy_contact_tracing_people:
            self.__tab_proxy_contact_tracing_people.append(device_id)
            
            for device in devices_list:
                if device.getDeviceId() == device_id:
                    device_rolling_proximity_identifier = device.GetRollingProximityIdentifierKey()
                    if device_rolling_proximity_identifier not in self.__tab_rolling_proximity_identifiers:
                        self.__tab_rolling_proximity_identifiers.append(device_rolling_proximity_identifier)
                    
            for device in devices_list:
                device.addContacts(self.__tab_rolling_proximity_identifiers)
                
            print('HI---------------------\n', self.__tab_rolling_proximity_identifiers)
# TEST MEETHING

devices = []

d2 = Device("<NAME>", 222)
devices.append(d2)

d3 = Device("<NAME>", 333)
devices.append(d3)

meeting_1 = Meeting()

meeting_1.addContact(222, devices)
meeting_1.addContact(333, devices)


d2.checkIfContactedInfectedDiagKeys()
d3.checkIfContactedInfectedDiagKeys()
print(devices)
meeting_1.getAllRollingProximityIdentifiers()
d2.testedPositive()
d2.checkIfContactedInfectedDiagKeys()
d3.checkIfContactedInfectedDiagKeys()
d3.GetRollingProximityIdentifierKey()
meeting_1.addContact(333, devices)
meeting_1.getAllRollingProximityIdentifiers()<jupyter_output>b'141'
{'fraction_non_alphanumeric': 0.08558139534883721, 'fraction_numerical': 0.012868217054263565, 'mean_word_length': 2.5945945945945947, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '457883', 'license': 'MIT', 'n_tokens_mistral': 4130, 'n_tokens_neox': 3847, 'n_words': 750, 'path': 'contact_tracing_simple.ipynb', 'repo_name': 'mk4dev/concat_tracing_coivd_in_python', 'size': '13000'}