bec_lib.file_utils.FileWriter#

class FileWriter(*args, service_config: dict = None, connector: RedisConnector = None, scan_bundle: int = 1000, leading_zeros: int = 5, **kwargs)[source]#

Bases: object

FileWriter for creating file paths and directories for services and devices.

Initialize the file writer mixin class. If no RedisConnector is provided, the class will not be able to communicate with the REDIS server.

In this case, it will fall back to a default base path in the current directory.

Parameters:

service_config (dict) – File writer service config. Must at least contain base_path.
connector (Redisconnector, optional) – Connector class to use. Defaults to RedisConnector.
scan_bundle (int, optional) – Scan bundle size. Defaults to 1000.
leading_zeros (int, optional) – Number of leading zeros. Defaults to 5.

Methods

`compile_full_filename`	Compile a full filename for a given scan number and suffix.
`configure_file_writer`	Configure the file writer mixin class in case service_config is provided
`get_scan_directory`	Get the scan directory for a given scan number and scan bundle.
`get_scan_msg`	Get the scan message for the next scan

Attributes

`leading_zeros`	Get the number of leading zeros.
`scan_bundle`	Get the scan bundle size.

compile_full_filename(suffix: str, create_dir: bool = True) → str[source]#

Compile a full filename for a given scan number and suffix.

This method should only be called after a scan has been opened, i.e. preferable in stage and not during __init__.

The method will use the last scan message received in REDIS, and will return an empty string if no scan message is available.

Parameters:

suffix (str) – Filename suffix including extension. We allow alphanumeric, ascii characters - and _.
file_type (str) – Optional, will default to h5.
create_dir (bool, optional) – Create the directory if it does not exist. Defaults to True.

Returns:

Full filename

Return type:

str

configure_file_writer()[source]#: Configure the file writer mixin class in case service_config is provided

get_scan_directory(scan_number: int, scan_bundle: int, leading_zeros: int | None = None, user_suffix: str | None = None) → str[source]#

Get the scan directory for a given scan number and scan bundle.

Parameters:

scan_bundle (int) – Scan bundle size
scan_number (int) – Scan number
leading_zeros (int, optional) – Number of leading zeros. Defaults to None.
user_suffix (str, optional) – User defined suffix. Defaults to None.

Returns:

Scan directory

Return type:

str

Examples

>>> get_scan_directory(1234, 1000, 5)
'S01000-01999/S01234'
>>> get_scan_directory(1234, 1000, 5, 'sampleA')
'S01000-01999/S01234_sampleA'

get_scan_msg()[source]#: Get the scan message for the next scan

property leading_zeros: int#: Get the number of leading zeros.

property scan_bundle: int#: Get the scan bundle size.