"""
Provides information on the kinematics involved in the data.
Uses the PLOTTING file specification.
"""
from collections import namedtuple
import logging
import numpy as np
import pandas as pd
from reportengine import collect
from reportengine.checks import check_positive
from reportengine.table import table
from validphys.core import CutsPolicy
from validphys.plotoptions import core as plotoptions_core
log = logging.getLogger(__name__)
[docs]@check_positive('titlelevel')
def describe_kinematics(commondata, titlelevel: int = 1):
"""Output a markdown text describing the stored metadata for a given
commondata.
titlelevel can be used to control the header level of the title.
"""
import inspect
cd = commondata
info = plotoptions_core.get_info(cd)
proc = cd.load_commondata().commondataproc
src = inspect.getsource(info.kinematics_override.xq2map)
titlespec = '#' * titlelevel
return f"""
{titlespec} {cd}
{info.dataset_label}
Stored data:
- Process type: **{proc}** ({info.process_description})
- variables:
* k1: {info.kinlabels[0]}
* k2: {info.kinlabels[1]}
* k3: {info.kinlabels[2]}
Map:
```python
{src}
```
"""
describe_kinematics.highlight = 'markdown'
nfittedlabel = '$N_{fitted}$'
ndatalabel = '$N_{data}$'
[docs]def kinlimits(commondata, cuts, use_cuts, use_kinoverride: bool = True):
"""Return a mapping containing the number of fitted and used datapoints, as
well as the label, minimum and maximum value for each of the three
kinematics. If ``use_kinoverride`` is set to False, the PLOTTING files will
be ignored and the kinematics will be interpred based on the process type
only. If use_cuts is 'CutsPolicy.NOCUTS', the information on the total
number of points will be displayed, instead of the fitted ones."""
info = plotoptions_core.get_info(commondata, cuts=None, use_plotfiles=use_kinoverride)
kintable = plotoptions_core.kitable(commondata, info)
ndata = len(kintable)
if cuts:
kintable = kintable.loc[cuts.load()]
nfitted = len(kintable)
elif use_cuts is not CutsPolicy.NOCUTS:
nfitted = len(kintable)
else:
nfitted = '-'
d = {'dataset': commondata, ndatalabel: ndata, nfittedlabel: nfitted}
for i, key in enumerate(['k1', 'k2', 'k3']):
kmin = kintable[key].min()
kmax = kintable[key].max()
label = info.kinlabels[i]
d[key] = label
d[key + ' min'] = kmin
d[key + ' max'] = kmax
return d
all_kinlimits = collect(kinlimits, ('dataset_inputs',))
[docs]@table
def all_kinlimits_table(all_kinlimits, use_kinoverride: bool = True):
"""Return a table with the kinematic limits for the datasets given as input
in dataset_inputs. If the PLOTTING overrides are not used, the information on
sqrt(k2) will be displayed."""
table = pd.DataFrame(
all_kinlimits,
columns=[
'dataset',
'$N_{data}$',
'$N_{fitted}$',
'k1',
'k1 min',
'k1 max',
'k2',
'k2 min',
'k2 max',
'k3',
'k3 min',
'k3 max',
],
)
# We really want to see the square root of the scale
if not use_kinoverride:
table['k2'] = 'sqrt(' + table['k2'] + ')'
table['k2 min'] = np.sqrt(table['k2 min'])
table['k2 max'] = np.sqrt(table['k2 max'])
# renaming the columns is overly complicated
cols = list(table.columns)
cols[6:9] = ['sqrt(k2)', 'sqrt(k2) min', 'sqrt(k2) max']
table.columns = cols
return table
[docs]@table
def all_commondata_grouping(all_commondata, metadata_group):
"""Return a table with the grouping specified
by `metadata_group` key for each dataset for all available commondata.
"""
records = []
for cd in all_commondata:
records.append(
{
'dataset': str(cd),
metadata_group: getattr(plotoptions_core.get_info(cd), metadata_group),
}
)
df = pd.DataFrame.from_records(records, index='dataset')
# sort first by grouping alphabetically and then dataset name
return df.sort_values([metadata_group, 'dataset'])
[docs]def total_fitted_points(all_kinlimits_table) -> int:
"""Print the total number of fitted points in a given set of data"""
tb = all_kinlimits_table
return int(tb[nfittedlabel].sum())
XQ2Map = namedtuple('XQ2Map', ('experiment', 'commondata', 'fitted', 'masked', "group"))
[docs]def xq2map_with_cuts(commondata, cuts, group_name=None):
"""Return two (x,Q²) tuples: one for the fitted data and one for the
cut data. If `display_cuts` is false or all data passes the cuts, the second
tuple will be empty."""
info = plotoptions_core.get_info(commondata)
kintable = plotoptions_core.kitable(commondata, info)
if cuts:
mask = cuts.load()
boolmask = np.zeros(len(kintable), dtype=bool)
boolmask[mask] = True
fitted_kintable = kintable.loc[boolmask]
masked_kitable = kintable.loc[~boolmask]
xq2fitted = plotoptions_core.get_xq2map(fitted_kintable, info)
xq2masked = plotoptions_core.get_xq2map(masked_kitable, info)
else:
xq2fitted = plotoptions_core.get_xq2map(kintable, info)
xq2masked = (np.array([]), np.array([]))
return XQ2Map(info.experiment, commondata, xq2fitted, xq2masked, group_name)
dataset_inputs_by_groups_xq2map = collect(
xq2map_with_cuts, ('group_dataset_inputs_by_metadata', 'data_input')
)
[docs]def kinematics_table_notable(commondata, cuts, show_extra_labels: bool = False):
"""
Table containing the kinematics of a commondata object,
indexed by their datapoint id. The kinematics will be tranfsormed as per the
PLOTTING file of the dataset or process type, and the column headers will
be the labels of the variables defined in the metadata.
If ``show_extra_labels`` is ``True`` then extra label defined in the
PLOTTING files will be displayed. Otherwise only the original three
kinematics will be shown.
"""
info = plotoptions_core.get_info(commondata, cuts=cuts)
res = plotoptions_core.kitable(commondata, info, cuts=cuts)
res.columns = [*info.kinlabels, *res.columns[3:]]
if not show_extra_labels:
res = res.iloc[:, :3]
return res
[docs]@table
def kinematics_table(kinematics_table_notable):
"""Same as kinematics_table_notable but writing the table to file"""
return kinematics_table_notable