core.network package


core.network.auth

core.network.auth.get_auth(name: str) dict[source]

Returns a dictionary with credentials, using .netrc

name is the identifier (= machine in .netrc). This allows for several accounts on a single machine. The url is returned as account

core.network.auth.get_auth_dhus(name)[source]

core.network.download

core.network.download.download_nextcloud(product_name: str, output_dir: Path | str, nextcloud_dir: Path | str = '', sharelink: str = 'https://docs.hygeos.com/s/Fy2bYLpaxGncgPM/', wget_opts='', check_function=None, verbose: bool = True, if_exists: Literal['skip', 'overwrite', 'backup', 'error'] = 'skip', **kwargs)[source]

Function for downloading data from Nextcloud contained in the data/eoread directory

Parameters:
  • product_name (str) – Name of the product with the extension

  • output_dir (Path | str) – Directory where to store downloaded data

  • nextcloud_dir (Path | str, optional) – Sub Nextcloud repository in which the product are stored. Defaults to ‘’.

  • sharelink (str, optional) – Nextcloud public link. By defaults, it is public link to eoread repository.

Returns:

Output path of the downloaded data

Return type:

Path

core.network.download.download_url(url: str, dirname: Path, wget_opts='', check_function=None, verbose=True, if_exists: Literal['skip', 'overwrite', 'backup', 'error'] = 'skip', **kwargs) Path[source]

Download url to dirname with wget

Options wget_opts are added to wget Uses a filegen wrapper Other kwargs are passed to filegen (lock_timeout, tmpdir, if_exists)

Returns the path to the downloaded file

core.network.ftp

core.network.ftp.ftp_create_dir(ftp: FTP, path_server: Path | str)[source]
core.network.ftp.ftp_download(ftp: FTP, file_local: Path, dir_server: str | Path, verbose=True)[source]

Downloads file_local on ftp, from server directory dir_server

The file name on the server is determined by file_local.name

Example

ftp = FTP(**get_auth_ftp(‘my_server’)) local_file = Path(‘./data.txt’) ftp_download(ftp, local_file, ‘/remote/folder’) ftp.quit()

Refs:

https://stackoverflow.com/questions/19692739/ https://stackoverflow.com/questions/73534659/

core.network.ftp.ftp_file_exists(ftp: FTP, path_server: Path | str) bool[source]
core.network.ftp.ftp_list(ftp: FTP, dir_server: str, pattern: str = '*')[source]

Returns the list of fles matching pattern on dir_server

core.network.ftp.ftp_upload(ftp: FTP, file_local: Path, dir_server: str, if_exists='skip', blocksize=8192, verbose=True)[source]

FTP upload function

  • Use temporary files

  • Create remote directories

  • if_exists:

    ‘skip’: skip the existing file ‘error’: raise an error on existing file ‘overwrite’: overwrite existing file

core.network.ftp.get_auth_ftp(name) Dict[source]

get netrc credentials for use with pyfilesystem’s FTPFS or ftplib’s FTP

Ex: FTP(**get_auth_ftp(<name>))

core.network.ftp.get_url_ftpfs(name)[source]

get netrc credentials for use with pyfilesystem’s fs.open_fs

Ex: fs.open_fs(get_url_ftpfs(<name>))