modules.py

#!/usr/bin/env python3
# encoding: utf-8
#
# Copyright (C) 2022 Max Planck Institute for Multidisclplinary Sciences
# Copyright (C) 2022 University Medical Center Goettingen
# Copyright (C) 2022 Ajinkya Kulkarni <ajinkya.kulkarni@mpinat.mpg.de>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# Visit https://docs.indiscale.com/ for CaosDB Documentation.

# UX/UI recommendations provided by Radhika Bhagwat (radhika.bhagwat3@gmail.com, Product Designer) and Md Motiur Rahman Sagar <motiur.sagar@mpinat.mpg.de>

########################################################################################

import os

import boto3
from botocore.exceptions import ClientError

import caosdb as db
import urllib3
urllib3.disable_warnings() # Disable the HTTPS warnings for CaosDB authentication

import sys
# Don't generate the __pycache__ folder locally
sys.dont_write_bytecode = True 
# Print exception without the buit-in python warning
sys.tracebacklimit = 0 

# Initialize the logger
import logging
logger = logging.getLogger()
logger.setLevel(logging.ERROR)

########################################################################################

from PASSWORDS import *

########################################################################################

def make_multipart_upload(tiff_file, bucket_name, amazon_bucket_target_name, FolderPathKey):

	"""
	This function creates a multipart upload in the specified S3 bucket and uploads the specified tiff file to the specified target name in the bucket.

	Parameters:
		tiff_file (str): The name of the tiff file to be uploaded.
		bucket_name (str): The name of the S3 bucket where the file will be uploaded.
		amazon_bucket_target_name (str): The target name for the file in the S3 bucket.
		FolderPathKey (str): The path to the folder where the tiff file is located.
		
	Returns:
		None
	"""

	# Create a boto3 client for S3
	s3 = boto3.client('s3', endpoint_url = AMAZON_S3_ENDPOINT_URL, aws_access_key_id = AMAZON_S3_ACCESS_KEY, aws_secret_access_key = AMAZON_S3_SECRET_KEY)

	try:
		# Create a multipart upload
		response = s3.create_multipart_upload(Bucket=bucket_name, Key=amazon_bucket_target_name)

		# Get the upload ID
		upload_id = response['UploadId']

		# Initialize the part number and the parts list
		part_number = 1
		parts = []
		
		try:
			# Upload the part
			response = s3.upload_part(Body=open(os.path.join(FolderPathKey, tiff_file), 'rb'), Bucket=bucket_name, Key=amazon_bucket_target_name, PartNumber=part_number, UploadId=upload_id)

			# Append the part to the parts list
			parts.append({'ETag': response['ETag'], 'PartNumber': part_number})

			# Increment the part number
			part_number += 1

		except ClientError as e:

			logger.error("Failed to upload %s", tiff_file, exc_info=True)
			pass
		try:
			# Complete the multipart upload
			response = s3.complete_multipart_upload(Bucket=bucket_name, Key=amazon_bucket_target_name, UploadId=upload_id, MultipartUpload={'Parts': parts})
		except ClientError as e:
			logger.error("Failed to complete multipart upload", exc_info=True)
			pass
	finally:
		s3.close()

########################################################################################

# Function which checks for erroneous input(s) for Aperture(s), Exposure Time(s) and Active Channels and returns the appropriate exception

def SanityChecks(my_list1, my_list2, my_list3):
	"""
	This function performs sanity checks on the provided lists.
	Parameters:
		my_list1 (list): A list of strings representing whether a channel is active or not. The string should be 'Yes' or 'No'.
		my_list2 (list): A list of integers representing the aperture of the corresponding channel in my_list1.
		my_list3 (list): A list of integers representing the exposure time of the corresponding channel in my_list1.
	Returns:
		None
	Raises:
		ValueError: If aperture or exposure time is not as expected for the corresponding active/non-active channel.
	"""
	for i in range(len(my_list1)):
		is_active = my_list1[i] == 'Yes'
		aperture = int(my_list2[i])
		exposure_time = int(my_list3[i])
		if is_active:
			if aperture == 0:
				raise ValueError("Aperture(s) should not be 0 for active channels")
			if exposure_time == 0:
				raise ValueError("Exposure Time(s) should not be 0 for active channels")
		else:
			if aperture != 0:
				raise ValueError("Aperture(s) should be 0 for non-active channels")
			if exposure_time != 0:
				raise ValueError("Exposure Time(s) should be 0 for non-active channels")

#######################################################################################