Clean1.0

[Louis3030]

Automated Intelligent Inventory System for Encircle

This project automates the inventory process using AI and integrates with the Encircle API to streamline item documentation for disaster recovery jobs.

Features

• AI-driven photo analysis for item recognition.
• Damage assessment based on image data.
• Automatic data submission to the Encircle platform via API.
• User verification before final submission.

Setup

Requirements

• Python 3.8+
• Libraries: TensorFlow, OpenCV, Requests

Install dependencies:

pip install -r requirements.txt

[Louis3030]

python main.py --image_path /path/to/image.jpg --model_path /path/to/model.h5 --claim_id 12345678

[Louis3030]

Once you've created and edited the README.md, commit and push it to the repository:

git add README.md
git commit -m "Added project description in README"
git push

[Louis3030]

/project-root │ ├── /src # Contains all source code │ ├── main.py │ ├── model.py │ ├── api.py │ ├── damage_assessment.py │ ├── /models # Stores pre-trained models │ └── model.h5 │ ├── /tests # Unit tests for your code │ └── test_main.py │ └── README.md

[Louis3030]

Ignore Python compiled files

pycache/ *.pyc

Ignore pre-trained models

/models/

Ignore environment files

*.env

[Louis3030]

name: Python application

on: [push]

jobs: build:

runs-on: ubuntu-latest

steps:
- uses: actions/checkout@v2
- name: Set up Python 3.8
  uses: actions/setup-python@v2
  with:
    python-version: 3.8
- name: Install dependencies
  run: |
    python -m pip install --upgrade pip
    pip install -r requirements.txt
- name: Run tests
  run: |
    pytest

[Louis3030]

test_main.py

[Louis3030]

mkdir tests touch tests/test_main.py

[Louis3030]

import unittest from src.model import load_model, process_image, predict_item from src.api import upload_inventory_data, generate_item_payload from src.damage_assessment import assess_damage

class TestInventoryAutomation(unittest.TestCase):

def setUp(self):
    # Called before each test, set up initial state
    self.model_path = 'models/test_model.h5'
    self.image_path = 'tests/sample_image.jpg'
    self.claim_id = '12345678'

def test_load_model(self):
    # Test model loading function
    model = load_model(self.model_path)
    self.assertIsNotNone(model, "Model failed to load")

def test_process_image(self):
    # Test image processing
    image = process_image(self.image_path)
    self.assertEqual(image.shape, (1, 224, 224, 3), "Image not processed correctly")

def test_predict_item(self):
    # Test item prediction (mock or use a real model)
    model = load_model(self.model_path)
    image = process_image(self.image_path)
    item = predict_item(model, image)
    self.assertIsNotNone(item, "Prediction failed")

def test_assess_damage(self):
    # Test damage assessment (this is simple, so we mock it)
    image = process_image(self.image_path)
    damage_level = assess_damage(image)
    self.assertIn(damage_level, [0, 1, 2], "Damage assessment returned invalid value")

def test_generate_item_payload(self):
    # Test payload generation for API submission
    item = 1  # Example predicted item class
    damage_level = 2  # Example damage level
    description = "Test description"
    payload = generate_item_payload(item, damage_level, description)
    self.assertEqual(payload['item'], item, "Item mismatch in payload")
    self.assertEqual(payload['damage_level'], damage_level, "Damage level mismatch in payload")

def test_upload_inventory_data(self):
    # Test API upload function (using a mock or a sandbox)
    item_data = {
        'item': 1,
        'damage_level': 2,
        'description': 'Test item description'
    }
    response = upload_inventory_data(item_data, self.claim_id)
    self.assertIn('status', response, "API response missing status")
    self.assertEqual(response['status'], 'success', "Failed API upload")

if name == "main": unittest.main()

[Louis3030]

import unittest from unittest.mock import patch from src.model import load_model, process_image, predict_item from src.api import upload_inventory_data, generate_item_payload from src.damage_assessment import assess_damage

class TestInventoryAutomation(unittest.TestCase):

def setUp(self):
    # Called before each test, set up initial state
    self.model_path = 'models/test_model.h5'
    self.image_path = 'tests/sample_image.jpg'
    self.claim_id = '12345678'

def test_load_model(self):
    # Test model loading function
    model = load_model(self.model_path)
    self.assertIsNotNone(model, "Model failed to load")

