Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import openslide
- import boto3
- from io import BytesIO
- from slugify import slugify
- import json
- import os
- import glob
- class SVSFileProcessor:
- def __init__(self, svs_file_path, file_name, aws_access_key, aws_secret_key, s3_bucket_name, s3_region):
- self.svs_file_path = svs_file_path
- self.original_filename = os.path.basename(svs_file_path)
- self.s3_bucket_name = s3_bucket_name
- self.s3_region = s3_region
- self.safe_file_name = slugify(file_name)
- self.s3_client = None
- self.wsi = None
- # Initialize S3 client
- try:
- self.s3_client = boto3.client(
- 's3',
- aws_access_key_id=aws_access_key,
- aws_secret_access_key=aws_secret_key,
- region_name=self.s3_region
- )
- except Exception as e:
- print(f"Kesalahan Kredensial AWS: {str(e)}")
- # Open the SVS file
- try:
- self.wsi = openslide.OpenSlide(self.svs_file_path)
- print(f"File SVS '{self.original_filename}' berhasil dibuka.")
- except openslide.OpenSlideError as e:
- print(f"Gagal membuka file SVS '{self.original_filename}': {e}")
- def extract_properties(self):
- if not self.wsi:
- return []
- data = []
- try:
- properties = self.wsi.properties
- for key, value in properties.items():
- # Clean up key for better compatibility (e.g., in JSON)
- clean_key = key.replace(' ', '-').replace('.', '-')
- data.append({clean_key: str(value)})
- except Exception as e:
- print(f"Terjadi kesalahan saat mengekstrak properti: {e}")
- return data
- def generate_and_upload_thumbnail(self, thumb_size=(256, 256)):
- if not self.wsi or not self.s3_client:
- print("Tidak dapat membuat thumbnail. File SVS atau klien S3 tidak diinisialisasi.")
- return None
- try:
- thumb_image = self.wsi.get_thumbnail(thumb_size)
- thumb_buffer = BytesIO()
- thumb_image.save(thumb_buffer, "PNG")
- thumb_buffer.seek(0)
- s3_key = f"thumbnails/{self.safe_file_name}_thumb.png"
- print(f"Mengunggah thumbnail ke bucket S3 '{self.s3_bucket_name}' dengan key '{s3_key}'...")
- self.s3_client.upload_fileobj(
- thumb_buffer,
- self.s3_bucket_name,
- s3_key,
- )
- s3_url = f"https://{self.s3_bucket_name}.s3.{self.s3_region}.amazonaws.com/{s3_key}"
- print(f"Thumbnail berhasil diunggah. URL: {s3_url}")
- return s3_url
- except Exception as e:
- print(f"Gagal mengunggah thumbnail: {e}")
- return None
- def process_file(self):
- if not self.wsi:
- return {
- "filename": self.original_filename,
- "properties": [],
- "thumbnail": None,
- "error": "File SVS tidak dapat dibaca atau rusak."
- }
- properties_data = self.extract_properties()
- thumbnail_url = self.generate_and_upload_thumbnail()
- return {
- "filename": self.original_filename,
- "properties": properties_data,
- "thumbnail": thumbnail_url
- }
- def __del__(self):
- if self.wsi:
- self.wsi.close()
- def process_svs_batch(directory_path, aws_access_key, aws_secret_key, s3_bucket_name, s3_region):
- search_pattern = os.path.join(directory_path, '*.svs')
- file_paths = glob.glob(search_pattern)
- if not file_paths:
- print(f"Peringatan: Tidak ada file .svs yang ditemukan di '{directory_path}'")
- return json.dumps([], indent=4)
- print(f"Ditemukan {len(file_paths)} file .svs untuk diproses.")
- all_results = []
- for file_path in file_paths:
- print("\n" + "="*20 + f" Memproses {os.path.basename(file_path)} " + "="*20)
- file_name_base = os.path.splitext(os.path.basename(file_path))[0]
- processor = SVSFileProcessor(
- svs_file_path=file_path,
- file_name=file_name_base,
- aws_access_key=aws_access_key,
- aws_secret_key=aws_secret_key,
- s3_bucket_name=s3_bucket_name,
- s3_region=s3_region
- )
- result = processor.process_file()
- all_results.append(result)
- return json.dumps(all_results, indent=4)
- # --- CONTOH PENGGUNAAN ---
- if __name__ == '__main__':
- SVS_DIRECTORY = input("Masukkan path lengkap ke direktori yang berisi file .svs .tiff \n contoh /Users/f/keselyoleren/neurabot/product/data/slide : ")
- AWS_ACCESS_KEY_ID = os.environ.get('AWS_ACCESS_KEY_ID', 'AKIA5XSKJQZJZCKNAUKI')
- AWS_SECRET_ACCESS_KEY = os.environ.get('AWS_SECRET_ACCESS_KEY', '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')
- S3_BUCKET = os.environ.get('S3_BUCKET', 'mylab-production')
- S3_REGION = os.environ.get('S3_REGION', 'ap-southeast-3')
- if not os.path.isdir(SVS_DIRECTORY):
- print("="*50)
- print(f"Error: Direktori tidak ditemukan di '{SVS_DIRECTORY}'")
- print("Harap jalankan kembali skrip dan masukkan path folder yang benar.")
- print("Contoh /Users/f/keselyoleren/neurabot/product/data/slide")
- print("="*50)
- elif 'YOUR_AWS' in AWS_ACCESS_KEY_ID or 'your-s3' in S3_BUCKET:
- print("="*50)
- print(f"Peringatan: Kredensial AWS atau nama bucket belum diatur.")
- print("Harap perbarui variabel kredensial AWS atau atur melalui environment variables.")
- print("="*50)
- else:
- json_output = process_svs_batch(
- directory_path=SVS_DIRECTORY,
- aws_access_key=AWS_ACCESS_KEY_ID,
- aws_secret_key=AWS_SECRET_ACCESS_KEY,
- s3_bucket_name=S3_BUCKET,
- s3_region=S3_REGION
- )
- output_filename = 'svs_metadata_output.json'
- print(f"\n\n--- HASIL OUTPUT JSON GABUNGAN ---")
- print(f"Menyimpan hasil ke file '{output_filename}'...")
- with open(output_filename, 'w') as f:
- f.write(json_output)
- print("Selesai.")
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement