02_extract_log_scan.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "4493f2ef-302d-4924-b8cf-72678d653dc7",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Python library for loading a log of the Consumer Physics SCIO\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "67005b5a-8221-45ef-ba05-24f187824037",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import struct\n",
    "import datetime\n",
    "import os\n",
    "import csv\n",
    "import base64\n",
    "import numpy as np\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "888b6c86-1d8e-45f1-a6a0-28bcf1813905",
   "metadata": {},
   "outputs": [],
   "source": [
    "class scio_log:\n",
    "    def __init__(self):\n",
    "        self.device_info = {}\n",
    "        self.scan_rawdata = {}\n",
    "        self.b64_data = {}\n",
    "        self.spectral_data = {}\n",
    "        self.calibration_data = {}\n",
    "        self.temperature = {}\n",
    "        self.key = b''\n",
    "        self.cmd_info = r\"DeviceInfo{(.*?)}\"\n",
    "        self.cmd_scan = r\"ResponseCommandParse\tbytes:( [\\s\\w]+)\"\n",
    "        self.cmd_b64  = r'^{\"device_id\"(.*)$' #r\"({\\\"device_id\\\".*?\\\"})\"\n",
    "        self.cmd_spec = r\"Response: {\\\"_type\\\":\\\"POST spectroscan2\\\",\\\"spectrum\\\":(.*?)}\"\n",
    "    \n",
    "    def load_log(self, log_fn, out_path, silent=True):\n",
    "        next_task = 'dev_info'\n",
    "        \n",
    "        scan_data_started = True\n",
    "        scan_data = ''\n",
    "        end_of_scan = False\n",
    "        \n",
    "        scan_nb = 0\n",
    "        i = 0\n",
    "        \n",
    "        with open(log_fn, \"r\") as file:\n",
    "            for line in file:\n",
    "                sample_done = False\n",
    "                # Search for the pattern in each line\n",
    "                bytes_match = re.search(self.cmd_scan, line)\n",
    "                info_match  = re.search(self.cmd_info, line)\n",
    "                b64_match   = re.search(self.cmd_b64, line)\n",
    "                spec_match  = re.search(self.cmd_spec, line)\n",
    "                if bytes_match:\n",
    "                    # Extract the bytes portion from the matched line\n",
    "                    hex_str = bytes_match.group(1).replace(\" 0x\", \"\")\n",
    "                    if('01BA04' in hex_str):\n",
    "                        print('Detected scan ' + str(scan_nb) + ':')\n",
    "                        scan_nb += 1\n",
    "                        self.temperature = self.read_temperature(hex_str.strip())\n",
    "                        print('  - Temperature: ' + str(self.temperature))\n",
    "                    elif('01BA02' in hex_str):\n",
    "                        scan_data = hex_str.strip()[10:]\n",
    "                        scan_data_started = False\n",
    "                    elif((~scan_data_started) and (hex_str[0:2] != '01')):\n",
    "                        scan_data_started = True\n",
    "                        scan_data = scan_data + hex_str.strip()[2:]\n",
    "                        if(len(hex_str.strip()) < 40):\n",
    "                            scan_data_started = True\n",
    "                            if(i == 0):\n",
    "                                sample = scan_data\n",
    "                                print('  - Sample bytes:          ', str(int(len(sample)/2)))\n",
    "                                self.scan_rawdata.update({'sample_dark': sample})\n",
    "                            elif(i == 1):\n",
    "                                sample_dark = scan_data\n",
    "                                print('  - Sample dark bytes:     ', str(int(len(sample_dark)/2)))\n",
    "                                self.scan_rawdata.update({'sample': sample_dark})\n",
    "                            elif(i == 2):\n",
    "                                sample_gradient = scan_data\n",
    "                                print('  - Sample gradient bytes: ', str(int(len(sample_gradient)/2)))\n",
    "                                self.scan_rawdata.update({'sample_gradient': sample_gradient})\n",
    "                                i = -1\n",
    "                            i += 1\n",
    "                    else:\n",
    "                        print(\"Bytes: {}\".format(hex_str.strip()))\n",
    "                if info_match:\n",
    "                    self.parse_device_info(info_match.group(1))\n",
    "                if b64_match:\n",
    "                    self.b64_data = json.loads(b64_match.group())\n",
    "                if spec_match:\n",
    "                    self.spectral_data = spec_match.group(1)\n",
    "                    print('  - Saving scan data')\n",
    "                    self.write_data_file('scan_from_log', out_path)\n",
    "        #print('  - Saving scan data')\n",
    "        #self.write_data_file('scan_from_log', out_path)\n",
    "        return\n",
    "    \n",
    "    def parse_device_info(self, match_str):\n",
    "        info = str(match_str).replace(\"\\'\", \"\").split(\", \")\n",
    "        for item in info:\n",
    "            key, value = item.split(\"=\")\n",
    "            if value.isdigit():\n",
    "                self.device_info[key] = int(value)\n",
    "            elif value.lower() == \"true\":\n",
    "                self.device_info[key] = True\n",
    "            elif value.lower() == \"false\":\n",
    "                self.device_info[key] = False\n",
    "            else:\n",
    "                self.device_info[key] = value\n",
    "        return(self.device_info)\n",
    "    \n",
    "    def read_temperature(self, hex_string):\n",
    "        response_data_raw = bytearray.fromhex(hex_string[10:])\n",
    "        response_length = len(response_data_raw)\n",
    "        # Parse response\n",
    "        num_vars = response_length / 4 # divide by 4 because we are dealing with longs\n",
    "        data_struct = '<' + str(int(num_vars)) + 'L' # This is usually '<3l' or '<lll'\n",
    "        # Convert bytes\n",
    "        response_data = struct.unpack(data_struct, response_data_raw)\n",
    "        # Convert to actual temperatures\n",
    "        cmosTemperature   = (response_data[0] - 375.22) / 1.4092 # From the disassembled Android app\n",
    "        chipTemperature   = response_data[1] / 100\n",
    "        objectTemperature = response_data[2] / 100\n",
    "        temperatures = {'cmos_t': cmosTemperature, 'chip_t': chipTemperature, 'obj_t': objectTemperature}\n",
    "        return(temperatures)\n",
    "    \n",
    "    # Saves a single scan as raw data to JSON\n",
    "    def write_data_file(self, output_fn_base, out_path):\n",
    "        # Check if output path exists\n",
    "        if not os.path.exists(out_path):\n",
    "            os.makedirs(out_path)\n",
    "        \n",
    "        # Create file name\n",
    "        timestamp = self.b64_data['sampled_at'][:19].replace('T', '_').replace('-', '').replace(':', '')\n",
    "        output_fn = output_fn_base + '_' + timestamp + '.json'\n",
    "        \n",
    "        # Create the json string\n",
    "        jsondata = {}\n",
    "        jsondata['device'] = self.device_info\n",
    "        jsondata['temperature'] = self.temperature\n",
    "        jsondata['raw_data'] = self.scan_rawdata\n",
    "        jsondata['b64_data'] = self.b64_data\n",
    "        jsondata['spec_data'] = self.spectral_data\n",
    "        \n",
    "        # Empty the data after saving, so that it doesn't get mixed up in the output\n",
    "        self.scan_rawdata = {}\n",
    "        self.b64_data = {}\n",
    "        self.spectral_data = {}\n",
    "        \n",
    "        # Write data to JSON file\n",
    "        with open(out_path + output_fn, 'w') as outfile:\n",
    "            json.dump(jsondata, outfile, indent=4)\n",
    "        return\n",
    "    \n",
    "    def encode_b64(self, bytestring):\n",
    "        urlSafeEncodedBytes = base64.urlsafe_b64encode(bytestring)\n",
    "        urlSafeEncodedStr = str(urlSafeEncodedBytes, 'utf-8')\n",
    "        return(urlSafeEncodedStr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "44a7e31e-904c-4dc9-844d-a96993f7cc68",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_fn  = './01_rawdata/log_files/log_20200604-bark.txt'\n",
    "#log_fn  = './01_rawdata/log_files/log_20200604-soil-crust.txt'\n",
    "#log_fn = './01_rawdata/log_files/log_20211020_skin.txt'\n",
    "#log_fn = './01_rawdata/log_files/log_20200604-rock.txt'\n",
    "#log_fn = './01_rawdata/log_files/log_20200604-calibrationbox.txt'\n",
    "output_path = './01_rawdata/log_extracted/bark/'\n",
    "\n",
    "# Now set up device\n",
    "scio = scio_log() # Create an instance for this device address\n",
    "scio.load_log(log_fn, output_path, silent=False)\n",
    "print('Done...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f56c5115-1db3-4aa1-b5f5-b5c466358df4",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "scio",
   "language": "python",
   "name": "scio"
  },
  "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}