def test_process_image(self):
    # Test image processing
    image = process_image(self.image_path)
    self.assertEqual(image.shape, (1, 224, 224, 3), "Image not processed correctly")

def test_predict_item(self):
    # Test item prediction (mock or use a real model)
    model = load_model(self.model_path)
    image = process_image(self.image_path)​⬤

[Louis3030]

import unittest from unittest.mock import patch from src.model import load_model, process_image, predict_item from src.api import upload_inventory_data, generate_item_payload from src.damage_assessment import assess_damage

class TestInventoryAutomation(unittest.TestCase):

def setUp(self):
    # Called before each test, set up initial state
    self.model_path = 'models/test_model.h5'
    self.image_path = 'tests/sample_image.jpg'
    self.claim_id = '12345678'

def test_load_model(self):
    # Test model loading function
    model = load_model(self.model_path)
    self.assertIsNotNone(model, "Model failed to load")

def test_process_image(self):
    # Test image processing
    image = process_image(self.image_path)
    self.assertEqual(image.shape, (1, 224, 224, 3), "Image not processed correctly")

def test_predict_item(self):
    # Test item prediction (mock or use a real model)
    model = load_model(self.model_path)
    image = process_image(self.image_path)​⬤

[Louis3030]

pip install requests responses

[Louis3030]

import requests import responses

Define the URL of the Encircle API endpoint you're testing

ENCIRCLE_API_URL = "https://api.encircleapp.com/v1/items"

Mock API response data for testing

MOCK_RESPONSE = { "status": "success", "data": [ { "item_id": 123, "name": "Wooden Chair", "condition": "Damaged", "notes": "Burn marks on the legs", } ] }

Mock API POST request

@responses.activate def test_create_item():

Mock the POST response from Encircle API

responses.add(
    responses.POST,
    ENCIRCLE_API_URL,
    json=MOCK_RESPONSE,
    status=200
)

# Test the API call
response = requests.post(ENCIRCLE_API_URL, json={"name": "Wooden Chair", "condition": "Damaged"})

# Assert the response matches the mock data
assert response.status_code == 200
assert response.json() == MOCK_RESPONSE

print("Test passed: API call matched the mock response")

Run the test

test_create_item()

[Louis3030]

@responses.activate def test_item_not_found(): responses.add( responses.GET, f"{ENCIRCLE_API_URL}/999", json={"error": "Item not found"}, status=404 )

response = requests.get(f"{ENCIRCLE_API_URL}/999")
assert response.status_code == 404
assert response.json() == {"error": "Item not found"}

print("Test passed: Item not found handled correctly")

[Louis3030]

pytest test_encircle_integration.py

[Louis3030]

pip install requests responses

import requests import responses

Define the URL of the Encircle API endpoint you're testing

ENCIRCLE_API_URL = "https://api.encircleapp.com/v1/items"

Mock API response data for testing

MOCK_RESPONSE = { "status": "success", "data": [ { "item_id": 123, "name": "Wooden Chair", "condition": "Damaged", "notes": "Burn marks on the legs", } ] }

Mock API POST request

@responses.activate def test_create_item():

Mock the POST response from Encircle API

responses.add(
    responses.POST,
    ENCIRCLE_API_URL,
    json=MOCK_RESPONSE,
    status=200
)

# Test the API call
response = requests.post(ENCIRCLE_API_URL, json={"name": "Wooden Chair", "condition": "Damaged"})

# Assert the response matches the mock data
assert response.status_code == 200
assert response.json() == MOCK_RESPONSE

print("Test passed: API call matched the mock response")

Run the test

test_create_item()

@responses.activate def test_item_not_found(): responses.add( responses.GET, f"{ENCIRCLE_API_URL}/999", json={"error": "Item not found"}, status=404 )

response = requests.get(f"{ENCIRCLE_API_URL}/999")
assert response.status_code == 404
assert response.json() == {"error": "Item not found"}

print("Test passed: Item not found handled correctly")

pytest test_encircle_integration.py

[Louis3030]

from unittest.mock import patch import requests

def test_create_item_with_mock(): mock_response = { "status": "success", "data": {"item_id": 123, "name": "Wooden Chair", "condition": "Damaged"} }

with patch('requests.post') as mocked_post:
    mocked_post.return_value.status_code = 200
    mocked_post.return_value.json.return_value = mock_response

    response = requests.post(ENCIRCLE_API_URL, json={"name": "Wooden Chair", "condition": "Damaged"})

    assert response.status_code == 200
    assert response.json() == mock_response

    print("Test passed: API call using unittest.mock")

