Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
Bliss
bliss
Commits
a9924641
Commit
a9924641
authored
Jun 03, 2021
by
Wout De Nolf
Browse files
writer: get technique from the icat metadata (no longer in SCAN_SAVING)
parent
a6900e15
Changes
5
Hide whitespace changes
Inline
Side-by-side
bliss/scanning/scan.py
View file @
a9924641
...
...
@@ -664,6 +664,10 @@ class Scan:
scan_saving
=
current_session
.
scan_saving
self
.
__scan_saving
=
scan_saving
.
clone
()
@
property
def
scan_saving
(
self
):
return
self
.
__scan_saving
def
_init_scan_display
(
self
):
self
.
__scan_display
=
current_session
.
scan_display
.
clone
()
...
...
nexus_writer_service/subscribers/scan_writer_publish.py
View file @
a9924641
...
...
@@ -31,6 +31,7 @@ from bliss.controllers.lima.roi import (
)
from
bliss.common.counter
import
SamplingCounter
from
..utils
import
config_utils
from
..utils
import
session_utils
from
..utils
import
scan_utils
...
...
@@ -58,7 +59,10 @@ def fill_instrument_name(scan):
:param bliss.scanning.scan.Scan scan:
"""
logger
.
debug
(
"fill instrument name"
)
return
{
"instrument"
:
config_utils
.
instrument
()}
default
=
session_utils
.
scan_saving_get
(
"beamline"
,
scan_saving
=
scan
.
scan_saving
,
default
=
""
)
return
{
"instrument"
:
config_utils
.
instrument
(
default
=
default
)}
def
fill_technique_info
(
scan
):
...
...
@@ -66,7 +70,7 @@ def fill_technique_info(scan):
:param bliss.scanning.scan.Scan scan:
"""
logger
.
debug
(
"fill technique info"
)
return
{
"technique"
:
current_technique_definition
()}
return
{
"technique"
:
current_technique_definition
(
scan
.
scan_saving
)}
def
fill_masterfiles
(
scan
):
...
...
@@ -205,7 +209,11 @@ def writer_config():
:returns dict:
"""
return
config_utils
.
static_root_find
(
"nexus_definitions"
,
default
=
{})
cfgnode
=
config_utils
.
static_root_find
(
"nexus_definitions"
)
if
cfgnode
is
None
:
return
dict
()
else
:
return
cfgnode
.
to_dict
()
def
writer_config_get
(
name
,
default
=
None
):
...
...
@@ -244,13 +252,15 @@ def default_technique():
return
technique_info_get
(
"default"
,
"undefined"
)
def
current_technique
():
def
current_technique
(
scan_saving
=
None
):
"""
Active technique from the session's scan saving object
:returns str:
"""
return
config_utils
.
scan_saving_get
(
"technique"
,
default_technique
())
return
session_utils
.
dataset_get
(
"definition"
,
default_technique
(),
scan_saving
=
scan_saving
)
def
techniques
():
...
...
@@ -280,13 +290,13 @@ def technique_definition(technique):
"plots"
:
plots
,
"plotselect"
:
""
,
}
technique_
info
=
technique_info_get
(
"techniques"
,
{}).
get
(
technique
,
{})
if
technique_
info
is
None
:
technique_
info
=
{}
info
=
technique_info_get
(
"techniques"
,
{}).
get
(
technique
,
{})
if
info
is
None
:
info
=
{}
# Get the application definitions selected for this technique
applicationdict
=
technique_info_get
(
"applications"
,
{})
for
name
in
technique_
info
.
get
(
"applications"
,
[]):
for
name
in
info
.
get
(
"applications"
,
[]):
definition
=
applicationdict
.
get
(
name
,
{})
# for example {'xrf':{'I0': 'iodet',
# 'It': 'idet',
...
...
@@ -298,7 +308,7 @@ def technique_definition(technique):
# Get the plots selected for this technique
plotdict
=
technique_info_get
(
"plots"
,
{})
plotselect
=
""
# this first one is the default
for
name
in
technique_
info
.
get
(
"plots"
,
[]):
for
name
in
info
.
get
(
"plots"
,
[]):
plotdefinition
=
plotdict
.
get
(
name
,
{})
# for examples:
# {'personal_name': 'counters', 'items': ['iodet', 'xmap1:deadtime_det2', ...]}
...
...
@@ -318,10 +328,10 @@ def technique_definition(technique):
return
ret
def
current_technique_definition
():
def
current_technique_definition
(
scan_saving
=
None
):
"""
Current technique definition from the technique info
:returns dict(dict): technique:definition (str:dict)
"""
return
technique_definition
(
current_technique
())
return
technique_definition
(
current_technique
(
scan_saving
=
scan_saving
))
nexus_writer_service/utils/config_utils.py
View file @
a9924641
...
...
@@ -10,57 +10,41 @@
# Distributed under the GNU LGPLv3. See LICENSE for more info.
"""
Bliss session
configuration utilities
Static Bliss
configuration utilities
"""
import
os
import
re
from
functools
import
wraps
from
bliss
import
current_session
from
bliss.config
import
static
from
bliss.scanning.scan_saving
import
ScanSaving
,
with_eval_dict
def
static_config
():
"""
Get static session configuration
:returns bliss.config.static.Config:
"""
return
static
.
get_config
()
def
static_root
():
"""
Get static session configuration
:returns bliss.config.static.ConfigNode:
"""
return
static_config
().
root
def
static_root_get
(
name
,
default
=
None
):
def
static_root
(
root
=
None
):
"""
Get attribute from the static session configuration
:returns str:
:returns ConfigNode:
"""
return
static_root
().
get
(
name
,
default
)
if
root
is
None
:
return
static_config
().
root
else
:
return
root
def
static_root_find
(
name
,
default
=
None
,
paren
t
=
None
):
def
static_root_find
(
name
,
roo
t
=
None
):
"""
:param bliss.config.static.ConfigNode parent:
:returns dict:
:returns ConfigNode:
"""
if
parent
is
None
:
parent
=
static_root
()
if
parent
.
children
:
for
node
in
parent
.
children
:
root
=
static_root
(
root
=
root
)
if
root
.
children
:
for
node
in
root
.
children
:
if
node
.
get
(
"name"
,
None
)
==
name
:
return
node
.
to_dict
()
return
node
nodes
=
[]
for
node
in
paren
t
.
values
():
for
node
in
roo
t
.
values
():
if
isinstance
(
node
,
static
.
ConfigNode
):
nodes
.
append
(
node
)
elif
isinstance
(
node
,
list
):
...
...
@@ -69,90 +53,45 @@ def static_root_find(name, default=None, parent=None):
nodes
.
append
(
nodei
)
for
node
in
nodes
:
if
node
.
get
(
"name"
,
None
)
==
name
:
return
node
.
to_dict
()
ret
=
static_root_find
(
name
,
paren
t
=
node
)
if
ret
:
return
node
ret
=
static_root_find
(
name
,
roo
t
=
node
)
if
ret
is
not
None
:
return
ret
return
{}
return
None
def
with_scan_saving
(
func
):
"""Pass the current session's SCAN_SAVING instance as a named argument
:param callable func:
:returns callable:
def
beamline
(
root
=
None
,
default
=
"id00"
):
"""
@
wraps
(
func
)
def
wrapper
(
*
args
,
**
kwargs
):
scan_saving
=
kwargs
.
get
(
"scan_saving"
)
if
scan_saving
is
None
:
kwargs
[
"scan_saving"
]
=
ScanSaving
(
current_session
.
name
)
return
func
(
*
args
,
**
kwargs
)
return
wrapper
@
with_scan_saving
def
scan_saving_get
(
attr
,
default
=
None
,
scan_saving
=
None
):
"""
Get attribute from the session's scan saving object
:param str attr:
:param default:
:param bliss.scanning.scan.ScanSaving scan_saving:
:returns str:
"""
return
getattr
(
scan_saving
,
attr
,
default
)
@
with_eval_dict
@
with_scan_saving
def
scan_saving_eval
(
template
,
scan_saving
=
None
,
eval_dict
=
None
):
"""
Evaluate template with SCAN_SAVING attributes and properties.
:param str template:
:param bliss.scanning.scan.ScanSaving scan_saving:
:param dict eval_dict:
:returns str:
"""
return
scan_saving
.
eval_template
(
template
,
eval_dict
=
eval_dict
)
def
beamline
():
"""
:returns str:
"""
name
=
"id00"
name
=
default
for
k
in
"BEAMLINENAME"
,
"BEAMLINE"
:
name
=
os
.
environ
.
get
(
k
,
name
)
name
=
static_root_get
(
"beamline"
,
name
)
scan_saving
=
static_root_get
(
"scan_saving"
,
{})
name
=
scan_saving
.
get
(
"beamline"
,
name
)
root
=
static_root
(
root
=
root
)
name
=
root
.
get
(
"beamline"
,
name
)
scan_saving
=
static_root_find
(
"scan_saving"
,
root
=
root
)
if
scan_saving
is
not
None
:
name
=
scan_saving
.
get
(
"beamline"
,
name
)
return
name
.
lower
()
def
institute
():
def
institute
(
root
=
None
,
default
=
""
):
"""
:returns str:
"""
root
=
static_root
()
name
=
""
root
=
static_root
(
root
=
root
)
name
=
default
name
=
root
.
get
(
"institute"
,
name
)
name
=
root
.
get
(
"laboratory"
,
name
)
name
=
root
.
get
(
"synchrotron"
,
name
)
return
name
def
instrument
():
def
instrument
(
root
=
None
,
default
=
""
):
"""
:returns str:
"""
root
=
static_root
()
name
=
""
name
=
root
.
get
(
"institute"
,
name
)
name
=
root
.
get
(
"laboratory"
,
name
)
name
=
root
.
get
(
"synchrotron"
,
name
)
root
=
static_root
(
root
=
root
)
name
=
institute
(
root
=
root
,
default
=
default
)
name
=
root
.
get
(
"instrument"
,
name
)
return
name
nexus_writer_service/utils/scan_utils.py
View file @
a9924641
...
...
@@ -11,7 +11,7 @@
import
os
import
gevent
from
.
import
config
_utils
from
.
import
session
_utils
from
.
import
data_policy
from
.logging_utils
import
print_out
...
...
@@ -98,7 +98,7 @@ def scan_filename(scan):
return
scan_info_get
(
scan
,
"filename"
)
@
config
_utils
.
with_scan_saving
@
session
_utils
.
with_scan_saving
def
session_filename
(
scan_saving
=
None
):
"""
Name of the file that contains the scan data of the current BLISS session
...
...
@@ -106,7 +106,7 @@ def session_filename(scan_saving=None):
:param bliss.scanning.scan.ScanSaving scan_saving:
:returns str or None:
"""
return
config
_utils
.
scan_saving_get
(
return
session
_utils
.
scan_saving_get
(
"filename"
,
default
=
None
,
scan_saving
=
scan_saving
)
...
...
@@ -125,7 +125,7 @@ def scan_master_filenames(scan, config=True):
return
info
.
get
(
"masterfiles"
,
{})
@
config
_utils
.
with_scan_saving
@
session
_utils
.
with_scan_saving
def
session_master_filenames
(
scan_saving
=
None
,
config
=
True
):
"""
Names of the files that contain links to the scan data of the current BLISS session
...
...
@@ -165,7 +165,7 @@ def scan_filenames(scan, config=True):
return
filenames
@
config
_utils
.
with_scan_saving
@
session
_utils
.
with_scan_saving
def
session_filenames
(
scan_saving
=
None
,
config
=
True
):
"""
Names of the files that contain links to the scan data (raw or as links) of the current BLISS session
...
...
nexus_writer_service/utils/session_utils.py
0 → 100644
View file @
a9924641
# -*- coding: utf-8 -*-
#
# This file is part of the nexus writer service of the BLISS project.
#
# Code is maintained by the ESRF Data Analysis Unit.
#
# Original author: Wout de Nolf
#
# Copyright (c) 2015-2020 ESRF
# Distributed under the GNU LGPLv3. See LICENSE for more info.
"""
Activate Bliss session utilities
"""
from
functools
import
wraps
from
bliss
import
current_session
def
with_scan_saving
(
func
):
"""Pass the current session's SCAN_SAVING instance as a named argument
:param callable func:
:returns callable:
"""
@
wraps
(
func
)
def
wrapper
(
*
args
,
**
kwargs
):
scan_saving
=
kwargs
.
get
(
"scan_saving"
)
if
scan_saving
is
None
:
if
current_session
:
kwargs
[
"scan_saving"
]
=
current_session
.
scan_saving
else
:
raise
RuntimeError
(
"No activate Bliss session"
)
return
func
(
*
args
,
**
kwargs
)
return
wrapper
@
with_scan_saving
def
scan_saving_get
(
attr
,
default
=
None
,
scan_saving
=
None
):
"""Get attribute from the session's scan saving object
:param str attr:
:param default:
:param bliss.scanning.scan.ScanSaving scan_saving:
:returns str:
"""
return
getattr
(
scan_saving
,
attr
,
default
)
@
with_scan_saving
def
dataset_get
(
attr
,
default
=
None
,
scan_saving
=
None
):
"""Get attribute from the session's dataset object
:param str attr:
:param default:
:param bliss.scanning.scan.ScanSaving scan_saving:
:returns str:
"""
try
:
return
scan_saving
.
dataset
[
attr
]
except
(
AttributeError
,
KeyError
):
return
default
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment