diff --git a/Readme.md b/Readme.md --- a/Readme.md +++ b/Readme.md @@ -142,4 +142,9 @@ For help, run `python -m owi.cli --help` or `owi --help`. ## Screencast -A demonstration is available via this [screencast demo](https://syncandshare.lrz.de/getlink/fi9Dat4pLDLtE9bZsiGqFd/owi-demo.mov). \ No newline at end of file +A demonstration is available via this [screencast demo](https://syncandshare.lrz.de/getlink/fi9Dat4pLDLtE9bZsiGqFd/owi-demo.mov). + +# Development + +- `poetry add --group dev sphinx myst-parser ` +- `poetry run sphinx-build -b html docs/source/ docs/build/html` diff --git a/pyproject.toml b/pyproject.toml --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [tool.poetry] name = "owilix" version = "0.3.0" -description = "OWILIX - the Command Line Interface for the Open Web Index. Your standing stone, to access the Open Web Index (and please, don't throw it at people)." +description = "OWILIX - the Command Line Interface for slicing and consuming the Open Web Index. " authors = ["Michael Granitzer "] license = "MIT" maintainers = [ @@ -33,6 +33,7 @@ [tool.poetry.group.dev.dependencies] sphinx = "^7.3.7" myst-parser = "^3.0.1" +pytest = "^8.2.0" [build-system] requires = ["poetry-core"] @@ -41,6 +42,8 @@ [tool.poetry.scripts] owi = "owi.cli:main" owilix = "owi.cli:main" +test = "pytest" + diff --git a/docs/Makefile b/docs/Makefile new file mode 100644 --- /dev/null +++ b/docs/Makefile @@ -0,0 +1,20 @@ +# Minimal makefile for Sphinx documentation +# + +# You can set these variables from the command line, and also +# from the environment for the first two. +SPHINXOPTS ?= +SPHINXBUILD ?= sphinx-build +SOURCEDIR = source +BUILDDIR = build + +# Put it first so that "make" without argument is like "make help". +help: + @$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O) + +.PHONY: help Makefile + +# Catch-all target: route all unknown targets to Sphinx using the new +# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS). +%: Makefile + @$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O) diff --git a/docs/make.bat b/docs/make.bat new file mode 100644 --- /dev/null +++ b/docs/make.bat @@ -0,0 +1,35 @@ +@ECHO OFF + +pushd %~dp0 + +REM Command file for Sphinx documentation + +if "%SPHINXBUILD%" == "" ( + set SPHINXBUILD=sphinx-build +) +set SOURCEDIR=source +set BUILDDIR=build + +%SPHINXBUILD% >NUL 2>NUL +if errorlevel 9009 ( + echo. + echo.The 'sphinx-build' command was not found. Make sure you have Sphinx + echo.installed, then set the SPHINXBUILD environment variable to point + echo.to the full path of the 'sphinx-build' executable. Alternatively you + echo.may add the Sphinx directory to PATH. + echo. + echo.If you don't have Sphinx installed, grab it from + echo.https://www.sphinx-doc.org/ + exit /b 1 +) + +if "%1" == "" goto help + +%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O% +goto end + +:help +%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O% + +:end +popd diff --git a/owilix/__init__.py b/owilix/__init__.py --- a/owilix/__init__.py +++ b/owilix/__init__.py @@ -0,0 +1,31 @@ +""" +OWI CLI Module +============== + +The OWI (Open Web Index) CLI is a command-line interface designed to manage data slices for the Open Web Index. This module facilitates operations such as pulling data, listing files with specific criteria, and querying datasets using DuckDB. + +Installation: +------------- +OWI CLI can be installed via Conda environment using a package URL, directly from the repository, or using Poetry with support for optional plugins. It requires Python 3.10 or 3.11 due to compatibility issues with newer versions in some dependencies. + +Usage: +------ +The CLI provides functionalities to pull and list dataset slices from specified data centers on specified dates or periods. Advanced options allow for filtering and detailed querying with the aid of DuckDB. The default path for storing data can be modified using the OWS_OWI_PATH environment variable. + +Commands include: +- `owi pull ` to retrieve data. +- `owi ls ` for listing and filtering available files. +- `owi duckdb ` to perform SQL queries on retrieved data. + +Advanced Usage: +--------------- +Examples of advanced usage involve setting up sessions, authenticating, and executing detailed queries and filters using Python's integration with fsspec and DuckDB. + +Troubleshooting: +---------------- +Common issues involve authentication problems, Python version incompatibilities, and DuckDB operational errors. Solutions include file deletions and dependency reinstalls. + +For detailed documentation and a screencast demo, visit [OpenWebIndex documentation](http://openwebindex.eu). + +Developed with Poetry, this module also includes Sphinx documentation setup for easy maintenance and extension of usage docs. +""" diff --git a/owilix/cli.py b/owilix/cli.py --- a/owilix/cli.py +++ b/owilix/cli.py @@ -6,8 +6,8 @@ import owilix.plugin import pandas as pd from rich.console import Console -from owilix.core import OWSProject -from owilix.core.commands import show_dataframe, read_config, write_config, DownloadManager, IRODSBasedAccess +from owilix.core import OWIlixProject +from owilix.core.commands import show_dataframe, read_config, write_config, LexisDownloadManager, IRODSBasedCommands from owilix.core.scripts import load_and_check # TODO add unpack and consider tar.gz when checking for consistency @@ -39,8 +39,9 @@ @click.option('--yes', is_flag=True, default=False, help='Automatically confirm all prompts') @click.option('--nparallel', default=4, help='Maximum number of parallel threads used for download') @click.option('--verbose', is_flag=True, default=False, help='Set verbose output') +@click.option('--slice', default="main", help='Local slice name of the index') @click.pass_context -def cli(ctx, fields, tablefmt, target, yes, nparallel, verbose): +def cli(ctx, fields, tablefmt, target, yes, nparallel, verbose, slice): """ Main command line interface group for OWI management tools. @@ -61,7 +62,8 @@ ctx.obj['AUTOYES'] = yes ctx.obj['NPARALLEL'] = nparallel ctx.obj['VERBOSE'] = verbose - ctx.obj['OWI'] = OWSProject(target, target + os.path.sep + ".logs") + ctx.obj['SLICE'] = slice + ctx.obj['OWI'] = OWIlixProject(target, target + os.path.sep + ".logs") def collect_kwargs(ctx, param, value): """Collects values into a dictionary, maintaining them under keys as lists if repeated.""" @@ -127,10 +129,10 @@ Parameters: specifier (str): A specifier that filters datasets according to data center, date, or metadata. """ - dm = DownloadManager(ctx.obj['OWI'], **ctx.obj) + dm = LexisDownloadManager(ctx.obj['OWI'], **ctx.obj) try: #dm.pull_logic(specifier, file_select, no_sync, no_ext, details=details, **ctx.obj) - _datasets, _files = dm.get_datasets_and_files(specifier, files, True, details=details, return_files=files is not None) + _datasets, _files = dm.get_datasets_and_files(specifier, files, True, details=details, use_irods=False) _console = ctx.obj["CONSOLE"] _console.print(f"Found {len(_datasets)} Datasets") if len(_datasets)>0: @@ -166,7 +168,7 @@ This command is used to download and optionally extract datasets for further local analysis or processing. """ - dm = DownloadManager(ctx.obj['OWI'], **ctx.obj) + dm = LexisDownloadManager(ctx.obj['OWI'], **ctx.obj) try: #dm.pull_logic(specifier, files, no_sync, details, details=details) results = dm.pull_file_based(specifier, files, no_sync, details) @@ -176,6 +178,7 @@ ctx.obj["CONSOLE"].print(f"Download for {len(df[df['Downloaded']])} out of {len(df)} " f"successfully finished. {len(df[~df['Prepared']])} failed in prepration.") if details: + ctx.obj["CONSOLE"].print("Following files were downloaded:") show_dataframe(df, **ctx.obj) else: if (~df['Downloaded']).any(): @@ -192,7 +195,7 @@ def duckdb(ctx, specifier, files): """ """ - dm = IRODSBasedAccess(ctx.obj['OWI'], **ctx.obj) + dm = IRODSBasedCommands(ctx.obj['OWI'], **ctx.obj) try: #dm.pull_logic(specifier, files, no_sync, details, details=details, **ctx.obj) results = dm.duckdb_shell(specifier, files) @@ -200,7 +203,7 @@ ctx.obj['CONSOLE'].print("Pull aborted by user") -@click.command(help='runs SQL Script via duckdb: ' + IRODSBasedAccess.sql.__doc__) +@click.command(help='runs SQL Script via duckdb sql : ' + IRODSBasedCommands.sql.__doc__) @click.argument('specifier', default="all") @click.argument('script', default="Select url from read_parquet({owi_remote_files}) where url_suffix= 'de'") @click.option('--files', default="all", help='Specifier to select specific files within the datasets.') @@ -212,7 +215,7 @@ """ Executes SQL using duckdb """ - dm = IRODSBasedAccess(ctx.obj['OWI'], **ctx.obj) + dm = IRODSBasedCommands(ctx.obj['OWI'], **ctx.obj) try: #dm.pull_logic(specifier, files, no_sync, details, details=details, **ctx.obj) params = {} @@ -221,6 +224,28 @@ if ff: script = load_and_check(script, **params) dm.sql(specifier, files, script ,ctx.obj.get("VERBOSE",False), explain=explain, **params) + except UserWarning as uw: + ctx.obj['CONSOLE'].print("Pull aborted by user") + except Exception as e: + raise e + +@click.command(help='Allows to slice OWI data vertically and horizontally. use fillet ' + IRODSBasedCommands.fillet.__doc__) +@click.argument('specifier', default="all") +@click.argument('fields', default="id, url, plain_text, url_suffix") +@click.option('--files', default="**/*", help='Specifier to select specific files within the datasets.') +@click.option('--no_overwrite', is_flag=True, default=False, help='overwrite is not allowed if true') +@click.option('--where', default="", help="SQL Where clause" ) +@click.option('--no_sync', is_flag=True, help='If set, skips the synchronization check against local versions.') +@click.option('--partitions', default="", help="fields to partition the stored data (HIVE like partitoining)" ) +@click.pass_context +def fillet(ctx, specifier, fields, files, where, partitions, no_sync=False, no_overwrite=False, ): + """ + Executes SQL using duckdb + """ + dm = IRODSBasedCommands(ctx.obj['OWI'], **ctx.obj) + try: + dm.fillet( specifier, files, fields, where, partitioned_by=partitions, + overwrite_ignore=not no_overwrite, verbose= ctx.obj.get("VERBOSE",False)) except UserWarning as uw: ctx.obj['CONSOLE'].print("Pull aborted by user") except Exception as e: @@ -270,7 +295,7 @@ def main(): - cmds = [local, stats, ls, pull, events, clean, config, remote, duckdb, sql] + cmds = [local, stats, ls, pull, events, clean, config, remote, duckdb, sql, fillet] for i in cmds: cli.add_command(i) cli(obj={}) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 --- /dev/null +++ b/tests/__init__.py diff --git a/tests/owilix_project.py b/tests/owilix_project.py new file mode 100644 --- /dev/null +++ b/tests/owilix_project.py @@ -0,0 +1,88 @@ +import tempfile + +import pytest +from unittest.mock import MagicMock, patch +import pandas as pd +from datetime import datetime +from owilix.core import OWIlixProject # Adjust this import to the correct module path + +# Sample DataFrame returned by the mocked ls_http method +mocked_df = pd.DataFrame({ + 'Title': ['TOWI-The Open Web Index-2023-12-25@lrz-eng-multi', 'TOWI-The Open Web Index-2023-12-25@lrz-eng', + 'owi test data bigger', 'parquet-test', 'OWI-TEST', 'OWI-TEST'], + 'Access': ['public', 'public', 'public', 'public', 'project', 'project'], + 'Project': ['openwebsearch', 'openwebsearch', 'openwebsearch', 'openwebsearch', 'openwebsearch', 'openwebsearch'], + 'Zone': ['IT4ILexisV2'] * 6, + 'InternalID': ['f3ffccb4-dc63-11ee-a201-0242c0a87004', 'c46e4d62-dc5f-11ee-ad7e-0242c0a87004', + 'e6a2de78-dc57-11ee-ad7e-0242c0a87004', '5aa5c3dc-dc56-11ee-ad7e-0242c0a87004', + '6d4bcdec-e2a5-11ee-9017-0242c0a8f009', 'ba368b68-e2ac-11ee-9017-0242c0a8f009'], + 'CreationDate': ['2024-03-07 09:20:36', '2024-03-07 08:50:38', '2024-03-07 07:54:20', '2024-03-07 07:43:14', + '2024-03-15 08:24:23', '2024-03-15 09:16:42'], + 'Owner': [[['The OpenWebSearch.eu Consortium']], [['The OpenWebSearch.eu Consortium']], + [['mgrani']], [['mgrani']], [], []], + 'Creator': [[['The OpenWebSearch.eu Consortium']], [['The OpenWebSearch.eu Consortium']], + [['mgrani']], [['mgrani']], [], []], + 'Contributor': [[['The OpenWebSearch.eu Consortium']], [['The OpenWebSearch.eu Consortium']], + [['mgrani']], [['mgrani']], [], []], + 'Publisher': [[['The OpenWebSearch.eu Consortium']], [['The OpenWebSearch.eu Consortium']], + [['mgrani']], [['mgrani']], [], []], + 'PublicationYear': ['UNKNOWN Publication Year'] * 6, + 'ResourceType': ['Open Web Index V1.0', 'TOWI', 'OWI Data Set', 'OWI Data Set', 'UNKNOWN Resource Type', + 'UNKNOWN Resource Type'], + 'Compression': ['no'] * 6, + 'Encryption': ['no'] * 6, + 'DataCenter': [None] * 6, + 'Date': [pd.NaT] * 6 +}) + +mocked_files = {} + +@pytest.fixture +def project(): + """Fixture to create an OWIlixProject instance for testing with a temporary directory.""" + with tempfile.TemporaryDirectory() as tmp_owi_path: + with tempfile.TemporaryDirectory() as tmp_log_path: + yield OWIlixProject(owi_path=tmp_owi_path, logpath=tmp_log_path) + +@patch('py4lexis.session.LexisSession', autospec=True) +@patch('owilix.core.OWIRemoteData.ls_http') +def test_stats(mock_ls_http, MockLexisSession, project): + """Test the stats method.""" + # Set up the mock return value for ls_http + mock_ls_http.return_value = (mocked_df, mocked_files) + + # Call the stats method with the mocked dependencies + stats_df = project.stats(day='2024-05-01', duration=1) + + # Assertions to check the correct statistics + assert not stats_df.empty + assert stats_df['Key'].iloc[0] == "Total" + assert stats_df['Count'].sum() == 64 # Two datasets returned in mocked_df + +@pytest.mark.parametrize("specifier, expected", [ + ("BAdW-LRZ:2024-01-03", {'data_center': 'BAdW-LRZ', 'day': datetime(2024, 1, 3), 'duration': None, 'query': {}}), + ("all:2024-01-03", {'data_center': 'all', 'day': datetime(2024, 1, 3), 'duration': None, 'query': {}}), + ("it4i:latest", {'data_center': 'it4i', 'day': datetime(2024, 5, 10), 'duration': None, 'query': {}}), + ("csc/Access=public", {'data_center': 'csc', 'day': None, 'duration': None, 'query': {'Access': 'public'}}), + ("BAdW-LRZ:2024-01-03#4", {'data_center': 'BAdW-LRZ', 'day': datetime(2024, 1, 3), 'duration': 4, 'query': {}}), + ("csc:*#*/Access=public", {'data_center': 'csc', 'day': None, 'duration': None, 'query': {'Access': 'public'}}), + ("BAdW-LRZ:2024-01-03/Access=public;ResourceType=warc", {'data_center': 'BAdW-LRZ', 'day': datetime(2024, 1, 3), 'duration': None, 'query': {'Access': 'public', 'ResourceType': 'warc'}}), + ("BAdW-LRZ", {'data_center': 'BAdW-LRZ', 'day': None, 'duration': None, 'query': {}}), + ("all", {'data_center': 'all', 'day': None, 'duration': None, 'query': {}}) +]) +def test_parse_specifier(project, specifier, expected): + """Test the parse_specifier method with various specifier formats.""" + parsed = project.parse_specifier(specifier) + + # Correct 'latest' handling, convert to actual date if needed + if parsed['day'] == 'latest': + # Mock today's date for test consistency; in real scenario, replace with actual logic as needed + parsed['day'] = datetime(2024, 5, 10) + + # Assert conditions for each field + assert parsed['data_center'] == expected['data_center'] + assert parsed['day'] == expected['day'] + assert parsed['duration'] == expected['duration'] + assert parsed['query'] == expected['query'] + +# Sample \ No newline at end of file diff --git a/tests/owilix_remotedata.py b/tests/owilix_remotedata.py new file mode 100644 --- /dev/null +++ b/tests/owilix_remotedata.py @@ -0,0 +1,84 @@ +import os +import tempfile + +import pytest +from unittest.mock import MagicMock, patch +from pandas import DataFrame + +# Assuming the class OWIRemoteData and related classes are in the module `owilix.core.remote_data` +from owilix.core import OWIRemoteData +from owilix.core import OWIlixProject + + +@pytest.fixture +def mock_project(): + project = MagicMock(spec=OWIlixProject) + project.session = MagicMock() + project.name = "test_project" + return project + +@pytest.fixture +def remote_data(mock_project): + return OWIRemoteData(project=mock_project) + +def test_ls(remote_data): + with tempfile.TemporaryDirectory() as tmp_owi_path: + prj = OWIlixProject(owi_path=tmp_owi_path, logpath=tmp_owi_path+os.path.sep+'.logs') + _di, _fi = prj.remote_data.ls_irods("2024-05-01", 1) + _dh, _fh = prj.remote_data.ls_http("2023-11-29", 0) + assert isinstance(_di, DataFrame) + assert isinstance(_fi, DataFrame) + assert isinstance(_dh, DataFrame) + assert isinstance(_fh, DataFrame) + assert _fi=={} + assert _fh=={} + + _di, _fi = prj.remote_data.ls_irods("2024-05-01", 0, files_glob="**/*") + _dh, _fh = prj.remote_data.ls_http("2023-11-29", 0, files_glob="**/*") + + assert isinstance(_di, DataFrame) + assert isinstance(_fi, DataFrame) + assert isinstance(_dh, DataFrame) + assert isinstance(_fh, DataFrame) + assert len(_fi)==len(_di) + assert len(_fh)==len(_dh) + +def test_ls_irods_empty_return(remote_data): + with patch.object(remote_data.irods.filesystem, 'ls', return_value=[]): + df, files = remote_data.ls_irods() + assert isinstance(df, DataFrame) + assert df.empty + assert isinstance(files, dict) + assert not files # files should be empty + +def test_ls_http_empty_return(remote_data): + with patch.object(remote_data.dscli, 'get_all_datasets', return_value=DataFrame()): + df, files = remote_data.ls_http() + assert isinstance(df, DataFrame) + assert df.empty + assert isinstance(files, dict) + assert not files # files should be empty + +def test_ls_irods_with_filters(remote_data): + # Assuming `remote_data.irods.filesystem.ls` needs to return a list of paths + test_data = ["/zone/public/proj862c5962623246664c1fda27b7afb108/dataset1"] + with patch.object(remote_data.irods.session.collections, 'get', return_value=MagicMock(metadata={"id": "dataset1", "title": "Test Dataset"})): + with patch.object(remote_data.irods.filesystem, 'ls', return_value=test_data): + df, files = remote_data.ls_irods(day='2022-01-01', duration=10, data_center='test_center') + assert not df.empty + assert 'dataset1' in df['InternalID'].values + +def test_ls_http_with_query(remote_data): + # Setup test DataFrame returned by dscli.get_all_datasets + test_data = DataFrame({'Title': ['test_center @Test-2022/01/01'], + 'Project': ['test_project'], + 'InternalID': ['1234']}) + test_data['Title'] = test_data['Title'].astype(str) # Ensure correct type for extraction + + with patch.object(remote_data.dscli, 'get_all_datasets', return_value=test_data): + df, files = remote_data.ls_http(day='2022-01-01', duration=10, data_center='test_center', query={'InternalID': '1234'}) + assert not df.empty + assert '1234' in df['InternalID'].values + assert 'test_center' in df['DataCenter'].values + +# Additional tests could include checking the effects of different `files_glob` patterns, `cb_progress` usage, etc. diff --git a/docs/source/commands.md b/docs/source/commands.md new file mode 100644 --- /dev/null +++ b/docs/source/commands.md @@ -0,0 +1,12 @@ +# OWIlix Commands + +The OWIlix CLI includes custom commands that can be executed within the Flask application context. +These commands are defined in the `owilix/core/commands.py` file and are imported and parameterized in `owilix/cli.py`. + +```{automodule} owilix.core.commands +:members: +:undoc-members: +:private-members: +:special-members: __init__ +:show-inheritance: +``` \ No newline at end of file diff --git a/docs/source/conf.py b/docs/source/conf.py new file mode 100644 --- /dev/null +++ b/docs/source/conf.py @@ -0,0 +1,37 @@ +# Configuration file for the Sphinx documentation builder. +# +# For the full list of built-in configuration values, see the documentation: +# https://www.sphinx-doc.org/en/master/usage/configuration.html + +# -- Project information ----------------------------------------------------- +# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information + +project = 'Owilix' +copyright = '2024, Michael Granitzer' +author = 'Michael Granitzer' +release = '0.3.0' + +# -- General configuration --------------------------------------------------- +# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration + +extensions = [] + +templates_path = ['_templates'] +exclude_patterns = [] + +extensions = [ + 'myst_parser', + 'sphinx.ext.autodoc', + 'sphinx.ext.napoleon', # If you use Google or Numpy style docstrings + 'sphinx.ext.viewcode', # Include links to source code +] + +# -- Options for HTML output ------------------------------------------------- +# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output + +html_theme = 'alabaster' +html_static_path = ['_static'] + +import os +import sys +sys.path.insert(0, os.path.abspath('../..')) # Adjust the path according to the location of your package relative to the docs directory diff --git a/docs/source/index.rst b/docs/source/index.rst new file mode 100644 --- /dev/null +++ b/docs/source/index.rst @@ -0,0 +1,31 @@ +.. Owilix documentation master file, created by + sphinx-quickstart on Sat May 11 09:52:31 2024. + You can adapt this file completely to your liking, but it should at least + contain the root `toctree` directive. + +Welcome to Owilix's documentation! +================================== + +.. image:: ./_static/owili-image-v1.webp + :alt: Logo + :align: center + +.. toctree:: + :maxdepth: 2 + :caption: Contents: + + commands.md + +.. automodule:: owilix + :members: + :undoc-members: + :show-inheritance: + + + +Indices and tables +================== + +* :ref:`genindex` +* :ref:`modindex` +* :ref:`search` diff --git a/owilix/core/__init__.py b/owilix/core/__init__.py --- a/owilix/core/__init__.py +++ b/owilix/core/__init__.py @@ -1,2 +1,5 @@ -from .project import OWSProject -from .render import * \ No newline at end of file + +from .render import * +from .datasets import OWIRemoteData +from .local import OWILocal +from .project import OWIlixProject \ No newline at end of file diff --git a/owilix/core/commands.py b/owilix/core/commands.py --- a/owilix/core/commands.py +++ b/owilix/core/commands.py @@ -2,13 +2,17 @@ Module containing different commands that go beyond calling a single function from OWIProject. It uses rich ui python library for interaction. - BaseCommand: Base class for all commands. It contains common methods and attributes used by all commands, particularly in resolviing context parameters like getting the rich console etc. +- LexisDownloadManager: class for managing download of partitoins from the Open Web Index. It uses the Lexis Rest API and does not rely on iRODS for file access. +- IRODSBasedCommands: class for accessing the iRODS datasets and defining a set of commands using duckdb plus IRODS. """ import configparser import os import random import re import time +from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed +import traceback import pandas as pd from rich.progress import Progress, TextColumn, BarColumn, TimeRemainingColumn @@ -59,11 +63,10 @@ This will display columns 'FirstName' and 'Age' for the given DataFrame. """ # Split the patterns and compile them to regex objects for efficient matching - if "FIELDS" not in kwargs or kwargs['FIELDS'] is None or kwargs['FIELDS'] =="": + if "FIELDS" not in kwargs or kwargs['FIELDS'] is None or kwargs['FIELDS'] == "": patterns = [] else: patterns = [re.compile(pattern) for pattern in kwargs['FIELDS'].split(',')] - def find_excluded_columns(columns, patterns): """ @@ -81,6 +84,7 @@ # Filter the DataFrame to only include matched columns filtered_df = datasets[excluded_columns] render_dataframe(filtered_df, kwargs["CONSOLE"]) # Render the filtered DataFrame + def read_config(target): """ @@ -108,6 +112,7 @@ if os.path.exists(config_path): config.read(config_path) return config + def write_config(target, section, option, value): """ @@ -145,6 +150,7 @@ with open(config_path, 'w') as configfile: config.write(configfile) + class BaseCommand: def __init__(self, owi, **kwargs): @@ -161,114 +167,102 @@ self.owi = owi self.kwargs = kwargs self.console = kwargs["CONSOLE"] - self.target = kwargs["TARGET"] + self._target = kwargs["TARGET"] self.autoyes = kwargs["AUTOYES"] self.n_parallel = kwargs.get("NPARALLEL", 4) + self.slice = kwargs.get("SLICE", "main") + + @property + def target(self): + return os.path.join(self._target, self.slice) def show_dataframe(self, dataframe): show_dataframe(dataframe, **self.kwargs) - def get_datasets_and_files(self, specifier, file_select, no_sync, details=False, return_files=False): + def get_datasets_and_files(self, specifier, file_select=None, no_sync=False, details=False, use_irods=False): """ - Retrieve and analyze datasets based on a specified criteria. + Retrieve and analyze datasets based on a provided dataframe and a file glob (i.e. files is a astring) or a + dict of {id:[files]} per dataset. Generalises irods and lexis datasets. Args: - specifier (str): Criteria to filter datasets. - file_select (str): Pattern or filter to select specific files within datasets. + df_datasets (Dataframe): Criteria to filter datasets. + files (str or dict): Pattern or filter to select specific files within datasets. or a dict containing the files for a dataset id no_sync (bool): If True, do not synchronize with the remote source and only use local datasets. details (bool, optional): If True, collect detailed information about dataset files. Defaults to False. - return_files (bool, optional): If True, return detailed file information per dataset. Defaults to False. + use_irods (bool, optional): If True, use irods to fetch the datasets. Defaults to False. Returns: tuple: A tuple containing: - datasets (DataFrame): DataFrame containing the filtered datasets with additional information. - - _dataset_files (dict): Dictionary containing the detailed files information per dataset. + - _dataset_files (dict{str:DataFrame}): Dictionary containing the detailed files information per dataset. """ # List datasets according to the provided specifier. - datasets = self.owi.datasets.ls(**self.owi.parse_specifier(specifier)) - # Compare locally with target to identify differences. - datasets_full = datasets if no_sync else self.owi.datasets.match(datasets, self.target) - # If not no_sync, filter to only include datasets that are modified or new. - if not no_sync: - datasets = datasets[(datasets["FileStatus"] == "*") | (datasets["FileStatus"] == "M")] - if len(datasets) == 0: - # If no datasets need updating, print the status and return. - self.console.print(f"Found the following {len(datasets_full)} datasets, which " - f"also have local versions (content and metadata). Everything seems up-to-date") - self.show_dataframe(datasets_full) - return datasets_full, None - # Initialize the dictionary to hold dataset file details. - _dataset_files = {} - # Set default file selection pattern. - file_select = ".*" if file_select is None or file_select.lower() == "all" else file_select - # If details are required or file selection is specified, analyze the datasets. - if details and len(datasets) > 0 or file_select != ".*": - task = None - with Progress(transient=True) as progress: - if task is None: - task = progress.add_task(f"Estimating details for datasets", total=len(datasets)) - datasets["n_parquet"] = 0 - datasets["n_ciff"] = 0 - datasets["n_groups_parquet"] = 0 - datasets["n_groups_ciff"] = 0 - datasets["n_to_sync"] = 0 - for ix, dataset in datasets.iterrows(): - progress.update(task, advance=0, description=f"Listing files for {dataset['Title']}") - try: - # Fetch files from the remote dataset. - files = self.owi.datasets.files(dataset['InternalID'], dataset['Access'], - zone=dataset['Zone'], - path_filter=file_select) - files = files[files["Type"] == "file"] - progress.update(task, advance=1, - description=f"Files fetched. Doing analysis on {dataset['Title']}") - # Retrieve local files from the target directory and match them. - local_path = f"{self.target}/{dataset['InternalID']}/" - files['Local'] = files['Path'].apply( - lambda x: '*' if not os.path.exists(os.path.join(local_path, x)) else '' - ) - # Analyze file types and count groups. - files['Directory'] = files['Path'].str.rsplit('/', n=1).str[0].where( - files['Path'].str.contains('/'), '') - df_parquet = files[files['Path'].str.contains(r'\.parquet$', regex=True)] - df_ciff = files[files['Path'].str.contains('ciff', regex=True)] + def fn_cb_filter(df): + df = self.owi.remote_data.match(df, self.target) + if no_sync: + return df + else: + return df[(df["FileStatus"] == "*") | (df["FileStatus"] == "M")] - # Count the number of files and groups. - datasets.at[ix, "n_parquet"] = len(df_parquet) - datasets.at[ix, "n_ciff"] = len(df_ciff) - datasets.at[ix, "n_groups_parquet"] = df_parquet['Directory'].nunique() - datasets.at[ix, "n_groups_ciff"] = df_ciff['Directory'].nunique() - datasets.at[ix, "n_to_sync"] = len(files[files['Local'] == "*"]) + cb_filter = None if no_sync else fn_cb_filter + spec = self.owi.parse_specifier(specifier) # TODO: needs to be implemented - # If required, collect file details for this dataset. - if return_files: - if no_sync: - _dataset_files[dataset['InternalID']] = files - else: - _dataset_files[dataset['InternalID']] = files[files["Local"] == "*"] - except Exception as e: - # Handle exceptions by setting negative counts. - self.console.print( - f"[warning]Warning[\warning]: files could not be obtained for {dataset['Title']} reason: {str(e)}") - datasets.at[ix, "n_parquet"] = -1 - datasets.at[ix, "n_ciff"] = -1 - datasets.at[ix, "n_groups_parquet"] = -1 - datasets.at[ix, "n_groups_ciff"] = -1 - _dataset_files[dataset['InternalID']] = None + with Progress(transient=True) as progress: + task = progress.add_task(f"Estimating details for datasets", total=1) - return datasets, _dataset_files + def fn_cb_progress(current, total, id, title): + progress.update(task, advance=current / total, description=f"Listing files for {id}:{title}") + if details and file_select is None: file_select = "**/*" + if use_irods: + _datasets, _dataset_files = self.owi.remote_data.ls_irods(**spec, files_glob=file_select, + cb_progress=fn_cb_progress, + cb_filter=cb_filter) + else: + _datasets, _dataset_files = self.owi.remote_data.ls_http(**spec, files_glob=file_select, + cb_progress=fn_cb_progress, + cb_filter=cb_filter) -class DownloadManager(BaseCommand): + if (details or file_select is not None) and len(_datasets) > 0: + _datasets[["n_parquet", "n_ciff", "n_groups_parquet", "n_groups_ciff", "n_to_sync"]] = 0 + for ix, _d in _datasets.iterrows(): + # Retrieve local files from the target directory and match them. + local_path = f"{self.target}/{_d['InternalID']}/" + _files = _dataset_files[_d['InternalID']] + _files['Local'] = _files['Path'].apply( + lambda x: '*' if not os.path.exists(os.path.join(local_path, x)) else '') + # Analyze file types and count groups. + _files['Directory'] = _files['Path'].str.rsplit('/', n=1).str[0].where(_files['Path'].str.contains('/'), + '') + df_parquet = _files[_files['Path'].str.contains(r'\.parquet$', regex=True)] + df_ciff = _files[_files['Path'].str.contains('ciff', regex=True)] + # Count the number of _files and groups. + _datasets.loc[ix, ["n_parquet", "n_ciff", "n_groups_parquet", "n_groups_ciff", "n_to_sync"]] = [ + len(df_parquet), len(df_ciff), df_parquet['Directory'].nunique(), df_ciff['Directory'].nunique(), + len(_files[_files['Local'] == "*"])] + # If required, collect file details for this dataset. + _dataset_files[_d['InternalID']] = _files if no_sync else _files[_files["Local"] == "*"] + + #if len(_datasets) == 0: + # # If no _datasets need updating, print the status and return. + # _ds = df_full["dataset"] if df_full['dataset'] is not None else _datasets + # self.console.print(f"Found the following {len(_ds)} datasets, which " + # f"also have local versions (content and metadata). Everything seems up-to-date") + # self.show_dataframe(_ds) + # return _ds, {} # todo: is that correct here? mixes up semantics. + # todo check that datasets and _dataset_files have the minimum needed columns and types and consistency + return _datasets, _dataset_files + +class LexisDownloadManager(BaseCommand): """ DownloadManager is a class that provides methods to download datasets and files from the Open Web Index. It uses the Lexis Rest API and does not rely on iRODS for file access. """ - def __init__(self, owi, slice="main", max_retries=500, sleep=1, **kwargs): + + def __init__(self, owi, max_retries=500, sleep=1, **kwargs): """ Initialize the DownloadManager with the specified parameters. owi (OWIProject): The OWIProject instance to use for downloading datasets and files. - slice (str): The slice of the dataset to download, note that the main slices are stored under main max_retries (int): The maximum number of retries to attempt when waiting for a file to be prepared. sleep (int): The time to sleep between retries in seconds. kwargs (dict): Additional keyword arguments to pass to the DownloadManager which will be resolved by the BaseCommand class @@ -276,22 +270,17 @@ super().__init__(owi, **kwargs) self.max_retries = max_retries self.sleep = sleep - self.slice = slice - - @property - def target(self): - return os.path.join(super().target, self.slice) def prepare_and_download_file(self, item, progress, prep_task): try: if "request" not in item or item["request"] is None: - item["request"] = self.owi.datasets.start_transfer(item["dataset"]["InternalID"], - item["dataset"]['Access'], - zone=item["dataset"]['Zone'], - path=item["file"]["Path"]) + item["request"] = self.owi.remote_data.start_transfer(item["dataset"]["InternalID"], + item["dataset"]['Access'], + zone=item["dataset"]['Zone'], + path=item["file"]["Path"]) for i in range(self.max_retries): - time.sleep(self.sleep+random.uniform(0,1)) - status = self.owi.datasets.get_transfer_status(item["request"]['request_id'], False) + time.sleep(self.sleep + random.uniform(0, 1)) + status = self.owi.remote_data.get_transfer_status(item["request"]['request_id'], False) item["request"] = status if status["task_state"] == "SUCCESS": progress.update(prep_task, advance=1, current_item=item["file"]["Path"]) @@ -300,23 +289,22 @@ raise Exception("Failed or error state during preparation") # Simulate download logic with a sleep - fn = self.owi.datasets.download_file_with_request(item["request"]["request_id"], - item["dataset"]['InternalID'], - item["dataset"]["Access"], - self.target, - item["file"]["Path"]) + fn = self.owi.remote_data.download_file_with_request(item["request"]["request_id"], + item["dataset"]['InternalID'], + item["dataset"]["Access"], + self.target, + item["file"]["Path"]) item["owilstatus"] = {"download": True, "preparation": True, "local_file": fn} return item except Exception as e: _req = item.get("request") item["owilstatus"] = {"download": False, - "preparation": _req is not None and _req.get("task_state")=="SUCCESS"} + "preparation": _req is not None and _req.get("task_state") == "SUCCESS"} return item def pull_file_based(self, specifier, file_select, no_sync, show_files): # Get files and datasets and user ok - _datasets, _files = self.get_datasets_and_files(specifier, file_select, no_sync, - details=True, return_files=True) + _datasets, _files = self.get_datasets_and_files(specifier, file_select, no_sync, details=True, use_irods=False) if show_files: self._ask_or_raise(_datasets, _files) else: @@ -342,18 +330,18 @@ TextColumn("{task.fields[current_item]}"), # Display the current item ) as progress: # Create progress tasks - _prep_task = progress.add_task("Files prepared", total=len(_downloads)+1, current_item="Setup") + _prep_task = progress.add_task("Files prepared", total=len(_downloads) + 1, current_item="Setup") _download_task = progress.add_task("Downloading files", total=len(_downloads), current_item="Setup") _failed_task = progress.add_task("Failed downloads", total=len(_downloads), current_item="None") # Prepare and start download results = [] - all_transfers = self.owi.datasets.get_transfer_status() + all_transfers = self.owi.remote_data.get_transfer_status() progress.update(_prep_task, advance=1, current_item="Past preparation obtained") for ix, t in all_transfers.iterrows(): if t["DatasetID"] in _ds_ids and t["Filename"] in _file_paths: for _d in _downloads: - if _d["dataset"]["InternalID"]==t["DatasetID"] and _d["file"]["Path"] == t["Filename"]: + if _d["dataset"]["InternalID"] == t["DatasetID"] and _d["file"]["Path"] == t["Filename"]: _d["request"] = t with ThreadPoolExecutor(max_workers=self.n_parallel) as executor: @@ -390,7 +378,7 @@ raise UserWarning("User Aborted Dataset Download") -class IRODSBasedAccess(BaseCommand): +class IRODSBasedCommands(BaseCommand): """ The class provides direct access to the IRODS datasets (if irods is available) and defines a set of commands using duckdb plus IRODS. @@ -398,55 +386,63 @@ def __init__(self, owi, **kwargs): super().__init__(owi, **kwargs) - if self.owi.datasets.irods is None: + if self.owi.remote_data.irods is None: raise IOError("IRODS connection not available. Direct Access is not possible") - def duckdb_shell(self, specifier, file_select): - # Get files and datasets and user ok + """ + Starts an interactive shell with duckdb and the irods filesystem registered. Path to OWI files are provided + based on the specifier and file_select. The datasets are listed and available as 'ds' in the shell. + + specifier (str): Criteria to filter datasets. + file_select (str): Pattern or filter to select specific files within datasets. + """ self.console.print(f"Fetching filenames for {specifier} selecting files for {file_select}") - _datasets = self.owi.datasets.list_irods_datasets() + _datasets, _files = self.get_datasets_and_files(specifier, file_select, no_sync=True, details=True, + use_irods=False) + import duckdb as d - d.register_filesystem(self.owi.datasets.filesystem) + d.register_filesystem(self.owi.remote_data.filesystem) from IPython import embed - for _d in _datasets: - _d["files"] =["irods://"+i for i in self.owi.datasets.filesystem.glob(os.path.join(_d["path"], "**", file_select))] - ds = _datasets - self.console.print("Starting interactive shell. Duckdb is available as 'd' and 'ds' contains " - "available datasets and files. Example would be:\n "+ - """d.sql(f"Select url FROM read_parquet('{ds[-1]['files'][0]}') LIMIT 5").df()""" + d, f = _datasets, _files + self.console.print("Starting interactive shell. Duckdb is available as 'd' and 'f' contains " + "available datasets and files. Example would be:\n " + + """d.sql(f"Select url FROM read_parquet('{f.values()[0].iloc(0, "Path")}') LIMIT 5").df()""" ) c = self.console embed() - #_file = "irods:///IT4ILexisV2/public/proj862c5962623246664c1fda27b7afb108/5aa5c3dc-dc56-11ee-ad7e-0242c0a87004/metadata_2.parquet" - #parquet = set(_files[_files["Filename"].str.endswith(".parquet")]["Filename"]) - #from = "read_parquet(["+",".join([f"'{f}'" for f in _parquet])+"])" - #select = "Select * " - #self.show_dataframe(duckdb.sql(_select + " FROM " +_from + " LIMIT 5 " ).df()) - #self.show_dataframe(db.sql(f"SELECT url FROM read_parquet({_file}) LIMIT 5 ").df()) - + # _file = "irods:///IT4ILexisV2/public/proj862c5962623246664c1fda27b7afb108/5aa5c3dc-dc56-11ee-ad7e-0242c0a87004/metadata_2.parquet" + # parquet = set(_files[_files["Filename"].str.endswith(".parquet")]["Filename"]) + # from = "read_parquet(["+",".join([f"'{f}'" for f in _parquet])+"])" + # select = "Select * " + # self.show_dataframe(duckdb.sql(_select + " FROM " +_from + " LIMIT 5 " ).df()) + # self.show_dataframe(db.sql(f"SELECT url FROM read_parquet({_file}) LIMIT 5 ").df()) def sql(self, specifier, file_select, script, verbose=True, explain=False, sink=None, **kwargs): """ Executes an SQL Script via duckdb on the irods data defined by the specifier and file_select. + sql intends to provide a generic interface to the duckdb sql engine, which is connected to the irods filesystem + and to re-use the results in an easy, yet flexible manner. The SQL script can use {owi_remote_files} as placeholder, which will be substituted for an list of files formated as python array ['fiel1, 'file2', ...]. Files are prefixed with the irods filesystem, which is registered with duckdb, i.e. irods:// access the Open Web Index irods storage. - SQL Scripts can contain param + SQL Scripts can contain paramemeters, which are assumed to be in the kwargs dictionary. + + The script output can be re-used in a sink, which then receives the return of the duckdb.sql command. + If sink is None, the output is displayed as Dataframe in the console if duckdb.sql returns non None (i.e. calling duckdb.sql().df())) """ # Get files and datasets and user ok self.console.print(f"Fetching filenames for {specifier} selecting files for {file_select}") - _datasets = self.owi.datasets.list_irods_datasets() + _datasets, _files = self.get_datasets_and_files(specifier, os.path.join("**", file_select), + no_sync=True, details=True, use_irods=False) # self.console.print(f"Found {len(_datasets)} datasets. Starting DuckDB.") if verbose: - self.show_dataframe(pd.DataFrame([d["metadata"] for d in _datasets])) + self.show_dataframe(_datasets) import duckdb - duckdb.register_filesystem(self.owi.datasets.filesystem) - for _d in _datasets: - _d["files"] =["irods://"+i for i in self.owi.datasets.filesystem.glob(os.path.join(_d["path"], "**", file_select))] - _files = [f for d in _datasets for f in d["files"]] + duckdb.register_filesystem(self.owi.remote_data.filesystem) + _files = [x['Path'] for f in _files.values() for i, x in f.iterrows()] if len(_files) == 0: self.console.print("[warning]No files found[\warning]. This might impact your script.") _kwargs = SafeTemplateDict(**kwargs) @@ -458,7 +454,101 @@ _explained = _sql.explain() self.console.print("Query Plan:", _explained) if sink is None: - if _sql and hasattr(_sql, "df"): + if _sql is not None and _sql and hasattr(_sql, "df"): self.show_dataframe(_sql.df()) - else: raise NotImplementedError("Other sinks are not implemented yet") + else: + raise NotImplementedError("Other sinks are not implemented yet") + def fillet(self, specifier, file_select, fields, where=";", partitioned_by="", + no_sync=False, overwrite_ignore=True, verbose=True): + """ + the fillet command creates vertical (fields/columns) and horizontal (daily partitions, files, row_selectors) + slices of the Open Web Index and stores it under the slice name configured in the Base Command (SLICE param). + """ + # todos: list_irods_dataset must return similar datastructure than list_datasets, write log file. Sync datasets, i.e. only update non existing datasets + if self.slice is None or self.slice == "main": + self.console.print( + "[error]Slice name to store the fillet under must be specified. Main or none is not allowed [/error]") + return + self.console.print( + f"Fetching datasets and files to fillet using specifier {specifier} and file filter {file_select}") + file_select = file_select if file_select is not None else "**/*" + _datasets, _files = self.get_datasets_and_files(specifier, file_select, + no_sync=no_sync, details=True, use_irods=True) + self.console.print(f"Found {len(_datasets)} datasets.") + if (len(_datasets) == 0): + return + if verbose: + self.show_dataframe(_datasets) + import duckdb + duckdb.register_filesystem(self.owi.remote_data.filesystem) + _sql_template = """ + BEGIN; + DROP TABLE IF EXISTS fillet; + CREATE TABLE fillet AS SELECT {fields} FROM read_parquet({owi_remote_files}) {where} + COPY fillet to '{store}' (FORMAT 'parquet', + OVERWRITE_OR_IGNORE {overwrite_ignore} {partitioned_by}); + COMMIT; + """ + if partitioned_by is not None and partitioned_by != "": + partitioned_by = f", PARTITIONED BY ({partitioned_by})" + if where is not None and where != "" and where != ";": + where = f"WHERE {where};" + else: + where = ";" + + with Progress( + TextColumn("[progress.description]{task.description}"), + BarColumn(), + "[progress.percentage]{task.percentage:>3.0f}%", + TimeRemainingColumn(), + TextColumn("{task.fields[current_item]}"), # Display the current item + ) as _progress: + # Create _progress tasks + _total_files = sum([len(f) for f in _files.values() if f is not None]) + _ds_task = _progress.add_task("Ducked Datasets", total=len(_datasets), current_item="Setup") + _files_task = _progress.add_task("Ducked Partitions/Files", total=_total_files, current_item="Setup") + _failed_files_task = _progress.add_task("Ducks crashed (Files)", total=_total_files, current_item="None") + for ix, _d in _datasets.iterrows(): + # group files based on the same leaf directory they occur in + _dataset_id = _d['InternalID'] + _grouped_files = defaultdict(list) + _df_file = _files.get(_dataset_id) + if _df_file is not None: + try: + for p in list(_df_file["Path"]): + _grouped_files["/".join(p.split(_dataset_id + "/")[-1].split("/")[:-1])].append(p) + for _g, _gfiles in _grouped_files.items(): + try: + _kwargs = SafeTemplateDict() + _kwargs["owi_remote_files"] = str(_gfiles) + _kwargs["fields"] = fields + _kwargs["where"] = where + _kwargs["partitioned_by"] = partitioned_by + _kwargs["overwrite_ignore"] = str(overwrite_ignore).lower() + _kwargs["store"] = os.path.join(f"{self.target}/{_dataset_id}/", _g) + if not os.path.exists(_kwargs["store"]): + os.makedirs(_kwargs["store"]) + _kwargs["store"] = os.path.join(_kwargs["store"], "metadata.parquet") + _sql_script = _sql_template.format_map(_kwargs) + _sql = duckdb.sql(_sql_script) + _progress.update(_files_task, advance=len(_gfiles), current_item=_g) + except Exception as e: + traceback.print_exc() + _progress.update(_failed_files_task, advance=len(_gfiles), current_item=_g) + self.owi.update_log({"status": {"success": False, "reason": str(e)}, + "dataset": _d.to_dict(), "files": _gfiles}, slice=self.slice) + except Exception as e: + traceback.print_exc() + self.owi.update_log({"status": {"success": False, "reason": str(e)}, + "dataset": _d.to_dict(), "files": _files.get(_dataset_id, [])}, + slice=self.slice) + _progress.update(_failed_files_task, advance=len(_d["files"]), current_item=p) + _progress.update(_ds_task, advance=1, current_item=_d["Title"]) + self.owi.update_log({"status": {"success": True, "reason": "Succesfully filleted"}, + "parameters": {"fields": f"{fields}", "where": f"{where}", + "partitioned_by": f"{partitioned_by}", + "overwrite_ignore": f"{overwrite_ignore}"}, + "dataset": _d.to_dict(), + "files": list(_df_file["Path"]) if _df_file is not None else []}, + slice=self.slice) diff --git a/owilix/core/datasets.py b/owilix/core/datasets.py --- a/owilix/core/datasets.py +++ b/owilix/core/datasets.py @@ -1,3 +1,4 @@ +import fnmatch import logging import tarfile import threading @@ -15,7 +16,7 @@ from owilix.core.lexis import OWILexisDataset -def filter_by_date_range(df, start_date=None, end_date=None): +def _filter_by_date_range(df, start_date=None, end_date=None): if start_date and end_date: mask = (df['Date'] >= start_date) & (df['Date'] <= end_date) elif start_date: @@ -26,42 +27,7 @@ return df # No filtering, return all return df[mask] - -class OWSData: - - def __init__(self, project_name, session=None, lexis_id="proj862c5962623246664c1fda27b7afb108"): - self.session =session if session is not None else LexisSession(in_cli=True) - self.dscli = OWILexisDataset(self.session) - self.lexis_id = lexis_id - self.__project_name = project_name - - def _to_datetime(self, date: str|datetime): - return datetime.strptime(date, "%Y-%m-%d") if isinstance(date,str) else date - - @property - def irods(self): - return self.dscli.irods - - @property - def filesystem(self): - return self.irods.filesystem - - def list_irods_datasets(self): - if self.irods is None: return [] - zone = self.irods.zone - datasets = self.irods.filesystem.ls(os.path.join("/", zone,"public",self.lexis_id)) # todo: make configurable - returns = [] - for d in datasets: - do = self.irods.session.collections.get(d) - returns.append({"metadata":{k:v for k,v in do.metadata.items()},"path":d}) - return returns - - def ls(self, day: str|datetime|None=None, duration:int = 0, data_center=None, query=None): - df = self.dscli.get_all_datasets(content_as_pandas=True) - df= df[df["Project"] == self.__project_name] - pattern = r'@(.+?)[-\s](\d{4}/\d{1,2}/\d{1,2})$' - df[['DataCenter', 'Date']] = df['Title'].str.extract(pattern) - df['Date'] = pd.to_datetime(df["Date"]) +def _apply_dataset_filter(df, day, duration, data_center, query): # data center filter if data_center is not None and data_center != "all" and data_center != "*": df = df[df["DataCenter"] == data_center] @@ -75,11 +41,136 @@ # time filter if day=="latest": day=df["Date"].max() - else: - day = self._to_datetime(day) + elif day is not None: + day = _to_datetime(day) if duration is None: duration=0 startdate = None if day is None else day - timedelta(days=duration) - return filter_by_date_range(df, pd.to_datetime(startdate), pd.to_datetime(day)) + + return _filter_by_date_range(df, pd.to_datetime(startdate), pd.to_datetime(day)) + +def _to_datetime(date: str|datetime): + """Internal: Get string format for the date used in all of the data (YYYY-MM-DD)""" + return datetime.strptime(date, "%Y-%m-%d") if isinstance(date,str) else date + + +class OWIRemoteData: + """ + Class managing remote data of the Open Web Index via either Lexis HTTP API or the LEXIS IRODS API. + The class access local data via the OWILocal class provided through the OWIlixProject class. The class relies on + OWILexisDataset for the actual data access, which wrapps the PY4Lexis Datasets, + """ + def __init__(self, project, lexis_id="proj862c5962623246664c1fda27b7afb108"): + self.session =project.session if project.session is not None else LexisSession(in_cli=True) + self.dscli = OWILexisDataset(self.session) + self.lexis_id = lexis_id + self._project_name = project.name + self._project = project + + @property + def irods(self): + """Access to the irods client for the project (class IRODSOWI). Allows maybe faster and more direct operations, including accessing irods metadata""" + return self.dscli.irods + + @property + def filesystem(self): + """get irods as fsspec filesystem""" + return self.irods.filesystem + + def ls_irods(self, day: str | datetime | None=None, duration:int = 0, data_center=None, query=None, + files_glob=None, cb_progress = None, cb_filter=None): + """ + List the datasets in the irods zone. If files_glob is not None, the files in the dataset are also listed. + + Args: + day: the day to list the datasets for. If "latest" the latest day is used. + duration: the duration to go back from the day in the past (0 based, i.e. 0 means only the day) + data_center: the data center to filter the datasets for + query: a query to filter the datasets for. Query is a dictionary with the column names as keys and the values as values + files_glob: a glob pattern to filter the files in the dataset + cb_progress: a callback function to report the progress of the listing. + The callback function should accept four arguments: + the current dataset and the maximum number of datasets, + the internal id and the title of the dataset. + + Returns: + tuple (DataFrame, DataFrame): a DataFrame with the metadata of the datasets and a second Datframe with the files per dataset (key=InternalID) + """ + if self.irods is None: return [] + _path = os.path.join("/", self.irods.zone,"public",self.lexis_id) + _datasets = self.irods.filesystem.ls(_path) # todo: make configurable + _metadata = [] + for _i, _d in enumerate(_datasets): + do = self.irods.session.collections.get(_d) + _metadata.append({"metadata":{k[0].upper()+k[1:]:v for k,v in do.metadata.items()},"path":_d}) + _metadata[-1]["metadata"]["InternalID"] = _d.split("/")[-1] + _metadata[-1]["metadata"]["Path"] = _d + + + _df = pd.DataFrame([d["metadata"] for d in _metadata]) + # todo needs to be changed when the data is finally there + _df["Date"] = pd.to_datetime(day) + _df["DataCenter"] = "it4i" + _df["Access"]="public" + _df = _apply_dataset_filter(_df, day, duration, data_center, query) + # callback on filter (e.g. match) + if cb_filter is not None: + _df = cb_filter(_df) + # now get files + _files = {} + if files_glob is not None: + for _i, _d in _df.iterrows(): + if cb_progress is not None: + cb_progress(_i, len(_datasets), _d["InternalID"], _d["Title"]) + _f = ["irods://"+i for i in self.filesystem.glob(os.path.join(_d["Path"], files_glob))] + _files[_d["InternalID"]] = pd.DataFrame([{"Dir/File-name": f, "Path": f, "Type": "file", + "Size": 0, "CreateTime": 0} for f in _f], + columns=["Dir/File-name", "Path", "Type", "Size", "CreateTime"]) + # todo: ensure the returned dataframe do have the same minimum colums as in ls_http + return _df, _files + + def ls_http(self, day: str|datetime|None=None, duration:int = 0, data_center=None, query=None, + files_glob=None, cb_progress = None, cb_filter = None): + """ + List the datasets using the Lexis HTTP API. If files_glob is not None, the files in the dataset are also listed. + + Args: + day: the day to list the datasets for. If "latest" the latest day is used. + duration: the duration to go back from the day in the past (0 based, i.e. 0 means only the day) + data_center: the data center to filter the datasets for + query: a query to filter the datasets for. Query is a dictionary with the column names as keys and the values as values + files_glob: a glob pattern to filter the files in the dataset + cb_progress: a callback function to report the progress of the listing. + The callback function should accept four arguments: + the current dataset and the maximum number of datasets, + the internal id and the title of the dataset. + + Returns: + tuple (DataFrame, DataFrame): a DataFrame with the metadata of the datasets and a second Datframe with the files per dataset (key=InternalID) + """ + _df = self.dscli.get_all_datasets(content_as_pandas=True) + # convert metadata to schema + _df["Path"] = _df.apply( lambda row: "/" + row["Zone"] + "/" + + ("public" if row["Access"] == "public" else "project") + "/" + + self.lexis_id + "/" + row["InternalID"], + axis=1) + _df= _df[_df["Project"] == self._project_name] + pattern = r'@(.+?)[-\s](\d{4}/\d{1,2}/\d{1,2})$' + _df[['DataCenter', 'Date']] = _df['Title'].str.extract(pattern) + _df['Date'] = pd.to_datetime(_df["Date"]) + _df = _apply_dataset_filter(_df, day, duration, data_center, query) + # callback on filter (e.g. match) + if cb_filter is not None: + _df = cb_filter(_df) + _files = {} + if files_glob is not None: + _cnt = 0 + for i, row in _df.iterrows(): + if cb_progress is not None: + cb_progress(_cnt, len(_df), row["InternalID"], row["Title"]) + _cnt += 1 + _f = self.files(row["InternalID"], "public", zone=row["Zone"], path_filter=files_glob) + _files[row["InternalID"]] =_f[_f["Type"]=="file"] + return _df, _files @@ -118,14 +209,11 @@ that fetches the dataset's content based on given parameters and returns it as a pandas DataFrame. """ - df = self.dscli.get_content_of_dataset(internalId, access, self.__project_name, zone=zone, + df = self.dscli.get_content_of_dataset(internalId, access, self._project_name, zone=zone, content_as_pandas=True) if path_filter is not None and path_filter != "" and path_filter != ".*": - df = df[df['Path'].str.contains(path_filter, regex=True)] + df =df[df['Path'].apply(lambda x: fnmatch.fnmatch(x, path_filter))] return df - - def download(self, datasets:DataFrame, language: List[str]|None = None): - pass def get_transfer_status(self, request_id=None, as_pandas=True, exclude_error=True): returns = self.dscli.get_dataset_status(request_id, as_pandas) @@ -138,15 +226,15 @@ def start_transfer(self, internalId:str, access:str, zone=None, path=None): return self.dscli.start_transfer(internalId, access, zone=zone, path=path, - project=self.__project_name) + project=self._project_name) - def download_with_request(self,request_id, internalId:str, access:str, dest="./", meta=dict()): - fn = os.path.join(dest, internalId) - self.dscli.download_dataset_with_request_id(request_id, internalId, access, destination_filepath=fn + ".tar.gz") - with open(fn + ".json", "w") as file: - json.dump(meta, file) - return fn + ".tar.gz", fn + ".json" - + #def download_with_request(self,request_id, internalId:str, access:str, dest="./", meta=dict()): + # fn = os.path.join(dest, internalId) + # self.dscli.download_dataset_with_request_id(request_id, internalId, access, destination_filepath=fn + ".tar.gz") + # with open(fn + ".json", "w") as file: + # json.dump(meta, file) + # return fn + ".tar.gz", fn + ".json" +# def download_file_with_request(self,request_id, internalId:str, access:str, dest, file_name): fn = os.path.join(dest, internalId, file_name) self.dscli.download_dataset_with_request_id(request_id, internalId, access, destination_filepath=fn) @@ -160,7 +248,7 @@ # irods.test() # just for testing how direct irods access could look like self.dscli.download_dataset(internalId, access, zone=zone, destination_filepath=fn+".tar.gz", - project=self.__project_name, + project=self._project_name, status_callback=status_callback) with open(fn+".json","w") as file: json.dump(meta,file) @@ -170,8 +258,8 @@ def match(self, remote, local_dest, file_selector=None): """ Adds a 'FileStatus' column to the DataFrame indicating the presence of files. - '*' if the .tar.gz file is missing. - 'M' if the .json file is missing but the .tar.gz file exists. + '*' if the directory is missing. + 'M' if the .json file is missing but the directory with dataset-id exists. 'D' if the files according to the provided file_selector do not match. Will not be checked if file_selector is None Args: @@ -180,7 +268,8 @@ """ # Verify local_dest is a valid directory if not os.path.isdir(local_dest): - raise ValueError(f"{local_dest} is not a valid directory") + remote['FileStatus'] = '*' + return remote # Function to check file existence and determine status def check_files(internal_id): @@ -203,38 +292,6 @@ # sync files here for all * or ''. pass return remote - - - def download_files(self, files: DataFrame): - - pass - - def query_deprecated(self, command): - # Split the command into the main parts - parts = command.split(':') - df = self.ls() - if len(parts) == 2: - data_center, date_or_latest = parts - if data_center == "all": - # All data centers, specific date - filtered_df = df[df['Date'] == date_or_latest] - elif date_or_latest == "latest": - # Latest date for a specific data center - filtered_df = df[df['DataCenter'] == data_center] - if not filtered_df.empty: - latest_date = filtered_df['Date'].max() - filtered_df = filtered_df[filtered_df['Date'] == latest_date] - else: - # Specific data center, specific date - filtered_df = df[(df['DataCenter'] == data_center) & (df['Date'] == date_or_latest)] - else: - # Assume the command is in the form 'data_center/resource_type' - data_center, resource_type = command.split('/') - filtered_df = df[(df['DataCenter'] == data_center) & (df['ResourceType'] == resource_type)] - if not filtered_df.empty: - latest_date = filtered_df['Date'].max() - filtered_df = filtered_df[filtered_df['Date'] == latest_date] - return filtered_df def extract(self, target, in_parallel=True): """ diff --git a/owilix/core/lexis.py b/owilix/core/lexis.py --- a/owilix/core/lexis.py +++ b/owilix/core/lexis.py @@ -16,6 +16,9 @@ class OWILexisDataset(Datasets): + """ + Wrapper class around Py4LEXIS Dataset to get access to private methods of Py4LEXIS Dataset class + """ def __init__(self, session: LexisSession, print_content: bool = False, diff --git a/owilix/core/local.py b/owilix/core/local.py --- a/owilix/core/local.py +++ b/owilix/core/local.py @@ -8,6 +8,8 @@ _req_keys = {"InternalID", "Title", "Access", "Zone", "Date", "DataCenter"} + + class DirectoryInspector: def __init__(self, path, @@ -181,4 +183,5 @@ lists the locally available datasets """ _inspector = DirectoryInspector(self.project.owi_path) - return pd.DataFrame(_inspector.list(False)) \ No newline at end of file + return pd.DataFrame(_inspector.list(False)) + diff --git a/owilix/core/project.py b/owilix/core/project.py --- a/owilix/core/project.py +++ b/owilix/core/project.py @@ -38,7 +38,7 @@ logpath (str): The path where log files are stored. logfile (str): The file path for logging events in JSON format. _session (LexisSession or None): Private attribute for managing a LexisNexis session. - _datasets (OWIRemoteData or None): Private attribute to hold an OWIRemoteData instance. + _remote_data (OWIRemoteData or None): Private attribute to hold an OWIRemoteData instance. _local (OWILocal or None): Private attribute to hold an OWILocal instance. Properties: @@ -101,7 +101,7 @@ if not os.path.exists(logpath): os.makedirs(logpath) self.logfile = os.path.join(logpath,"events.json") self._session = None - self._datasets = None + self._remote_data = None self._local = None @@ -125,9 +125,9 @@ Returns: OWIRemoteData: Remote dataset manager instance. """ - if self._datasets is None: - self._datasets = OWIRemoteData(self) - return self._datasets + if self._remote_data is None: + self._remote_data = OWIRemoteData(self) + return self._remote_data @property def session(self): @@ -156,7 +156,7 @@ pd.DataFrame: Summary DataFrame with statistics on the queried datasets. """ # todo: extend by statistics for files - df = self.remote_data.ls_http(day, duration, data_center, query) + df, _files = self.remote_data.ls_http(day, duration, data_center, query) # Initialize an empty DataFrame for the summary summary_df = pd.DataFrame({'Kind':["Datasets"], "Key":["Total"], "Value":["Number Datasets"], "Count":[len(df)]}) if df is None or len(df)==0: return summary_df diff --git a/docs/source/_static/owili-image-v1.prompt.txt b/docs/source/_static/owili-image-v1.prompt.txt new file mode 100644 --- /dev/null +++ b/docs/source/_static/owili-image-v1.prompt.txt @@ -0,0 +1,18 @@ +Dalle 4 + + +Draw a figure similar to Obelix (From Asterix and Obelix) for a python software library called owilix. the library is about slicing and dicing the web graph, which should be shown metaphorically in the figure by Obelix, eating a digital cooked boar from a pile of digital boars. The skin of the boars are a web-graph tatoo. + +User +Draw a figure similar to Obelix (From Asterix and Obelix) for an intro page of a python software library called owilix. the owilix library is about slicing and dicing the web using duckdb. Obelix usually wears a small gallian helmet, has red hairs with braids, is obese wearing blue-white vertically stripped trousers and naked upper body. +the figure should show: +- Obelix sitting on a bench and eating a roasted boar which has a skin showing matrix like digital symbols +- in front of him is campfire with blue stones in the form of stars resembling the European flag with two more wild boars being roasted. +- on the other side of the campfire, there is a pile of roasted boars again with a digital skin showing a digital graph/ network resembling the web graph +- above Obelix there are ducks flying bringing him a magic drink, where the form has the bottle of a db symbol +Obeliks should be a little bit more muscular, the European flag should be covering the floor with the campfire being in the middle and there should be two wild boars being roasted on a lancet above the campfire + + +User +Create an image of a muscular, red-haired man with braids and a small helmet, reminiscent of a classic comic Gaul. He wears blue and white striped trousers and is seated on a wooden bench, eating a roasted boar with a skin containing a web-graph tattoo. He is eating with bare hands, holding the boar in both hands and happy to dig in. In front of him, a campfire with blue star-shaped stones mimics the European flag, with two boars roasting above it. Opposite the fire, a pile of boars showcases digital network graph patterns. Above, very close to the head of Obelix, ducks carry magical drink bottles shaped like database symbols. The ducks head towards Obleix. The scene is set on a European flag-covered ground, blending mystical and technological themes. As usual, Obelix is extremley pleased with eating the roasted board with bare hands, digging in fully. +take the right image, but make it more comic style. Obelix should have no clothing above the belly, while the belly can be a bit larger (still remaining a muscular upper body). There is a pile of boars prepared for roasting and one boar is roasted over the campfier on a lancet. Add the ducks flying on top bringing a bottle shaped like a database. \ No newline at end of file diff --git a/docs/source/_static/owili-image-v1.webp b/docs/source/_static/owili-image-v1.webp new file mode 100644 --- /dev/null +++ b/docs/source/_static/owili-image-v1.webp diff --git a/docs/source/_static/owili-image-v2.webp b/docs/source/_static/owili-image-v2.webp new file mode 100644 --- /dev/null +++ b/docs/source/_static/owili-image-v2.webp