Run the test

test_create_item_with_mock()

pytest test_encircle_integration.py

[Louis3030]

name: Encircle Automation Integration

on: push: branches:

• main pull_request: branches:
• main

jobs: build: runs-on: ubuntu-latest

steps:
  - name: Checkout code
    uses: actions/checkout@v2

  - name: Set up Python
    uses: actions/setup-python@v2
    with:
      python-version: '3.x'  # Specify the version of Python you're using

  - name: Install dependencies
    run: |
      pip install -r requirements.txt  # Install any dependencies from your requirements.txt file

  - name: Run tests
    run: |
      pytest  # Run your automated tests (if any)

  - name: Run Encircle Integration Script
    run: |
      python main.py  # Replace with the actual script you're using for Encircle automation

[Louis3030]

import requests

Example of making an authenticated API request

ENCIRCLE_API_KEY = 'your_api_key_here' headers = { 'Authorization': f'Bearer {ENCIRCLE_API_KEY}', 'Content-Type': 'application/json' }

def fetch_encircle_data(): url = "https://api.encircleapp.com/v1/your-endpoint" response = requests.get(url, headers=headers) if response.status_code == 200: return response.json() else: raise Exception(f"Failed to fetch data: {response.status_code} - {response.text}")

Your main logic here

if name == "main": data = fetch_encircle_data() print(data) # Do something with the fetched data

[Louis3030]

def validate_item_data(item): """Validates that the necessary fields in the item data are present and correct.""" required_fields = ['id', 'description', 'condition', 'quantity'] for field in required_fields: if field not in item or not item[field]: raise ValueError(f"Missing or invalid {field} in item data: {item}") return True

[Louis3030]

import time import requests

def api_request_with_retries(url, headers, retries=3, delay=5): """Tries to make an API request multiple times with a delay between retries.""" for attempt in range(retries): try: response = requests.get(url, headers=headers) response.raise_for_status() # Will raise an HTTPError for bad responses return response.json() except requests.exceptions.RequestException as e: print(f"Attempt {attempt+1} failed: {e}") if attempt < retries - 1: time.sleep(delay) # Wait before retrying else: raise Exception("Max retries reached, request failed.")

[Louis3030]

import logging

Configure logging to log to a file

logging.basicConfig(filename='integration.log', level=logging.INFO)

def log_event(message): """Logs a message to a file for debugging and tracking purposes.""" logging.info(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {message}")

[Louis3030]

def process_items_in_batches(items, batch_size=10): """Processes items in batches to avoid overloading the API.""" for i in range(0, len(items), batch_size): batch = items[i:i+batch_size]

Process or upload each batch here

    upload_items(batch)

def upload_items(items): """Uploads a batch of items to Encircle.""" url = "https://api.encircleapp.com/v1/upload-items" response = requests.post(url, json={'items': items}, headers=headers) if response.status_code == 200: log_event(f"Uploaded batch of {len(items)} items successfully.") else: log_event(f"Failed to upload batch: {response.text}")

[Louis3030]

def generate_item_description(item): """Generates a standardized description for an item based on attributes.""" description = f"{item['quantity']}x {item['condition']} {item['type']} from {item['room']}" return description

def generate_descriptions_for_items(items): """Automatically generate descriptions for a list of items.""" for item in items: item['description'] = generate_item_description(item) return items

[Louis3030]

import smtplib from email.mime.text import MIMEText

def send_error_report_via_email(error_message): """Sends an error report to a specified email address.""" msg = MIMEText(f"Error occurred: {error_message}") msg['Subject'] = "Encircle Integration Error" msg['From'] = "your-email@example.com" msg['To'] = "recipient-email@example.com"

with smtplib.SMTP('smtp.example.com') as server:
    server.login("your-email@example.com", "your-password")
    server.sendmail(msg['From'], [msg['To']], msg.as_string())

[Louis3030]

def summarize_run_success(items_processed, items_failed): """Outputs a summary of the run, including success and failure counts.""" total_items = len(items_processed) + len(items_failed) print(f"Run Summary: {len(items_processed)} items processed successfully out of {total_items}.") if items_failed: print(f"{len(items_failed)} items failed to process.")

[Louis3030]

def create_job_file(policy_holder_first_name, policy_holder_last_name, use_last_name_only=True): """Creates a job file using the policy holder's name. Uses last name by default.""" if use_last_name_only: job_name = policy_holder_last_name else: job_name = f"{policy_holder_first_name} {policy_holder_last_name}"

# API call to create a job file in Encircle (replace with actual API)
url = "https://api.encircleapp.com/v1/jobs"
payload = {
    "title": job_name,
    "policy_holder_first_name": policy_holder_first_name,
    "policy_holder_last_name": policy_holder_last_name
}
response = requests.post(url, json=payload, headers=headers)

if response.status_code == 201:
    print(f"Job file '{job_name}' created successfully.")
else:
    print(f"Failed to create job file: {response.status_code} - {response.text}")
return response.json()

[Louis3030]

def assign_employee_to_job(job_id, employee_id): """Assigns an employee to the specified job.""" url = f"https://api.encircleapp.com/v1/jobs/{job_id}/assign" payload = { "employee_id": employee_id } response = requests.post(url, json=payload, headers=headers)

if response.status_code == 200:
    print(f"Employee {employee_id} successfully assigned to job {job_id}.")
else:
    print(f"Failed to assign employee {employee_id} to job {job_id}: {response.text}")

[Louis3030]

def assign_employees_to_job(job_id, employee_ids): """Assigns a list of employees to the specified job.""" for employee_id in employee_ids: assign_employee_to_job(job_id, employee_id)

[Louis3030]

import requests

Example headers for authorization, make sure to store your API key securely

ENCIRCLE_API_KEY = 'your_encircle_api_key_here' headers = { 'Authorization': f'Bearer {ENCIRCLE_API_KEY}', 'Content-Type': 'application/json' }

def create_job_file(policy_holder_first_name, policy_holder_last_name, use_last_name_only=True): """Creates a job file using the policyholder's name. Defaults to last name.""" if use_last_name_only: job_name = policy_holder_last_name else: job_name = f"{policy_holder_first_name} {policy_holder_last_name}"

url = "https://api.encircleapp.com/v1/jobs"  # Adjust the URL to the correct endpoint
payload = {
    "title": job_name,
    "policy_holder_first_name": policy_holder_first_name,
    "policy_holder_last_name": policy_holder_last_name
}

response = requests.post(url, json=payload, headers=headers)

if response.status_code == 201:
    print(f"Job file '{job_name}' created successfully.")
else:
    print(f"Failed to create job file: {response.status_code} - {response.text}")

return response.json() if response.status_code == 201 else None

[Louis3030]

def assign_employee_to_job(job_id, employee_id): """Assigns an employee to the specified job in Encircle.""" url = f"https://api.encircleapp.com/v1/jobs/{job_id}/assign" payload = { "employee_id": employee_id # Replace with actual employee data structure }

response = requests.post(url, json=payload, headers=headers)

if response.status_code == 200:
    print(f"Employee {employee_id} successfully assigned to job {job_id}.")
else:
    print(f"Failed to assign employee {employee_id}: {response.text}")

def assign_employees_to_job(job_id, employee_ids): """Assigns a list of employees to the specified job in Encircle.""" for employee_id in employee_ids: assign_employee_to_job(job_id, employee_id)

[Louis3030]

def automate_job_creation_and_assignment(policy_holder_first_name, policy_holder_last_name, employee_ids, use_last_name_only=True): """Automates the process of job creation and employee assignment."""

# Step 1: Create the job file
job_data = create_job_file(policy_holder_first_name, policy_holder_last_name, use_last_name_only)

if job_data and 'id' in job_data:
    job_id = job_data['id']  # Get the job ID from the response

    # Step 2: Assign employees to the job
    assign_employees_to_job(job_id, employee_ids)
else:
    print("Failed to create job file, skipping employee assignment.")

[Louis3030]

if name == "main": policy_holder_first_name = input("Enter policy holder's first name (or leave blank): ") policy_holder_last_name = input("Enter policy holder's last name: ")

# Use last name only?
use_last_name_only = input("Use last name only? (y/n): ").lower() == 'y'

# Example list of employee IDs
employee_ids = ['emp_1', 'emp_2', 'emp_3']  # Replace with actual employee IDs

# Call the automation function
automate_job_creation_and_assignment(policy_holder_first_name, policy_holder_last_name, employee_ids, use_last_name_only)